first commit

services/real-time/app/js/AuthorizationManager.js

@@ -0,0 +1,81 @@
+const { NotAuthorizedError } = require('./Errors')
+
+let AuthorizationManager
+module.exports = AuthorizationManager = {
+  assertClientCanViewProject(client, callback) {
+    AuthorizationManager._assertClientHasPrivilegeLevel(
+      client,
+      ['readOnly', 'readAndWrite', 'review', 'owner'],
+      callback
+    )
+  },
+
+  assertClientCanEditProject(client, callback) {
+    AuthorizationManager._assertClientHasPrivilegeLevel(
+      client,
+      ['readAndWrite', 'owner'],
+      callback
+    )
+  },
+
+  assertClientCanReviewProject(client, callback) {
+    AuthorizationManager._assertClientHasPrivilegeLevel(
+      client,
+      ['readAndWrite', 'owner', 'review'],
+      callback
+    )
+  },
+
+  _assertClientHasPrivilegeLevel(client, allowedLevels, callback) {
+    if (allowedLevels.includes(client.ol_context.privilege_level)) {
+      callback(null)
+    } else {
+      callback(new NotAuthorizedError())
+    }
+  },
+
+  assertClientCanViewProjectAndDoc(client, docId, callback) {
+    AuthorizationManager.assertClientCanViewProject(client, function (error) {
+      if (error) {
+        return callback(error)
+      }
+      AuthorizationManager._assertClientCanAccessDoc(client, docId, callback)
+    })
+  },
+
+  assertClientCanEditProjectAndDoc(client, docId, callback) {
+    AuthorizationManager.assertClientCanEditProject(client, function (error) {
+      if (error) {
+        return callback(error)
+      }
+      AuthorizationManager._assertClientCanAccessDoc(client, docId, callback)
+    })
+  },
+
+  assertClientCanReviewProjectAndDoc(client, docId, callback) {
+    AuthorizationManager.assertClientCanReviewProject(client, function (error) {
+      if (error) {
+        return callback(error)
+      }
+      AuthorizationManager._assertClientCanAccessDoc(client, docId, callback)
+    })
+  },
+
+  _assertClientCanAccessDoc(client, docId, callback) {
+    if (client.ol_context[`doc:${docId}`] === 'allowed') {
+      callback(null)
+    } else {
+      callback(new NotAuthorizedError())
+    }
+  },
+
+  addAccessToDoc(client, docId, callback) {
+    client.ol_context[`doc:${docId}`] = 'allowed'
+    callback(null)
+  },
+
+  removeAccessToDoc(client, docId, callback) {
+    delete client.ol_context[`doc:${docId}`]
+    callback(null)
+  },
+}

services/real-time/app/js/ChannelManager.js

@@ -0,0 +1,101 @@
+const logger = require('@overleaf/logger')
+const metrics = require('@overleaf/metrics')
+const settings = require('@overleaf/settings')
+const OError = require('@overleaf/o-error')
+
+const ClientMap = new Map() // for each redis client, store a Map of subscribed channels (channelname -> subscribe promise)
+
+// Manage redis pubsub subscriptions for individual projects and docs, ensuring
+// that we never subscribe to a channel multiple times. The socket.io side is
+// handled by RoomManager.
+
+module.exports = {
+  getClientMapEntry(rclient) {
+    // return the per-client channel map if it exists, otherwise create and
+    // return an empty map for the client.
+    return (
+      ClientMap.get(rclient) || ClientMap.set(rclient, new Map()).get(rclient)
+    )
+  },
+
+  subscribe(rclient, baseChannel, id) {
+    const clientChannelMap = this.getClientMapEntry(rclient)
+    const channel = `${baseChannel}:${id}`
+    const actualSubscribe = function () {
+      // subscribe is happening in the foreground and it should reject
+      return rclient
+        .subscribe(channel)
+        .finally(function () {
+          if (clientChannelMap.get(channel) === subscribePromise) {
+            clientChannelMap.delete(channel)
+          }
+        })
+        .then(function () {
+          logger.debug({ channel }, 'subscribed to channel')
+          metrics.inc(`subscribe.${baseChannel}`)
+        })
+        .catch(function (err) {
+          logger.error({ channel, err }, 'failed to subscribe to channel')
+          metrics.inc(`subscribe.failed.${baseChannel}`)
+          // add context for the stack-trace at the call-site
+          throw new OError('failed to subscribe to channel', {
+            channel,
+          }).withCause(err)
+        })
+    }
+
+    const pendingActions = clientChannelMap.get(channel) || Promise.resolve()
+    const subscribePromise = pendingActions.then(
+      actualSubscribe,
+      actualSubscribe
+    )
+    clientChannelMap.set(channel, subscribePromise)
+    logger.debug({ channel }, 'planned to subscribe to channel')
+    return subscribePromise
+  },
+
+  unsubscribe(rclient, baseChannel, id) {
+    const clientChannelMap = this.getClientMapEntry(rclient)
+    const channel = `${baseChannel}:${id}`
+    const actualUnsubscribe = function () {
+      // unsubscribe is happening in the background, it should not reject
+      return rclient
+        .unsubscribe(channel)
+        .finally(function () {
+          if (clientChannelMap.get(channel) === unsubscribePromise) {
+            clientChannelMap.delete(channel)
+          }
+        })
+        .then(function () {
+          logger.debug({ channel }, 'unsubscribed from channel')
+          metrics.inc(`unsubscribe.${baseChannel}`)
+        })
+        .catch(function (err) {
+          logger.error({ channel, err }, 'unsubscribed from channel')
+          metrics.inc(`unsubscribe.failed.${baseChannel}`)
+        })
+    }
+
+    const pendingActions = clientChannelMap.get(channel) || Promise.resolve()
+    const unsubscribePromise = pendingActions.then(
+      actualUnsubscribe,
+      actualUnsubscribe
+    )
+    clientChannelMap.set(channel, unsubscribePromise)
+    logger.debug({ channel }, 'planned to unsubscribe from channel')
+    return unsubscribePromise
+  },
+
+  publish(rclient, baseChannel, id, data) {
+    let channel
+    metrics.summary(`redis.publish.${baseChannel}`, data.length)
+    if (id === 'all' || !settings.publishOnIndividualChannels) {
+      channel = baseChannel
+    } else {
+      channel = `${baseChannel}:${id}`
+    }
+    // we publish on a different client to the subscribe, so we can't
+    // check for the channel existing here
+    rclient.publish(channel, data)
+  },
+}

services/real-time/app/js/ConnectedUsersManager.js

@@ -0,0 +1,249 @@
+const async = require('async')
+const Settings = require('@overleaf/settings')
+const logger = require('@overleaf/logger')
+const redis = require('@overleaf/redis-wrapper')
+const OError = require('@overleaf/o-error')
+const Metrics = require('@overleaf/metrics')
+const rclient = redis.createClient(Settings.redis.realtime)
+const Keys = Settings.redis.realtime.key_schema
+
+const ONE_HOUR_IN_S = 60 * 60
+const ONE_DAY_IN_S = ONE_HOUR_IN_S * 24
+const FOUR_DAYS_IN_S = ONE_DAY_IN_S * 4
+
+const USER_TIMEOUT_IN_S = ONE_HOUR_IN_S / 4
+const REFRESH_TIMEOUT_IN_S = 10 // only show clients which have responded to a refresh request in the last 10 seconds
+
+function recordProjectNotEmptySinceMetric(res, status) {
+  const diff = Date.now() / 1000 - parseInt(res, 10)
+  const BUCKETS = [
+    0,
+    ONE_HOUR_IN_S,
+    2 * ONE_HOUR_IN_S,
+    ONE_DAY_IN_S,
+    2 * ONE_DAY_IN_S,
+    7 * ONE_DAY_IN_S,
+    30 * ONE_DAY_IN_S,
+  ]
+  Metrics.histogram('project_not_empty_since', diff, BUCKETS, { status })
+}
+
+module.exports = {
+  // Use the same method for when a user connects, and when a user sends a cursor
+  // update. This way we don't care if the connected_user key has expired when
+  // we receive a cursor update.
+  updateUserPosition(projectId, clientId, user, cursorData, callback) {
+    logger.debug({ projectId, clientId }, 'marking user as joined or connected')
+
+    const multi = rclient.multi()
+
+    multi.sadd(Keys.clientsInProject({ project_id: projectId }), clientId)
+    multi.scard(Keys.clientsInProject({ project_id: projectId }))
+    multi.expire(
+      Keys.clientsInProject({ project_id: projectId }),
+      FOUR_DAYS_IN_S
+    )
+
+    multi.hset(
+      Keys.connectedUser({ project_id: projectId, client_id: clientId }),
+      'last_updated_at',
+      Date.now()
+    )
+    multi.hset(
+      Keys.connectedUser({ project_id: projectId, client_id: clientId }),
+      'user_id',
+      user._id
+    )
+    multi.hset(
+      Keys.connectedUser({ project_id: projectId, client_id: clientId }),
+      'first_name',
+      user.first_name || ''
+    )
+    multi.hset(
+      Keys.connectedUser({ project_id: projectId, client_id: clientId }),
+      'last_name',
+      user.last_name || ''
+    )
+    multi.hset(
+      Keys.connectedUser({ project_id: projectId, client_id: clientId }),
+      'email',
+      user.email || ''
+    )
+
+    if (cursorData) {
+      multi.hset(
+        Keys.connectedUser({ project_id: projectId, client_id: clientId }),
+        'cursorData',
+        JSON.stringify(cursorData)
+      )
+    }
+    multi.expire(
+      Keys.connectedUser({ project_id: projectId, client_id: clientId }),
+      USER_TIMEOUT_IN_S
+    )
+
+    multi.exec(function (err, res) {
+      if (err) {
+        err = new OError('problem marking user as connected').withCause(err)
+      }
+      const [, nConnectedClients] = res
+      Metrics.inc('editing_session_mode', 1, {
+        method: cursorData ? 'update' : 'connect',
+        status: nConnectedClients === 1 ? 'single' : 'multi',
+      })
+      callback(err)
+    })
+  },
+
+  refreshClient(projectId, clientId) {
+    logger.debug({ projectId, clientId }, 'refreshing connected client')
+    const multi = rclient.multi()
+    multi.hset(
+      Keys.connectedUser({ project_id: projectId, client_id: clientId }),
+      'last_updated_at',
+      Date.now()
+    )
+    multi.expire(
+      Keys.connectedUser({ project_id: projectId, client_id: clientId }),
+      USER_TIMEOUT_IN_S
+    )
+    multi.exec(function (err) {
+      if (err) {
+        logger.err(
+          { err, projectId, clientId },
+          'problem refreshing connected client'
+        )
+      }
+    })
+  },
+
+  markUserAsDisconnected(projectId, clientId, callback) {
+    logger.debug({ projectId, clientId }, 'marking user as disconnected')
+    const multi = rclient.multi()
+    multi.srem(Keys.clientsInProject({ project_id: projectId }), clientId)
+    multi.scard(Keys.clientsInProject({ project_id: projectId }))
+    multi.expire(
+      Keys.clientsInProject({ project_id: projectId }),
+      FOUR_DAYS_IN_S
+    )
+    multi.del(
+      Keys.connectedUser({ project_id: projectId, client_id: clientId })
+    )
+    multi.exec(function (err, res) {
+      if (err) {
+        err = new OError('problem marking user as disconnected').withCause(err)
+      }
+      const [, nConnectedClients] = res
+      const status =
+        nConnectedClients === 0
+          ? 'empty'
+          : nConnectedClients === 1
+            ? 'single'
+            : 'multi'
+      Metrics.inc('editing_session_mode', 1, {
+        method: 'disconnect',
+        status,
+      })
+      if (status === 'empty') {
+        rclient.getdel(Keys.projectNotEmptySince({ projectId }), (err, res) => {
+          if (err) {
+            logger.warn(
+              { err, projectId },
+              'could not collect projectNotEmptySince'
+            )
+          } else if (res) {
+            recordProjectNotEmptySinceMetric(res, status)
+          }
+        })
+      } else {
+        // Only populate projectNotEmptySince when more clients remain connected.
+        const nowInSeconds = Math.ceil(Date.now() / 1000).toString()
+        // We can go back to SET GET after upgrading to redis 7.0+
+        const multi = rclient.multi()
+        multi.get(Keys.projectNotEmptySince({ projectId }))
+        multi.set(
+          Keys.projectNotEmptySince({ projectId }),
+          nowInSeconds,
+          'NX',
+          'EX',
+          31 * ONE_DAY_IN_S
+        )
+        multi.exec((err, res) => {
+          if (err) {
+            logger.warn(
+              { err, projectId },
+              'could not get/set projectNotEmptySince'
+            )
+          } else if (res[0]) {
+            recordProjectNotEmptySinceMetric(res[0], status)
+          }
+        })
+      }
+      callback(err)
+    })
+  },
+
+  _getConnectedUser(projectId, clientId, callback) {
+    rclient.hgetall(
+      Keys.connectedUser({ project_id: projectId, client_id: clientId }),
+      function (err, result) {
+        if (err) {
+          err = new OError('problem fetching connected user details', {
+            other_client_id: clientId,
+          }).withCause(err)
+          return callback(err)
+        }
+        if (!(result && result.user_id)) {
+          result = {
+            connected: false,
+            client_id: clientId,
+          }
+        } else {
+          result.connected = true
+          result.client_id = clientId
+          result.client_age =
+            (Date.now() - parseInt(result.last_updated_at, 10)) / 1000
+          if (result.cursorData) {
+            try {
+              result.cursorData = JSON.parse(result.cursorData)
+            } catch (e) {
+              OError.tag(e, 'error parsing cursorData JSON', {
+                other_client_id: clientId,
+                cursorData: result.cursorData,
+              })
+              return callback(e)
+            }
+          }
+        }
+        callback(err, result)
+      }
+    )
+  },
+
+  getConnectedUsers(projectId, callback) {
+    const self = this
+    rclient.smembers(
+      Keys.clientsInProject({ project_id: projectId }),
+      function (err, results) {
+        if (err) {
+          err = new OError('problem getting clients in project').withCause(err)
+          return callback(err)
+        }
+        const jobs = results.map(
+          clientId => cb => self._getConnectedUser(projectId, clientId, cb)
+        )
+        async.series(jobs, function (err, users) {
+          if (err) {
+            OError.tag(err, 'problem getting connected users')
+            return callback(err)
+          }
+          users = users.filter(
+            user =>
+              user && user.connected && user.client_age < REFRESH_TIMEOUT_IN_S
+          )
+          callback(null, users)
+        })
+      }
+    )
+  },
+}

