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

import logger from '@overleaf/logger'
import settings from '@overleaf/settings'
import EventLogger from './EventLogger.js'
import metrics from '@overleaf/metrics'

let DocumentUpdaterController

export default DocumentUpdaterController = {
  // DocumentUpdaterController broadcasts applied ops (and errors) from
  // document-updater to socket.io clients. The messages arrive on the
  // per-project editor-events channel and are forwarded here by the
  // WebsocketLoadBalancer.

  // Handle an already-parsed message from document-updater (an applied op
  // or an error) and broadcast it to the project room (roomId = project_id).
  handleAppliedOpMessage(io, message, roomId) {
    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,
        roomId,
        message.doc_id,
        message.op
      )
    } else if (message.error) {
      DocumentUpdaterController._processErrorFromDocumentUpdater(
        io,
        roomId,
        message.doc_id,
        message.error,
        message
      )
    }
  },

  _applyUpdateFromDocumentUpdater(io, roomId, docId, update) {
    let client
    const clientList = io.sockets.clients(roomId)
    // 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, roomId, docId, error, message) {
    for (const client of io.sockets.clients(roomId)) {
      // When broadcasting via the project room, only clients that have
      // joined the doc are affected by the error. (Always true for clients
      // in the doc room.)
      if (client.ol_context?.[`doc:${docId}`] !== 'allowed') {
        continue
      }
      logger.warn(
        { err: error, docId, clientId: client.id },
        'error from document updater, disconnecting client'
      )
      client.emit('otUpdateError', error, message)
      client.disconnect()
    }
  },
}