/* eslint-disable no-unused-vars, */ // TODO: This file was created by bulk-decaffeinate. // Fix any style issues and re-enable lint. /* * decaffeinate suggestions: * DS101: Remove unnecessary use of Array.from * DS102: Remove unnecessary code created because of implicit returns * DS207: Consider shorter variations of null checks * Full docs: https://github.com/decaffeinate/decaffeinate/blob/master/docs/suggestions.md */ const Settings = require('@overleaf/settings') const { promisifyAll } = require('@overleaf/promise-utils') const rclient = require('@overleaf/redis-wrapper').createClient( Settings.redis.documentupdater ) const pubsubClient = require('@overleaf/redis-wrapper').createClient( Settings.redis.pubsub ) const Keys = Settings.redis.documentupdater.key_schema const logger = require('@overleaf/logger') const os = require('node:os') const crypto = require('node:crypto') const metrics = require('./Metrics') const HOST = os.hostname() const RND = crypto.randomBytes(4).toString('hex') // generate a random key for this process let COUNT = 0 const MAX_OPS_PER_ITERATION = 8 // process a limited number of ops for safety /** * Drain a batch of updates from a pending-updates list. * * The MULTI only operates on the single passed key (which has an id in curly * braces), so all of its operations run on the same node in a cluster * environment. * * @param {string} key - the redis key of the queue * @param {function(Error, Array=): void} callback */ function getPendingUpdatesFromQueue(key, callback) { const multi = rclient.multi() multi.llen(key) multi.lrange(key, 0, MAX_OPS_PER_ITERATION - 1) multi.ltrim(key, MAX_OPS_PER_ITERATION, -1) multi.exec(function (error, replys) { if (error != null) { return callback(error) } const [llen, jsonUpdates, _trimResult] = replys metrics.histogram( 'redis.pendingUpdates.llen', llen, [0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 15, 20, 25, 50, 75, 100], { path: 'project' } ) for (const jsonUpdate of jsonUpdates) { // record metric for each update removed from queue metrics.summary('redis.pendingUpdates', jsonUpdate.length, { status: 'pop', path: 'project', }) } const updates = [] for (const jsonUpdate of jsonUpdates) { let update try { update = JSON.parse(jsonUpdate) } catch (e) { return callback(e) } updates.push(update) } return callback(error, updates) }) } const RealTimeRedisManager = { /** * Drain a batch of updates from a project's per-project queue. Each update * on the per-project queue carries its own `doc` id so the caller knows * which doc to apply it to. * * @param {string} projectId * @param {function(Error, Array=): void} callback */ getPendingProjectUpdates(projectId, callback) { getPendingUpdatesFromQueue( Keys.pendingProjectUpdates({ project_id: projectId }), callback ) }, /** * Get the length of a project's per-project queue. * * @param {string} projectId * @param {function(Error, number=): void} callback */ getProjectUpdatesLength(projectId, callback) { rclient.llen( Keys.pendingProjectUpdates({ project_id: projectId }), callback ) }, /** * Publish an applied op or an error for a doc back to real-time. * * The message is published on the project's editor-events channel, which * real-time subscribes to for the lifetime of every connected project. It is * broadcast to the project room in real-time; clients filter by doc id. * * @param {Object} data * @param {string} data.project_id - routes the message to the project's * editor-events channel and to the project room in real-time * @param {string} data.doc_id - identifies the doc for clients * @param {Object} [data.op] - the applied op * @param {string} [data.error] - the error message when applying failed */ sendData(data) { // create a unique message id using a counter const messageId = `doc:${HOST}:${RND}-${COUNT++}` data._id = messageId // the message name routes the message to the applied-ops handling in // real-time's editor-events processing data.message = data.op ? 'otUpdateApplied' : 'otUpdateError' const blob = JSON.stringify(data) metrics.summary('redis.publish.applied-ops', blob.length, { path: 'editor-events', }) // publish on a per-project channel when configured (needs realtime to be // configured for this too), otherwise on the base editor-events channel. if (Settings.publishOnIndividualChannels) { return pubsubClient.publish(`editor-events:${data.project_id}`, blob) } else { return pubsubClient.publish('editor-events', blob) } }, } module.exports = RealTimeRedisManager module.exports.promises = promisifyAll(RealTimeRedisManager, { without: ['sendData'], })