services/real-time/app/js/DeploymentManager.js

@@ -0,0 +1,62 @@
+const logger = require('@overleaf/logger')
+const settings = require('@overleaf/settings')
+const fs = require('node:fs')
+
+// Monitor a status file (e.g. /etc/real_time_status) periodically and close the
+// service if the file contents don't contain the matching deployment colour.
+
+const FILE_CHECK_INTERVAL = 5000
+const statusFile = settings.deploymentFile
+const deploymentColour = settings.deploymentColour
+
+let serviceCloseTime
+
+function updateDeploymentStatus(fileContent) {
+  const closed = fileContent && !fileContent.includes(deploymentColour)
+  if (closed && !settings.serviceIsClosed) {
+    settings.serviceIsClosed = true
+    serviceCloseTime = Date.now() + 60 * 1000 // delay closing by 1 minute
+    logger.info({ fileContent }, 'closing service')
+  } else if (!closed && settings.serviceIsClosed) {
+    settings.serviceIsClosed = false
+    logger.info({ fileContent }, 'opening service')
+  }
+}
+
+function pollStatusFile() {
+  fs.readFile(statusFile, { encoding: 'utf8' }, (err, fileContent) => {
+    if (err) {
+      logger.error(
+        { file: statusFile, fsErr: err },
+        'error reading service status file'
+      )
+      return
+    }
+    updateDeploymentStatus(fileContent)
+  })
+}
+
+function checkStatusFileSync() {
+  // crash on start up if file does not exist
+  const content = fs.readFileSync(statusFile, { encoding: 'utf8' })
+  updateDeploymentStatus(content)
+  if (settings.serviceIsClosed) {
+    serviceCloseTime = Date.now() // skip closing delay on start up
+  }
+}
+
+module.exports = {
+  initialise() {
+    if (statusFile && deploymentColour) {
+      logger.info(
+        { statusFile, deploymentColour, interval: FILE_CHECK_INTERVAL },
+        'monitoring deployment status file'
+      )
+      checkStatusFileSync() // perform an initial synchronous check at start up
+      setInterval(pollStatusFile, FILE_CHECK_INTERVAL) // continue checking periodically
+    }
+  },
+  deploymentIsClosed() {
+    return settings.serviceIsClosed && Date.now() >= serviceCloseTime
+  },
+}

services/real-time/app/js/DocumentUpdaterController.js

@@ -0,0 +1,184 @@
+const logger = require('@overleaf/logger')
+const settings = require('@overleaf/settings')
+const RedisClientManager = require('./RedisClientManager')
+const SafeJsonParse = require('./SafeJsonParse')
+const EventLogger = require('./EventLogger')
+const HealthCheckManager = require('./HealthCheckManager')
+const RoomManager = require('./RoomManager')
+const ChannelManager = require('./ChannelManager')
+const metrics = require('@overleaf/metrics')
+
+let DocumentUpdaterController
+module.exports = DocumentUpdaterController = {
+  // DocumentUpdaterController is responsible for updates that come via Redis
+  // Pub/Sub from the document updater.
+  rclientList: RedisClientManager.createClientList(settings.redis.pubsub),
+
+  listenForUpdatesFromDocumentUpdater(io) {
+    logger.debug(
+      { rclients: this.rclientList.length },
+      'listening for applied-ops events'
+    )
+    for (const rclient of this.rclientList) {
+      rclient.subscribe('applied-ops')
+      rclient.on('message', function (channel, message) {
+        metrics.inc('rclient', 0.001) // global event rate metric
+        if (settings.debugEvents > 0) {
+          EventLogger.debugEvent(channel, message)
+        }
+        DocumentUpdaterController._processMessageFromDocumentUpdater(
+          io,
+          channel,
+          message
+        )
+      })
+    }
+    // create metrics for each redis instance only when we have multiple redis clients
+    if (this.rclientList.length > 1) {
+      this.rclientList.forEach((rclient, i) => {
+        // per client event rate metric
+        const metricName = `rclient-${i}`
+        rclient.on('message', () => metrics.inc(metricName, 0.001))
+      })
+    }
+    this.handleRoomUpdates(this.rclientList)
+  },
+
+  handleRoomUpdates(rclientSubList) {
+    const roomEvents = RoomManager.eventSource()
+    roomEvents.on('doc-active', function (docId) {
+      const subscribePromises = rclientSubList.map(rclient =>
+        ChannelManager.subscribe(rclient, 'applied-ops', docId)
+      )
+      RoomManager.emitOnCompletion(subscribePromises, `doc-subscribed-${docId}`)
+    })
+    roomEvents.on('doc-empty', docId =>
+      rclientSubList.map(rclient =>
+        ChannelManager.unsubscribe(rclient, 'applied-ops', docId)
+      )
+    )
+  },
+
+  _processMessageFromDocumentUpdater(io, channel, message) {
+    SafeJsonParse.parse(message, function (error, message) {
+      if (error) {
+        logger.error({ err: error, channel }, 'error parsing JSON')
+        return
+      }
+      if (message.op) {
+        if (message._id && settings.checkEventOrder) {
+          const status = EventLogger.checkEventOrder(
+            'applied-ops',
+            message._id,
+            message
+          )
+          if (status === 'duplicate') {
+            return // skip duplicate events
+          }
+        }
+        DocumentUpdaterController._applyUpdateFromDocumentUpdater(
+          io,
+          message.doc_id,
+          message.op
+        )
+      } else if (message.error) {
+        DocumentUpdaterController._processErrorFromDocumentUpdater(
+          io,
+          message.doc_id,
+          message.error,
+          message
+        )
+      } else if (message.health_check) {
+        logger.debug(
+          { message },
+          'got health check message in applied ops channel'
+        )
+        HealthCheckManager.check(channel, message.key)
+      }
+    })
+  },
+
+  _applyUpdateFromDocumentUpdater(io, docId, update) {
+    let client
+    const clientList = io.sockets.clients(docId)
+    // avoid unnecessary work if no clients are connected
+    if (clientList.length === 0) {
+      return
+    }
+
+    update.meta = update.meta || {}
+    const { tsRT: realTimeIngestionTime } = update.meta
+    delete update.meta.tsRT
+
+    // send updates to clients
+    logger.debug(
+      {
+        docId,
+        version: update.v,
+        source: update.meta && update.meta.source,
+        socketIoClients: clientList.map(client => client.id),
+      },
+      'distributing updates to clients'
+    )
+    const seen = {}
+    // send messages only to unique clients (due to duplicate entries in io.sockets.clients)
+    for (client of clientList) {
+      if (!seen[client.id]) {
+        seen[client.id] = true
+        if (client.publicId === update.meta.source) {
+          logger.debug(
+            {
+              docId,
+              version: update.v,
+              source: update.meta.source,
+            },
+            'distributing update to sender'
+          )
+          metrics.histogram(
+            'update-processing-time',
+            performance.now() - realTimeIngestionTime,
+            [
+              0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 20, 50, 100, 200, 500, 1000,
+              2000, 5000, 10000,
+            ],
+            { path: 'sharejs' }
+          )
+          client.emit('otUpdateApplied', { v: update.v, doc: update.doc })
+        } else if (!update.dup) {
+          // Duplicate ops should just be sent back to sending client for acknowledgement
+          logger.debug(
+            {
+              docId,
+              version: update.v,
+              source: update.meta.source,
+              clientId: client.id,
+            },
+            'distributing update to collaborator'
+          )
+          client.emit('otUpdateApplied', update)
+        }
+      }
+    }
+    if (Object.keys(seen).length < clientList.length) {
+      metrics.inc('socket-io.duplicate-clients', 0.1)
+      logger.debug(
+        {
+          docId,
+          socketIoClients: clientList.map(client => client.id),
+        },
+        'discarded duplicate clients'
+      )
+    }
+  },
+
+  _processErrorFromDocumentUpdater(io, docId, error, message) {
+    for (const client of io.sockets.clients(docId)) {
+      logger.warn(
+        { err: error, docId, clientId: client.id },
+        'error from document updater, disconnecting client'
+      )
+      client.emit('otUpdateError', error, message)
+      client.disconnect()
+    }
+  },
+}

services/real-time/app/js/DocumentUpdaterManager.js

