[8081] Data Farm > Hosted Instances 3 - subscriber

forge/routes/api/project.js

@@ -233,7 +233,11 @@ module.exports = async function (app) {
         const projectViewPromise = app.db.views.Project.project(project)
         const projectStatePromise = project.liveState()
 
-        app.comms?.team?.notifyEntityLifecycle(team.hashid, 'p', project.id, 'created', await app.db.views.Project.project(project, { includeSettings: false }))
+        if (app.comms?.team) {
+            const createdInstanceView = await app.db.views.Project.project(project, { includeSettings: false })
+            createdInstanceView.application = app.db.views.Application.applicationSummary(application)
+            app.comms.team.notifyEntityLifecycle(team.hashid, 'p', project.id, 'created', createdInstanceView)
+        }
 
         reply.send({ ...await projectViewPromise, ...await projectStatePromise })
     })

frontend/src/stores/data-farm-hosted-instances.ts

@@ -48,6 +48,24 @@ export const useDataFarmHostedInstancesStore = defineStore('data-farm-hosted-ins
         currentPageIds.value = currentPageIds.value.filter(instanceId => instanceId !== id)
     }
 
+    function upsertInstance (instance: StoreInstance): void {
+        if (!instance?.id) return
+        instancesById.value[instance.id] = { ...instancesById.value[instance.id], ...instance }
+    }
+
+    function applyRealtimeEvent (event: { id?: string, action?: string, data?: StoreInstance }): void {
+        if (!event?.id || !event.action) return
+        if (event.action === 'deleted') {
+            removeInstance(event.id)
+            return
+        }
+        if (!event.data) return
+        upsertInstance(event.data)
+        if (event.action === 'created' && !currentPageIds.value.includes(event.id)) {
+            currentPageIds.value.push(event.id)
+        }
+    }
+
     async function fetchTeamInstancesPage (teamId: string, query: PageQuery = {}): Promise<void> {
         if (!teamId) return
         const response = await teamApi.getInstances(teamId, {
@@ -147,6 +165,7 @@ export const useDataFarmHostedInstancesStore = defineStore('data-farm-hosted-ins
         currentPageInstances,
         fetchTeamInstancesPage,
         removeInstance,
+        applyRealtimeEvent,
         applyLiveStatus,
         applyPolledStatus,
         startInstance,

frontend/src/subscribers/hosted-instances.subscriber.ts

@@ -0,0 +1,47 @@
+import { defineSubscriberSingleton } from './subscriber.factory'
+import { SubscriberRoute, TeamSubscriber } from './team-subscriber.contract'
+
+import { useDataFarmHostedInstancesStore } from '@/stores/data-farm-hosted-instances'
+import type { InstanceSummary } from '@/types'
+import type { CreateSubscriberOptions, TeamSubscriberI } from '@/types/subscribers/subscriber.types'
+
+const INSTANCE_LIFECYCLE_TOPIC_REGEX = /^ff\/v1\/[^/]+\/p\/[^/]+\/(?:created|updated|deleted)$/
+
+class HostedInstancesSubscriber extends TeamSubscriber implements TeamSubscriberI {
+    constructor ({ app, router, transport, subscribers }: CreateSubscriberOptions) {
+        super({
+            name: 'hostedInstances',
+            app,
+            router,
+            transport,
+            subscribers
+        })
+    }
+
+    protected _topics (teamId: string): string[] {
+        return [
+            `ff/v1/${teamId}/p/+/created`,
+            `ff/v1/${teamId}/p/+/updated`,
+            `ff/v1/${teamId}/p/+/deleted`
+        ]
+    }
+
+    protected _routes (): SubscriberRoute[] {
+        return [
+            { pattern: INSTANCE_LIFECYCLE_TOPIC_REGEX, handle: (payload) => this._onLifecycle(payload) }
+        ]
+    }
+
+    protected _onLifecycle (payload: { id?: string, action?: string, data?: InstanceSummary }): void {
+        if (!payload?.id || !payload.action) return
+        try {
+            useDataFarmHostedInstancesStore().applyRealtimeEvent(payload)
+        } catch {}
+    }
+}
+
+const { create: createHostedInstancesSubscriber, destroy: destroyHostedInstancesSubscriber } = defineSubscriberSingleton(HostedInstancesSubscriber)
+
+export { createHostedInstancesSubscriber, destroyHostedInstancesSubscriber }
+
+export default createHostedInstancesSubscriber

frontend/src/subscribers/subscriber.registry.ts

@@ -1,9 +1,11 @@
 import { createApplicationsSubscriber } from './applications.subscriber'
+import { createHostedInstancesSubscriber } from './hosted-instances.subscriber'
 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: 'liveStatus' as const, create: createLiveStatusSubscriber, requiredLifecycle: ['destroy'] as const },
-    { key: 'applications' as const, create: createApplicationsSubscriber, requiredLifecycle: ['destroy'] as const }
+    { key: 'applications' as const, create: createApplicationsSubscriber, requiredLifecycle: ['destroy'] as const },
+    { key: 'hostedInstances' as const, create: createHostedInstancesSubscriber, requiredLifecycle: ['destroy'] as const }
 ]

frontend/src/types/subscribers/subscriber.types.ts

@@ -23,6 +23,7 @@ export type SubscriberInstances = {
     teamChannel: TeamSubscriberI | null
     liveStatus: TeamSubscriberI | null
     applications: TeamSubscriberI | null
+    hostedInstances: TeamSubscriberI | null
 }
 
 export interface CreateSubscriberOptions<TTransport extends Transport = Transport> {

test/unit/forge/routes/api/project_spec.js

@@ -1275,6 +1275,9 @@ describe('Project API', function () {
             const data = notifySpy.getCall(0).args[4]
             data.should.have.property('id', result.id)
             data.should.not.have.property('settings')
+            // owning application must be in the payload so the realtime row isn't "unassigned"
+            data.should.have.property('application')
+            data.application.should.have.property('id', TestObjects.ApplicationA.hashid)
         })
 
         it('publishes updated on a synchronous update without leaking settings', async function () {

test/unit/frontend/services/app.orchestrator.spec.js

@@ -7,6 +7,7 @@ const mockCreateMqttService = vi.fn()
 const mockCreateTeamChannelSubscriber = vi.fn()
 const mockCreateLiveStatusSubscriber = vi.fn()
 const mockCreateApplicationsSubscriber = vi.fn()
+const mockCreateHostedInstancesSubscriber = vi.fn()
 const mockCreateMqttTransport = vi.fn()
 
 vi.mock('../../../../frontend/src/services/automations.service.js', () => {
@@ -57,6 +58,12 @@ vi.mock('../../../../frontend/src/subscribers/applications.subscriber.js', () =>
     }
 })
 
+vi.mock('../../../../frontend/src/subscribers/hosted-instances.subscriber.js', () => {
+    return {
+        createHostedInstancesSubscriber: mockCreateHostedInstancesSubscriber
+    }
+})
+
 async function loadOrchestratorModule () {
     vi.resetModules()
     return await import('../../../../frontend/src/services/app.orchestrator.ts')
@@ -70,6 +77,7 @@ function seedServices () {
     const teamChannelSubscriber = { name: 'teamChannel', destroy: vi.fn().mockResolvedValue() }
     const liveStatusSubscriber = { name: 'liveStatus', destroy: vi.fn().mockResolvedValue() }
     const applicationsSubscriber = { name: 'applications', destroy: vi.fn().mockResolvedValue() }
+    const hostedInstancesSubscriber = { name: 'hostedInstances', destroy: vi.fn().mockResolvedValue() }
     const transport = { name: 'mqtt-transport' }
 
     mockCreateAutomationsService.mockReturnValue(automationsService)
@@ -80,8 +88,9 @@ function seedServices () {
     mockCreateTeamChannelSubscriber.mockReturnValue(teamChannelSubscriber)
     mockCreateLiveStatusSubscriber.mockReturnValue(liveStatusSubscriber)
     mockCreateApplicationsSubscriber.mockReturnValue(applicationsSubscriber)
+    mockCreateHostedInstancesSubscriber.mockReturnValue(hostedInstancesSubscriber)
 
-    return { automationsService, bootstrapService, postMessageService, mqttService, teamChannelSubscriber, liveStatusSubscriber, applicationsSubscriber, transport }
+    return { automationsService, bootstrapService, postMessageService, mqttService, teamChannelSubscriber, liveStatusSubscriber, applicationsSubscriber, hostedInstancesSubscriber, transport }
 }
 
 describe('AppOrchestrator', () => {
@@ -163,7 +172,7 @@ describe('AppOrchestrator', () => {
             mqtt: null,
             automations: null
         })
-        expect(orchestrator.$subscribers).toEqual({ teamChannel: null, liveStatus: null, applications: null })
+        expect(orchestrator.$subscribers).toEqual({ teamChannel: null, liveStatus: null, applications: null, hostedInstances: null })
         expect(orchestrator.$app).toBeNull()
         expect(orchestrator.$router).toBeNull()
         expect(orchestrator.$cleanupRegistered).toBe(false)

test/unit/frontend/stores/data-farm-hosted-instances.spec.js

@@ -162,6 +162,65 @@ describe('data-farm-hosted-instances store', () => {
         })
     })
 
+    describe('applyRealtimeEvent', () => {
+        it('created surfaces the instance on the current page (cross-session add)', async () => {
+            vi.spyOn(teamApi, 'getInstances').mockResolvedValue({ projects: [instance('i1', 'running')], meta: { total: 1 } })
+            const store = useDataFarmHostedInstancesStore()
+            await store.fetchTeamInstancesPage('team-1')
+
+            store.applyRealtimeEvent({ id: 'i2', action: 'created', data: instance('i2', 'stopped') })
+
+            expect(store.currentPageIds).toEqual(['i1', 'i2'])
+            expect(store.instancesById.i2.name).toBe('i2')
+        })
+
+        it('updated merges in place and does not pull an off-page instance onto the current page', async () => {
+            vi.spyOn(teamApi, 'getInstances').mockResolvedValue({ projects: [instance('i1', 'running')], meta: { total: 1 } })
+            const store = useDataFarmHostedInstancesStore()
+            await store.fetchTeamInstancesPage('team-1')
+
+            store.applyRealtimeEvent({ id: 'i1', action: 'updated', data: { id: 'i1', name: 'Renamed' } })
+            expect(store.instancesById.i1.name).toBe('Renamed')
+
+            store.applyRealtimeEvent({ id: 'i9', action: 'updated', data: instance('i9', 'running') })
+            expect(store.currentPageIds).toEqual(['i1'])
+            expect(store.instancesById.i9).toBeDefined()
+        })
+
+        it('deleted removes the instance from byId and the current page', async () => {
+            vi.spyOn(teamApi, 'getInstances').mockResolvedValue({ projects: [instance('i1'), instance('i2')], meta: { total: 2 } })
+            const store = useDataFarmHostedInstancesStore()
+            await store.fetchTeamInstancesPage('team-1')
+
+            store.applyRealtimeEvent({ id: 'i1', action: 'deleted' })
+
+            expect(store.currentPageIds).toEqual(['i2'])
+            expect(store.instancesById.i1).toBeUndefined()
+        })
+
+        it('ignores events missing id or action', () => {
+            const store = useDataFarmHostedInstancesStore()
+            store.applyRealtimeEvent({ action: 'created', data: instance('x') })
+            store.applyRealtimeEvent({ id: 'x' })
+            expect(store.currentPageIds).toEqual([])
+            expect(store.instancesById.x).toBeUndefined()
+        })
+
+        it('created is idempotent and a created event with no data is ignored', async () => {
+            vi.spyOn(teamApi, 'getInstances').mockResolvedValue({ projects: [instance('i1')], meta: { total: 1 } })
+            const store = useDataFarmHostedInstancesStore()
+            await store.fetchTeamInstancesPage('team-1')
+
+            store.applyRealtimeEvent({ id: 'i2', action: 'created', data: instance('i2') })
+            store.applyRealtimeEvent({ id: 'i2', action: 'created', data: instance('i2') })
+            expect(store.currentPageIds).toEqual(['i1', 'i2'])
+
+            store.applyRealtimeEvent({ id: 'i3', action: 'created' })
+            expect(store.currentPageIds).toEqual(['i1', 'i2'])
+            expect(store.instancesById.i3).toBeUndefined()
+        })
+    })
+
     describe('lifecycle actions', () => {
         it('startInstance sets optimistic then pending-from-server on success', async () => {
             vi.spyOn(instanceApi, 'startInstance').mockResolvedValue({})

test/unit/frontend/subscribers/hosted-instances.subscriber.spec.js

@@ -0,0 +1,135 @@
+/* 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 applyRealtimeEvent = vi.fn()
+const useDataFarmHostedInstancesStore = vi.fn(() => ({ applyRealtimeEvent }))
+
+vi.mock('@/api/team.js', () => ({
+    default: { getTeamCommsCreds: (...args) => getTeamCommsCreds(...args) }
+}))
+vi.mock('@/stores/account-auth.js', () => ({ useAccountAuthStore }))
+vi.mock('@/stores/data-farm-hosted-instances', () => ({ useDataFarmHostedInstancesStore }))
+
+function makeTransport () {
+    return {
+        attach: vi.fn().mockImplementation(async (key) => ({ key, id: 1 })),
+        subscribe: vi.fn().mockResolvedValue(undefined),
+        detach: vi.fn().mockResolvedValue(undefined)
+    }
+}
+
+describe('HostedInstancesSubscriber', async () => {
+    const mod = await import('../../../../frontend/src/subscribers/hosted-instances.subscriber.ts')
+    const { createHostedInstancesSubscriber, destroyHostedInstancesSubscriber } = mod
+
+    function createSubscriber ({ transport = makeTransport() } = {}) {
+        const subscribers = { teamChannel: null, liveStatus: null, applications: null, hostedInstances: null }
+        const subscriber = createHostedInstancesSubscriber({ app: {}, router: {}, transport, subscribers })
+        subscribers.hostedInstances = subscriber
+        return { subscriber, transport }
+    }
+
+    beforeEach(async () => {
+        getTeamCommsCreds.mockReset()
+        useAccountAuthStore.mockClear().mockReturnValue({ user: { id: 'user-hashid-1' }, getSessionId: () => 'session-test-id' })
+        applyRealtimeEvent.mockClear()
+        await destroyHostedInstancesSubscriber()
+    })
+
+    afterEach(async () => {
+        await destroyHostedInstancesSubscriber()
+    })
+
+    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/+/created|updated|deleted wildcards with qos 1', async () => {
+            const { transport, onConnect } = await connectAndCaptureOnConnect()
+            await onConnect()
+            expect(transport.subscribe).toHaveBeenCalledWith(
+                'team:team-1',
+                [
+                    'ff/v1/team-1/p/+/created',
+                    'ff/v1/team-1/p/+/updated',
+                    'ff/v1/team-1/p/+/deleted'
+                ],
+                { qos: 1 }
+            )
+        })
+    })
+
+    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('created topic forwards the event to applyRealtimeEvent', async () => {
+            const { onMessage } = await connectAndCaptureOnMessage()
+            const event = { id: 'inst-1', action: 'created', data: { id: 'inst-1', name: 'One' } }
+            onMessage('ff/v1/team-1/p/inst-1/created', Buffer.from(JSON.stringify(event)))
+            expect(applyRealtimeEvent).toHaveBeenCalledWith(event)
+        })
+
+        test('updated topic forwards the event to applyRealtimeEvent', async () => {
+            const { onMessage } = await connectAndCaptureOnMessage()
+            const event = { id: 'inst-1', action: 'updated', data: { id: 'inst-1', name: 'Renamed' } }
+            onMessage('ff/v1/team-1/p/inst-1/updated', Buffer.from(JSON.stringify(event)))
+            expect(applyRealtimeEvent).toHaveBeenCalledWith(event)
+        })
+
+        test('deleted topic forwards the event to applyRealtimeEvent', async () => {
+            const { onMessage } = await connectAndCaptureOnMessage()
+            const event = { id: 'inst-1', action: 'deleted' }
+            onMessage('ff/v1/team-1/p/inst-1/deleted', Buffer.from(JSON.stringify(event)))
+            expect(applyRealtimeEvent).toHaveBeenCalledWith(event)
+        })
+
+        test('ignores an event missing id or action', async () => {
+            const { onMessage } = await connectAndCaptureOnMessage()
+            onMessage('ff/v1/team-1/p/inst-1/created', Buffer.from(JSON.stringify({ action: 'created' })))
+            onMessage('ff/v1/team-1/p/inst-1/created', Buffer.from(JSON.stringify({ id: 'inst-1' })))
+            expect(applyRealtimeEvent).not.toHaveBeenCalled()
+        })
+
+        test('ignores unrelated topics (p/+/state owned by live-status, and team-channel)', async () => {
+            const { onMessage } = await connectAndCaptureOnMessage()
+            onMessage('ff/v1/team-1/p/inst-1/state', Buffer.from(JSON.stringify({ id: 'inst-1', meta: { state: 'running' } })))
+            onMessage('ff/v1/team-1/t/updated', Buffer.from('{}'))
+            expect(applyRealtimeEvent).not.toHaveBeenCalled()
+        })
+
+        test('does not throw on malformed JSON payloads', async () => {
+            const { onMessage } = await connectAndCaptureOnMessage()
+            expect(() => onMessage('ff/v1/team-1/p/inst-1/created', Buffer.from('not json'))).not.toThrow()
+            expect(applyRealtimeEvent).not.toHaveBeenCalled()
+        })
+    })
+
+    describe('disconnect / destroy', () => {
+        test('disconnect detaches the transport', 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(subscriber.isConnected()).toBe(false)
+        })
+    })
+})