feat: polling as fallback for SSE (#150)

* feat: polling as fallback for SSE

* feat: polling as fallback for SSE

* feat: polling as fallback for SSE

* chore(deps): combined dependency update

* debug comment

.github/workflows/build-release.yml

@@ -32,6 +32,11 @@ jobs:
           RELEASE_VERSION: ${{ steps.next.outputs.version }}
         run: npm version $RELEASE_VERSION --allow-same-version
 
+      - name: Setup node
+        uses: actions/setup-node@v3
+        with:
+          node-version: 20.x
+
       - name: Install
         working-directory: app
         shell: bash

.github/workflows/build.yml

@@ -15,7 +15,7 @@ jobs:
       - name: Setup node
         uses: actions/setup-node@v3
         with:
-          node-version: 16.x
+          node-version: 20.x
 
       - name: Install
         working-directory: app

Dockerfile

@@ -1,4 +1,4 @@
-FROM node:18.16-alpine
+FROM node:20.5-alpine
 COPY app/dist /opt/app/
 WORKDIR /opt/app/
 

app/lib/app.ts

@@ -1,9 +1,10 @@
 import EventSource from "eventsource"
 import cron from "node-cron"
+import { getAppConfig } from "./config/config"
 import { registerConnectionCheck, unregisterConnectionCheck } from "./connection"
 import { log } from "./logger"
-import { login, needsRefresh } from "./miele/login/login"
-import { smallMessage } from "./miele/miele"
+import { getCurrentToken, login, needsRefresh } from "./miele/login/login"
+import { fetchDevices, smallMessage } from "./miele/miele"
 import { startSSE } from "./miele/sse-client"
 import { connectMqtt, publish } from "./mqtt/mqtt-client"
 
@@ -14,6 +15,26 @@ export const triggerFullUpdate = async () => {
     }
 }
 