@@ -0,0 +1,156 @@
+const request = require('request')
+const _ = require('lodash')
+const OError = require('@overleaf/o-error')
+const logger = require('@overleaf/logger')
+const settings = require('@overleaf/settings')
+const metrics = require('@overleaf/metrics')
+const {
+  ClientRequestedMissingOpsError,
+  DocumentUpdaterRequestFailedError,
+  NullBytesInOpError,
+  UpdateTooLargeError,
+} = require('./Errors')
+
+const rclient = require('@overleaf/redis-wrapper').createClient(
+  settings.redis.documentupdater
+)
+const Keys = settings.redis.documentupdater.key_schema
+
+const DocumentUpdaterManager = {
+  getDocument(projectId, docId, fromVersion, callback) {
+    const timer = new metrics.Timer('get-document')
+    const url = `${settings.apis.documentupdater.url}/project/${projectId}/doc/${docId}?fromVersion=${fromVersion}`
+    logger.debug(
+      { projectId, docId, fromVersion },
+      'getting doc from document updater'
+    )
+    request.get(url, function (err, res, body) {
+      timer.done()
+      if (err) {
+        OError.tag(err, 'error getting doc from doc updater')
+        return callback(err)
+      }
+      if (res.statusCode >= 200 && res.statusCode < 300) {
+        logger.debug(
+          { projectId, docId },
+          'got doc from document document updater'
+        )
+        try {
+          body = JSON.parse(body)
+        } catch (error) {
+          OError.tag(error, 'error parsing doc updater response')
+          return callback(error)
+        }
+        body = body || {}
+        callback(
+          null,
+          body.lines,
+          body.version,
+          body.ranges,
+          body.ops,
+          body.ttlInS
+        )
+      } else if (res.statusCode === 422 && body?.firstVersionInRedis) {
+        callback(new ClientRequestedMissingOpsError(422, body))
+      } else if ([404, 422].includes(res.statusCode)) {
+        callback(new ClientRequestedMissingOpsError(res.statusCode))
+      } else {
+        callback(
+          new DocumentUpdaterRequestFailedError('getDocument', res.statusCode)
+        )
+      }
+    })
+  },
+
+  checkDocument(projectId, docId, callback) {
+    // in this call fromVersion = -1 means get document without docOps
+    DocumentUpdaterManager.getDocument(projectId, docId, -1, callback)
+  },
+
+  flushProjectToMongoAndDelete(projectId, callback) {
+    // this method is called when the last connected user leaves the project
+    logger.debug({ projectId }, 'deleting project from document updater')
+    const timer = new metrics.Timer('delete.mongo.project')
+    // flush the project in the background when all users have left
+    const url =
+      `${settings.apis.documentupdater.url}/project/${projectId}?background=true` +
+      (settings.shutDownInProgress ? '&shutdown=true' : '')
+    request.del(url, function (err, res) {
+      timer.done()
+      if (err) {
+        OError.tag(err, 'error deleting project from document updater')
+        callback(err)
+      } else if (res.statusCode >= 200 && res.statusCode < 300) {
+        logger.debug({ projectId }, 'deleted project from document updater')
+        callback(null)
+      } else {
+        callback(
+          new DocumentUpdaterRequestFailedError(
+            'flushProjectToMongoAndDelete',
+            res.statusCode
+          )
+        )
+      }
+    })
+  },
+
+  _getPendingUpdateListKey() {
+    const shard = _.random(0, settings.pendingUpdateListShardCount - 1)
+    if (shard === 0) {
+      return 'pending-updates-list'
+    } else {
+      return `pending-updates-list-${shard}`
+    }
+  },
+
+  queueChange(projectId, docId, change, callback) {
+    const allowedKeys = [
+      'doc',
+      'op',
+      'v',
+      'dupIfSource',
+      'meta',
+      'lastV',
+      'hash',
+    ]
+    change = _.pick(change, allowedKeys)
+    const jsonChange = JSON.stringify(change)
+    if (jsonChange.indexOf('\u0000') !== -1) {
+      // memory corruption check
+      return callback(new NullBytesInOpError(jsonChange))
+    }
+
+    const updateSize = jsonChange.length
+    if (updateSize > settings.maxUpdateSize) {
+      return callback(new UpdateTooLargeError(updateSize))
+    }
+
+    // record metric for each update added to queue
+    metrics.summary('redis.pendingUpdates', updateSize, { status: 'push' })
+
+    const docKey = `${projectId}:${docId}`
+    // Push onto pendingUpdates for doc_id first, because once the doc updater
+    // gets an entry on pending-updates-list, it starts processing.
+    rclient.rpush(
+      Keys.pendingUpdates({ doc_id: docId }),
+      jsonChange,
+      function (error) {
+        if (error) {
+          error = new OError('error pushing update into redis').withCause(error)
+          return callback(error)
+        }
+        const queueKey = DocumentUpdaterManager._getPendingUpdateListKey()
+        rclient.rpush(queueKey, docKey, function (error) {
+          if (error) {
+            error = new OError('error pushing doc_id into redis')
+              .withInfo({ queueKey })
+              .withCause(error)
+          }
+          callback(error)
+        })
+      }
+    )
+  },
+}
+
+module.exports = DocumentUpdaterManager

services/real-time/app/js/DrainManager.js

@@ -0,0 +1,59 @@
+const logger = require('@overleaf/logger')
+
+module.exports = {
+  startDrainTimeWindow(io, minsToDrain, callback) {
+    const drainPerMin = io.sockets.clients().length / minsToDrain
+    // enforce minimum drain rate
+    this.startDrain(io, Math.max(drainPerMin / 60, 4), callback)
+  },
+
+  startDrain(io, rate, callback) {
+    // Clear out any old interval
+    clearInterval(this.interval)
+    logger.info({ rate }, 'starting drain')
+    if (rate === 0) {
+      return
+    }
+    let pollingInterval
+    if (rate < 1) {
+      // allow lower drain rates
+      // e.g. rate=0.1 will drain one client every 10 seconds
+      pollingInterval = 1000 / rate
+      rate = 1
+    } else {
+      pollingInterval = 1000
+    }
+    this.interval = setInterval(() => {
+      const requestedAllClientsToReconnect = this.reconnectNClients(io, rate)
+      if (requestedAllClientsToReconnect && callback) {
+        callback()
+        callback = undefined
+      }
+    }, pollingInterval)
+  },
+
+  RECONNECTED_CLIENTS: {},
+  reconnectNClients(io, N) {
+    let drainedCount = 0
+    for (const client of io.sockets.clients()) {
+      if (!this.RECONNECTED_CLIENTS[client.id]) {
+        this.RECONNECTED_CLIENTS[client.id] = true
+        logger.debug(
+          { clientId: client.id },
+          'Asking client to reconnect gracefully'
+        )
+        client.emit('reconnectGracefully')
+        drainedCount++
+      }
+      const haveDrainedNClients = drainedCount === N
+      if (haveDrainedNClients) {
+        break
+      }
+    }
+    if (drainedCount < N) {
+      logger.info('All clients have been told to reconnectGracefully')
+      return true
+    }
+    return false
+  },
+}

services/real-time/app/js/Errors.js

@@ -0,0 +1,104 @@
+const OError = require('@overleaf/o-error')
+
+class ClientRequestedMissingOpsError extends OError {
+  constructor(statusCode, info = {}) {
+    super('doc updater could not load requested ops', {
+      statusCode,
+      ...info,
+    })
+  }
+}
+
+class CodedError extends OError {
+  constructor(message, code) {
+    super(message, { code })
+  }
+}
+
+class CorruptedJoinProjectResponseError extends OError {
+  constructor() {
+    super('no data returned from joinProject request')
+  }
+}
+
+class DataTooLargeToParseError extends OError {
+  constructor(data) {
+    super('data too large to parse', {
+      head: data.slice(0, 1024),
+      length: data.length,
+    })
+  }
+}
+
+class DocumentUpdaterRequestFailedError extends OError {
+  constructor(action, statusCode) {
+    super('doc updater returned a non-success status code', {
+      action,
+      statusCode,
+    })
+  }
+}
+
+class JoinLeaveEpochMismatchError extends OError {
+  constructor() {
+    super('joinLeaveEpoch mismatch')
+  }
+}
+
+class MissingSessionError extends OError {
+  constructor() {
+    super('could not look up session by key')
+  }
+}
+
+class NotAuthorizedError extends OError {
+  constructor() {
+    super('not authorized')
+  }
+}
+
+class NotJoinedError extends OError {
+  constructor() {
+    super('no project_id found on client')
+  }
+}
+
+class NullBytesInOpError extends OError {
+  constructor(jsonChange) {
+    super('null bytes found in op', { jsonChange })
+  }
+}
+
+class UnexpectedArgumentsError extends OError {
+  constructor() {
+    super('unexpected arguments')
+  }
+}
+
+class UpdateTooLargeError extends OError {
+  constructor(updateSize) {
+    super('update is too large', { updateSize })
+  }
+}
+
+class WebApiRequestFailedError extends OError {
+  constructor(statusCode) {
+    super('non-success status code from web', { statusCode })
+  }
+}
+
+module.exports = {
+  CodedError,
+  CorruptedJoinProjectResponseError,
+  ClientRequestedMissingOpsError,
+  DataTooLargeToParseError,
+  DocumentUpdaterRequestFailedError,
+  JoinLeaveEpochMismatchError,
+  MissingSessionError,
+  NotAuthorizedError,
+  NotJoinedError,
+  NullBytesInOpError,
+  UnexpectedArgumentsError,
+  UpdateTooLargeError,
+  WebApiRequestFailedError,
+}

services/real-time/app/js/EventLogger.js

@@ -0,0 +1,81 @@
+let EventLogger
+const logger = require('@overleaf/logger')
+const metrics = require('@overleaf/metrics')
+const settings = require('@overleaf/settings')
+
+// keep track of message counters to detect duplicate and out of order events
+// messsage ids have the format "UNIQUEHOSTKEY-COUNTER"
+
+const EVENT_LOG_COUNTER = {}
+const EVENT_LOG_TIMESTAMP = {}
+let EVENT_LAST_CLEAN_TIMESTAMP = 0
+
+// counter for debug logs
+let COUNTER = 0
+
+module.exports = EventLogger = {
+  MAX_STALE_TIME_IN_MS: 3600 * 1000,
+
+  debugEvent(channel, message) {
+    if (settings.debugEvents > 0) {
+      logger.info({ channel, message, counter: COUNTER++ }, 'logging event')
+      settings.debugEvents--
+    }
+  },
+
+  checkEventOrder(channel, messageId) {
+    if (typeof messageId !== 'string') {
+      return
+    }
+    let result
+    if (!(result = messageId.match(/^(.*)-(\d+)$/))) {
+      return
+    }
+    const key = result[1]
+    const count = parseInt(result[2], 0)
+    if (!(count >= 0)) {
+      // ignore checks if counter is not present
+      return
+    }
+    // store the last count in a hash for each host
+    const previous = EventLogger._storeEventCount(key, count)
+    if (!previous || count === previous + 1) {
+      metrics.inc(`event.${channel}.valid`)
+      return // order is ok
+    }
+    if (count === previous) {
+      metrics.inc(`event.${channel}.duplicate`)
+      logger.warn({ channel, messageId }, 'duplicate event')
+      return 'duplicate'
+    } else {
+      metrics.inc(`event.${channel}.out-of-order`)
+      logger.warn(
+        { channel, messageId, key, previous, count },
+        'out of order event'
+      )
+      return 'out-of-order'
+    }
+  },
+
+  _storeEventCount(key, count) {
+    const previous = EVENT_LOG_COUNTER[key]
+    const now = Date.now()
+    EVENT_LOG_COUNTER[key] = count
+    EVENT_LOG_TIMESTAMP[key] = now
+    // periodically remove old counts
+    if (now - EVENT_LAST_CLEAN_TIMESTAMP > EventLogger.MAX_STALE_TIME_IN_MS) {
+      EventLogger._cleanEventStream(now)
+      EVENT_LAST_CLEAN_TIMESTAMP = now
+    }
+    return previous
+  },
+
+  _cleanEventStream(now) {
+    Object.entries(EVENT_LOG_TIMESTAMP).forEach(([key, timestamp]) => {
+      if (now - timestamp > EventLogger.MAX_STALE_TIME_IN_MS) {
+        delete EVENT_LOG_COUNTER[key]
+        delete EVENT_LOG_TIMESTAMP[key]
+      }
+    })
+  },
+}

services/real-time/app/js/HealthCheckManager.js

@@ -0,0 +1,77 @@
+const metrics = require('@overleaf/metrics')
+const logger = require('@overleaf/logger')
+
+const os = require('node:os')
+const HOST = os.hostname()
+const PID = process.pid
+let COUNT = 0
+
+const CHANNEL_MANAGER = {} // hash of event checkers by channel name
+const CHANNEL_ERROR = {} // error status by channel name
+
+module.exports = class HealthCheckManager {
+  // create an instance of this class which checks that an event with a unique
+  // id is received only once within a timeout
+  constructor(channel, timeout) {
+    // unique event string
+    this.channel = channel
+    this.id = `host=${HOST}:pid=${PID}:count=${COUNT++}`
+    // count of number of times the event is received
+    this.count = 0
+    // after a timeout check the status of the count
+    this.handler = setTimeout(() => {
+      this.setStatus()
+    }, timeout || 1000)
+    // use a timer to record the latency of the channel
+    this.timer = new metrics.Timer(`event.${this.channel}.latency`)
+    // keep a record of these objects to dispatch on
+    CHANNEL_MANAGER[this.channel] = this
+  }
+
+  processEvent(id) {
+    // if this is our event record it
+    if (id === this.id) {
+      this.count++
+      if (this.timer) {
+        this.timer.done()
+      }
+      this.timer = undefined // only time the latency of the first event
+    }
+  }
+
+  setStatus() {
+    // if we saw the event anything other than a single time that is an error
+    const isFailing = this.count !== 1
+    if (isFailing) {
+      logger.err(
+        { channel: this.channel, count: this.count, id: this.id },
+        'redis channel health check error'
+      )
+    }
+    CHANNEL_ERROR[this.channel] = isFailing
+  }
+
+  // class methods
+  static check(channel, id) {
+    // dispatch event to manager for channel
+    if (CHANNEL_MANAGER[channel]) {
+      CHANNEL_MANAGER[channel].processEvent(id)
+    }
+  }
+
+  static status() {
+    // return status of all channels for logging
+    return CHANNEL_ERROR
+  }
+
+  static isFailing() {
+    // check if any channel status is bad
+    for (const channel in CHANNEL_ERROR) {
+      const error = CHANNEL_ERROR[channel]
+      if (error === true) {
+        return true
+      }
+    }
+    return false
+  }
+}

