[web] add optional pub/sub based sync mechanism for system messages (#31328)

* [web] add optional pub/sub based sync mechanism for system messages

* [web] refactor lazy creation of redis connection

GitOrigin-RevId: 5f69428b8e8dac59aad73e8f16cd25625f2daae4

services/web/app/src/Features/SystemMessages/SystemMessageManager.mjs

@@ -2,6 +2,11 @@ import { SystemMessage } from '../../models/SystemMessage.mjs'
 import { addRequiredCleanupHandlerBeforeDrainingConnections } from '../../infrastructure/GracefulShutdown.mjs'
 import { callbackifyAll } from '@overleaf/promise-utils'
 import logger from '@overleaf/logger'
+import RedisWrapper from '../../infrastructure/RedisWrapper.mjs'
+import Settings from '@overleaf/settings'
+import redis from '@overleaf/redis-wrapper'
+
+const PUB_SUB_CHANNEL_TRIGGER_REFRESH = 'refresh-system-messages'
 
 const SystemMessageManager = {
   _cachedMessages: [],
@@ -16,15 +21,32 @@ const SystemMessageManager = {
 
   async clearMessages() {
     await SystemMessage.deleteMany({}).exec()
+    await SystemMessageManager.notifyOtherPods()
     await this.refreshCache()
   },
 
   async createMessage(content) {
     const message = new SystemMessage({ content })
     await message.save()
+    await SystemMessageManager.notifyOtherPods()
     await this.refreshCache()
   },
 
+  async notifyOtherPods() {
+    if (!Settings.notifyOnSystemMessageChanges) return
+    const redisFeatureSettings = Settings.redis.pubsub || Settings.redis.web
+    const rclient = redis.createClient(redisFeatureSettings)
+    try {
+      await rclient.publish(PUB_SUB_CHANNEL_TRIGGER_REFRESH, '')
+    } finally {
+      try {
+        await rclient.disconnect()
+      } catch (err) {
+        logger.err({ err }, 'failed to disconnect pub/sub redis client')
+      }
+    }
+  },
+
   async refreshCache() {
     this._cachedMessages = await this.getMessagesFromDB()
   },
@@ -50,6 +72,23 @@ addRequiredCleanupHandlerBeforeDrainingConnections(
   }
 )
 
+if (Settings.notifyOnSystemMessageChanges) {
+  const pubSubClient = RedisWrapper.client('pubsub')
+  pubSubClient.subscribe(PUB_SUB_CHANNEL_TRIGGER_REFRESH).catch(err => {
+    logger.error(
+      { err },
+      'falling back to background refresh for system messages'
+    )
+  })
+  pubSubClient.on('message', () =>
+    SystemMessageManager.refreshCacheInBackground()
+  )
+  addRequiredCleanupHandlerBeforeDrainingConnections(
+    `pub-sub ${PUB_SUB_CHANNEL_TRIGGER_REFRESH}`,
+    () => pubSubClient.unsubscribe(PUB_SUB_CHANNEL_TRIGGER_REFRESH)
+  )
+}
+
 export default {
   getMessages: SystemMessageManager.getMessages.bind(SystemMessageManager),
   ...callbackifyAll(SystemMessageManager, { without: ['getMessages'] }),

services/web/config/settings.defaults.js

@@ -370,6 +370,9 @@ module.exports = {
     preservePath: process.env.MULTER_PRESERVE_PATH,
   },
 
+  notifyOnSystemMessageChanges:
+    process.env.NOTIFY_ON_SYSTEM_MESSAGE_CHANGES === 'true',
+
   // start failing the health check if active handles exceeds this limit
   maxActiveHandles: process.env.MAX_ACTIVE_HANDLES
     ? parseInt(process.env.MAX_ACTIVE_HANDLES, 10)