Shared Cache implementation

forge/caches/index.js

@@ -0,0 +1,35 @@
+/**
+ * The a pluggable object cache
+ *
+ * @namespace cache
+ * @memberof forge
+ */
+
+/**
+ * @typedef {Object} forge.Status
+ * @property {string} status
+ */
+const fp = require('fastify-plugin')
+
+const CACHE_DRIVERS = {
+    memory: './memory-cache.js',
+    redis: './redis-cache.js'
+}
+
+module.exports = fp(async function (app, _opts) {
+    const cacheType = app.config.cache?.driver || 'memory'
+    const cacheModule = CACHE_DRIVERS[cacheType]
+    try {
+        app.log.info(`Cache driver: ${cacheType}`)
+        const driver = require(cacheModule)
+        await driver.initCache(app.config.cache?.options || {})
+        app.decorate('caches', driver)
+        app.addHook('onClose', async (_) => {
+            app.log.info('Driver shutdown')
+            await driver.closeCache()
+        })
+    } catch (err) {
+        app.log.error(`Failed to load the cache driver: ${cacheType}`)
+        throw err
+    }
+}, { name: 'app.caches' })

forge/caches/memory-cache.js

@@ -0,0 +1,45 @@
+const caches = {}
+
+async function initCache () {}
+
+async function closeCache () {}
+
+function getCache (name, options) {
+    if (!caches[name]) {
+        caches[name] = new Cache(name, options)
+    }
+    return caches[name]
+}
+
+class Cache {
+    constructor (name, options) {
+        this.holder = {}
+    }
+
+    async get (key) {
+        return this.holder[key]
+    }
+
+    async set (key, value) {
+        this.holder[key] = value
+        return value
+    }
+
+    async del (key) {
+        delete this.holder[key]
+    }
+
+    async keys () {
+        return Object.keys(this.holder)
+    }
+
+    async all () {
+        return this.holder
+    }
+}
+
+module.exports = {
+    initCache,
+    getCache,
+    closeCache
+}

forge/caches/redis-cache.js

@@ -0,0 +1,66 @@
+const { createClient } = require('@redis/client')
+
+const caches = {}
+
+let client
+
+async function initCache (options) {
+    client = createClient(options)
+    await client.connect()
+}
+
+function getCache (name, options) {
+    if (!caches[name]) {
+        caches[name] = new Cache(name, { client, ...options })
+    }
+    return caches[name]
+}
+
+async function closeCache () {
+    client.close()
+}
+
+class Cache {
+    constructor (name, options) {
+        this.name = name
+        this.client = options.client
+    }
+
+    async get (key) {
+        const val = JSON.parse(await this.client.hGet(this.name, key))
+        if (val !== null) {
+            return val
+        } else {
+            return undefined
+        }
+    }
+
+    async set (key, value) {
+        await this.client.hSet(this.name, key, JSON.stringify(value))
+        return value
+    }
+
+    async del (key) {
+        await this.client.hDel(this.name, key)
+    }
+
+    async keys () {
+        const keys = await this.client.hKeys(this.name)
+        return keys
+    }
+
+    async all () {
+        const values = await this.client.hGetAll(this.name)
+        const newObj = {}
+        for (const k of Object.keys(values)) {
+            newObj[k] = JSON.parse(values[k])
+        }
+        return newObj
+    }
+}
+
+module.exports = {
+    initCache,
+    getCache,
+    closeCache
+}

forge/comms/devices.js

@@ -235,7 +235,7 @@ class DeviceCommsHandler {
             if (inFlightCommand) {
                 // This command is known to the local instance - process it
                 inFlightCommand.resolve(message.payload)
-                delete this.inFlightCommands[response.correlationData]
+                delete this.inFlightCommands[message.correlationData]
             }
         }
     }
@@ -334,7 +334,7 @@ class DeviceCommsHandler {
                 resolve(payload)
                 delete this.inFlightCommands[inFlightCommand.correlationData]
             }
