[7320] Live team updates over MQTT/WS

forge/comms/aclManager.js

@@ -115,6 +115,31 @@ module.exports = function (app) {
                 return false
             }
         },
+        checkUserIsTeamMember: async function (requestParts, usernameParts) {
+            // requestParts = [ fullTopic , <teamHash> [, <userHash>] ]
+            // usernameParts = [ 'team-frontend', <userHash>, <teamHash>, <sessionId> ]
+            const topicTeamHash = requestParts[1]
+            const usernameUserHash = usernameParts[1]
+            const usernameTeamHash = usernameParts[2]
+            if (topicTeamHash !== usernameTeamHash) {
+                return false
+            }
+            // membership topic: user capture must match the credential's user
+            if (requestParts[2] !== undefined && requestParts[2] !== usernameUserHash) {
+                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)
+                return !!membership
+            } catch (error) {
+                app.log.error('Unexpected error during team-channel 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>] ]
@@ -290,6 +315,11 @@ module.exports = function (app) {
                 // Send commands to all application-assigned devices
                 // - ff/v1/+/a/+/command
                 { topic: /^ff\/v1\/[^/]+\/a\/[^/]+\/command$/ },
+                // Team channel broadcasts to subscribed team members
+                // - ff/v1/<team>/team/updated
+                { topic: /^ff\/v1\/[^/]+\/team\/updated$/ },
+                // - ff/v1/<team>/u/<user>/membership
+                { topic: /^ff\/v1\/[^/]+\/u\/[^/]+\/membership$/ },
                 // ff/v1/platform/sync
                 { topic: /^ff\/v1\/platform\/sync$/ },
                 // ff/v1/platform/leader
@@ -347,6 +377,16 @@ module.exports = function (app) {
                 { topic: /^ff\/v1\/([^/]+)\/d\/([^/]+)\/resources\/heartbeat$/, verify: 'checkDeviceIsAssigned' }
             ]
         },
+        // browser-side team channel (per-tab, per-team)
+        teamFrontend: {
+            sub: [
+                // - 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' }
+            ],
+            pub: []
+        },
         // frontend client (user)
         expertClient: {
             sub: [
@@ -384,6 +424,7 @@ module.exports = function (app) {
             // - project:<teamid>:<projectid>
             // - device:<teamid>:<deviceid>
             // - frontend:<teamid>:<deviceid>
+            // - team-frontend:<userid>:<teamid>:<sessionid>
             // - expert-client:<userid>:<sessionid>
             // - expert-agent:<userid>:<apiversion>
 
@@ -399,6 +440,8 @@ module.exports = function (app) {
                 aclList = ACLS.project[aclType]
             } else if (/^device:/.test(username)) {
                 aclList = ACLS.device[aclType]
+            } else if (/^team-frontend:/.test(username)) {
+                aclList = ACLS.teamFrontend[aclType]
             } else if (/^frontend:/.test(username)) {
                 aclList = ACLS.frontend[aclType]
             } else if (/^expert-agent:/.test(username)) {

forge/comms/index.js

@@ -60,6 +60,18 @@ module.exports = fp(async function (app, _opts) {
                         client.publish('ff/v1/platform/leader', JSON.stringify(msg))
                     }
                 }
+            },
+            team: {
+                notify: function (teamHash, reason, srcId) {
+                    if (!teamHash) return
+                    const msg = { reason: reason || null, srcId: srcId || null }
+                    client.publish(`ff/v1/${teamHash}/team/updated`, JSON.stringify(msg))
+                },
+                notifyMembership: function (teamHash, userHash, reason, srcId) {
+                    if (!teamHash || !userHash) return
+                    const msg = { reason: reason || null, srcId: srcId || null }
+                    client.publish(`ff/v1/${teamHash}/u/${userHash}/membership`, JSON.stringify(msg))
+                }
             }
         })
 

forge/comms/v2AuthRoutes.js

@@ -34,6 +34,7 @@ module.exports = async function (app) {
         if ((username.startsWith('device:') && password.startsWith('ffbd_')) ||
             (username.startsWith('project:') && password.startsWith('ffbp_')) ||
             (username.startsWith('frontend:') && password.startsWith('ffbf_')) ||
+            (username.startsWith('team-frontend:') && password.startsWith('ffbtf_')) ||
             (username.startsWith('expert-agent:') && password.startsWith('ffbea_')) ||
             (username.startsWith('expert-client:') && password.startsWith('ffbec_')) ||
             (username === 'forge_platform')) {
@@ -138,6 +139,7 @@ module.exports = async function (app) {
         if ((username.startsWith('device:') ||
             username.startsWith('project:') ||
             username.startsWith('frontend:') ||
+            username.startsWith('team-frontend:') ||
             username.startsWith('expert-agent:') ||
             username.startsWith('expert-client:') ||
             username === 'forge_platform') && !username.includes('@')) {

forge/db/controllers/BrokerClient.js

@@ -10,7 +10,7 @@ module.exports = {
             attributes: ['username', 'password']
         })
         if (compareHash(password || '', user ? user.password : '')) {
-            if (username.startsWith('frontend:') || username.startsWith('expert-client:')) {
+            if (username.startsWith('frontend:') || username.startsWith('expert-client:') || username.startsWith('team-frontend:')) {
                 await user.destroy()
             }
             return true
@@ -182,6 +182,31 @@ module.exports = {
         return null
     },
 
+    createClientForTeamFrontend: async function (app, user, team, sessionId) {
+        if (app.comms) {
+            const username = `team-frontend:${user.hashid}:${team.hashid}:${sessionId}`
+            const existingClient = await app.db.models.BrokerClient.findOne({
+                where: { username }
+            })
+            if (existingClient) {
+                await existingClient.destroy()
+            }
+            const password = generateToken(32, 'ffbtf')
+            await app.db.models.BrokerClient.create({
+                username,
+                password,
+                ownerId: '' + user.id,
+                ownerType: 'team-frontend'
+            })
+            return {
+                url: app.config.broker.public_url || app.config.broker.url || null,
+                username,
+                password
+            }
+        }
+        return null
+    },
+
     createClientForExpertClient: async function (app, user, sessionId) {
         if (app.comms) {
             const existingClient = await app.db.models.BrokerClient.findOne({

forge/ee/routes/billing/index.js

@@ -268,6 +268,9 @@ module.exports = async function (app) {
                 } else {
                     app.log.warn(`Stripe subscription ${stripeSubscriptionId} has transitioned in Stripe to a state not currently handled: '${stripeSubscriptionStatus}'`)
                 }
+                if (team) {
+                    app.comms?.team?.notify(team.hashid, 'billing-updated')
+                }
 
                 break
             }
@@ -286,6 +289,7 @@ module.exports = async function (app) {
                     response.status(200).send()
                     return
                 }
+                app.comms?.team?.notify(team.hashid, 'billing-deleted')
 
                 // Suspend all projects of that team
                 const projects = await app.db.models.Project.byTeam(team.hashid)
@@ -444,6 +448,7 @@ module.exports = async function (app) {
                     await team.save()
                 }
             }
+            app.comms?.team?.notify(team.hashid, 'billing-manual-enabled')
             response.code(200).send({})
         } catch (err) {
             // Standard errors
@@ -472,6 +477,7 @@ module.exports = async function (app) {
         const team = request.team
         try {
             await app.billing.disableManualBilling(team)
+            app.comms?.team?.notify(team.hashid, 'billing-manual-disabled')
             response.code(200).send({})
         } catch (err) {
             // Standard errors

forge/routes/api/team.js

@@ -841,6 +841,7 @@ module.exports = async function (app) {
                         }
                         teamAuditFunc(request.session.User, null, request.team)
                         platformAuditFunc(request.session.User, null, request.team)
+                        app.comms?.team?.notify(request.team.hashid, request.body.suspended ? 'suspended' : 'unsuspended')
                         reply.send(app.db.views.Team.team(request.team))
                         return
                     } catch (err) {
@@ -894,6 +895,7 @@ module.exports = async function (app) {
             // Only log if something changes
             if (updates.length > 0) {
                 auditLogFunc(request.session.User, null, request.team, updates)
+                app.comms?.team?.notify(request.team.hashid, 'updated')
             }
             reply.send(app.db.views.Team.team(request.team))
         } catch (err) {
@@ -914,6 +916,63 @@ module.exports = async function (app) {
         }
     })
 
+    /**
+     * Issue MQTT/WS credentials for the team-level browser channel.
+     * @name /api/v1/teams/:teamId/comms-credentials
+     */
+    app.post('/:teamId/comms-credentials', {
+        preHandler: app.needsPermission('team:read'),
+        schema: {
+            summary: 'Issue team-channel broker credentials for the current user/session',
+            tags: ['Teams'],
+            params: {
+                type: 'object',
+                properties: {
+                    teamId: { type: 'string' }
+                }
+            },
+            body: {
+                type: 'object',
+                properties: {
+                    sessionId: { type: 'string' }
+                }
+            },
+            response: {
+                200: {
+                    type: 'object',
+                    properties: {
+                        url: { type: 'string' },
+                        username: { type: 'string' },
+                        password: { type: 'string' }
+                    }
+                },
+                '4xx': {
+                    $ref: 'APIError'
+                }
+            }
+        }
+    }, async (request, reply) => {
+        if (!app.comms) {
+            reply.code(503).send({ code: 'comms_unavailable', error: 'Broker not configured' })
+            return
+        }
+        const sessionId = request.body?.sessionId
+        if (!sessionId || typeof sessionId !== 'string' || sessionId.length < 8) {
+            reply.code(400).send({ code: 'invalid_request', error: 'sessionId is required' })
+            return
+        }
+        const creds = await app.db.controllers.BrokerClient.createClientForTeamFrontend(
+            request.session.User,
+            request.team,
+            sessionId
+        )
+        if (!creds) {
+            reply.code(503).send({ code: 'comms_unavailable', error: 'Broker not configured' })
+            return
+        }
+        reply.send(creds)
+    })
+
     /**
      * Get the session users team membership
      * @name /api/v1/team/:teamId/user

forge/routes/api/teamMembers.js

@@ -120,6 +120,7 @@ module.exports = async function (app) {
             const result = await app.db.controllers.Team.removeUser(request.team, request.user, request.userRole)
             if (result) {
                 await app.auditLog.Team.team.user.removed(request.session.User, null, request.team, request.user)
+                app.comms?.team?.notifyMembership(request.team.hashid, request.user.hashid, 'removed')
             }
             reply.send({ status: 'okay' })
         } catch (err) {
@@ -188,6 +189,7 @@ module.exports = async function (app) {
                             // might want to make this only if it drop under Member
                             await app.db.controllers.StorageSession.removeUserFromTeamSessions(request.user, request.team)
                         }
+                        app.comms?.team?.notifyMembership(request.team.hashid, request.user.hashid, 'role-changed')
                     }
                     reply.send({ status: 'okay' })
                 } catch (err) {

frontend/src/api/team.js

@@ -320,6 +320,11 @@ const getTeamAuditLog = async (teamId, params, cursor, limit) => {
     const url = paginateUrl(`/api/v1/teams/${teamId}/audit-log`, cursor, limit)
     return client.get(url, { params }).then(res => res.data)
 }
+const getTeamCommsCreds = (teamId, sessionId) => {
+    return client.post(`/api/v1/teams/${teamId}/comms-credentials`, { sessionId })
+        .then(res => res.data)
+}
+
 const getTeamUserMembership = (teamId) => {
     return client.get(`/api/v1/teams/${teamId}/user`).then(res => res.data)
 }
@@ -564,6 +569,7 @@ export default {
     resendTeamInvitation,
     getTeamAuditLog,
     getTeamUserMembership,
+    getTeamCommsCreds,
     getTeamDevices,
     getTeamRegistry,
     generateRegistryUserToken,

frontend/src/services/service.registry.ts

@@ -1,9 +1,11 @@
 import { createBootstrapService } from './bootstrap.service'
 import { createMqttService } from './mqtt.service'
 import { createMessagingService } from './post-message.service'
+import { createTeamChannelService } from './team-channel.service'
 
 export default [
     { key: 'bootstrap' as const, create: createBootstrapService, requiredLifecycle: ['init', 'destroy'] as const },
     { key: 'postMessage' as const, create: createMessagingService, requiredLifecycle: ['destroy'] as const },
-    { key: 'mqtt' as const, create: createMqttService, requiredLifecycle: ['destroy'] as const }
+    { key: 'mqtt' as const, create: createMqttService, requiredLifecycle: ['destroy'] as const },
+    { key: 'teamChannel' as const, create: createTeamChannelService, requiredLifecycle: ['destroy'] as const }
 ]