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

import _ from 'lodash'
import OError from '@overleaf/o-error'
import logger from '@overleaf/logger'
import settings from '@overleaf/settings'
import metrics from '@overleaf/metrics'
import RedisWrapper from '@overleaf/redis-wrapper'
import Errors from './Errors.js'
import {
  fetchJson,
  fetchNothing,
  RequestFailedError,
} from '@overleaf/fetch-utils'
import { callbackify } from 'node:util'

const {
  ClientRequestedMissingOpsError,
  DocumentUpdaterRequestFailedError,
  NullBytesInOpError,
  UpdateTooLargeError,
} = Errors

const rclient = RedisWrapper.createClient(settings.redis.documentupdater)
const Keys = settings.redis.documentupdater.key_schema

async function getDocument(projectId, docId, fromVersion) {
  const timer = new metrics.Timer('get-document')
  const url = `${settings.apis.documentupdater.url}/project/${projectId}/doc/${docId}?fromVersion=${fromVersion}&historyOTSupport=true`
  logger.debug(
    { projectId, docId, fromVersion },
    'getting doc from document updater'
  )
  try {
    const body = await fetchJson(url)
    timer.done()
    logger.debug({ projectId, docId }, 'got doc from document document updater')
    return {
      lines: body?.lines,
      version: body?.version,
      ranges: body?.ranges,
      ops: body?.ops,
      ttlInS: body?.ttlInS,
      type: body?.type,
    }
  } catch (err) {
    timer.done()
    if (err instanceof RequestFailedError) {
      const { response, body } = err
      let parsedErrBody = null
      try {
        parsedErrBody = JSON.parse(body)
      } catch (error) {
        // ignore parse error
      }
      if (response.status === 422 && parsedErrBody?.firstVersionInRedis) {
        throw new ClientRequestedMissingOpsError(422, parsedErrBody)
      } else if ([404, 422].includes(response.status)) {
        throw new ClientRequestedMissingOpsError(response.status)
      } else {
        throw new DocumentUpdaterRequestFailedError(
          'getDocument',
          response.status
        )
      }
    }
    OError.tag(err, 'error getting doc from doc updater')
    throw err
  }
}

async function checkDocument(projectId, docId) {
  // in this call fromVersion = -1 means get document without docOps
  return await getDocument(projectId, docId, -1)
}

async function flushProjectToMongoAndDelete(projectId) {
  // 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' : '')

  try {
    await fetchNothing(url, { method: 'DELETE' })
    logger.debug({ projectId }, 'deleted project from document updater')
    timer.done()
  } catch (err) {
    timer.done()
    if (err instanceof RequestFailedError) {
      throw new DocumentUpdaterRequestFailedError(
        'flushProjectToMongoAndDelete',
        err.response.status
      )
    }
    OError.tag(err, 'error deleting project from document updater')
    throw err
  }
}

function _getPendingUpdateListKey() {
  const shard = _.random(0, settings.pendingUpdateListShardCount - 1)
  if (shard === 0) {
    return 'pending-updates-list'
  } else {
    return `pending-updates-list-${shard}`
  }
}

async function queueChange(projectId, docId, change) {
  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
    throw new NullBytesInOpError(jsonChange)
  }

  const updateSize = jsonChange.length
  if (updateSize > settings.maxUpdateSize) {
    throw 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.
  try {
    await rclient.rpush(Keys.pendingUpdates({ doc_id: docId }), jsonChange)
  } catch (error) {
    throw new OError('error pushing update into redis').withCause(error)
  }

  const queueKey = _getPendingUpdateListKey()
  try {
    await rclient.rpush(queueKey, docKey)
  } catch (error) {
    throw new OError('error pushing doc_id into redis')
      .withInfo({ queueKey })
      .withCause(error)
  }
}

export default {
  getDocument: callbackify(getDocument),
  checkDocument: callbackify(checkDocument),
  flushProjectToMongoAndDelete: callbackify(flushProjectToMongoAndDelete),
  _getPendingUpdateListKey,
  queueChange: callbackify(queueChange),
  promises: {
    getDocument,
    checkDocument,
    flushProjectToMongoAndDelete,
    queueChange,
  },
}