-            inFlightCommand.reject = (err) => {
+            inFlightCommand.reject = async (err) => {
                 inFlightCommand.rejected = true
                 clearTimeout(inFlightCommand.timer)
                 reject(err)

forge/db/controllers/Project.js

@@ -8,11 +8,11 @@ const { KEY_SETTINGS } = require('../models/ProjectSettings')
  * is no need to store that in the database. But we do need to know it so the
  * information can be returned on the API.
  */
-const inflightProjectState = { }
+const inflightProjectState = 'project-inflightProjectState'
 
-const latestProjectState = { }
+const latestProjectState = 'project-latestProjectState'
 
-const inflightDeploys = new Set()
+const inflightDeploys = 'project-inflightDeploys'
 
 module.exports = {
     /**
@@ -21,8 +21,8 @@ module.exports = {
      * @param {*} project
      * @returns the in-flight state
      */
-    getInflightState: function (app, project) {
-        return inflightProjectState[project.id]
+    getInflightState: async function (app, project) {
+        return await app.caches.getCache(inflightProjectState).get(project.id)
     },
 
     /**
@@ -31,36 +31,37 @@ module.exports = {
      * @param {*} project
      * @param {*} state
      */
-    setInflightState: function (app, project, state) {
-        inflightProjectState[project.id] = state
+    setInflightState: async function (app, project, state) {
+        await app.caches.getCache(inflightProjectState).set(project.id, state)
     },
 
     /**
      * Check whether an instance is currently flagged as deploying
      * @param {*} app
      * @param {*} instance
      */
-    isDeploying: function (app, instance) {
-        return inflightDeploys.has(instance.id)
+    isDeploying: async function (app, instance) {
+        const has = await app.caches.getCache(inflightDeploys).get(instance.id)
+        return has === true
     },
 
     /**
      * Mark an instance as currently being deployed
      * @param {*} app
      * @param {*} instance
      */
-    setInDeploy: function (app, instance) {
-        inflightDeploys.add(instance.id)
+    setInDeploy: async function (app, instance) {
+        await app.caches.getCache(inflightDeploys).set(instance.id, true)
     },
 
     /**
      * Set the in-flight state of a project
      * @param {*} app
      * @param {*} project
      */
-    clearInflightState: function (app, project) {
-        delete inflightProjectState[project.id]
-        inflightDeploys.delete(project.id)
+    clearInflightState: async function (app, project) {
+        await app.caches.getCache(inflightProjectState).del(project.id)
+        await app.caches.getCache(inflightDeploys).del(project.id)
     },
 
     /**
@@ -570,7 +571,7 @@ module.exports = {
             throw error
         }
         if (project.state === 'running') {
-            app.db.controllers.Project.setInflightState(project, 'restarting')
+            await app.db.controllers.Project.setInflightState(project, 'restarting')
             project.state = 'running'
             await project.save()
             const result = await app.containers.restartFlows(project)
@@ -699,8 +700,9 @@ module.exports = {
      * @param {string|Number} projectId - The unique identifier of the project whose latest state is to be retrieved.
      * @returns {string} The latest state of the specified project.
      */
-    getLatestProjectState: function (app, projectId) {
-        return latestProjectState[projectId]
+    getLatestProjectState: async function (app, projectId) {
+        return await app.caches.getCache(latestProjectState).get(projectId)
+        // return latestProjectState[projectId]
     },
 
     /**
@@ -710,8 +712,8 @@ module.exports = {
      * @param {string|number} projectId - The unique identifier of the project whose state is being updated.
      * @param {any} state - The new state to be assigned to the project.
      */
-    setLatestProjectState: function (app, projectId, state) {
-        latestProjectState[projectId] = state
+    setLatestProjectState: async function (app, projectId, state) {
+        app.caches.getCache(latestProjectState).set(projectId, state)
     },
 
     /**
@@ -723,8 +725,8 @@ module.exports = {
      *
      * @param {String|Number} projectId - The project id whose latest state should be cleared.
      */
-    clearLatestProjectState: function (app, projectId) {
-        delete latestProjectState[projectId]
+    clearLatestProjectState: async function (app, projectId) {
+        return await app.caches.getCache(latestProjectState).del(projectId)
     },
 
     /**
@@ -733,11 +735,11 @@ module.exports = {
      * @param {String|Number} projectId - The project id related to the driver state update.
      * @param {string} state - The new state to update, can be 'running', 'stopped', or other valid states.
      */
-    updateLatestProjectState: function (app, projectId, state) {
+    updateLatestProjectState: async function (app, projectId, state) {
         if (['running'].includes(state)) {
-            this.clearLatestProjectState(app, projectId)
+            await this.clearLatestProjectState(app, projectId)
         } else {
-            this.setLatestProjectState(app, projectId, state)
+            await this.setLatestProjectState(app, projectId, state)
         }
     }
 }

forge/db/models/Project.js

@@ -328,8 +328,8 @@ module.exports = {
                 async liveState ({ omitStorageFlows = false } = { }) {
                     const result = {}
 
-                    const inflightState = Controllers.Project.getInflightState(this)
-                    const isDeploying = Controllers.Project.isDeploying(this)
+                    const inflightState = await Controllers.Project.getInflightState(this)
+                    const isDeploying = await Controllers.Project.isDeploying(this)
 
                     if (!omitStorageFlows) {
                         let storageFlow = this.StorageFlow
@@ -353,7 +353,7 @@ module.exports = {
                         result.meta = await app.containers.details(this) || { state: 'unknown' }
 
                         if (result.meta.state !== this.state) {
-                            Controllers.Project.setLatestProjectState(this.id, result.meta.state)
+                            await Controllers.Project.setLatestProjectState(this.id, result.meta.state)
                         }
 
                         if (result.meta.versions) {
@@ -713,14 +713,14 @@ module.exports = {
                     const teamRbacEnabled = team.TeamType.getFeatureProperty('rbacApplication', false)
                     const rbacEnabled = platformRbacEnabled && teamRbacEnabled
 
-                    results.forEach((project) => {
+                    for (const project of results) {
                         if (rbacEnabled && !app.hasPermission(membership, 'project:read', { applicationId: project.Application.hashid })) {
                             // This instance is not accessible to this user, do not include in states map
-                            return
+                            continue
                         }
-                        const state = Controllers.Project.getLatestProjectState(project.id) ?? project.state
+                        const state = await Controllers.Project.getLatestProjectState(project.id) ?? project.state
                         statesMap[state] = (statesMap[state] || 0) + 1
-                    })
+                    }
 
                     return Object.entries(statesMap).map(([state, count]) => ({ state, count }))
                 },

forge/ee/db/controllers/Pipeline.js

@@ -536,11 +536,10 @@ module.exports = {
 
         const restartTargetInstance = targetInstance?.state === 'running'
 
-        app.db.controllers.Project.setInflightState(targetInstance, 'importing')
-        app.db.controllers.Project.setInDeploy(targetInstance)
-
         // Complete heavy work async
         return (async function () {
+            await app.db.controllers.Project.setInflightState(targetInstance, 'importing')
+            await app.db.controllers.Project.setInDeploy(targetInstance)
             try {
                 const setAsTargetForDevices = deployToDevices ?? false
                 const targetSnapshot = await copySnapshot(app, sourceSnapshot, targetInstance, {
@@ -559,9 +558,9 @@ module.exports = {
                 await app.auditLog.Project.project.imported(user.id, null, targetInstance, sourceInstance, sourceDevice) // technically this isn't a project event
                 await app.auditLog.Project.project.snapshot.imported(user.id, null, targetInstance, sourceInstance, sourceDevice, targetSnapshot)
 
-                app.db.controllers.Project.clearInflightState(targetInstance)
+                await app.db.controllers.Project.clearInflightState(targetInstance)
             } catch (err) {
-                app.db.controllers.Project.clearInflightState(targetInstance)
+                await app.db.controllers.Project.clearInflightState(targetInstance)
 
                 await app.auditLog.Project.project.imported(user.id, null, targetInstance, sourceInstance, sourceDevice) // technically this isn't a project event
                 await app.auditLog.Project.project.snapshot.imported(user.id, err, targetInstance, sourceInstance, sourceDevice, null)

forge/ee/lib/billing/trialTask.js

@@ -123,12 +123,12 @@ module.exports.init = function (app) {
                 // There is some DRY code here with projectActions.js suspend logic.
                 // TODO: consider move to controllers.Project
                 try {
-                    app.db.controllers.Project.setInflightState(project, 'suspending')
+                    await app.db.controllers.Project.setInflightState(project, 'suspending')
                     await app.containers.stop(project)
-                    app.db.controllers.Project.clearInflightState(project)
+                    await app.db.controllers.Project.clearInflightState(project)
                     await app.auditLog.Project.project.suspended(null, null, project)
                 } catch (err) {
-                    app.db.controllers.Project.clearInflightState(project)
+                    await app.db.controllers.Project.clearInflightState(project)
                     const resp = { code: 'unexpected_error', error: err.toString() }
                     await app.auditLog.Project.project.suspended(null, resp, project)
                 }