From dcf013b85e27a75d97fbf91b6da37b22c883da1e Mon Sep 17 00:00:00 2001 From: Noley Holland Date: Thu, 16 Jul 2026 11:11:37 -0700 Subject: [PATCH 1/2] Split live-status into its own subscriber sharing the team MQTT connection via attach/detach --- frontend/src/stores/account-auth.js | 6 +- frontend/src/stores/account.js | 11 +- .../src/subscribers/live-status.subscriber.ts | 66 ++++++ .../src/subscribers/subscriber.contract.ts | 40 ---- .../src/subscribers/subscriber.factory.ts | 23 +++ .../src/subscribers/subscriber.registry.ts | 4 +- .../subscribers/team-channel.subscriber.ts | 157 ++------------ .../subscribers/team-subscriber.contract.ts | 173 ++++++++++++++++ frontend/src/transport/mqtt.transport.ts | 10 +- .../src/types/subscribers/subscriber.types.ts | 5 +- .../src/types/transport/transport.types.ts | 9 +- .../services/app.orchestrator.spec.js | 18 +- .../live-status.subscriber.spec.js | 195 ++++++++++++++++++ .../team-channel.subscriber.spec.js | 157 ++++++-------- 14 files changed, 575 insertions(+), 299 deletions(-) create mode 100644 frontend/src/subscribers/live-status.subscriber.ts delete mode 100644 frontend/src/subscribers/subscriber.contract.ts create mode 100644 frontend/src/subscribers/subscriber.factory.ts create mode 100644 frontend/src/subscribers/team-subscriber.contract.ts create mode 100644 test/unit/frontend/subscribers/live-status.subscriber.spec.js diff --git a/frontend/src/stores/account-auth.js b/frontend/src/stores/account-auth.js index 94af237402..ceb1f52198 100644 --- a/frontend/src/stores/account-auth.js +++ b/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(() => { diff --git a/frontend/src/stores/account.js b/frontend/src/stores/account.js index 8e3da9e584..628cfaa98b 100644 --- a/frontend/src/stores/account.js +++ b/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 }, diff --git a/frontend/src/subscribers/live-status.subscriber.ts b/frontend/src/subscribers/live-status.subscriber.ts new file mode 100644 index 0000000000..6b610fef7d --- /dev/null +++ b/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 } }): void { + if (!payload?.id || !payload.meta?.state) return + try { + useLiveStatusStore().setInstanceStatus(payload.id, payload.meta.state) + } 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 diff --git a/frontend/src/subscribers/subscriber.contract.ts b/frontend/src/subscribers/subscriber.contract.ts deleted file mode 100644 index 75d0e345a9..0000000000 --- a/frontend/src/subscribers/subscriber.contract.ts +++ /dev/null @@ -1,40 +0,0 @@ -import type { App } from 'vue' -import type { Router } from 'vue-router' - -import { Maybe } from '@/types/common/types' -import type { CreateSubscriberOptions, SubscriberInstances } from '@/types/subscribers/subscriber.types' -import type { Transport } from '@/types/transport/transport.types' - -export abstract class BaseSubscriber { - protected $name: string - - protected $app: Maybe = null - - protected $router: Maybe = null - - protected $transport: Maybe = null - - protected $subscribers: Maybe = null - - protected constructor ({ - name, - app, - router, - transport, - subscribers - }: { name: string } & CreateSubscriberOptions) { - this.$name = name - this.$app = app - this.$router = router - this.$transport = transport - this.$subscribers = subscribers ?? null - } - - init () { - return undefined - } - - async destroy () { - return Promise.resolve() - } -} diff --git a/frontend/src/subscribers/subscriber.factory.ts b/frontend/src/subscribers/subscriber.factory.ts new file mode 100644 index 0000000000..171432a998 --- /dev/null +++ b/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 = new (options: CreateSubscriberOptions) => T + +export function defineSubscriberSingleton }> (SubscriberClass: SubscriberConstructor) { + let instance: Maybe = null + + function create (options: CreateSubscriberOptions): T { + if (!instance) { + instance = new SubscriberClass(options) + } + return instance + } + + async function destroy (): Promise { + if (!instance) return + await instance.destroy() + instance = null + } + + return { create, destroy } +} diff --git a/frontend/src/subscribers/subscriber.registry.ts b/frontend/src/subscribers/subscriber.registry.ts index 467f5b9e54..50cf0f233c 100644 --- a/frontend/src/subscribers/subscriber.registry.ts +++ b/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 } ] diff --git a/frontend/src/subscribers/team-channel.subscriber.ts b/frontend/src/subscribers/team-channel.subscriber.ts index 41e3af6bd2..e232db3ef5 100644 --- a/frontend/src/subscribers/team-channel.subscriber.ts +++ b/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 = 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): Promise { - 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 { - 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 { - await this.disconnect() - } - - protected async _onConnect (teamId: string, userId: string): Promise { - 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 } }) => 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 } }): void { - if (!payload?.id || !payload.meta?.state) return - try { - useLiveStatusStore().setInstanceStatus(payload.id, payload.meta.state) - } 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 = 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 { - if (!TeamChannelSubscriberInstance) return - await TeamChannelSubscriberInstance.destroy() - TeamChannelSubscriberInstance = null -} +export { createTeamChannelSubscriber, destroyTeamChannelSubscriber } export default createTeamChannelSubscriber diff --git a/frontend/src/subscribers/team-subscriber.contract.ts b/frontend/src/subscribers/team-subscriber.contract.ts new file mode 100644 index 0000000000..72d3890aa1 --- /dev/null +++ b/frontend/src/subscribers/team-subscriber.contract.ts @@ -0,0 +1,173 @@ +import type { App } from 'vue' +import type { Router } from 'vue-router' + +import teamApi from '@/api/team.js' +import { useAccountAuthStore } from '@/stores/account-auth.js' +import { Maybe } from '@/types/common/types' +import type { CreateSubscriberOptions, SubscriberInstances, TeamRef } from '@/types/subscribers/subscriber.types' +import type { Transport, TransportAttachmentHandle } from '@/types/transport/transport.types' + +export function connectionKey (teamId: string): string { + return `team:${teamId}` +} + +type SubscriberPayload = { reason?: string, id?: string, meta?: { state?: string } } + +export interface SubscriberRoute { + pattern: RegExp + handle: (payload: SubscriberPayload) => void +} + +export abstract class TeamSubscriber { + protected $name: string + + protected $app: Maybe = null + + protected $router: Maybe = null + + protected $transport: Maybe = null + + protected $subscribers: Maybe = null + + protected $connectedTeamId: Maybe = null + + protected $attachment: Maybe = null + + protected $operation: Maybe> = null + + protected constructor ({ + name, + app, + router, + transport, + subscribers + }: { name: string } & CreateSubscriberOptions) { + this.$name = name + this.$app = app + this.$router = router + this.$transport = transport + this.$subscribers = subscribers ?? null + } + + init () { + return undefined + } + + isConnected (): boolean { + return this.$connectedTeamId !== null + } + + async connect (team: Maybe): Promise { + await this.runSubscriberOperation(() => this._connect(team)) + } + + async disconnect (): Promise { + await this.runSubscriberOperation(() => this._disconnect()) + } + + async destroy (): Promise { + await this.disconnect() + } + + async runSubscriberOperation (operation: () => Promise | T): Promise { + const previous = this.$operation ?? Promise.resolve() + const current = previous + .catch(() => {}) + .then(operation) + + const tracked = current.finally(() => { + if (this.$operation === tracked) { + this.$operation = null + } + }) + this.$operation = tracked + + return tracked + } + + protected async _connect (team: Maybe): Promise { + 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 { + this.$attachment = await transport.attach(key, { + getCredentials: () => teamApi.getTeamCommsCreds(teamId, sessionId), + onMessage: (topic, message) => this._onMessage(topic, message), + onConnect: () => this._onConnect(teamId, userId), + onClose: () => {}, + onOffline: () => {}, + onDisconnect: () => {}, + onError: () => {} + }) + this.$connectedTeamId = teamId + } catch { + this.$connectedTeamId = null + this.$attachment = null + } + } + + protected async _disconnect (): Promise { + if (!this.$connectedTeamId) return + const transport = this.$transport + const attachment = this.$attachment + this.$connectedTeamId = null + this.$attachment = null + if (!transport || !attachment) return + try { + await transport.detach(attachment) + this._onDisconnected() + } catch { + // ignore teardown failures + } + } + + protected async _onConnect (teamId: string, userId: string): Promise { + const transport = this.$transport + if (!transport) return + try { + await transport.subscribe(connectionKey(teamId), this._topics(teamId, userId), { qos: 1 }) + this._onSubscribed() + } catch { + // non-fatal — the transport replays subscriptions on reconnect + } + } + + protected _onMessage (topic: string, message: Buffer | Uint8Array | string): void { + let payload: SubscriberPayload = {} + try { + const raw = typeof message === 'string' + ? message + : (message?.toString ? message.toString() : '') + payload = raw ? JSON.parse(raw) : {} + } catch { + payload = {} + } + + for (const { pattern, handle } of this._routes()) { + if (pattern.test(topic)) { + handle(payload) + return + } + } + } + + protected abstract _topics (teamId: string, userId: string): string[] + + protected abstract _routes (): SubscriberRoute[] + + protected _onSubscribed (): void {} + + protected _onDisconnected (): void {} +} diff --git a/frontend/src/transport/mqtt.transport.ts b/frontend/src/transport/mqtt.transport.ts index 2e08430510..5384917ad6 100644 --- a/frontend/src/transport/mqtt.transport.ts +++ b/frontend/src/transport/mqtt.transport.ts @@ -1,5 +1,5 @@ import type { MqttConnectionOptions, MqttServiceI } from '@/types/services/mqtt.types' -import type { Transport, TransportConnectOptions, TransportSubscribeOptions } from '@/types/transport/transport.types' +import type { Transport, TransportAttachmentHandle, TransportConnectOptions, TransportSubscribeOptions } from '@/types/transport/transport.types' export class MqttTransport implements Transport { private $mqtt: MqttServiceI @@ -8,16 +8,16 @@ export class MqttTransport implements Transport { this.$mqtt = mqtt } - async connect (key: string, options: TransportConnectOptions): Promise { - await this.$mqtt.createClient(key, options as Partial) + attach (key: string, options: TransportConnectOptions): Promise { + return this.$mqtt.attachClient(key, options as Partial) } subscribe (key: string, topics: string[], options: TransportSubscribeOptions = {}): Promise { return this.$mqtt.subscribe(key, topics, options) } - disconnect (key: string): Promise { - return this.$mqtt.destroyClient(key) + detach (handle: TransportAttachmentHandle): Promise { + return this.$mqtt.detachClient(handle) } } diff --git a/frontend/src/types/subscribers/subscriber.types.ts b/frontend/src/types/subscribers/subscriber.types.ts index 532dc747f2..1110bc68e8 100644 --- a/frontend/src/types/subscribers/subscriber.types.ts +++ b/frontend/src/types/subscribers/subscriber.types.ts @@ -12,7 +12,7 @@ export interface Subscriber { destroy?: () => (void | Promise) } -export interface TeamChannelSubscriberI extends Subscriber { +export interface TeamSubscriberI extends Subscriber { connect(team: TeamRef | null | undefined): Promise disconnect(): Promise destroy(): Promise @@ -20,7 +20,8 @@ export interface TeamChannelSubscriberI extends Subscriber { } export type SubscriberInstances = { - teamChannel: TeamChannelSubscriberI | null + teamChannel: TeamSubscriberI | null + liveStatus: TeamSubscriberI | null } export interface CreateSubscriberOptions { diff --git a/frontend/src/types/transport/transport.types.ts b/frontend/src/types/transport/transport.types.ts index eb08ee9d36..9c14f38edf 100644 --- a/frontend/src/types/transport/transport.types.ts +++ b/frontend/src/types/transport/transport.types.ts @@ -12,8 +12,13 @@ export interface TransportSubscribeOptions { qos?: 0 | 1 | 2 } +export interface TransportAttachmentHandle { + key: string + id: number +} + export interface Transport { - connect(key: string, options: TConnect): Promise + attach(key: string, options: TConnect): Promise subscribe(key: string, topics: string[], options?: TransportSubscribeOptions): Promise - disconnect(key: string): Promise + detach(handle: TransportAttachmentHandle): Promise } diff --git a/test/unit/frontend/services/app.orchestrator.spec.js b/test/unit/frontend/services/app.orchestrator.spec.js index 26ea53e8f8..1d77b266ad 100644 --- a/test/unit/frontend/services/app.orchestrator.spec.js +++ b/test/unit/frontend/services/app.orchestrator.spec.js @@ -5,6 +5,7 @@ const mockCreateBootstrapService = vi.fn() const mockCreateMessagingService = vi.fn() const mockCreateMqttService = vi.fn() const mockCreateTeamChannelSubscriber = vi.fn() +const mockCreateLiveStatusSubscriber = vi.fn() const mockCreateMqttTransport = vi.fn() vi.mock('../../../../frontend/src/services/automations.service.js', () => { @@ -43,6 +44,12 @@ vi.mock('../../../../frontend/src/subscribers/team-channel.subscriber.js', () => } }) +vi.mock('../../../../frontend/src/subscribers/live-status.subscriber.js', () => { + return { + createLiveStatusSubscriber: mockCreateLiveStatusSubscriber + } +}) + async function loadOrchestratorModule () { vi.resetModules() return await import('../../../../frontend/src/services/app.orchestrator.ts') @@ -54,6 +61,7 @@ function seedServices () { const postMessageService = { name: 'postMessage', destroy: vi.fn().mockResolvedValue() } const mqttService = { name: 'mqtt', destroy: vi.fn().mockResolvedValue() } const teamChannelSubscriber = { name: 'teamChannel', destroy: vi.fn().mockResolvedValue() } + const liveStatusSubscriber = { name: 'liveStatus', destroy: vi.fn().mockResolvedValue() } const transport = { name: 'mqtt-transport' } mockCreateAutomationsService.mockReturnValue(automationsService) @@ -62,8 +70,9 @@ function seedServices () { mockCreateMqttService.mockReturnValue(mqttService) mockCreateMqttTransport.mockReturnValue(transport) mockCreateTeamChannelSubscriber.mockReturnValue(teamChannelSubscriber) + mockCreateLiveStatusSubscriber.mockReturnValue(liveStatusSubscriber) - return { automationsService, bootstrapService, postMessageService, mqttService, teamChannelSubscriber, transport } + return { automationsService, bootstrapService, postMessageService, mqttService, teamChannelSubscriber, liveStatusSubscriber, transport } } describe('AppOrchestrator', () => { @@ -74,6 +83,7 @@ describe('AppOrchestrator', () => { mockCreateMqttService.mockReset() mockCreateMqttTransport.mockReset() mockCreateTeamChannelSubscriber.mockReset() + mockCreateLiveStatusSubscriber.mockReset() }) test('init boots services + subscribers, injects shared instances, and provides them on the app', async () => { @@ -116,7 +126,7 @@ describe('AppOrchestrator', () => { }) test('dispose destroys services + subscribers and resets internal state', async () => { - const { bootstrapService, postMessageService, mqttService, teamChannelSubscriber } = seedServices() + const { bootstrapService, postMessageService, mqttService, teamChannelSubscriber, liveStatusSubscriber } = seedServices() mqttService.destroy.mockRejectedValueOnce(new Error('destroy failed')) const app = { provide: vi.fn(), unmount: vi.fn() } @@ -130,9 +140,11 @@ describe('AppOrchestrator', () => { postMessageService.destroy.mockClear() mqttService.destroy.mockClear() teamChannelSubscriber.destroy.mockClear() + liveStatusSubscriber.destroy.mockClear() await orchestrator.dispose() expect(teamChannelSubscriber.destroy).toHaveBeenCalledTimes(1) + expect(liveStatusSubscriber.destroy).toHaveBeenCalledTimes(1) expect(bootstrapService.destroy).toHaveBeenCalledTimes(1) expect(postMessageService.destroy).toHaveBeenCalledTimes(1) expect(mqttService.destroy).toHaveBeenCalledTimes(1) @@ -142,7 +154,7 @@ describe('AppOrchestrator', () => { mqtt: null, automations: null }) - expect(orchestrator.$subscriberInstances).toEqual({ teamChannel: null }) + expect(orchestrator.$subscriberInstances).toEqual({ teamChannel: null, liveStatus: null }) expect(orchestrator.$app).toBeNull() expect(orchestrator.$router).toBeNull() expect(orchestrator.$cleanupRegistered).toBe(false) diff --git a/test/unit/frontend/subscribers/live-status.subscriber.spec.js b/test/unit/frontend/subscribers/live-status.subscriber.spec.js new file mode 100644 index 0000000000..2a05cdecd6 --- /dev/null +++ b/test/unit/frontend/subscribers/live-status.subscriber.spec.js @@ -0,0 +1,195 @@ +/* eslint-env browser */ +import { afterEach, beforeEach, describe, expect, test, vi } from 'vitest' + +const getTeamCommsCreds = vi.fn() +const useAccountAuthStore = vi.fn(() => ({ user: { id: 'user-hashid-1' }, getSessionId: () => 'session-test-id' })) +const setInstanceStatus = vi.fn() +const setDeviceStatus = vi.fn() +const setLive = vi.fn() +const clear = vi.fn() +const useLiveStatusStore = vi.fn(() => ({ setInstanceStatus, setDeviceStatus, setLive, clear })) + +vi.mock('@/api/team.js', () => ({ + default: { getTeamCommsCreds: (...args) => getTeamCommsCreds(...args) } +})) +vi.mock('@/stores/account-auth.js', () => ({ useAccountAuthStore })) +vi.mock('@/stores/live-status', () => ({ useLiveStatusStore })) + +function makeTransport () { + return { + attach: vi.fn().mockImplementation(async (key) => ({ key, id: 1 })), + subscribe: vi.fn().mockResolvedValue(undefined), + detach: vi.fn().mockResolvedValue(undefined) + } +} + +describe('LiveStatusSubscriber', async () => { + const mod = await import('../../../../frontend/src/subscribers/live-status.subscriber.ts') + const { createLiveStatusSubscriber, destroyLiveStatusSubscriber } = mod + + function createSubscriber ({ transport = makeTransport() } = {}) { + const subscribers = { teamChannel: null, liveStatus: null } + const subscriber = createLiveStatusSubscriber({ app: {}, router: {}, transport, subscribers }) + subscribers.liveStatus = subscriber + return { subscriber, transport } + } + + beforeEach(async () => { + getTeamCommsCreds.mockReset() + useAccountAuthStore.mockClear().mockReturnValue({ user: { id: 'user-hashid-1' }, getSessionId: () => 'session-test-id' }) + setInstanceStatus.mockClear() + setDeviceStatus.mockClear() + setLive.mockClear() + clear.mockClear() + await destroyLiveStatusSubscriber() + }) + + afterEach(async () => { + await destroyLiveStatusSubscriber() + }) + + describe('connect', () => { + test('attaches the transport keyed by team id', async () => { + const { subscriber, transport } = createSubscriber() + await subscriber.connect({ id: 'team-1' }) + expect(transport.attach).toHaveBeenCalledTimes(1) + expect(transport.attach.mock.calls[0][0]).toBe('team:team-1') + expect(subscriber.isConnected()).toBe(true) + }) + + test('no-ops without a team or logged-in user', async () => { + useAccountAuthStore.mockReturnValue({ user: null, getSessionId: () => 'session-test-id' }) + const { subscriber, transport } = createSubscriber() + await subscriber.connect(null) + await subscriber.connect({ id: 'team-1' }) + expect(transport.attach).not.toHaveBeenCalled() + }) + + test('degrades gracefully when attach rejects', async () => { + const { subscriber, transport } = createSubscriber() + transport.attach.mockRejectedValue(new Error('comms_unavailable')) + await expect(subscriber.connect({ id: 'team-1' })).resolves.toBeUndefined() + expect(subscriber.isConnected()).toBe(false) + expect(setLive).not.toHaveBeenCalled() + }) + }) + + describe('subscribe on connect', () => { + async function connectAndCaptureOnConnect () { + const { subscriber, transport } = createSubscriber() + let onConnect + transport.attach.mockImplementation(async (key, opts) => { + onConnect = opts.onConnect + return { key, id: 1 } + }) + await subscriber.connect({ id: 'team-1' }) + return { subscriber, transport, onConnect } + } + + test('subscribes to the p/+/state and d/+/state wildcards with qos 1', async () => { + const { transport, onConnect } = await connectAndCaptureOnConnect() + await onConnect() + expect(transport.subscribe).toHaveBeenCalledWith( + 'team:team-1', + [ + 'ff/v1/team-1/p/+/state', + 'ff/v1/team-1/d/+/state' + ], + { qos: 1 } + ) + }) + + test('marks the status channel live once subscribed', async () => { + const { onConnect } = await connectAndCaptureOnConnect() + await onConnect() + expect(setLive).toHaveBeenCalledWith(true) + }) + + test('does not mark live when the subscribe fails', async () => { + const { subscriber, transport } = createSubscriber() + transport.subscribe.mockRejectedValue(new Error('subscribe failed')) + let onConnect + transport.attach.mockImplementation(async (key, opts) => { + onConnect = opts.onConnect + return { key, id: 1 } + }) + await subscriber.connect({ id: 'team-1' }) + await onConnect() + expect(setLive).not.toHaveBeenCalled() + }) + }) + + describe('message routing (dispatches to the store)', () => { + async function connectAndCaptureOnMessage () { + const { subscriber, transport } = createSubscriber() + let onMessage + transport.attach.mockImplementation(async (key, opts) => { + onMessage = opts.onMessage + return { key, id: 1 } + }) + await subscriber.connect({ id: 'team-1' }) + return { subscriber, onMessage } + } + + test('instance state routes id + state into setInstanceStatus', async () => { + const { onMessage } = await connectAndCaptureOnMessage() + onMessage('ff/v1/team-1/p/inst-1/state', Buffer.from(JSON.stringify({ id: 'inst-1', meta: { state: 'running' } }))) + expect(setInstanceStatus).toHaveBeenCalledWith('inst-1', 'running') + expect(setDeviceStatus).not.toHaveBeenCalled() + }) + + test('device state routes id + state into setDeviceStatus', async () => { + const { onMessage } = await connectAndCaptureOnMessage() + onMessage('ff/v1/team-1/d/dev-1/state', Buffer.from(JSON.stringify({ id: 'dev-1', meta: { state: 'stopped' } }))) + expect(setDeviceStatus).toHaveBeenCalledWith('dev-1', 'stopped') + expect(setInstanceStatus).not.toHaveBeenCalled() + }) + + test('ignores a state message missing id or state', async () => { + const { onMessage } = await connectAndCaptureOnMessage() + onMessage('ff/v1/team-1/p/inst-1/state', Buffer.from(JSON.stringify({ id: 'inst-1' }))) + onMessage('ff/v1/team-1/d/dev-1/state', Buffer.from(JSON.stringify({ meta: { state: 'running' } }))) + expect(setInstanceStatus).not.toHaveBeenCalled() + expect(setDeviceStatus).not.toHaveBeenCalled() + }) + + test('ignores membership/team-updated topics (owned by the team-channel subscriber)', async () => { + const { onMessage } = await connectAndCaptureOnMessage() + onMessage('ff/v1/team-1/t/updated', Buffer.from('{}')) + onMessage('ff/v1/team-1/u/user-hashid-1/membership', Buffer.from(JSON.stringify({ reason: 'removed' }))) + expect(setInstanceStatus).not.toHaveBeenCalled() + expect(setDeviceStatus).not.toHaveBeenCalled() + }) + + test('does not throw on malformed JSON payloads', async () => { + const { onMessage } = await connectAndCaptureOnMessage() + expect(() => onMessage('ff/v1/team-1/p/inst-1/state', Buffer.from('not json'))).not.toThrow() + expect(setInstanceStatus).not.toHaveBeenCalled() + }) + }) + + describe('disconnect / destroy', () => { + test('disconnect detaches and clears the live-status store', async () => { + const { subscriber, transport } = createSubscriber() + await subscriber.connect({ id: 'team-1' }) + await subscriber.disconnect() + expect(transport.detach).toHaveBeenCalledWith(expect.objectContaining({ key: 'team:team-1' })) + expect(clear).toHaveBeenCalledTimes(1) + expect(subscriber.isConnected()).toBe(false) + }) + + test('disconnect is a no-op when not connected', async () => { + const { subscriber, transport } = createSubscriber() + await subscriber.disconnect() + expect(transport.detach).not.toHaveBeenCalled() + expect(clear).not.toHaveBeenCalled() + }) + + test('destroy disconnects the active attachment', async () => { + const { subscriber, transport } = createSubscriber() + await subscriber.connect({ id: 'team-1' }) + await subscriber.destroy() + expect(transport.detach).toHaveBeenCalledWith(expect.objectContaining({ key: 'team:team-1' })) + }) + }) +}) diff --git a/test/unit/frontend/subscribers/team-channel.subscriber.spec.js b/test/unit/frontend/subscribers/team-channel.subscriber.spec.js index 7dae63f4d1..027875f8a7 100644 --- a/test/unit/frontend/subscribers/team-channel.subscriber.spec.js +++ b/test/unit/frontend/subscribers/team-channel.subscriber.spec.js @@ -6,24 +6,18 @@ const refreshTeam = vi.fn().mockResolvedValue(undefined) const onTeamChannelMembership = vi.fn().mockResolvedValue(undefined) const useContextStore = vi.fn(() => ({ refreshTeam, onTeamChannelMembership })) const useAccountAuthStore = vi.fn(() => ({ user: { id: 'user-hashid-1' }, getSessionId: () => 'session-test-id' })) -const setInstanceStatus = vi.fn() -const setDeviceStatus = vi.fn() -const setLive = vi.fn() -const clear = vi.fn() -const useLiveStatusStore = vi.fn(() => ({ setInstanceStatus, setDeviceStatus, setLive, clear })) vi.mock('@/api/team.js', () => ({ default: { getTeamCommsCreds: (...args) => getTeamCommsCreds(...args) } })) vi.mock('@/stores/context.js', () => ({ useContextStore })) vi.mock('@/stores/account-auth.js', () => ({ useAccountAuthStore })) -vi.mock('@/stores/live-status', () => ({ useLiveStatusStore })) function makeTransport () { return { - connect: vi.fn(), + attach: vi.fn().mockImplementation(async (key) => ({ key, id: 1 })), subscribe: vi.fn().mockResolvedValue(undefined), - disconnect: vi.fn().mockResolvedValue(undefined) + detach: vi.fn().mockResolvedValue(undefined) } } @@ -39,7 +33,7 @@ describe('TeamChannelSubscriber', async () => { const { createTeamChannelSubscriber, destroyTeamChannelSubscriber } = mod function createSubscriber ({ transport = makeTransport(), router = makeRouter() } = {}) { - const subscribers = { teamChannel: null } + const subscribers = { teamChannel: null, liveStatus: null } const subscriber = createTeamChannelSubscriber({ app: {}, router, transport, subscribers }) subscribers.teamChannel = subscriber return { subscriber, transport, router } @@ -51,10 +45,6 @@ describe('TeamChannelSubscriber', async () => { onTeamChannelMembership.mockClear() useContextStore.mockClear() useAccountAuthStore.mockClear().mockReturnValue({ user: { id: 'user-hashid-1' }, getSessionId: () => 'session-test-id' }) - setInstanceStatus.mockClear() - setDeviceStatus.mockClear() - setLive.mockClear() - clear.mockClear() await destroyTeamChannelSubscriber() }) @@ -69,7 +59,7 @@ describe('TeamChannelSubscriber', async () => { await subscriber.connect({}) await subscriber.connect({ id: '' }) await subscriber.connect({ id: 123 }) - expect(transport.connect).not.toHaveBeenCalled() + expect(transport.attach).not.toHaveBeenCalled() expect(subscriber.isConnected()).toBe(false) }) @@ -77,16 +67,15 @@ describe('TeamChannelSubscriber', async () => { useAccountAuthStore.mockReturnValue({ user: null, getSessionId: () => 'session-test-id' }) const { subscriber, transport } = createSubscriber() await subscriber.connect({ id: 'team-1' }) - expect(transport.connect).not.toHaveBeenCalled() + expect(transport.attach).not.toHaveBeenCalled() expect(subscriber.isConnected()).toBe(false) }) - test('connects the transport keyed by team id', async () => { + test('attaches the transport keyed by team id', async () => { const { subscriber, transport } = createSubscriber() - transport.connect.mockResolvedValue(undefined) await subscriber.connect({ id: 'team-1' }) - expect(transport.connect).toHaveBeenCalledTimes(1) - const [key, opts] = transport.connect.mock.calls[0] + expect(transport.attach).toHaveBeenCalledTimes(1) + const [key, opts] = transport.attach.mock.calls[0] expect(key).toBe('team:team-1') expect(typeof opts.getCredentials).toBe('function') expect(typeof opts.onMessage).toBe('function') @@ -96,9 +85,8 @@ describe('TeamChannelSubscriber', async () => { test('wires the full lifecycle handler surface', async () => { const { subscriber, transport } = createSubscriber() - transport.connect.mockResolvedValue(undefined) await subscriber.connect({ id: 'team-1' }) - const opts = transport.connect.mock.calls[0][1] + const opts = transport.attach.mock.calls[0][1] expect(typeof opts.onMessage).toBe('function') expect(typeof opts.onConnect).toBe('function') expect(typeof opts.onClose).toBe('function') @@ -109,9 +97,8 @@ describe('TeamChannelSubscriber', async () => { test('quiet lifecycle handlers do not throw', async () => { const { subscriber, transport } = createSubscriber() - transport.connect.mockResolvedValue(undefined) await subscriber.connect({ id: 'team-1' }) - const opts = transport.connect.mock.calls[0][1] + const opts = transport.attach.mock.calls[0][1] expect(() => opts.onClose()).not.toThrow() expect(() => opts.onOffline()).not.toThrow() expect(() => opts.onDisconnect()).not.toThrow() @@ -121,45 +108,69 @@ describe('TeamChannelSubscriber', async () => { test('credential callback forwards team id + per-tab session id', async () => { getTeamCommsCreds.mockResolvedValue({ url: 'wss://broker', username: 'u', password: 'p' }) const { subscriber, transport } = createSubscriber() - transport.connect.mockResolvedValue(undefined) await subscriber.connect({ id: 'team-1' }) - const opts = transport.connect.mock.calls[0][1] + const opts = transport.attach.mock.calls[0][1] await opts.getCredentials() expect(getTeamCommsCreds).toHaveBeenCalledWith('team-1', 'session-test-id') }) test('skips reconnect when already connected to the same team', async () => { const { subscriber, transport } = createSubscriber() - transport.connect.mockResolvedValue(undefined) await subscriber.connect({ id: 'team-1' }) await subscriber.connect({ id: 'team-1' }) - expect(transport.connect).toHaveBeenCalledTimes(1) + expect(transport.attach).toHaveBeenCalledTimes(1) }) - test('disconnects from the previous team when switching teams', async () => { + test('detaches from the previous team when switching teams', async () => { const { subscriber, transport } = createSubscriber() - transport.connect.mockResolvedValue(undefined) await subscriber.connect({ id: 'team-1' }) await subscriber.connect({ id: 'team-2' }) - expect(transport.disconnect).toHaveBeenCalledWith('team:team-1') - expect(transport.connect).toHaveBeenCalledTimes(2) - expect(transport.connect.mock.calls[1][0]).toBe('team:team-2') + expect(transport.detach).toHaveBeenCalledWith(expect.objectContaining({ key: 'team:team-1' })) + expect(transport.attach).toHaveBeenCalledTimes(2) + expect(transport.attach.mock.calls[1][0]).toBe('team:team-2') }) - test('degrades gracefully when connect rejects (e.g. no broker)', async () => { + test('degrades gracefully when attach rejects (e.g. no broker)', async () => { const { subscriber, transport } = createSubscriber() - transport.connect.mockRejectedValue(new Error('comms_unavailable')) + transport.attach.mockRejectedValue(new Error('comms_unavailable')) await expect(subscriber.connect({ id: 'team-1' })).resolves.toBeUndefined() expect(subscriber.isConnected()).toBe(false) }) + + test('serializes overlapping connects to the same team (no orphaned attachment)', async () => { + const { subscriber, transport } = createSubscriber() + let resolveAttach + const gate = new Promise((resolve) => { resolveAttach = resolve }) + transport.attach.mockImplementation((key) => gate.then(() => ({ key, id: 1 }))) + + const first = subscriber.connect({ id: 'team-1' }) + const second = subscriber.connect({ id: 'team-1' }) + resolveAttach() + await Promise.all([first, second]) + + expect(transport.attach).toHaveBeenCalledTimes(1) + expect(subscriber.isConnected()).toBe(true) + }) + + test('serialized switch detaches the old team exactly once before attaching the new one', async () => { + const { subscriber, transport } = createSubscriber() + const first = subscriber.connect({ id: 'team-1' }) + const second = subscriber.connect({ id: 'team-2' }) + await Promise.all([first, second]) + + expect(transport.detach).toHaveBeenCalledTimes(1) + expect(transport.detach).toHaveBeenCalledWith(expect.objectContaining({ key: 'team:team-1' })) + expect(transport.attach.mock.calls.map(c => c[0])).toEqual(['team:team-1', 'team:team-2']) + }) }) describe('subscribe on connect', () => { - test('subscribes to t/updated, membership and the state wildcards with qos 1', async () => { + test('subscribes to t/updated and membership with qos 1', async () => { const { subscriber, transport } = createSubscriber() let onConnect - transport.connect.mockImplementation(async (_key, opts) => { + transport.attach.mockImplementation(async (key, opts) => { onConnect = opts.onConnect + return { key, id: 1 } }) await subscriber.connect({ id: 'team-1' }) await onConnect() @@ -167,43 +178,19 @@ describe('TeamChannelSubscriber', async () => { 'team:team-1', [ 'ff/v1/team-1/t/updated', - 'ff/v1/team-1/u/user-hashid-1/membership', - 'ff/v1/team-1/p/+/state', - 'ff/v1/team-1/d/+/state' + 'ff/v1/team-1/u/user-hashid-1/membership' ], { qos: 1 } ) }) - test('marks the status channel live once subscribed', async () => { - const { subscriber, transport } = createSubscriber() - let onConnect - transport.connect.mockImplementation(async (_key, opts) => { - onConnect = opts.onConnect - }) - await subscriber.connect({ id: 'team-1' }) - await onConnect() - expect(setLive).toHaveBeenCalledWith(true) - }) - - test('does not mark live when the subscribe fails', async () => { - const { subscriber, transport } = createSubscriber() - transport.subscribe.mockRejectedValue(new Error('subscribe failed')) - let onConnect - transport.connect.mockImplementation(async (_key, opts) => { - onConnect = opts.onConnect - }) - await subscriber.connect({ id: 'team-1' }) - await onConnect() - expect(setLive).not.toHaveBeenCalled() - }) - test('swallows subscribe errors (transport reconnect will retry)', async () => { const { subscriber, transport } = createSubscriber() transport.subscribe.mockRejectedValue(new Error('subscribe failed')) let onConnect - transport.connect.mockImplementation(async (_key, opts) => { + transport.attach.mockImplementation(async (key, opts) => { onConnect = opts.onConnect + return { key, id: 1 } }) await subscriber.connect({ id: 'team-1' }) await expect(onConnect()).resolves.toBeUndefined() @@ -214,8 +201,9 @@ describe('TeamChannelSubscriber', async () => { async function connectAndCaptureOnMessage () { const { subscriber, transport, router } = createSubscriber() let onMessage - transport.connect.mockImplementation(async (_key, opts) => { + transport.attach.mockImplementation(async (key, opts) => { onMessage = opts.onMessage + return { key, id: 1 } }) await subscriber.connect({ id: 'team-1' }) return { subscriber, transport, router, onMessage } @@ -254,60 +242,35 @@ describe('TeamChannelSubscriber', async () => { expect(onTeamChannelMembership).not.toHaveBeenCalled() }) - test('instance state routes id + state into setInstanceStatus', async () => { + test('ignores instance/device state topics (owned by the live-status subscriber)', async () => { const { onMessage } = await connectAndCaptureOnMessage() onMessage('ff/v1/team-1/p/inst-1/state', Buffer.from(JSON.stringify({ id: 'inst-1', meta: { state: 'running' } }))) - expect(setInstanceStatus).toHaveBeenCalledWith('inst-1', 'running') - expect(setDeviceStatus).not.toHaveBeenCalled() - }) - - test('device state routes id + state into setDeviceStatus', async () => { - const { onMessage } = await connectAndCaptureOnMessage() onMessage('ff/v1/team-1/d/dev-1/state', Buffer.from(JSON.stringify({ id: 'dev-1', meta: { state: 'stopped' } }))) - expect(setDeviceStatus).toHaveBeenCalledWith('dev-1', 'stopped') - expect(setInstanceStatus).not.toHaveBeenCalled() - }) - - test('ignores a state message missing id or state', async () => { - const { onMessage } = await connectAndCaptureOnMessage() - onMessage('ff/v1/team-1/p/inst-1/state', Buffer.from(JSON.stringify({ id: 'inst-1' }))) - onMessage('ff/v1/team-1/d/dev-1/state', Buffer.from(JSON.stringify({ meta: { state: 'running' } }))) - expect(setInstanceStatus).not.toHaveBeenCalled() - expect(setDeviceStatus).not.toHaveBeenCalled() + expect(refreshTeam).not.toHaveBeenCalled() + expect(onTeamChannelMembership).not.toHaveBeenCalled() }) }) describe('disconnect / destroy', () => { - test('disconnect tears down the client and clears connected state', async () => { + test('disconnect tears down the attachment and clears connected state', async () => { const { subscriber, transport } = createSubscriber() - transport.connect.mockResolvedValue(undefined) await subscriber.connect({ id: 'team-1' }) await subscriber.disconnect() - expect(transport.disconnect).toHaveBeenCalledWith('team:team-1') + expect(transport.detach).toHaveBeenCalledWith(expect.objectContaining({ key: 'team:team-1' })) expect(subscriber.isConnected()).toBe(false) }) - test('disconnect clears the live-status store (no prior-team leak)', async () => { - const { subscriber, transport } = createSubscriber() - transport.connect.mockResolvedValue(undefined) - await subscriber.connect({ id: 'team-1' }) - await subscriber.disconnect() - expect(clear).toHaveBeenCalledTimes(1) - }) - test('disconnect is a no-op when not connected', async () => { const { subscriber, transport } = createSubscriber() await subscriber.disconnect() - expect(transport.disconnect).not.toHaveBeenCalled() - expect(clear).not.toHaveBeenCalled() + expect(transport.detach).not.toHaveBeenCalled() }) - test('destroy disconnects the active client', async () => { + test('destroy disconnects the active attachment', async () => { const { subscriber, transport } = createSubscriber() - transport.connect.mockResolvedValue(undefined) await subscriber.connect({ id: 'team-1' }) await subscriber.destroy() - expect(transport.disconnect).toHaveBeenCalledWith('team:team-1') + expect(transport.detach).toHaveBeenCalledWith(expect.objectContaining({ key: 'team:team-1' })) }) }) }) From 91a07e2841537c0295d84abde8462a193a730463 Mon Sep 17 00:00:00 2001 From: Noley Holland Date: Fri, 17 Jul 2026 09:34:26 -0700 Subject: [PATCH 2/2] Fix calling of method in transport --- frontend/src/transport/mqtt.transport.ts | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/frontend/src/transport/mqtt.transport.ts b/frontend/src/transport/mqtt.transport.ts index 5384917ad6..d66d104ce0 100644 --- a/frontend/src/transport/mqtt.transport.ts +++ b/frontend/src/transport/mqtt.transport.ts @@ -9,7 +9,7 @@ export class MqttTransport implements Transport { } attach (key: string, options: TransportConnectOptions): Promise { - return this.$mqtt.attachClient(key, options as Partial) + return this.$mqtt.attachClientObserver(key, options as Partial) } subscribe (key: string, topics: string[], options: TransportSubscribeOptions = {}): Promise { @@ -17,7 +17,7 @@ export class MqttTransport implements Transport { } detach (handle: TransportAttachmentHandle): Promise { - return this.$mqtt.detachClient(handle) + return this.$mqtt.detachClientObserver(handle) } }