feat: add sse watchdog

app/lib/SSEClient.ts

@@ -1,8 +1,12 @@
 import EventSource from "eventsource"
 import { log } from "./logger"
 import { getAppConfig } from "./config/config"
+import { takeEvent } from "./state/state-event-handler"
 
-export const startSSE = () => {
+let sse: EventSource | null = null
+let lastEvent = Date.now()
+
+const startSSE = () => {
     log.info("Starting Server-Sent events")
 
     const config = getAppConfig()
@@ -17,9 +21,52 @@ export const startSSE = () => {
     const baserUrl = `https://${config.hue.host}:${config.hue.port}`
     const sse = new EventSource(`${baserUrl}/eventstream/clip/v2`, eventSourceInitDict)
     sse.onerror = (err: any) => {
-        if (err) {
-            log.error("SSE Error", err)
-        }
+        log.error("SSE Error", err)
     }
     return sse
 }
+
+const sleep = (ms: number) => new Promise(resolve => setTimeout(resolve, ms))
+
+const registerSSEWatchDog = () => {
+    const millis = getAppConfig().hue["sse-watchdog-millis"]
+    if (millis === 0) {
+        log.info("SSE watchdog disabled")
+        return
+    }
+
+    log.info(`SSE watchdog enabled with ${millis}ms`)
+    setInterval(() => {
+        log.debug("Checking for SSE watchdog", Date.now() - lastEvent, millis)
+        if (Date.now() - lastEvent > millis) {
+            log.error("SSE watchdog triggered")
+            sse?.close()
+            sse = startSSE()
+        }
+    }, millis / 2)
+}
+
+export const initSSE = () => {
+    sse = startSSE()
+    sse.addEventListener("message", event => {
+        lastEvent = Date.now()
+        for (const data of JSON.parse(event.data)) {
+            takeEvent(data)
+        }
+    })
+
+    sse.onerror = async (err: any) => {
+        log.error("SSE Error", err)
+        await sleep(5000)
+        log.info("Reconnecting to SSE")
+        sse = startSSE()
+    }
+
+    registerSSEWatchDog()
+}
+
+export const destroySSE = () => {
+    if (sse) {
+        sse.close()
+    }
+}

app/lib/app.ts

@@ -1,6 +1,5 @@
 import { connectMqtt } from "./mqtt/mqtt-client"
-import { startSSE } from "./SSEClient"
-import { takeEvent } from "./state/state-event-handler"
+import { destroySSE, initSSE } from "./SSEClient"
 import { log } from "./logger"
 import cron from "node-cron"
 import { initStateManagerFromHue } from "./state/state-manager"
@@ -15,12 +14,7 @@ export const startApp = async () => {
     const mqttCleanUp = await connectMqtt()
     await triggerFullUpdate()
 
-    const sse = startSSE()
-    sse.addEventListener("message", event => {
-        for (const data of JSON.parse(event.data)) {
-            takeEvent(data)
-        }
-    })
+    initSSE()
 
     log.info("Scheduling hourly-full-update.")
     const task = cron.schedule("0 * * * *", triggerFullUpdate)
@@ -30,7 +24,7 @@ export const startApp = async () => {
 
     return () => {
         mqttCleanUp()
-        sse.close()
+        destroySSE()
         task.stop()
     }
 }

app/lib/config/config.ts

@@ -17,6 +17,7 @@ export type ConfigHue = {
     "api-key": string
     port: number
     protocol: "http"|"https"
+    "sse-watchdog-millis": number
 }
 
 export type Config = {
@@ -37,7 +38,8 @@ const mqttDefaults = {
 
 const hueDefaults = {
     port: 443,
-    protocol: "https"
+    protocol: "https",
+    "sse-watchdog-millis": 0
 }
 
 const configDefaults = {