services/real-time/app/js/HttpApiController.js

@@ -0,0 +1,49 @@
+const WebsocketLoadBalancer = require('./WebsocketLoadBalancer')
+const DrainManager = require('./DrainManager')
+const logger = require('@overleaf/logger')
+
+module.exports = {
+  sendMessage(req, res) {
+    logger.debug({ message: req.params.message }, 'sending message')
+    if (Array.isArray(req.body)) {
+      for (const payload of req.body) {
+        WebsocketLoadBalancer.emitToRoom(
+          req.params.project_id,
+          req.params.message,
+          payload
+        )
+      }
+    } else {
+      WebsocketLoadBalancer.emitToRoom(
+        req.params.project_id,
+        req.params.message,
+        req.body
+      )
+    }
+    res.sendStatus(204)
+  },
+
+  startDrain(req, res) {
+    const io = req.app.get('io')
+    let rate = req.query.rate || '4'
+    rate = parseFloat(rate) || 0
+    logger.info({ rate }, 'setting client drain rate')
+    DrainManager.startDrain(io, rate)
+    res.sendStatus(204)
+  },
+
+  disconnectClient(req, res, next) {
+    const io = req.app.get('io')
+    const { client_id: clientId } = req.params
+    const client = io.sockets.sockets[clientId]
+
+    if (!client) {
+      logger.debug({ clientId }, 'api: client already disconnected')
+      res.sendStatus(404)
+      return
+    }
+    logger.info({ clientId }, 'api: requesting client disconnect')
+    client.on('disconnect', () => res.sendStatus(204))
+    client.disconnect()
+  },
+}

services/real-time/app/js/HttpController.js

@@ -0,0 +1,53 @@
+let HttpController
+module.exports = HttpController = {
+  // The code in this controller is hard to unit test because of a lot of
+  // dependencies on internal socket.io methods. It is not critical to the running
+  // of Overleaf, and is only used for getting stats about connected clients,
+  // and for checking internal state in acceptance tests. The acceptances tests
+  // should provide appropriate coverage.
+  _getConnectedClientView(ioClient) {
+    const clientId = ioClient.id
+    const {
+      project_id: projectId,
+      user_id: userId,
+      first_name: firstName,
+      last_name: lastName,
+      email,
+      connected_time: connectedTime,
+    } = ioClient.ol_context
+    const client = {
+      client_id: clientId,
+      project_id: projectId,
+      user_id: userId,
+      first_name: firstName,
+      last_name: lastName,
+      email,
+      connected_time: connectedTime,
+    }
+    client.rooms = Object.keys(ioClient.manager.roomClients[clientId] || {})
+      // drop the namespace
+      .filter(room => room !== '')
+      // room names are composed as '<NAMESPACE>/<ROOM>' and the default
+      //  namespace is empty (see comments in RoomManager), just drop the '/'
+      .map(fullRoomPath => fullRoomPath.slice(1))
+    return client
+  },
+
+  getConnectedClients(req, res) {
+    const io = req.app.get('io')
+    const ioClients = io.sockets.clients()
+
+    res.json(ioClients.map(HttpController._getConnectedClientView))
+  },
+
+  getConnectedClient(req, res) {
+    const { client_id: clientId } = req.params
+    const io = req.app.get('io')
+    const ioClient = io.sockets.sockets[clientId]
+    if (!ioClient) {
+      res.sendStatus(404)
+      return
+    }
+    res.json(HttpController._getConnectedClientView(ioClient))
+  },
+}

services/real-time/app/js/RedisClientManager.js

@@ -0,0 +1,19 @@
+const redis = require('@overleaf/redis-wrapper')
+const logger = require('@overleaf/logger')
+
+module.exports = {
+  createClientList(...configs) {
+    // create a dynamic list of redis clients, excluding any configurations which are not defined
+    return configs.filter(Boolean).map(x => {
+      const redisType = x.cluster
+        ? 'cluster'
+        : x.sentinels
+          ? 'sentinel'
+          : x.host
+            ? 'single'
+            : 'unknown'
+      logger.debug({ redis: redisType }, 'creating redis client')
+      return redis.createClient(x)
+    })
+  },
+}

services/real-time/app/js/RoomManager.js

@@ -0,0 +1,161 @@
+const logger = require('@overleaf/logger')
+const metrics = require('@overleaf/metrics')
+const { EventEmitter } = require('node:events')
+const OError = require('@overleaf/o-error')
+
+const IdMap = new Map() // keep track of whether ids are from projects or docs
+const RoomEvents = new EventEmitter() // emits {project,doc}-active and {project,doc}-empty events
+
+// Manage socket.io rooms for individual projects and docs
+//
+// The first time someone joins a project or doc we emit a 'project-active' or
+// 'doc-active' event.
+//
+// When the last person leaves a project or doc, we emit 'project-empty' or
+// 'doc-empty' event.
+//
+// The pubsub side is handled by ChannelManager
+
+module.exports = {
+  joinProject(client, projectId, callback) {
+    this.joinEntity(client, 'project', projectId, callback)
+  },
+
+  joinDoc(client, docId, callback) {
+    this.joinEntity(client, 'doc', docId, callback)
+  },
+
+  leaveDoc(client, docId) {
+    this.leaveEntity(client, 'doc', docId)
+  },
+
+  leaveProjectAndDocs(client) {
+    // what rooms is this client in? we need to leave them all. socket.io
+    // will cause us to leave the rooms, so we only need to manage our
+    // channel subscriptions... but it will be safer if we leave them
+    // explicitly, and then socket.io will just regard this as a client that
+    // has not joined any rooms and do a final disconnection.
+    const roomsToLeave = this._roomsClientIsIn(client)
+    logger.debug({ client: client.id, roomsToLeave }, 'client leaving project')
+    for (const id of roomsToLeave) {
+      const entity = IdMap.get(id)
+      this.leaveEntity(client, entity, id)
+    }
+  },
+
+  emitOnCompletion(promiseList, eventName) {
+    Promise.all(promiseList)
+      .then(() => RoomEvents.emit(eventName))
+      .catch(err => RoomEvents.emit(eventName, err))
+  },
+
+  eventSource() {
+    return RoomEvents
+  },
+
+  joinEntity(client, entity, id, callback) {
+    const beforeCount = this._clientsInRoom(client, id)
+    // client joins room immediately but joinDoc request does not complete
+    // until room is subscribed
+    client.join(id)
+    // is this a new room? if so, subscribe
+    if (beforeCount === 0) {
+      logger.debug({ entity, id }, 'room is now active')
+      RoomEvents.once(`${entity}-subscribed-${id}`, function (err) {
+        // only allow the client to join when all the relevant channels have subscribed
+        if (err) {
+          OError.tag(err, 'error joining', { entity, id })
+          return callback(err)
+        }
+        logger.debug(
+          { client: client.id, entity, id, beforeCount },
+          'client joined new room and subscribed to channel'
+        )
+        callback(err)
+      })
+      RoomEvents.emit(`${entity}-active`, id)
+      IdMap.set(id, entity)
+      // keep track of the number of listeners
+      metrics.gauge('room-listeners', RoomEvents.eventNames().length)
+    } else {
+      logger.debug(
+        { client: client.id, entity, id, beforeCount },
+        'client joined existing room'
+      )
+      callback()
+    }
+  },
+
+  leaveEntity(client, entity, id) {
+    // Ignore any requests to leave when the client is not actually in the
+    // room. This can happen if the client sends spurious leaveDoc requests
+    // for old docs after a reconnection.
+    // This can now happen all the time, as we skip the join for clients that
+    //  disconnect before joinProject/joinDoc completed.
+    if (!this._clientAlreadyInRoom(client, id)) {
+      logger.debug(
+        { client: client.id, entity, id },
+        'ignoring request from client to leave room it is not in'
+      )
+      return
+    }
+    client.leave(id)
+    const afterCount = this._clientsInRoom(client, id)
+    logger.debug(
+      { client: client.id, entity, id, afterCount },
+      'client left room'
+    )
+    // is the room now empty? if so, unsubscribe
+    if (!entity) {
+      logger.error({ entity: id }, 'unknown entity when leaving with id')
+      return
+    }
+    if (afterCount === 0) {
+      logger.debug({ entity, id }, 'room is now empty')
+      RoomEvents.emit(`${entity}-empty`, id)
+      IdMap.delete(id)
+      metrics.gauge('room-listeners', RoomEvents.eventNames().length)
+    }
+  },
+
+  // internal functions below, these access socket.io rooms data directly and
+  // will need updating for socket.io v2
+
+  // The below code makes some assumptions that are always true for v0
+  // - we are using the base namespace '', so room names are '/<ENTITY>'
+  //   https://github.com/socketio/socket.io/blob/e4d61b1be65ac3313a85da111a46777aa8d4aae3/lib/manager.js#L62
+  //   https://github.com/socketio/socket.io/blob/e4d61b1be65ac3313a85da111a46777aa8d4aae3/lib/manager.js#L1018
+  // - client.namespace is a Namespace
+  //   https://github.com/socketio/socket.io/blob/e4d61b1be65ac3313a85da111a46777aa8d4aae3/lib/namespace.js#L204
+  //   https://github.com/socketio/socket.io/blob/e4d61b1be65ac3313a85da111a46777aa8d4aae3/lib/socket.js#L40
+  // - client.manager is a Manager
+  //   https://github.com/socketio/socket.io/blob/e4d61b1be65ac3313a85da111a46777aa8d4aae3/lib/namespace.js#L204
+  //   https://github.com/socketio/socket.io/blob/e4d61b1be65ac3313a85da111a46777aa8d4aae3/lib/socket.js#L41
+  // - a Manager has
+  //   - `.rooms={'NAMESPACE/ENTITY': []}` and
+  //   - `.roomClients={'CLIENT_ID': {'...': true}}`
+  //   https://github.com/socketio/socket.io/blob/e4d61b1be65ac3313a85da111a46777aa8d4aae3/lib/manager.js#L287-L288
+  //   https://github.com/socketio/socket.io/blob/e4d61b1be65ac3313a85da111a46777aa8d4aae3/lib/manager.js#L444-L455
+
+  _clientsInRoom(client, room) {
+    const clients = client.manager.rooms['/' + room] || []
+    return clients.length
+  },
+
+  _roomsClientIsIn(client) {
+    const rooms = client.manager.roomClients[client.id] || {}
+    return (
+      Object.keys(rooms)
+        // drop the namespace
+        .filter(room => room !== '')
+        // room names are composed as '<NAMESPACE>/<ROOM>' and the default
+        //  namespace is empty (see comments above), just drop the '/'
+        .map(fullRoomPath => fullRoomPath.slice(1))
+    )
+  },
+
+  _clientAlreadyInRoom(client, room) {
+    const rooms = client.manager.roomClients[client.id] || {}
+    return !!rooms['/' + room]
+  },
+}

services/real-time/app/js/Router.js

