Add tab presence support to track user activity across sessions

forge/comms/aclManager.js

@@ -171,6 +171,25 @@ module.exports = function (app) {
                 return false
             }
         },
+        checkPresenceTopic: async function (requestParts, usernameParts) {
+            // requestParts = [ fullTopic, <userId>, <sessionId>, <messageType> ]
+            // usernameParts = [ 'fe-team', <userHash>, <teamHash>, <sessionId> ]
+            const topicUserId = requestParts[1]
+            const usernameUserHash = usernameParts[1]
+            if (topicUserId !== usernameUserHash) {
+                return false
+            }
+            try {
+                const user = await app.db.models.User.byId(usernameUserHash)
+                if (!user || user.suspended) {
+                    return false
+                }
+                return true
+            } catch (error) {
+                app.log.error('Unexpected error during presence topic ACL check', { requestParts, usernameParts, error })
+                return false
+            }
+        },
         checkExpertPlatformTopic: async function (topicParts, usernameParts, acl) {
             // topicParts = [ fullTopic , <userid>, <sessionid>, <command> ]
             // usernameParts = [ 'forge_platform' | 'expert-agent', <userid> [, <sessionid>] ]
@@ -467,7 +486,13 @@ module.exports = function (app) {
                 // ff/v1/platform/leader
                 { topic: /^ff\/v1\/platform\/leader$/ },
                 // platform can listen for Expert Agent requests
-                { topic: /^ff\/v1\/expert\/([^/]+)\/([^/]+)\/platform\/([^/]+)\/request$/, verify: 'checkExpertPlatformTopic', allowWildcard: { user: true, session: true, command: true }, isPlatform: true, isSub: true, agent: 'platform' }
+                { topic: /^ff\/v1\/expert\/([^/]+)\/([^/]+)\/platform\/([^/]+)\/request$/, verify: 'checkExpertPlatformTopic', allowWildcard: { user: true, session: true, command: true }, isPlatform: true, isSub: true, agent: 'platform' },
+                // platform can listen for browser tab presence (shared subscription)
+                // - ff/v1/browser/tab-presence/<userId>/<sessionId>/<heartbeat|context>
+                // Uses [^/]+ for the message-type segment because the subscription wildcard (+)
+                // is matched as a literal character. The publish-side ACL on teamFrontend
+                // already restricts to heartbeat|context.
+                { topic: /^ff\/v1\/browser\/tab-presence\/[^/]+\/[^/]+\/[^/]+$/, shared: true }
             ],
             pub: [
                 // Send commands to project launchers
@@ -566,7 +591,10 @@ module.exports = function (app) {
                 // - ff/v1/<team>/a/+/created|updated|deleted
                 { topic: /^ff\/v1\/([^/]+)\/a\/([^/]+)\/(created|updated|deleted)$/, verify: 'checkTeamStateSub' }
             ],
-            pub: []
+            pub: [
+                // ff/v1/browser/tab-presence/<userId>/<sessionId>/<heartbeat|context>
+                { topic: /^ff\/v1\/browser\/tab-presence\/([^/]+)\/([^/]+)\/(heartbeat|context)$/, verify: 'checkPresenceTopic' }
+            ]
         },
         // frontend client (user)
         expertClient: {
@@ -647,7 +675,7 @@ module.exports = function (app) {
                         isSharedSub = true
                         // This is a shared sub - validate the share group name
                         const shareGroup = sharedSubParts[1]
-                        if (shareGroup !== 'platform' && shareGroup !== usernameParts[2]) {
+                        if (shareGroup !== 'platform' && shareGroup !== 'browser' && shareGroup !== usernameParts[2]) {
                             return false
                         }
                         topic = sharedSubParts[2]

forge/comms/browserSessionPresence.js

@@ -0,0 +1,56 @@
+const CACHE_NAME = 'browserSessions'
+const CACHE_TTL = 135_000 // ~3x the 45s heartbeat interval
+const CACHE_MAX = 10_000
+
+class BrowserSessionPresenceHandler {
+    constructor (app, client) {
+        this.app = app
+        this.client = client
+        this.cache = app.caches.createCache(CACHE_NAME, { max: CACHE_MAX, ttl: CACHE_TTL })
+        this.setupEventHandlers()
+    }
+
+    setupEventHandlers () {
+        this.client.on('tab-presence', (msg) => this.handlePresence(msg))
+    }
+
+    async handlePresence ({ userId, sessionId, messageType, payload }) {
+        const cacheKey = `${userId}:${sessionId}`
+
+        if (messageType === 'heartbeat') {
+            const existing = await this.cache.get(cacheKey) || {}
+            await this.cache.set(cacheKey, {
+                ...existing,
+                userId,
+                sessionId,
+                lastSeen: Date.now(),
+                visibility: payload.visibility || 'visible'
+            })
+        } else if (messageType === 'context') {
+            const existing = await this.cache.get(cacheKey) || {}
+            await this.cache.set(cacheKey, {
+                ...existing,
+                userId,
+                sessionId,
+                lastSeen: Date.now(),
+                context: payload
+            })
+        }
+    }
+
+    async getSessionsByUser (userId) {
+        const allEntries = await this.cache.all()
+        const prefix = `${userId}:`
+        const sessions = []
+        for (const [key, value] of Object.entries(allEntries)) {
+            if (key.startsWith(prefix)) {
+                sessions.push(value)
+            }
+        }
+        return sessions
+    }
+}
+
+module.exports = {
+    BrowserSessionPresenceHandler: (app, client) => new BrowserSessionPresenceHandler(app, client)
+}

forge/comms/commsClient.js

@@ -52,6 +52,22 @@ class CommsClient extends EventEmitter {
                 const ownerId = topicParts[4]
                 const messageType = topicParts[5]
 
+                if (topicParts[2] === 'browser') {
+                    // ff/v1/browser/tab-presence/<userId>/<sessionId>/<messageType>
+                    const userId = topicParts[4]
+                    const sessionId = topicParts[5]
+                    const messageType = topicParts[6]
+                    let payload
+                    try {
+                        payload = JSON.parse(message.toString())
+                    } catch (err) {
+                        this.app.log.warn(`Ignoring malformed browser payload on ${topic}: ${err.message}`)
+                        return
+                    }
+                    this.emit('tab-presence', { userId, sessionId, messageType, payload })
+                    return
+                }
+
                 if (topicParts[2] === 'expert') {
                     const userId = topicParts[3]
                     const sessionId = topicParts[4]
@@ -249,7 +265,9 @@ class CommsClient extends EventEmitter {
                 // of consumers that share the workload, so keeping Expert separate from the
                 // "platform" group prevents unrelated features from sharing a consumer pool and
                 // allows them to scale independently.
-                '$share/expert/ff/v1/expert/+/+/platform/+/request'
+                '$share/expert/ff/v1/expert/+/+/platform/+/request',
+                // Browser tab presence - shared subscription
+                '$share/browser/ff/v1/browser/tab-presence/+/+/+'
             ])
         }
     }

forge/comms/index.js

@@ -1,6 +1,7 @@
 const fp = require('fastify-plugin')
 
 const ACLManager = require('./aclManager')
+const { BrowserSessionPresenceHandler } = require('./browserSessionPresence')
 const { CommsClient } = require('./commsClient')
 const { DeviceCommsHandler } = require('./devices')
 const { ExpertCommsHandler } = require('./expert')
@@ -36,6 +37,7 @@ module.exports = fp(async function (app, _opts) {
         const instanceCommsHandler = InstanceCommsHandler(app, client)
         const platformAutomationHandler = PlatformAutomationHandler(app, client)
         const expertCommsHandler = new ExpertCommsHandler(app, client)
+        const browserSessionPresenceHandler = BrowserSessionPresenceHandler(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`
@@ -50,6 +52,7 @@ module.exports = fp(async function (app, _opts) {
             aclManager: ACLManager(app),
             platformAutomation: platformAutomationHandler,
             expert: expertCommsHandler,
+            browserSessions: browserSessionPresenceHandler,
             platform: {
                 settings: {
                     sync: function (key) {

test/unit/forge/comms/authRoutesV2_spec.js

@@ -1575,6 +1575,38 @@ describe('Broker Auth v2 API', async function () {
                     teamLookupStub.restore()
                 }
             })
+
+            // Browser session presence topics
+            it('allows fe-team to publish heartbeat to own presence topic', async function () {
+                await allowWrite({
+                    username: teamFrontendUsername,
+                    topic: `ff/v1/browser/tab-presence/${TestObjects.alice.hashid}/session-abc12345/heartbeat`
+                })
+            })
+            it('allows fe-team to publish context to own presence topic', async function () {
+                await allowWrite({
+                    username: teamFrontendUsername,
+                    topic: `ff/v1/browser/tab-presence/${TestObjects.alice.hashid}/session-abc12345/context`
+                })
+            })
+            it('denies fe-team from publishing to another user\'s presence topic', async function () {
+                await denyWrite({
+                    username: teamFrontendUsername,
+                    topic: `ff/v1/browser/tab-presence/${bob.hashid}/session-abc12345/heartbeat`
+                })
+            })
+            it('denies fe-team from publishing to an invalid presence message type', async function () {
+                await denyWrite({
+                    username: teamFrontendUsername,
+                    topic: `ff/v1/browser/tab-presence/${TestObjects.alice.hashid}/session-abc12345/invalid`
+                })
+            })
+            it('allows forge_platform to subscribe to presence topics via shared subscription', async function () {
+                await allowRead({
+                    username: 'forge_platform',
+                    topic: `$share/browser/ff/v1/browser/tab-presence/${TestObjects.alice.hashid}/session-abc12345/heartbeat`
+                })
+            })
         })
     })
 })