Add caching to TeamBroker topic logging

forge/ee/lib/index.js

@@ -13,7 +13,7 @@ module.exports = fp(async function (app, opts) {
         require('./protectedInstance').init(app)
         require('./customHostnames').init(app)
         app.decorate('sso', await require('./sso').init(app))
-        require('./teamBroker').init(app)
+        await require('./teamBroker').init(app)
         app.decorate('gitops', await require('./gitops').init(app))
         // Set the MFA Feature Flag
         app.config.features.register('mfa', true, true)

forge/ee/lib/teamBroker/index.js

@@ -1,4 +1,6 @@
-module.exports.init = function (app) {
+const { LRUCache } = require('lru-cache')
+
+module.exports.init = async function (app) {
     // enable Team Broker Feature
     if (app.config.broker?.teamBroker?.enabled) {
         app.config.features.register('teamBroker', true, true)
@@ -39,15 +41,38 @@ module.exports.init = function (app) {
      */
     async function addUsedTopic (topic, team) {
         const teamId = app.db.models.Team.decodeHashid(team)
-        await app.db.models.MQTTTopicSchema.upsert({
-            topic,
-            TeamId: teamId,
-            BrokerCredentialsId: app.settings.get('team:broker:creds') ?? null
-        }, {
-            topic,
-            TeamId: teamId,
-            BrokerCredentialsId: app.settings.get('team:broker:creds') ?? null
+        const cacheHit = topicCache.get(`${teamId}#${topic}`)
+        if (!cacheHit) {
+            await app.db.models.MQTTTopicSchema.upsert({
+                topic,
+                TeamId: teamId,
+                BrokerCredentialsId: app.settings.get('team:broker:creds') ?? null
+            }, {
+                topic,
+                TeamId: teamId,
+                BrokerCredentialsId: app.settings.get('team:broker:creds') ?? null
+            })
+            topicCache.set(`${teamId}#${topic}`, true)
+        }
+    }
+
+    const topicCache = new LRUCache({
+        max: 5000,
+        ttl: 1000 * 60 * 30 // 30 min grace period
+    })
+    try {
+        const topics = await app.db.models.MQTTTopicSchema.findAll({
+            where: {
+                BrokerCredentialsId: app.settings.get('team:broker:creds')
+            },
+            order: [['updatedAt', 'DESC']],
+            limit: 5000
         })
+        for (const topic of topics) {
+            topicCache.set(`${topic.teamId}#${topic.topic}`, true)
+        }
+    } catch (err) {
+        app.log.debug(`Error populating Team Broker Topic Cache ${err.toString()}`)
     }
 
     app.decorate('teamBroker', {

test/unit/forge/comms/authRoutesV2_spec.js

@@ -599,6 +599,26 @@ describe('Broker Auth v2 API', async function () {
                         topic: 'ff/v1/abc/p/another-project/res-random/foo/bar'
                     })
                 })
+                describe('Team Broker Topic Update Cache', async function () {
+                    it('should not update if multiple calls to the same topic', async function () {
+                        const topic = 'update/topic/timestamp'
+                        await app.teamBroker.addUsedTopic(topic, TestObjects.ATeam.hashid)
+                        const firstTopic = await app.db.models.MQTTTopicSchema.findAll({
+                            where: {
+                                topic,
+                                TeamId: TestObjects.ATeam.id
+                            }
+                        })
+                        await app.teamBroker.addUsedTopic(topic, TestObjects.ATeam.hashid)
+                        const secondTopic = await app.db.models.MQTTTopicSchema.findAll({
+                            where: {
+                                topic,
+                                TeamId: TestObjects.ATeam.id
+                            }
+                        })
+                        secondTopic[0].updatedAt.toISOString().should.equal(firstTopic[0].updatedAt.toISOString())
+                    })
+                })
             })
         })