@@ -0,0 +1,607 @@
+const metrics = require('@overleaf/metrics')
+const logger = require('@overleaf/logger')
+const settings = require('@overleaf/settings')
+const WebsocketController = require('./WebsocketController')
+const HttpController = require('./HttpController')
+const HttpApiController = require('./HttpApiController')
+const WebsocketAddressManager = require('./WebsocketAddressManager')
+const bodyParser = require('body-parser')
+const base64id = require('base64id')
+const { UnexpectedArgumentsError } = require('./Errors')
+const Joi = require('joi')
+
+const HOSTNAME = require('node:os').hostname()
+const SERVER_PING_INTERVAL = 15000
+const SERVER_PING_LATENCY_THRESHOLD = 5000
+
+const JOI_OBJECT_ID = Joi.string()
+  .required()
+  .regex(/^[0-9a-f]{24}$/)
+  .message('invalid id')
+
+let Router
+module.exports = Router = {
+  _handleError(callback, error, client, method, attrs) {
+    attrs = attrs || {}
+    for (const key of ['project_id', 'user_id']) {
+      attrs[key] = attrs[key] || client.ol_context[key]
+    }
+    attrs.client_id = client.id
+    attrs.err = error
+    attrs.method = method
+    if (Joi.isError(error)) {
+      logger.info(attrs, 'validation error')
+      let message = 'invalid'
+      try {
+        message = error.details[0].message
+      } catch (e) {
+        // ignore unexpected errors
+        logger.warn({ error, e }, 'unexpected validation error')
+      }
+      const serializedError = { message }
+      metrics.inc('validation-error', 1, {
+        status: method,
+      })
+      callback(serializedError)
+    } else if (error.name === 'CodedError') {
+      logger.warn(attrs, error.message)
+      const serializedError = { message: error.message, code: error.info.code }
+      callback(serializedError)
+    } else if (error.message === 'unexpected arguments') {
+      // the payload might be very large; put it on level debug
+      logger.debug(attrs, 'unexpected arguments')
+      metrics.inc('unexpected-arguments', 1, { status: method })
+      const serializedError = { message: error.message }
+      callback(serializedError)
+    } else if (error.message === 'no project_id found on client') {
+      logger.debug(attrs, error.message)
+      const serializedError = { message: error.message }
+      callback(serializedError)
+    } else if (
+      [
+        'not authorized',
+        'joinLeaveEpoch mismatch',
+        'doc updater could not load requested ops',
+        'no project_id found on client',
+        'cannot join multiple projects',
+      ].includes(error.message)
+    ) {
+      logger.warn(attrs, error.message)
+      const serializedError = { message: error.message }
+      callback(serializedError)
+    } else {
+      logger.error(attrs, `server side error in ${method}`)
+      // Don't return raw error to prevent leaking server side info
+      const serializedError = {
+        message: 'Something went wrong in real-time service',
+      }
+      callback(serializedError)
+    }
+    if (attrs.disconnect) {
+      setTimeout(function () {
+        client.disconnect()
+      }, 100)
+    }
+  },
+
+  _handleInvalidArguments(client, method, args) {
+    const error = new UnexpectedArgumentsError()
+    let callback = args[args.length - 1]
+    if (typeof callback !== 'function') {
+      callback = function () {}
+    }
+    const attrs = { arguments: args }
+    Router._handleError(callback, error, client, method, attrs)
+  },
+
+  configure(app, io, session) {
+    app.set('io', io)
+
+    if (settings.behindProxy) {
+      app.set('trust proxy', settings.trustedProxyIps)
+    }
+    const websocketAddressManager = new WebsocketAddressManager(
+      settings.behindProxy,
+      settings.trustedProxyIps
+    )
+
+    app.get('/clients', HttpController.getConnectedClients)
+    app.get('/clients/:client_id', HttpController.getConnectedClient)
+
+    app.post(
+      '/project/:project_id/message/:message',
+      bodyParser.json({ limit: '5mb' }),
+      HttpApiController.sendMessage
+    )
+
+    app.post('/drain', HttpApiController.startDrain)
+    app.post(
+      '/client/:client_id/disconnect',
+      HttpApiController.disconnectClient
+    )
+
+    session.on('connection', function (error, client, session) {
+      // init client context, we may access it in Router._handleError before
+      //  setting any values
+      client.ol_context = {}
+      // bail out from joinDoc when a parallel joinDoc or leaveDoc is running
+      client.joinLeaveEpoch = 0
+
+      if (client) {
+        client.on('error', function (err) {
+          logger.err(
+            { clientErr: err, publicId: client.publicId, clientId: client.id },
+            'socket.io client error'
+          )
+          if (client.connected) {
+            client.emit('reconnectGracefully')
+            client.disconnect()
+          }
+        })
+      }
+
+      if (settings.shutDownInProgress) {
+        client.emit('connectionRejected', { message: 'retry' })
+        client.disconnect()
+        return
+      }
+
+      if (
+        client &&
+        error &&
+        error.message.match(/could not look up session by key/)
+      ) {
+        logger.warn(
+          { err: error, client: !!client, session: !!session },
+          'invalid session'
+        )
+        // tell the client to reauthenticate if it has an invalid session key
+        client.emit('connectionRejected', { message: 'invalid session' })
+        client.disconnect()
+        return
+      }
+
+      if (error) {
+        logger.err(
+          { err: error, client: !!client, session: !!session },
+          'error when client connected'
+        )
+        if (client) {
+          client.emit('connectionRejected', { message: 'error' })
+        }
+        if (client) {
+          client.disconnect()
+        }
+        return
+      }
+      const useServerPing =
+        !!client.handshake?.query?.esh &&
+        !!client.handshake?.query?.ssp &&
+        // No server ping with long-polling transports.
+        client.transport === 'websocket'
+      const isDebugging = !!client.handshake?.query?.debugging
+      const projectId = client.handshake?.query?.projectId
+
+      if (isDebugging) {
+        client.connectedAt = Date.now()
+        client.isDebugging = true
+      }
+
+      if (!isDebugging) {
+        try {
+          Joi.assert(projectId, JOI_OBJECT_ID)
+        } catch (error) {
+          metrics.inc('socket-io.connection', 1, {
+            status: client.transport,
+            method: projectId ? 'bad-project-id' : 'missing-project-id',
+          })
+          client.emit('connectionRejected', {
+            message: 'missing/bad ?projectId=... query flag on handshake',
+          })
+          client.disconnect()
+          return
+        }
+      }
+
+      // The client.id is security sensitive. Generate a publicId for sending to other clients.
+      client.publicId = 'P.' + base64id.generateId()
+
+      client.remoteIp = websocketAddressManager.getRemoteIp(client.handshake)
+      const headers = client.handshake && client.handshake.headers
+      client.userAgent = headers && headers['user-agent']
+
+      metrics.inc('socket-io.connection', 1, {
+        status: client.transport,
+        method: 'auto-join-project',
+      })
+      metrics.gauge('socket-io.clients', io.sockets.clients().length)
+
+      let user
+      if (session && session.passport && session.passport.user) {
+        ;({ user } = session.passport)
+      } else if (session && session.user) {
+        ;({ user } = session)
+      } else {
+        const anonymousAccessToken = session?.anonTokenAccess?.[projectId]
+        user = { _id: 'anonymous-user', anonymousAccessToken }
+      }
+
+      const info = {
+        userId: user._id,
+        projectId,
+        transport: client.transport,
+        publicId: client.publicId,
+        clientId: client.id,
+        isDebugging,
+      }
+      if (isDebugging) {
+        logger.info(info, 'client connected')
+      } else {
+        logger.debug(info, 'client connected')
+      }
+
+      const connectionDetails = {
+        userId: user._id,
+        projectId,
+        remoteIp: client.remoteIp,
+        publicId: client.publicId,
+        clientId: client.id,
+      }
+
+      let pingTimestamp
+      let pingId = -1
+      let pongId = -1
+      const pingTimer = useServerPing
+        ? setInterval(function () {
+            if (pongId !== pingId) {
+              logger.warn(
+                {
+                  ...connectionDetails,
+                  pingId,
+                  pongId,
+                  lastPingTimestamp: pingTimestamp,
+                },
+                'no client response to last ping'
+              )
+            }
+            pingTimestamp = Date.now()
+            client.emit(
+              'serverPing',
+              ++pingId,
+              pingTimestamp,
+              client.transport,
+              client.id
+            )
+          }, SERVER_PING_INTERVAL)
+        : null
+      client.on(
+        'clientPong',
+        function (
+          receivedPingId,
+          sentTimestamp,
+          serverTransport,
+          serverSessionId,
+          clientTransport,
+          clientSessionId
+        ) {
+          pongId = receivedPingId
+          const receivedTimestamp = Date.now()
+          if (
+            receivedPingId !== pingId ||
+            (serverSessionId && serverSessionId !== clientSessionId)
+          ) {
+            logger.warn(
+              {
+                ...connectionDetails,
+                receivedPingId,
+                pingId,
+                sentTimestamp,
+                receivedTimestamp,
+                latency: receivedTimestamp - sentTimestamp,
+                lastPingTimestamp: pingTimestamp,
+                serverTransport,
+                serverSessionId,
+                clientTransport,
+                clientSessionId,
+              },
+              'received pong with wrong counter'
+            )
+          } else if (
+            receivedTimestamp - sentTimestamp >
+            SERVER_PING_LATENCY_THRESHOLD
+          ) {
+            logger.warn(
+              {
+                ...connectionDetails,
+                receivedPingId,
+                pingId,
+                sentTimestamp,
+                receivedTimestamp,
+                latency: receivedTimestamp - sentTimestamp,
+                lastPingTimestamp: pingTimestamp,
+              },
+              'received pong with high latency'
+            )
+          }
+        }
+      )
+
+      if (settings.exposeHostname) {
+        client.on('debug.getHostname', function (callback) {
+          if (typeof callback !== 'function') {
+            return Router._handleInvalidArguments(
+              client,
+              'debug.getHostname',
+              arguments
+            )
+          }
+          callback(HOSTNAME)
+        })
+      }
+      client.on('debug', (data, callback) => {
+        if (typeof callback !== 'function') {
+          return Router._handleInvalidArguments(client, 'debug', arguments)
+        }
+
+        logger.info(
+          { publicId: client.publicId, clientId: client.id },
+          'received debug message'
+        )
+
+        const response = {
+          serverTime: Date.now(),
+          data,
+          client: {
+            publicId: client.publicId,
+            remoteIp: client.remoteIp,
+            userAgent: client.userAgent,
+            connected: !client.disconnected,
+            connectedAt: client.connectedAt,
+          },
+          server: {
+            hostname: settings.exposeHostname ? HOSTNAME : undefined,
+          },
+        }
+
+        callback(response)
+      })
+      const joinProject = function (callback) {
+        WebsocketController.joinProject(
+          client,
+          user,
+          projectId,
+          function (err, ...args) {
+            if (err) {
+              Router._handleError(callback, err, client, 'joinProject', {
+                project_id: projectId,
+                user_id: user._id,
+              })
+            } else {
+              callback(null, ...args)
+            }
+          }
+        )
+      }
+
+      client.on('disconnect', function () {
+        metrics.inc('socket-io.disconnect', 1, { status: client.transport })
+        metrics.gauge('socket-io.clients', io.sockets.clients().length)
+
+        if (client.isDebugging) {
+          const duration = Date.now() - client.connectedAt
+          metrics.timing('socket-io.debugging.duration', duration)
+          logger.info(
+            { duration, publicId: client.publicId, clientId: client.id },
+            'debug client disconnected'
+          )
+        } else {
+          clearInterval(pingTimer)
+        }
+
+        WebsocketController.leaveProject(io, client, function (err) {
+          if (err) {
+            Router._handleError(function () {}, err, client, 'leaveProject')
+          }
+        })
+      })
+
+      // Variadic. The possible arguments:
+      // doc_id, callback
+      // doc_id, fromVersion, callback
+      // doc_id, options, callback
+      // doc_id, fromVersion, options, callback
+      client.on('joinDoc', function (docId, fromVersion, options, callback) {
+        if (typeof fromVersion === 'function' && !options) {
+          callback = fromVersion
+          fromVersion = -1
+          options = {}
+        } else if (
+          typeof fromVersion === 'number' &&
+          typeof options === 'function'
+        ) {
+          callback = options
+          options = {}
+        } else if (
+          typeof fromVersion === 'object' &&
+          typeof options === 'function'
+        ) {
+          callback = options
+          options = fromVersion
+          fromVersion = -1
+        } else if (
+          typeof fromVersion === 'number' &&
+          typeof options === 'object' &&
+          typeof callback === 'function'
+        ) {
+          // Called with 4 args, things are as expected
+        } else {
+          return Router._handleInvalidArguments(client, 'joinDoc', arguments)
+        }
+        try {
+          Joi.assert(
+            { doc_id: docId, fromVersion, options },
+            Joi.object({
+              doc_id: JOI_OBJECT_ID,
+              fromVersion: Joi.number().integer(),
+              options: Joi.object().required(),
+            })
+          )
+        } catch (error) {
+          return Router._handleError(callback, error, client, 'joinDoc', {
+            disconnect: 1,
+          })
+        }
+        WebsocketController.joinDoc(
+          client,
+          docId,
+          fromVersion,
+          options,
+          function (err, ...args) {
+            if (err) {
+              Router._handleError(callback, err, client, 'joinDoc', {
+                doc_id: docId,
+                fromVersion,
+              })
+            } else {
+              callback(null, ...args)
+            }
+          }
+        )
+      })
+
+      client.on('leaveDoc', function (docId, callback) {
+        if (typeof callback !== 'function') {
+          return Router._handleInvalidArguments(client, 'leaveDoc', arguments)
+        }
+        try {
+          Joi.assert(docId, JOI_OBJECT_ID)
+        } catch (error) {
+          return Router._handleError(callback, error, client, 'joinDoc', {
+            disconnect: 1,
+          })
+        }
+        WebsocketController.leaveDoc(client, docId, function (err, ...args) {
+          if (err) {
+            Router._handleError(callback, err, client, 'leaveDoc', {
+              doc_id: docId,
+            })
+          } else {
+            callback(null, ...args)
+          }
+        })
+      })
+
+      client.on('clientTracking.getConnectedUsers', function (callback) {
+        if (typeof callback !== 'function') {
+          return Router._handleInvalidArguments(
+            client,
+            'clientTracking.getConnectedUsers',
+            arguments
+          )
+        }
+
+        WebsocketController.getConnectedUsers(client, function (err, users) {
+          if (err) {
+            Router._handleError(
+              callback,
+              err,
+              client,
+              'clientTracking.getConnectedUsers'
+            )
+          } else {
+            callback(null, users)
+          }
+        })
+      })
+
+      client.on(
+        'clientTracking.updatePosition',
+        function (cursorData, callback) {
+          if (!callback) {
+            callback = function () {
+              // NOTE: The frontend does not pass any callback to socket.io.
+              // Any error is already logged via Router._handleError.
+            }
+          }
+          if (typeof callback !== 'function') {
+            return Router._handleInvalidArguments(
+              client,
+              'clientTracking.updatePosition',
+              arguments
+            )
+          }
+
+          WebsocketController.updateClientPosition(
+            client,
+            cursorData,
+            function (err) {
+              if (err) {
+                Router._handleError(
+                  callback,
+                  err,
+                  client,
+                  'clientTracking.updatePosition'
+                )
+              } else {
+                callback()
+              }
+            }
+          )
+        }
+      )
+
+      client.on('applyOtUpdate', function (docId, update, callback) {
+        if (typeof callback !== 'function') {
+          return Router._handleInvalidArguments(
+            client,
+            'applyOtUpdate',
+            arguments
+          )
+        }
+        try {
+          Joi.assert(
+            { doc_id: docId, update },
+            Joi.object({
+              doc_id: JOI_OBJECT_ID,
+              update: Joi.object().required(),
+            })
+          )
+        } catch (error) {
+          return Router._handleError(callback, error, client, 'applyOtUpdate', {
+            disconnect: 1,
+          })
+        }
+        WebsocketController.applyOtUpdate(
+          client,
+          docId,
+          update,
+          function (err) {
+            if (err) {
+              Router._handleError(callback, err, client, 'applyOtUpdate', {
+                doc_id: docId,
+              })
+            } else {
+              callback()
+            }
+          }
+        )
+      })
+
+      if (!isDebugging) {
+        joinProject((err, project, permissionsLevel, protocolVersion) => {
+          if (err) {
+            client.emit('connectionRejected', err)
+            client.disconnect()
+            return
+          }
+          client.emit('joinProjectResponse', {
+            publicId: client.publicId,
+            project,
+            permissionsLevel,
+            protocolVersion,
+          })
+        })
+      }
+    })
+  },
+}

