[7324] Replace HTTP polling of instance / device status with MQTT-over-WS

forge/comms/aclManager.js

@@ -140,6 +140,58 @@ module.exports = function (app) {
                 return false
             }
         },
+        checkUserCanReadInstance: async function (requestParts, usernameParts) {
+            // requestParts = [ fullTopic, <teamHash>, <projectId> ]
+            // usernameParts = [ 'team-frontend', <userHash>, <teamHash>, <sessionId> ]
+            const topicTeamHash = requestParts[1]
+            const projectId = requestParts[2]
+            const usernameUserHash = usernameParts[1]
+            const usernameTeamHash = usernameParts[2]
+            if (topicTeamHash !== usernameTeamHash) {
+                return false
+            }
+            try {
+                const team = await app.db.models.Team.byId(usernameTeamHash)
+                if (!team) return false
+                const user = await app.db.models.User.byId(usernameUserHash)
+                if (!user) return false
+                const membership = await app.db.models.TeamMember.getTeamMembership(user.id, team.id, false)
+                if (!membership) return false
+                const project = await app.db.models.Project.byId(projectId)
+                if (!project || project.TeamId !== team.id) return false
+                const applicationId = app.db.models.Application.encodeHashid(project.ApplicationId)
+                return app.hasPermission(membership, 'project:read', { applicationId })
+            } catch (error) {
+                app.log.error('Unexpected error during instance-state ACL check', { requestParts, usernameParts, error })
+                return false
+            }
+        },
+        checkUserCanReadDevice: async function (requestParts, usernameParts) {
+            // requestParts = [ fullTopic, <teamHash>, <deviceId> ]
+            // usernameParts = [ 'team-frontend', <userHash>, <teamHash>, <sessionId> ]
+            const topicTeamHash = requestParts[1]
+            const deviceId = requestParts[2]
+            const usernameUserHash = usernameParts[1]
+            const usernameTeamHash = usernameParts[2]
+            if (topicTeamHash !== usernameTeamHash) {
+                return false
+            }
+            try {
+                const team = await app.db.models.Team.byId(usernameTeamHash)
+                if (!team) return false
+                const user = await app.db.models.User.byId(usernameUserHash)
+                if (!user) return false
+                const membership = await app.db.models.TeamMember.getTeamMembership(user.id, team.id, false)
+                if (!membership) return false
+                const device = await app.db.models.Device.byId(deviceId)
+                if (!device || device.TeamId !== team.id) return false
+                const applicationId = device.ApplicationId ? app.db.models.Application.encodeHashid(device.ApplicationId) : null
+                return app.hasPermission(membership, 'device:read', { applicationId })
+            } catch (error) {
+                app.log.error('Unexpected error during device-state ACL check', { requestParts, usernameParts, error })
+                return false
+            }
+        },
         checkExpertTopic: async function (topicParts, usernameParts, acl) {
             // topicParts = [ fullTopic , <userid>, <sessionid>, <entityType>, <entityId> [, <inflightType>] ]
             // usernameParts = [ 'expert-client' | 'expert-agent', <userid> [, <sessionid>] ]
@@ -320,6 +372,10 @@ module.exports = function (app) {
                 { topic: /^ff\/v1\/[^/]+\/team\/updated$/ },
                 // - ff/v1/<team>/u/<user>/membership
                 { topic: /^ff\/v1\/[^/]+\/u\/[^/]+\/membership$/ },
+                // - ff/v1/<team>/p/<project>/state
+                { topic: /^ff\/v1\/[^/]+\/p\/[^/]+\/state$/ },
+                // - ff/v1/<team>/d/<device>/state
+                { topic: /^ff\/v1\/[^/]+\/d\/[^/]+\/state$/ },
                 // ff/v1/platform/sync
                 { topic: /^ff\/v1\/platform\/sync$/ },
                 // ff/v1/platform/leader
@@ -383,7 +439,11 @@ module.exports = function (app) {
                 // - ff/v1/<team>/team/updated
                 { topic: /^ff\/v1\/([^/]+)\/team\/updated$/, verify: 'checkUserIsTeamMember' },
                 // - ff/v1/<team>/u/<user>/membership
-                { topic: /^ff\/v1\/([^/]+)\/u\/([^/]+)\/membership$/, verify: 'checkUserIsTeamMember' }
+                { topic: /^ff\/v1\/([^/]+)\/u\/([^/]+)\/membership$/, verify: 'checkUserIsTeamMember' },
+                // - ff/v1/<team>/p/<project>/state
+                { topic: /^ff\/v1\/([^/]+)\/p\/([^/]+)\/state$/, verify: 'checkUserCanReadInstance' },
+                // - ff/v1/<team>/d/<device>/state
+                { topic: /^ff\/v1\/([^/]+)\/d\/([^/]+)\/state$/, verify: 'checkUserCanReadDevice' }
             ],
             pub: []
         },

forge/comms/commsClient.js

@@ -47,17 +47,20 @@ class CommsClient extends EventEmitter {
             })
             this.client.on('message', (topic, message) => {
                 const topicParts = topic.split('/')
+                const teamId = topicParts[2]
                 const ownerType = topicParts[3]
                 const ownerId = topicParts[4]
                 const messageType = topicParts[5]
-                if (ownerType === 'p') {
+                if (ownerType === 'l' && messageType === 'status') {
                     this.emit('status/project', {
+                        teamId,
                         id: ownerId,
                         status: message.toString()
                     })
                 } else if (ownerType === 'd') {
                     if (messageType === 'status') {
                         this.emit('status/device', {
+                            teamId,
                             id: ownerId,
                             status: message.toString()
                         })

forge/comms/devices.js

@@ -123,6 +123,7 @@ class DeviceCommsHandler {
             try {
                 const payload = JSON.parse(status.status)
                 await this.app.db.controllers.Device.updateState(device, payload)
+                this.app.comms?.team?.publishDeviceState(teamId, deviceId, payload)
 
                 if (payload === null) {
                     // This device is busy updating - don't interrupt it

forge/comms/index.js

@@ -31,11 +31,27 @@ module.exports = fp(async function (app, _opts) {
         // Create the handler for any device-related messages
         const deviceCommsHandler = DeviceCommsHandler(app, client)
 
-        // Not in the current release, but when we handle Launcher status
-        // via MQTT, it will arrive here. Compare to the status/device handler in `devices.js`
-        // client.on('status/project', (status) => {
-        //     // console.info(status)
-        // })
+        function publishInstanceState (teamHash, instanceId, meta) {
+            if (!teamHash || !instanceId) return
+            const msg = { id: instanceId, meta: meta || null, ts: Date.now() }
+            client.publish(`ff/v1/${teamHash}/p/${instanceId}/state`, JSON.stringify(msg))
+        }
+        function publishDeviceState (teamHash, deviceId, meta) {
+            if (!teamHash || !deviceId) return
+            const msg = { id: deviceId, meta: meta || null, ts: Date.now() }
+            client.publish(`ff/v1/${teamHash}/d/${deviceId}/state`, JSON.stringify(msg))
+        }
+
+        client.on('status/project', (status) => {
+            try {
+                const meta = status.status ? JSON.parse(status.status) : null
+                publishInstanceState(status.teamId, status.id, meta)
+            } catch (err) {
+                if (!(err instanceof SyntaxError)) {
+                    app.log.error({ msg: 'Failed to relay instance state', project: status.id, team: status.teamId, err: err.message })
+                }
+            }
+        })
 
         // Setup the platform API for the comms component
         app.decorate('comms', {
@@ -71,7 +87,9 @@ module.exports = fp(async function (app, _opts) {
                     if (!teamHash || !userHash) return
                     const msg = { reason: reason || null, srcId: srcId || null }
                     client.publish(`ff/v1/${teamHash}/u/${userHash}/membership`, JSON.stringify(msg))
-                }
+                },
+                publishInstanceState,
+                publishDeviceState
             }
         })
 

frontend/src/components/DevicesBrowser.vue

@@ -352,6 +352,7 @@ import { markRaw } from 'vue'
 import deviceApi from '../api/devices.js'
 import teamApi from '../api/team.js'
 import DropdownMenu from '../components/DropdownMenu.vue'
+import { useMqttAvailability, useMqttResourceList } from '../composables/MqttTeamChannel.js'
 import usePermissions from '../composables/Permissions.js'
 import { getTeamProperty } from '../composables/TeamProperties.js'
 import deviceActionsMixin from '../mixins/DeviceActions.js'
@@ -429,8 +430,16 @@ export default {
     emits: ['instance-updated'],
     setup () {
         const { hasPermission } = usePermissions()
+        const { mqttAvailable, resolveMqttAvailability } = useMqttAvailability()
+        const { syncMqttSubscriptions, teardownMqttSubscriptions } = useMqttResourceList('device')
 
-        return { hasPermission }
+        return {
+            hasPermission,
+            mqttAvailable,
+            resolveMqttAvailability,
+            syncMqttSubscriptions,
+            teardownMqttSubscriptions
+        }
     },
     data () {
         return {
@@ -607,12 +616,18 @@ export default {
             }
         }
     },
-    mounted () {
+    async mounted () {
+        await this.resolveMqttAvailability()
         this.fullReloadOfData()
-        this.pollTimer = createPollTimer(this.pollTimerElapsed, POLL_TIME) // auto starts
+        if (!this.mqttAvailable) {
+            this.pollTimer = createPollTimer(this.pollTimerElapsed, POLL_TIME) // auto starts
+        }
     },
     async unmounted () {
-        this.pollTimer.stop()
+        if (this.pollTimer) {
+            this.pollTimer.stop()
+        }
+        this.teardownMqttSubscriptions()
         if (this.deviceCountDeltaSincePageLoad !== 0) {
             // Trigger a refresh of team info to resync following device
             // changes
@@ -854,6 +869,11 @@ export default {
         fullReloadOfData () {
             this.checkedDevices = []
             this.loadDevices(true)
+                .then(() => {
+                    this.resyncMqtt()
+                    return undefined
+                })
+                .catch(() => undefined)
             this.pollForDeviceStatuses(true)
         },
 
@@ -865,6 +885,7 @@ export default {
 
         async loadMoreDevices () {
             await this.fetchDevices()
+            this.resyncMqtt()
         },
 
         async pollForDeviceStatuses (reset) {
@@ -873,6 +894,22 @@ export default {
             }
         },
 
+        resyncMqtt () {
+            this.syncMqttSubscriptions(this.devices.keys(), this.mqttAvailable, this.onMqttDeviceState)
+        },
+        onMqttDeviceState (id, payload) {
+            const meta = (payload && payload.meta) || {}
+            const existing = this.allDeviceStatuses.get(id) || { id }
+            this.allDeviceStatuses.set(id, {
+                ...existing,
+                id,
+                status: meta.state || existing.status,
+                mode: meta.mode || existing.mode,
+                lastSeenAt: new Date().toISOString(),
+                lastSeenMs: 0
+            })
+        },
+
         async fetchDevices (resetPage = false) {
             if (resetPage) {
                 this.nextCursor = null

frontend/src/composables/MqttTeamChannel.js

@@ -0,0 +1,91 @@
+import { inject, ref } from 'vue'
+
+export function useMqttAvailability () {
+    const services = inject('$services', null)
+    const mqttAvailable = ref(false)
+
+    async function resolveMqttAvailability () {
+        const teamChannel = services?.teamChannel
+        if (!teamChannel) {
+            mqttAvailable.value = false
+            return false
+        }
+        try {
+            mqttAvailable.value = await teamChannel.ready()
+        } catch {
+            mqttAvailable.value = false
+        }
+        return mqttAvailable.value
+    }
+
+    return { mqttAvailable, resolveMqttAvailability }
+}
+
+function subscribeFor (teamChannel, kind, id, onPayload) {
+    return kind === 'instance'
+        ? teamChannel.subscribeInstance(id, onPayload)
+        : teamChannel.subscribeDevice(id, onPayload)
+}
+
+export function useMqttResourceSubscription (kind) {
+    const services = inject('$services', null)
+    let unsubscriber = null
+
+    function setupMqttSubscription (id, onPayload) {
+        teardownMqttSubscription()
+        if (!id || typeof onPayload !== 'function') return
+        const teamChannel = services?.teamChannel
+        if (!teamChannel) return
+        unsubscriber = subscribeFor(teamChannel, kind, id, onPayload)
+    }
+
+    function teardownMqttSubscription () {
+        if (!unsubscriber) return
+        try { unsubscriber() } catch {}
+        unsubscriber = null
+    }
+
+    return { setupMqttSubscription, teardownMqttSubscription }
+}
+
+export function useMqttResourceList (kind) {
+    const services = inject('$services', null)
+    const unsubscribers = new Map()
+
+    function syncMqttSubscriptions (visibleIds, available, onPayload) {
+        if (!available) {
+            teardownMqttSubscriptions()
+            return
+        }
+        const teamChannel = services?.teamChannel
+        if (!teamChannel) return
+        const visible = new Set(visibleIds)
+        for (const [id, unsub] of unsubscribers) {
+            if (!visible.has(id)) {
+                try { unsub() } catch {}
+                unsubscribers.delete(id)
+            }
+        }
+        for (const id of visible) {
+            if (!unsubscribers.has(id)) {
+                unsubscribers.set(id, subscribeFor(teamChannel, kind, id, (payload) => onPayload(id, payload)))
+            }
+        }
+    }
+
+    function dropMqttSubscription (id) {
+        const unsub = unsubscribers.get(id)
+        if (!unsub) return
+        try { unsub() } catch {}
+        unsubscribers.delete(id)
+    }
+
+    function teardownMqttSubscriptions () {
+        for (const unsub of unsubscribers.values()) {
+            try { unsub() } catch {}
+        }
+        unsubscribers.clear()
+    }
+
+    return { syncMqttSubscriptions, dropMqttSubscription, teardownMqttSubscriptions }
+}