+export const polling = async () => {
+    if (getAppConfig().miele.mode !== "polling") {
+        return
+    }
+
+    const token = getCurrentToken()
+    if (token) {
+        log.debug("Polling started")
+        const devices = await fetchDevices(token)
+        log.debug("Polling done")
+        for (const device of devices) {
+            publish(smallMessage(device), device.id)
+            publish(device.data, `${device.id}/full`)
+        }
+    }
+    else {
+        log.warn("Polling skipped (no token)")
+    }
+}
+
 const restart = async () => {
     eventSource?.close()
     unregisterConnectionCheck()
@@ -50,11 +71,15 @@ export const startApp = async () => {
         const task = cron.schedule("* * * * *", triggerFullUpdate)
         task.start()
 
+        const pollingTask = cron.schedule("* * * * *", polling)
+        pollingTask.start()
+
         return () => {
             mqttCleanUp()
             eventSource?.close()
             unregisterConnectionCheck()
             task.stop()
+            pollingTask.stop()
         }
     }
     catch (e) {

app/lib/connection.test.ts

@@ -33,20 +33,20 @@ describe("connection", () => {
 
         const fn = jest.fn()
 
-        const check = registerConnectionCheck(fn, config)
+        registerConnectionCheck(fn, config)
 
         await sleep(100)
 
         expect(fn).not.toHaveBeenCalled()
 
-        check?.unref()
+        unregisterConnectionCheck()
     })
 
     test("failed", async () => {
         __TEST_setCheck(() => Promise.resolve(false))
         const fn = jest.fn()
 
-        const check = registerConnectionCheck(fn, config)
+        registerConnectionCheck(fn, config)
 
         await sleep(100)
         expect(fn).not.toHaveBeenCalled()
@@ -56,14 +56,14 @@ describe("connection", () => {
         await sleep(100)
         expect(fn).toHaveBeenCalled()
 
-        check?.unref()
+        unregisterConnectionCheck()
     })
 
     test("check disabled", async () => {
         __TEST_setCheck(() => Promise.resolve(false))
         const fn = jest.fn()
 
-        const check = registerConnectionCheck(fn, { ...config, "connection-check-interval": 0 })
+        registerConnectionCheck(fn, { ...config, "connection-check-interval": 0 })
 
         await sleep(100)
         expect(fn).not.toHaveBeenCalled()
@@ -73,6 +73,18 @@ describe("connection", () => {
         await sleep(100)
         expect(fn).not.toHaveBeenCalled()
 
-        check?.unref()
+        unregisterConnectionCheck()
+    })
+
+    test("already registered", async () => {
+        __TEST_setCheck(() => Promise.resolve(false))
+        const fn = jest.fn()
+
+        const check1 = registerConnectionCheck(fn, config)
+        const check2 = registerConnectionCheck(fn, config)
+
+        expect(check1).toBe(check2)
+
+        unregisterConnectionCheck()
     })
 })

app/lib/connection.ts

@@ -2,11 +2,12 @@ import { getAppConfig } from "./config/config"
 import { log } from "./logger"
 import { ping } from "./miele/miele"
 
-let checkConnection: ReturnType<typeof setTimeout>
+let checkConnection: ReturnType<typeof setTimeout> | undefined
 let connectionLost = false
 
 export const unregisterConnectionCheck = () => {
     checkConnection?.unref()
+    checkConnection = undefined
 }
 
 let check = ping
@@ -18,6 +19,10 @@ export const __TEST_setCheck = (newCheck: () => Promise<boolean>) => {
 
 export const registerConnectionCheck = (restartHook: () => Promise<void>, config = getAppConfig().miele) => {
     const interval = config["connection-check-interval"]
+    if (checkConnection) {
+        log.debug("Connection check already registered")
+        return checkConnection
+    }
     if (interval === 0) {
         log.debug("Internet connection check disabled")
         return

app/lib/index.ts

@@ -16,8 +16,7 @@ log.info("Using config from file", configFile)
 const config = loadConfig(configFile)
 
 if (config.miele.mode === "polling") {
-    log.error("Polling mode is not supported for version >= 3.x. Please use version 2.x when you like to use the polling mode.")
-    process.exit(1)
+    log.info("Polling mode enabled. SSE is still active, using polling as fallback.")
 }
 
 startApp().then()

app/lib/integration/integration.test.ts

@@ -4,6 +4,7 @@ import { GenericContainer, StartedTestContainer, Wait } from "testcontainers"
 import { JEST_INTEGRATION_TIMEOUT, JEST_DEFAULT_TIMEOUT } from "../../test/test-utils"
 import { startApp } from "../app"
 import { applyConfig, ConfigMqtt, getAppConfig } from "../config/config"
+import { unregisterConnectionCheck } from "../connection"
 import { log } from "../logger"
 import { testConfig } from "../miele/miele-testutils"
 import { createMqttInstance, MqttInstance, subscribe } from "../mqtt/mqtt-client"
@@ -93,6 +94,7 @@ describe("Integration test", () => {
         await mqtt?.stop()
         jest.setTimeout(JEST_DEFAULT_TIMEOUT)
         await process.nextTick(() => {})
+        unregisterConnectionCheck()
     })
 
     test("Message is published", async () => {

app/lib/miele/login/code.ts

@@ -3,9 +3,12 @@ import axios from "axios"
 import { log } from "../../logger"
 
 export const codeUrl = "https://api.mcs3.miele.com/oauth/auth"
-// https://api.mcs3.miele.com/thirdparty/login/?redirect_uri=http://localhost:3000&client_id=76012106-c8c0-4901-8ff4-b3bf32696523&response_type=code&state=login&vgInformationSelector=de-DE
 export const fetchCode = async () => {
     log.debug("Fetching code")
+
+    // Debug this by visiting the following URL:
+    // https://api.mcs3.miele.com/thirdparty/login/?redirect_uri=/v1/&client_id=<your_client_id>&response_type=code
+    //
     const config: ConfigMiele = getAppConfig().miele
     const response = await axios.post(
         codeUrl,

app/lib/miele/login/login.ts

@@ -48,6 +48,8 @@ const assertConnection = async () => {
     }
 }
 
+let currentToken: string | undefined
+
 export const login = async (now = new Date()) => {
     log.debug("Logging in")
     let connected = await assertConnection()
@@ -74,9 +76,13 @@ export const login = async (now = new Date()) => {
         validUntil: token.expiresAt.toISOString()
     })
 
+    currentToken = token.access_token
+
     return token
 }
 
+export const getCurrentToken = () => currentToken
+
 export const getToken = async () => {
     if (!token || needsRefresh()) {
         await login()