services/real-time/app/js/SafeJsonParse.js

@@ -0,0 +1,17 @@
+const Settings = require('@overleaf/settings')
+const { DataTooLargeToParseError } = require('./Errors')
+
+module.exports = {
+  parse(data, callback) {
+    if (data.length > Settings.maxUpdateSize) {
+      return callback(new DataTooLargeToParseError(data))
+    }
+    let parsed
+    try {
+      parsed = JSON.parse(data)
+    } catch (e) {
+      return callback(e)
+    }
+    callback(null, parsed)
+  },
+}

services/real-time/app/js/SessionSockets.js

@@ -0,0 +1,45 @@
+const metrics = require('@overleaf/metrics')
+const OError = require('@overleaf/o-error')
+const { EventEmitter } = require('node:events')
+const { MissingSessionError } = require('./Errors')
+
+module.exports = function (io, sessionStore, cookieParser, cookieName) {
+  const missingSessionError = new MissingSessionError()
+
+  const sessionSockets = new EventEmitter()
+  function next(error, socket, session) {
+    sessionSockets.emit('connection', error, socket, session)
+  }
+
+  io.on('connection', function (socket) {
+    const req = socket.handshake
+    cookieParser(req, {}, function () {
+      const sessionId = req.signedCookies && req.signedCookies[cookieName]
+      if (!sessionId) {
+        metrics.inc('session.cookie', 1, {
+          // the cookie-parser middleware sets the signed cookie to false if the
+          // signature is invalid, so we can use this to detect bad signatures
+          status: sessionId === false ? 'bad-signature' : 'none',
+        })
+        return next(missingSessionError, socket)
+      }
+      sessionStore.get(sessionId, function (error, session) {
+        if (error) {
+          metrics.inc('session.cookie', 1, { status: 'error' })
+          OError.tag(error, 'error getting session from sessionStore', {
+            sessionId,
+          })
+          return next(error, socket)
+        }
+        if (!session) {
+          metrics.inc('session.cookie', 1, { status: 'missing' })
+          return next(missingSessionError, socket)
+        }
+        metrics.inc('session.cookie', 1, { status: 'signed' })
+        next(null, socket, session)
+      })
+    })
+  })
+
+  return sessionSockets
+}

services/real-time/app/js/WebApiManager.js

@@ -0,0 +1,63 @@
+const request = require('request')
+const OError = require('@overleaf/o-error')
+const settings = require('@overleaf/settings')
+const logger = require('@overleaf/logger')
+const {
+  CodedError,
+  CorruptedJoinProjectResponseError,
+  NotAuthorizedError,
+  WebApiRequestFailedError,
+} = require('./Errors')
+
+module.exports = {
+  joinProject(projectId, user, callback) {
+    const userId = user._id
+    logger.debug({ projectId, userId }, 'sending join project request to web')
+    const url = `${settings.apis.web.url}/project/${projectId}/join`
+    request.post(
+      {
+        url,
+        auth: {
+          user: settings.apis.web.user,
+          pass: settings.apis.web.pass,
+          sendImmediately: true,
+        },
+        json: {
+          userId,
+          anonymousAccessToken: user.anonymousAccessToken,
+        },
+        jar: false,
+      },
+      function (error, response, data) {
+        if (error) {
+          OError.tag(error, 'join project request failed')
+          return callback(error)
+        }
+        if (response.statusCode >= 200 && response.statusCode < 300) {
+          if (!(data && data.project)) {
+            return callback(new CorruptedJoinProjectResponseError())
+          }
+          const userMetadata = {
+            isRestrictedUser: data.isRestrictedUser,
+            isTokenMember: data.isTokenMember,
+            isInvitedMember: data.isInvitedMember,
+          }
+          callback(null, data.project, data.privilegeLevel, userMetadata)
+        } else if (response.statusCode === 429) {
+          callback(
+            new CodedError(
+              'rate-limit hit when joining project',
+              'TooManyRequests'
+            )
+          )
+        } else if (response.statusCode === 403) {
+          callback(new NotAuthorizedError())
+        } else if (response.statusCode === 404) {
+          callback(new CodedError('project not found', 'ProjectNotFound'))
+        } else {
+          callback(new WebApiRequestFailedError(response.statusCode))
+        }
+      }
+    )
+  },
+}

services/real-time/app/js/WebsocketAddressManager.js

@@ -0,0 +1,39 @@
+const proxyaddr = require('proxy-addr')
+
+module.exports = class WebsocketAddressManager {
+  constructor(behindProxy, trustedProxyIps) {
+    if (behindProxy) {
+      // parse trustedProxyIps comma-separated list the same way as express
+      this.trust = proxyaddr.compile(
+        trustedProxyIps ? trustedProxyIps.split(/ *, */) : []
+      )
+    }
+  }
+
+  getRemoteIp(clientHandshake) {
+    if (!clientHandshake) {
+      return 'client-handshake-missing'
+    } else if (this.trust) {
+      // create a dummy req object using the client handshake and
+      // connection.remoteAddress for the proxy-addr module to parse
+      try {
+        const addressPort = clientHandshake.address
+        const req = {
+          headers: {
+            'x-forwarded-for':
+              clientHandshake.headers &&
+              clientHandshake.headers['x-forwarded-for'],
+          },
+          connection: { remoteAddress: addressPort && addressPort.address },
+        }
+        // return the address parsed from x-forwarded-for
+        return proxyaddr(req, this.trust)
+      } catch (err) {
+        return 'client-handshake-invalid'
+      }
+    } else {
+      // return the address from the client handshake itself
+      return clientHandshake.address && clientHandshake.address.address
+    }
+  }
+}

services/real-time/app/js/WebsocketController.js

