Split live-status tracking into its own subscriber (and share one MQTT connection) - Part 2: Add LiveStatusSubscriber

frontend/src/stores/account-auth.js

@@ -201,8 +201,10 @@ export const useAccountAuthStore = defineStore('account-auth', {
             if (useAccountSettingsStore().settings['platform:sso:only']) {
                 logoutURL = useAccountSettingsStore().settings['platform:sso:only:logoutURL'] || '/'
             }
-            const teamChannel = getAppOrchestrator().$subscriberInstances.teamChannel
-            const disconnect = teamChannel ? teamChannel.disconnect().catch(() => {}) : Promise.resolve()
+            const subscribers = getAppOrchestrator().$subscriberInstances
+            const disconnect = Promise.all(
+                Object.values(subscribers).map(subscriber => subscriber?.disconnect().catch(() => {}))
+            )
             return disconnect
                 .then(() => userApi.logout())
                 .then(() => {

frontend/src/stores/account.js

@@ -11,8 +11,13 @@ import { useProductTablesStore } from '@/stores/product-tables.js'
 
 function ensureTeamChannelConnected (team) {
     if (!team?.id) return
-    const teamChannel = getAppOrchestrator().$subscriberInstances.teamChannel
-    teamChannel?.connect(team).catch(() => {})
+    const subscribers = getAppOrchestrator().$subscriberInstances
+    Object.values(subscribers).forEach(subscriber => subscriber?.connect(team).catch(() => {}))
+}
+
+function disconnectTeamSubscribers () {
+    const subscribers = getAppOrchestrator().$subscriberInstances
+    Object.values(subscribers).forEach(subscriber => subscriber?.disconnect().catch(() => {}))
 }
 
 export const useAccountStore = defineStore('account', {
@@ -94,7 +99,7 @@ export const useAccountStore = defineStore('account', {
             if (team?.id) {
                 ensureTeamChannelConnected(team)
             } else {
-                getAppOrchestrator().$subscriberInstances.teamChannel?.disconnect().catch(() => {})
+                disconnectTeamSubscribers()
             }
             this.pendingTeamChange = false
         },

frontend/src/subscribers/live-status.subscriber.ts

@@ -0,0 +1,66 @@
+import { defineSubscriberSingleton } from './subscriber.factory'
+import { SubscriberRoute, TeamSubscriber } from './team-subscriber.contract'
+
+import { useLiveStatusStore } from '@/stores/live-status'
+import type { CreateSubscriberOptions, TeamSubscriberI } from '@/types/subscribers/subscriber.types'
+
+const DEVICE_STATE_TOPIC_REGEX = /^ff\/v1\/[^/]+\/d\/[^/]+\/state$/
+const INSTANCE_STATE_TOPIC_REGEX = /^ff\/v1\/[^/]+\/p\/[^/]+\/state$/
+
+class LiveStatusSubscriber extends TeamSubscriber implements TeamSubscriberI {
+    constructor ({ app, router, transport, subscribers }: CreateSubscriberOptions) {
+        super({
+            name: 'liveStatus',
+            app,
+            router,
+            transport,
+            subscribers
+        })
+    }
+
+    protected _topics (teamId: string): string[] {
+        return [
+            `ff/v1/${teamId}/p/+/state`,
+            `ff/v1/${teamId}/d/+/state`
+        ]
+    }
+
+    protected _routes (): SubscriberRoute[] {
+        return [
+            { pattern: DEVICE_STATE_TOPIC_REGEX, handle: (payload) => this._onDeviceStatus(payload) },
+            { pattern: INSTANCE_STATE_TOPIC_REGEX, handle: (payload) => this._onInstanceStatus(payload) }
+        ]
+    }
+
+    protected _onSubscribed (): void {
+        try {
+            useLiveStatusStore().setLive(true)
+        } catch {}
+    }
+
+    protected _onDisconnected (): void {
+        try {
+            useLiveStatusStore().clear()
+        } catch {}
+    }
+
+    protected _onInstanceStatus (payload: { id?: string, meta?: { state?: string, versions?: Record<string, string> } }): void {
+        if (!payload?.id || !payload.meta?.state) return
+        try {
+            useLiveStatusStore().setInstanceStatus(payload.id, payload.meta.state, payload.meta.versions)
+        } catch {}
+    }
+
+    protected _onDeviceStatus (payload: { id?: string, meta?: { state?: string } }): void {
+        if (!payload?.id || !payload.meta?.state) return
+        try {
+            useLiveStatusStore().setDeviceStatus(payload.id, payload.meta.state)
+        } catch {}
+    }
+}
+
+const { create: createLiveStatusSubscriber, destroy: destroyLiveStatusSubscriber } = defineSubscriberSingleton(LiveStatusSubscriber)
+
+export { createLiveStatusSubscriber, destroyLiveStatusSubscriber }
+
+export default createLiveStatusSubscriber

frontend/src/subscribers/subscriber.factory.ts

@@ -0,0 +1,23 @@
+import { Maybe } from '@/types/common/types'
+import type { CreateSubscriberOptions } from '@/types/subscribers/subscriber.types'
+
+type SubscriberConstructor<T> = new (options: CreateSubscriberOptions) => T
+
+export function defineSubscriberSingleton<T extends { destroy(): Promise<void> }> (SubscriberClass: SubscriberConstructor<T>) {
+    let instance: Maybe<T> = null
+
+    function create (options: CreateSubscriberOptions): T {
+        if (!instance) {
+            instance = new SubscriberClass(options)
+        }
+        return instance
+    }
+
+    async function destroy (): Promise<void> {
+        if (!instance) return
+        await instance.destroy()
+        instance = null
+    }
+
+    return { create, destroy }
+}

frontend/src/subscribers/subscriber.registry.ts

@@ -1,5 +1,7 @@
+import { createLiveStatusSubscriber } from './live-status.subscriber'
 import { createTeamChannelSubscriber } from './team-channel.subscriber'
 
 export default [
-    { key: 'teamChannel' as const, create: createTeamChannelSubscriber, requiredLifecycle: ['destroy'] as const }
+    { key: 'teamChannel' as const, create: createTeamChannelSubscriber, requiredLifecycle: ['destroy'] as const },
+    { key: 'liveStatus' as const, create: createLiveStatusSubscriber, requiredLifecycle: ['destroy'] as const }
 ]

frontend/src/subscribers/team-channel.subscriber.ts

@@ -1,24 +1,13 @@
-import { BaseSubscriber } from './subscriber.contract'
+import { defineSubscriberSingleton } from './subscriber.factory'
+import { SubscriberRoute, TeamSubscriber } from './team-subscriber.contract'
 
-import teamApi from '@/api/team.js'
-import { useAccountAuthStore } from '@/stores/account-auth.js'
 import { useContextStore } from '@/stores/context.js'
-import { useLiveStatusStore } from '@/stores/live-status'
-import { Maybe } from '@/types/common/types'
-import type { CreateSubscriberOptions, TeamChannelSubscriberI, TeamRef } from '@/types/subscribers/subscriber.types'
+import type { CreateSubscriberOptions, TeamSubscriberI } from '@/types/subscribers/subscriber.types'
 
 const MEMBERSHIP_TOPIC_REGEX = /^ff\/v1\/[^/]+\/u\/([^/]+)\/membership$/
 const TEAM_UPDATED_TOPIC_REGEX = /^ff\/v1\/[^/]+\/t\/updated$/
-const DEVICE_STATE_TOPIC_REGEX = /^ff\/v1\/[^/]+\/d\/[^/]+\/state$/
-const INSTANCE_STATE_TOPIC_REGEX = /^ff\/v1\/[^/]+\/p\/[^/]+\/state$/
-
-function connectionKey (teamId: string): string {
-    return `team:${teamId}`
-}
-
-class TeamChannelSubscriber extends BaseSubscriber implements TeamChannelSubscriberI {
-    protected $connectedTeamId: Maybe<string> = null
 
+class TeamChannelSubscriber extends TeamSubscriber implements TeamSubscriberI {
     constructor ({ app, router, transport, subscribers }: CreateSubscriberOptions) {
         super({
             name: 'teamChannel',
@@ -29,124 +18,20 @@ class TeamChannelSubscriber extends BaseSubscriber implements TeamChannelSubscri
         })
     }
 
-    isConnected (): boolean {
-        return this.$connectedTeamId !== null
-    }
-
-    async connect (team: Maybe<TeamRef>): Promise<void> {
-        if (!team?.id || typeof team.id !== 'string' || team.id.length === 0) return
-        const authStore = useAccountAuthStore()
-        const userId = authStore.user?.id
-        if (!userId) return
-        if (this.$connectedTeamId === team.id) return
-
-        await this.disconnect()
-
-        const transport = this.$transport
-        if (!transport) return
-
-        const teamId = team.id
-        const sessionId = authStore.getSessionId()
-        const key = connectionKey(teamId)
-
-        try {
-            await transport.connect(key, {
-                getCredentials: () => teamApi.getTeamCommsCreds(teamId, sessionId),
-                onMessage: (topic, message) => this._onMessage(topic, message),
-                onConnect: () => this._onConnect(teamId, userId),
-                // close/offline/disconnect/error: nothing to do — the transport
-                // handles reconnect; terminal failures surface via this catch
-                onClose: () => {},
-                onOffline: () => {},
-                onDisconnect: () => {},
-                onError: () => {}
-            })
-            this.$connectedTeamId = teamId
-        } catch {
-            // graceful degrade — no broker, bad creds, or network
-            this.$connectedTeamId = null
-        }
-    }
-
-    async disconnect (): Promise<void> {
-        if (!this.$connectedTeamId) return
-        const transport = this.$transport
-        const key = connectionKey(this.$connectedTeamId)
-        this.$connectedTeamId = null
-        if (!transport) return
-        try {
-            await transport.disconnect(key)
-            useLiveStatusStore().clear()
-        } catch {
-            // ignore teardown failures
-        }
-    }
-
-    async destroy (): Promise<void> {
-        await this.disconnect()
-    }
-
-    protected async _onConnect (teamId: string, userId: string): Promise<void> {
-        const transport = this.$transport
-        if (!transport) return
-        try {
-            await transport.subscribe(connectionKey(teamId), [
-                `ff/v1/${teamId}/t/updated`,
-                `ff/v1/${teamId}/u/${userId}/membership`,
-                `ff/v1/${teamId}/p/+/state`,
-                `ff/v1/${teamId}/d/+/state`
-            ], { qos: 1 })
-
-            useLiveStatusStore().setLive(true)
-        } catch {
-            // non-fatal — the transport replays subscriptions on reconnect
-        }
-    }
-
-    protected _onMessage (topic: string, message: Buffer | Uint8Array | string): void {
-        let payload: { reason?: string } = {}
-        try {
-            const raw = typeof message === 'string'
-                ? message
-                : (message?.toString ? message.toString() : '')
-            payload = raw ? JSON.parse(raw) : {}
-        } catch {
-            payload = {}
-        }
-
-        for (const { pattern, handle } of this._topicRoutes()) {
-            if (pattern.test(topic)) {
-                handle(payload)
-                return
-            }
-        }
+    protected _topics (teamId: string, userId: string): string[] {
+        return [
+            `ff/v1/${teamId}/t/updated`,
+            `ff/v1/${teamId}/u/${userId}/membership`
+        ]
     }
 
-    // topic pattern → store action; the store owns interpretation (what a
-    // reason means, what to refresh). Add a row per new sync-able entity.
-    protected _topicRoutes (): Array<{ pattern: RegExp, handle: (payload: { reason?: string, id?: string, meta?: { state?: string, versions?: Record<string, string> } }) => void }> {
+    protected _routes (): SubscriberRoute[] {
         return [
             { pattern: MEMBERSHIP_TOPIC_REGEX, handle: (payload) => this._onMembership(payload) },
-            { pattern: TEAM_UPDATED_TOPIC_REGEX, handle: () => this._onTeamUpdated() },
-            { pattern: DEVICE_STATE_TOPIC_REGEX, handle: (payload) => this._onDeviceStatus(payload) },
-            { pattern: INSTANCE_STATE_TOPIC_REGEX, handle: (payload) => this._onInstanceStatus(payload) }
+            { pattern: TEAM_UPDATED_TOPIC_REGEX, handle: () => this._onTeamUpdated() }
         ]
     }
 
-    protected _onInstanceStatus (payload: { id?: string, meta?: { state?: string, versions?: Record<string, string> } }): void {
-        if (!payload?.id || !payload.meta?.state) return
-        try {
-            useLiveStatusStore().setInstanceStatus(payload.id, payload.meta.state, payload.meta.versions)
-        } catch {}
-    }
-
-    protected _onDeviceStatus (payload: { id?: string, meta?: { state?: string } }): void {
-        if (!payload?.id || !payload.meta?.state) return
-        try {
-            useLiveStatusStore().setDeviceStatus(payload.id, payload.meta.state)
-        } catch {}
-    }
-
     protected _onMembership (payload: { reason?: string }): void {
         try {
             useContextStore().onTeamChannelMembership(payload).catch(() => undefined)
@@ -160,24 +45,8 @@ class TeamChannelSubscriber extends BaseSubscriber implements TeamChannelSubscri
     }
 }
 
-let TeamChannelSubscriberInstance: Maybe<TeamChannelSubscriber> = null
-
-export function createTeamChannelSubscriber ({ app, router, transport, subscribers }: CreateSubscriberOptions): TeamChannelSubscriber {
-    if (!TeamChannelSubscriberInstance) {
-        TeamChannelSubscriberInstance = new TeamChannelSubscriber({
-            app,
-            router,
-            transport,
-            subscribers
-        })
-    }
-    return TeamChannelSubscriberInstance
-}
+const { create: createTeamChannelSubscriber, destroy: destroyTeamChannelSubscriber } = defineSubscriberSingleton(TeamChannelSubscriber)
 
-export async function destroyTeamChannelSubscriber (): Promise<void> {
-    if (!TeamChannelSubscriberInstance) return
-    await TeamChannelSubscriberInstance.destroy()
-    TeamChannelSubscriberInstance = null
-}
+export { createTeamChannelSubscriber, destroyTeamChannelSubscriber }
 
 export default createTeamChannelSubscriber