@@ -0,0 +1,657 @@
+const OError = require('@overleaf/o-error')
+const logger = require('@overleaf/logger')
+const metrics = require('@overleaf/metrics')
+const WebApiManager = require('./WebApiManager')
+const AuthorizationManager = require('./AuthorizationManager')
+const DocumentUpdaterManager = require('./DocumentUpdaterManager')
+const ConnectedUsersManager = require('./ConnectedUsersManager')
+const WebsocketLoadBalancer = require('./WebsocketLoadBalancer')
+const RoomManager = require('./RoomManager')
+const {
+  JoinLeaveEpochMismatchError,
+  NotAuthorizedError,
+  NotJoinedError,
+  ClientRequestedMissingOpsError,
+} = require('./Errors')
+
+const JOIN_DOC_CATCH_UP_LENGTH_BUCKETS = [
+  0, 5, 10, 25, 50, 100, 150, 200, 250, 500, 1000,
+]
+const JOIN_DOC_CATCH_UP_AGE = [
+  0,
+  1,
+  2,
+  5,
+  10,
+  20,
+  30,
+  60,
+  120,
+  240,
+  600,
+  60 * 60,
+  24 * 60 * 60,
+].map(x => x * 1000)
+
+let WebsocketController
+module.exports = WebsocketController = {
+  // If the protocol version changes when the client reconnects,
+  // it will force a full refresh of the page. Useful for non-backwards
+  // compatible protocol changes. Use only in extreme need.
+  PROTOCOL_VERSION: 2,
+
+  joinProject(client, user, projectId, callback) {
+    if (client.disconnected) {
+      metrics.inc('editor.join-project.disconnected', 1, {
+        status: 'immediately',
+      })
+      return callback()
+    }
+
+    const userId = user._id
+    logger.info(
+      {
+        userId,
+        projectId,
+        clientId: client.id,
+        remoteIp: client.remoteIp,
+        userAgent: client.userAgent,
+      },
+      'user joining project'
+    )
+    metrics.inc('editor.join-project', 1, { status: client.transport })
+    WebApiManager.joinProject(
+      projectId,
+      user,
+      function (error, project, privilegeLevel, userMetadata) {
+        if (error) {
+          return callback(error)
+        }
+        if (client.disconnected) {
+          logger.info(
+            { userId, projectId, clientId: client.id },
+            'client disconnected before joining project'
+          )
+          metrics.inc('editor.join-project.disconnected', 1, {
+            status: 'after-web-api-call',
+          })
+          return callback()
+        }
+
+        if (!privilegeLevel) {
+          return callback(new NotAuthorizedError())
+        }
+
+        client.ol_context = {}
+        client.ol_context.privilege_level = privilegeLevel
+        client.ol_context.user_id = userId
+        client.ol_context.project_id = projectId
+        client.ol_context.owner_id = project.owner && project.owner._id
+        client.ol_context.first_name = user.first_name
+        client.ol_context.last_name = user.last_name
+        client.ol_context.email = user.email
+        client.ol_context.connected_time = new Date()
+        client.ol_context.signup_date = user.signUpDate
+        client.ol_context.login_count = user.loginCount
+        client.ol_context.is_restricted_user = !!userMetadata.isRestrictedUser
+        client.ol_context.is_token_member = !!userMetadata.isTokenMember
+        client.ol_context.is_invited_member = !!userMetadata.isInvitedMember
+
+        RoomManager.joinProject(client, projectId, function (err) {
+          if (err) {
+            return callback(err)
+          }
+          logger.debug(
+            {
+              userId,
+              projectId,
+              clientId: client.id,
+              privilegeLevel,
+              userMetadata,
+            },
+            'user joined project'
+          )
+          callback(
+            null,
+            project,
+            privilegeLevel,
+            WebsocketController.PROTOCOL_VERSION
+          )
+        })
+
+        // No need to block for setting the user as connected in the cursor tracking
+        ConnectedUsersManager.updateUserPosition(
+          projectId,
+          client.publicId,
+          user,
+          null,
+          function (err) {
+            if (err) {
+              logger.warn(
+                { err, projectId, userId, clientId: client.id },
+                'background cursor update failed'
+              )
+            }
+          }
+        )
+      }
+    )
+  },
+
+  // We want to flush a project if there are no more (local) connected clients
+  // but we need to wait for the triggering client to disconnect. How long we wait
+  // is determined by FLUSH_IF_EMPTY_DELAY.
+  FLUSH_IF_EMPTY_DELAY: 500, // ms
+  leaveProject(io, client, callback) {
+    const { project_id: projectId, user_id: userId } = client.ol_context
+    if (!projectId) {
+      return callback()
+    } // client did not join project
+
+    metrics.inc('editor.leave-project', 1, { status: client.transport })
+    logger.info(
+      { projectId, userId, clientId: client.id },
+      'client leaving project'
+    )
+    WebsocketLoadBalancer.emitToRoom(
+      projectId,
+      'clientTracking.clientDisconnected',
+      client.publicId
+    )
+
+    // We can do this in the background
+    ConnectedUsersManager.markUserAsDisconnected(
+      projectId,
+      client.publicId,
+      function (err) {
+        if (err) {
+          logger.error(
+            { err, projectId, userId, clientId: client.id },
+            'error marking client as disconnected'
+          )
+        }
+      }
+    )
+
+    RoomManager.leaveProjectAndDocs(client)
+    setTimeout(function () {
+      const remainingClients = io.sockets.clients(projectId)
+      if (remainingClients.length === 0) {
+        // Flush project in the background
+        DocumentUpdaterManager.flushProjectToMongoAndDelete(
+          projectId,
+          function (err) {
+            if (err) {
+              logger.error(
+                { err, projectId, userId, clientId: client.id },
+                'error flushing to doc updater after leaving project'
+              )
+            }
+          }
+        )
+      }
+      callback()
+    }, WebsocketController.FLUSH_IF_EMPTY_DELAY)
+  },
+
+  joinDoc(client, docId, fromVersion, options, callback) {
+    if (client.disconnected) {
+      metrics.inc('editor.join-doc.disconnected', 1, { status: 'immediately' })
+      return callback()
+    }
+
+    const joinLeaveEpoch = ++client.joinLeaveEpoch
+    metrics.inc('editor.join-doc', 1, { status: client.transport })
+    const {
+      project_id: projectId,
+      user_id: userId,
+      is_restricted_user: isRestrictedUser,
+    } = client.ol_context
+    if (!projectId) {
+      return callback(new NotJoinedError())
+    }
+    logger.debug(
+      { userId, projectId, docId, fromVersion, clientId: client.id },
+      'client joining doc'
+    )
+
+    const emitJoinDocCatchUpMetrics = (
+      status,
+      { firstVersionInRedis, version, ttlInS }
+    ) => {
+      if (fromVersion === -1) return // full joinDoc call
+      if (typeof options.age !== 'number') return // old frontend
+      if (!ttlInS) return // old document-updater pod
+
+      const isStale = options.age > ttlInS * 1000
+      const method = isStale ? 'stale' : 'recent'
+      metrics.histogram(
+        'join-doc-catch-up-length',
+        version - fromVersion,
+        JOIN_DOC_CATCH_UP_LENGTH_BUCKETS,
+        { status, method, path: client.transport }
+      )
+      if (firstVersionInRedis) {
+        metrics.histogram(
+          'join-doc-catch-up-length-extra-needed',
+          firstVersionInRedis - fromVersion,
+          JOIN_DOC_CATCH_UP_LENGTH_BUCKETS,
+          { status, method, path: client.transport }
+        )
+      }
+      metrics.histogram(
+        'join-doc-catch-up-age',
+        options.age,
+        JOIN_DOC_CATCH_UP_AGE,
+        { status, path: client.transport }
+      )
+    }
+
+    WebsocketController._assertClientAuthorization(
+      client,
+      docId,
+      function (error) {
+        if (error) {
+          return callback(error)
+        }
+        if (client.disconnected) {
+          metrics.inc('editor.join-doc.disconnected', 1, {
+            status: 'after-client-auth-check',
+          })
+          // the client will not read the response anyways
+          return callback()
+        }
+        if (joinLeaveEpoch !== client.joinLeaveEpoch) {
+          // another joinDoc or leaveDoc rpc overtook us
+          return callback(new JoinLeaveEpochMismatchError())
+        }
+        // ensure the per-doc applied-ops channel is subscribed before sending the
+        // doc to the client, so that no events are missed.
+        RoomManager.joinDoc(client, docId, function (error) {
+          if (error) {
+            return callback(error)
+          }
+          if (client.disconnected) {
+            metrics.inc('editor.join-doc.disconnected', 1, {
+              status: 'after-joining-room',
+            })
+            // the client will not read the response anyways
+            return callback()
+          }
+
+          DocumentUpdaterManager.getDocument(
+            projectId,
+            docId,
+            fromVersion,
+            function (error, lines, version, ranges, ops, ttlInS) {
+              if (error) {
+                if (error instanceof ClientRequestedMissingOpsError) {
+                  emitJoinDocCatchUpMetrics('missing', error.info)
+                }
+                return callback(error)
+              }
+              emitJoinDocCatchUpMetrics('success', { version, ttlInS })
+              if (client.disconnected) {
+                metrics.inc('editor.join-doc.disconnected', 1, {
+                  status: 'after-doc-updater-call',
+                })
+                // the client will not read the response anyways
+                return callback()
+              }
+
+              if (isRestrictedUser && ranges && ranges.comments) {
+                ranges.comments = []
+              }
+
+              // Encode any binary bits of data so it can go via WebSockets
+              // See http://ecmanaut.blogspot.co.uk/2006/07/encoding-decoding-utf8-in-javascript.html
+              const encodeForWebsockets = text =>
+                unescape(encodeURIComponent(text))
+              const escapedLines = []
+              for (let line of lines) {
+                try {
+                  line = encodeForWebsockets(line)
+                } catch (err) {
+                  OError.tag(err, 'error encoding line uri component', { line })
+                  return callback(err)
+                }
+                escapedLines.push(line)
+              }
+              if (options.encodeRanges) {
+                try {
+                  for (const comment of (ranges && ranges.comments) || []) {
+                    if (comment.op.c) {
+                      comment.op.c = encodeForWebsockets(comment.op.c)
+                    }
+                  }
+                  for (const change of (ranges && ranges.changes) || []) {
+                    if (change.op.i) {
+                      change.op.i = encodeForWebsockets(change.op.i)
+                    }
+                    if (change.op.d) {
+                      change.op.d = encodeForWebsockets(change.op.d)
+                    }
+                  }
+                } catch (err) {
+                  OError.tag(err, 'error encoding range uri component', {
+                    ranges,
+                  })
+                  return callback(err)
+                }
+              }
+
+              AuthorizationManager.addAccessToDoc(client, docId, () => {})
+              logger.debug(
+                {
+                  userId,
+                  projectId,
+                  docId,
+                  fromVersion,
+                  clientId: client.id,
+                },
+                'client joined doc'
+              )
+              callback(null, escapedLines, version, ops, ranges)
+            }
+          )
+        })
+      }
+    )
+  },
+
+  _assertClientAuthorization(client, docId, callback) {
+    // Check for project-level access first
+    AuthorizationManager.assertClientCanViewProject(client, function (error) {
+      if (error) {
+        return callback(error)
+      }
+      // Check for doc-level access next
+      AuthorizationManager.assertClientCanViewProjectAndDoc(
+        client,
+        docId,
+        function (error) {
+          if (error) {
+            // No cached access, check docupdater
+            const { project_id: projectId } = client.ol_context
+            DocumentUpdaterManager.checkDocument(
+              projectId,
+              docId,
+              function (error) {
+                if (error) {
+                  return callback(error)
+                } else {
+                  // Success
+                  AuthorizationManager.addAccessToDoc(client, docId, callback)
+                }
+              }
+            )
+          } else {
+            // Access already cached
+            callback()
+          }
+        }
+      )
+    })
+  },
+
+  leaveDoc(client, docId, callback) {
+    // client may have disconnected, but we have to cleanup internal state.
+    client.joinLeaveEpoch++
+    metrics.inc('editor.leave-doc', 1, { status: client.transport })
+    const { project_id: projectId, user_id: userId } = client.ol_context
+    logger.debug(
+      { userId, projectId, docId, clientId: client.id },
+      'client leaving doc'
+    )
+    RoomManager.leaveDoc(client, docId)
+    // we could remove permission when user leaves a doc, but because
+    // the connection is per-project, we continue to allow access
+    // after the initial joinDoc since we know they are already authorised.
+    // # AuthorizationManager.removeAccessToDoc client, doc_id
+    callback()
+  },
+  updateClientPosition(client, cursorData, callback) {
+    if (client.disconnected) {
+      // do not create a ghost entry in redis
+      return callback()
+    }
+
+    metrics.inc('editor.update-client-position', 0.1, {
+      status: client.transport,
+    })
+    const {
+      project_id: projectId,
+      first_name: firstName,
+      last_name: lastName,
+      email,
+      user_id: userId,
+    } = client.ol_context
+    logger.debug(
+      { userId, projectId, clientId: client.id, cursorData },
+      'updating client position'
+    )
+
+    AuthorizationManager.assertClientCanViewProjectAndDoc(
+      client,
+      cursorData.doc_id,
+      function (error) {
+        if (error) {
+          logger.debug(
+            { err: error, clientId: client.id, projectId, userId },
+            "silently ignoring unauthorized updateClientPosition. Client likely hasn't called joinProject yet."
+          )
+          return callback()
+        }
+        cursorData.id = client.publicId
+        if (userId) {
+          cursorData.user_id = userId
+        }
+        if (email) {
+          cursorData.email = email
+        }
+        // Don't store anonymous users in redis to avoid influx
+        if (!userId || userId === 'anonymous-user') {
+          cursorData.name = ''
+          // consistent async behaviour
+          setTimeout(callback)
+        } else {
+          cursorData.name =
+            firstName && lastName
+              ? `${firstName} ${lastName}`
+              : firstName || lastName || ''
+          ConnectedUsersManager.updateUserPosition(
+            projectId,
+            client.publicId,
+            {
+              first_name: firstName,
+              last_name: lastName,
+              email,
+              _id: userId,
+            },
+            {
+              row: cursorData.row,
+              column: cursorData.column,
+              doc_id: cursorData.doc_id,
+            },
+            callback
+          )
+        }
+        WebsocketLoadBalancer.emitToRoom(
+          projectId,
+          'clientTracking.clientUpdated',
+          cursorData
+        )
+      }
+    )
+  },
+
+  CLIENT_REFRESH_DELAY: 1000,
+  getConnectedUsers(client, callback) {
+    if (client.disconnected) {
+      // they are not interested anymore, skip the redis lookups
+      return callback()
+    }
+
+    metrics.inc('editor.get-connected-users', { status: client.transport })
+    const {
+      project_id: projectId,
+      user_id: userId,
+      is_restricted_user: isRestrictedUser,
+    } = client.ol_context
+    if (isRestrictedUser) {
+      return callback(null, [])
+    }
+    if (!projectId) {
+      return callback(new NotJoinedError())
+    }
+    logger.debug(
+      { userId, projectId, clientId: client.id },
+      'getting connected users'
+    )
+    AuthorizationManager.assertClientCanViewProject(client, function (error) {
+      if (error) {
+        return callback(error)
+      }
+      WebsocketLoadBalancer.emitToRoom(projectId, 'clientTracking.refresh')
+      setTimeout(
+        () =>
+          ConnectedUsersManager.getConnectedUsers(
+            projectId,
+            function (error, users) {
+              if (error) {
+                return callback(error)
+              }
+              logger.debug(
+                { userId, projectId, clientId: client.id },
+                'got connected users'
+              )
+              callback(null, users)
+            }
+          ),
+        WebsocketController.CLIENT_REFRESH_DELAY
+      )
+    })
+  },
+
+  applyOtUpdate(client, docId, update, callback) {
+    // client may have disconnected, but we can submit their update to doc-updater anyways.
+    const { user_id: userId, project_id: projectId } = client.ol_context
+    if (!projectId) {
+      return callback(new NotJoinedError())
+    }
+
+    WebsocketController._assertClientCanApplyUpdate(
+      client,
+      docId,
+      update,
+      function (error) {
+        if (error) {
+          setTimeout(
+            () =>
+              // Disconnect, but give the client the chance to receive the error
+              client.disconnect(),
+            100
+          )
+          return callback(error)
+        }
+        if (!update.meta) {
+          update.meta = {}
+        }
+        update.meta.source = client.publicId
+        update.meta.user_id = userId
+        update.meta.tsRT = performance.now()
+        metrics.inc('editor.doc-update', 0.3, { status: client.transport })
+
+        logger.debug(
+          {
+            userId,
+            docId,
+            projectId,
+            clientId: client.id,
+            version: update.v,
+          },
+          'sending update to doc updater'
+        )
+
+        DocumentUpdaterManager.queueChange(
+          projectId,
+          docId,
+          update,
+          function (error) {
+            if ((error && error.message) === 'update is too large') {
+              metrics.inc('update_too_large')
+              const { updateSize } = error.info
+              logger.warn(
+                { userId, projectId, docId, updateSize },
+                'update is too large'
+              )
+
+              // mark the update as received -- the client should not send it again!
+              callback()
+
+              // trigger an out-of-sync error
+              const message = {
+                project_id: projectId,
+                doc_id: docId,
+                error: 'update is too large',
+              }
+              setTimeout(function () {
+                if (client.disconnected) {
+                  // skip the message broadcast, the client has moved on
+                  return metrics.inc('editor.doc-update.disconnected', 1, {
+                    status: 'at-otUpdateError',
+                  })
+                }
+                client.emit('otUpdateError', message.error, message)
+                client.disconnect()
+              }, 100)
+              return
+            }
+
+            if (error) {
+              OError.tag(error, 'document was not available for update', {
+                version: update.v,
+              })
+              client.disconnect()
+            }
+            callback(error)
+          }
+        )
+      }
+    )
+  },
+
+  _assertClientCanApplyUpdate(client, docId, update, callback) {
+    if (WebsocketController._isCommentUpdate(update)) {
+      return AuthorizationManager.assertClientCanViewProjectAndDoc(
+        client,
+        docId,
+        callback
+      )
+    } else if (update.meta?.tc) {
+      return AuthorizationManager.assertClientCanReviewProjectAndDoc(
+        client,
+        docId,
+        callback
+      )
+    } else {
+      return AuthorizationManager.assertClientCanEditProjectAndDoc(
+        client,
+        docId,
+        callback
+      )
+    }
+  },
+
+  _isCommentUpdate(update) {
+    if (!(update && update.op instanceof Array)) {
+      return false
+    }
+    for (const op of update.op) {
+      if (!op.c) {
+        return false
+      }
+    }
+    return true
+  },
+}

services/real-time/app/js/WebsocketLoadBalancer.js

@@ -0,0 +1,251 @@
+const Settings = require('@overleaf/settings')
+const logger = require('@overleaf/logger')
+const Metrics = require('@overleaf/metrics')
+const RedisClientManager = require('./RedisClientManager')
+const SafeJsonParse = require('./SafeJsonParse')
+const EventLogger = require('./EventLogger')
+const HealthCheckManager = require('./HealthCheckManager')
+const RoomManager = require('./RoomManager')
+const ChannelManager = require('./ChannelManager')
+const ConnectedUsersManager = require('./ConnectedUsersManager')
+
+const RESTRICTED_USER_MESSAGE_TYPE_PASS_LIST = [
+  'otUpdateApplied',
+  'otUpdateError',
+  'joinDoc',
+  'reciveNewDoc',
+  'reciveNewFile',
+  'reciveNewFolder',
+  'reciveEntityMove',
+  'reciveEntityRename',
+  'removeEntity',
+  'accept-changes',
+  'projectNameUpdated',
+  'rootDocUpdated',
+  'toggle-track-changes',
+  'projectRenamedOrDeletedByExternalSource',
+]
+const BANDWIDTH_BUCKETS = [0]
+// 64 bytes ... 8MB
+for (let i = 5; i <= 22; i++) {
+  BANDWIDTH_BUCKETS.push(2 << i)
+}
+
+let WebsocketLoadBalancer
+module.exports = WebsocketLoadBalancer = {
+  rclientPubList: RedisClientManager.createClientList(Settings.redis.pubsub),
+  rclientSubList: RedisClientManager.createClientList(Settings.redis.pubsub),
+
+  shouldDisconnectClient(client, message) {
+    const userId = client.ol_context.user_id
+    if (message?.message === 'userRemovedFromProject') {
+      if (message?.payload?.includes(userId)) {
+        return true
+      }
+    } else if (message?.message === 'project:publicAccessLevel:changed') {
+      const [info] = message.payload
+      if (
+        info.newAccessLevel === 'private' &&
+        !client.ol_context.is_invited_member
+      ) {
+        return true
+      }
+    } else if (message?.message === 'project:collaboratorAccessLevel:changed') {
+      const changedUserId = message.payload[0].userId
+      return userId === changedUserId
+    }
+    return false
+  },
+
+  emitToRoom(roomId, message, ...payload) {
+    if (!roomId) {
+      logger.warn(
+        { message, payload },
+        'no room_id provided, ignoring emitToRoom'
+      )
+      return
+    }
+    const data = JSON.stringify({
+      room_id: roomId,
+      message,
+      payload,
+    })
+    logger.debug(
+      { roomId, message, payload, length: data.length },
+      'emitting to room'
+    )
+
+    this.rclientPubList.map(rclientPub =>
+      ChannelManager.publish(rclientPub, 'editor-events', roomId, data)
+    )
+  },
+
+  emitToAll(message, ...payload) {
+    this.emitToRoom('all', message, ...payload)
+  },
+
+  listenForEditorEvents(io) {
+    logger.debug(
+      { rclients: this.rclientSubList.length },
+      'listening for editor events'
+    )
+    for (const rclientSub of this.rclientSubList) {
+      rclientSub.subscribe('editor-events')
+      rclientSub.on('message', function (channel, message) {
+        if (Settings.debugEvents > 0) {
+          EventLogger.debugEvent(channel, message)
+        }
+        WebsocketLoadBalancer._processEditorEvent(io, channel, message)
+      })
+    }
+    this.handleRoomUpdates(this.rclientSubList)
+  },
+
+  handleRoomUpdates(rclientSubList) {
+    const roomEvents = RoomManager.eventSource()
+    roomEvents.on('project-active', function (projectId) {
+      const subscribePromises = rclientSubList.map(rclient =>
+        ChannelManager.subscribe(rclient, 'editor-events', projectId)
+      )
+      RoomManager.emitOnCompletion(
+        subscribePromises,
+        `project-subscribed-${projectId}`
+      )
+    })
+    roomEvents.on('project-empty', projectId =>
+      rclientSubList.map(rclient =>
+        ChannelManager.unsubscribe(rclient, 'editor-events', projectId)
+      )
+    )
+  },
+
+  _processEditorEvent(io, channel, message) {
+    SafeJsonParse.parse(message, function (error, message) {
+      if (error) {
+        logger.error({ err: error, channel }, 'error parsing JSON')
+        return
+      }
+      if (message.room_id === 'all') {
+        io.sockets.emit(message.message, ...message.payload)
+      } else if (
+        message.message === 'clientTracking.refresh' &&
+        message.room_id
+      ) {
+        const clientList = io.sockets.clients(message.room_id)
+        logger.debug(
+          {
+            channel,
+            message: message.message,
+            roomId: message.room_id,
+            messageId: message._id,
+            socketIoClients: clientList.map(client => client.id),
+          },
+          'refreshing client list'
+        )
+        for (const client of clientList) {
+          ConnectedUsersManager.refreshClient(message.room_id, client.publicId)
+        }
+      } else if (message.message === 'canary-applied-op') {
+        const { ack, broadcast, source, projectId, docId } = message.payload
+
+        const estimateBandwidth = (room, path) => {
+          const seen = new Set()
+          for (const client of io.sockets.clients(room)) {
+            if (seen.has(client.id)) continue
+            seen.add(client.id)
+            let v = client.id === source ? ack : broadcast
+            if (v === 0) {
+              // Acknowledgements with update.dup===true will not get sent to other clients.
+              continue
+            }
+            v += `5:::{"name":"otUpdateApplied","args":[]}`.length
+            Metrics.histogram(
+              'estimated-applied-ops-bandwidth',
+              v,
+              BANDWIDTH_BUCKETS,
+              { path }
+            )
+          }
+        }
+
+        estimateBandwidth(projectId, 'per-project')
+        estimateBandwidth(docId, 'per-doc')
+      } else if (message.room_id) {
+        if (message._id && Settings.checkEventOrder) {
+          const status = EventLogger.checkEventOrder(
+            'editor-events',
+            message._id,
+            message
+          )
+          if (status === 'duplicate') {
+            return // skip duplicate events
+          }
+        }
+
+        const isRestrictedMessage =
+          !RESTRICTED_USER_MESSAGE_TYPE_PASS_LIST.includes(message.message)
+
+        // send messages only to unique clients (due to duplicate entries in io.sockets.clients)
+        const clientList = io.sockets.clients(message.room_id)
+
+        // avoid unnecessary work if no clients are connected
+        if (clientList.length === 0) {
+          return
+        }
+        logger.debug(
+          {
+            channel,
+            message: message.message,
+            roomId: message.room_id,
+            messageId: message._id,
+            socketIoClients: clientList.map(client => client.id),
+          },
+          'distributing event to clients'
+        )
+        const seen = new Map()
+        for (const client of clientList) {
+          if (!seen.has(client.id)) {
+            seen.set(client.id, true)
+            if (WebsocketLoadBalancer.shouldDisconnectClient(client, message)) {
+              logger.debug(
+                {
+                  message,
+                  userId: client?.ol_context?.user_id,
+                  projectId: client?.ol_context?.project_id,
+                },
+                'disconnecting client'
+              )
+              if (
+                message?.message !== 'project:collaboratorAccessLevel:changed'
+              ) {
+                client.emit('project:access:revoked')
+              }
+              client.disconnect()
+            } else {
+              if (isRestrictedMessage && client.ol_context.is_restricted_user) {
+                // hide restricted message
+                logger.debug(
+                  {
+                    message,
+                    clientId: client.id,
+                    userId: client.ol_context.user_id,
+                    projectId: client.ol_context.project_id,
+                  },
+                  'hiding restricted message from client'
+                )
+              } else {
+                client.emit(message.message, ...message.payload)
+              }
+            }
+          }
+        }
+      } else if (message.health_check) {
+        logger.debug(
+          { message },
+          'got health check message in editor events channel'
+        )
+        HealthCheckManager.check(channel, message.key)
+      }
+    })
+  },
+}