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

/* eslint-disable
    camelcase,
    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
 * DS205: Consider reworking code to avoid use of IIFEs
 * DS207: Consider shorter variations of null checks
 * Full docs: https://github.com/decaffeinate/decaffeinate/blob/master/docs/suggestions.md
 */
let DocumentUpdaterController;
const logger = require("logger-sharelatex");
const settings = require('settings-sharelatex');
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("metrics-sharelatex");

const MESSAGE_SIZE_LOG_LIMIT = 1024 * 1024; // 1Mb

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) {
		let i, rclient;
		logger.log({rclients: this.rclientList.length}, "listening for applied-ops events");
		for (i = 0; i < this.rclientList.length; i++) {
			rclient = this.rclientList[i];
			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); }
				return DocumentUpdaterController._processMessageFromDocumentUpdater(io, channel, message);
			});
		}
		// create metrics for each redis instance only when we have multiple redis clients
		if (this.rclientList.length > 1) {
			for (i = 0; i < this.rclientList.length; i++) {
				rclient = this.rclientList[i];
				((i => // per client event rate metric
                rclient.on("message", () => metrics.inc(`rclient-${i}`, 0.001))))(i);
			}
		}
		return this.handleRoomUpdates(this.rclientList);
	},

	handleRoomUpdates(rclientSubList) {
		const roomEvents = RoomManager.eventSource();
		roomEvents.on('doc-active', function(doc_id) {
			const subscribePromises = Array.from(rclientSubList).map((rclient) =>
				ChannelManager.subscribe(rclient, "applied-ops", doc_id));
			return RoomManager.emitOnCompletion(subscribePromises, `doc-subscribed-${doc_id}`);
		});
		return roomEvents.on('doc-empty', doc_id => Array.from(rclientSubList).map((rclient) =>
            ChannelManager.unsubscribe(rclient, "applied-ops", doc_id)));
	},

	_processMessageFromDocumentUpdater(io, channel, message) {
		return SafeJsonParse.parse(message, function(error, message) {
			if (error != null) {
				logger.error({err: error, channel}, "error parsing JSON");
				return;
			}
			if (message.op != null) {
				if ((message._id != null) && settings.checkEventOrder) {
					const status = EventLogger.checkEventOrder("applied-ops", message._id, message);
					if (status === 'duplicate') {
						return; // skip duplicate events
					}
				}
				return DocumentUpdaterController._applyUpdateFromDocumentUpdater(io, message.doc_id, message.op);
			} else if (message.error != null) {
				return DocumentUpdaterController._processErrorFromDocumentUpdater(io, message.doc_id, message.error, message);
			} else if (message.health_check != null) {
				logger.debug({message}, "got health check message in applied ops channel");
				return HealthCheckManager.check(channel, message.key);
			}
		});
	},

	_applyUpdateFromDocumentUpdater(io, doc_id, update) {
		let client;
		const clientList = io.sockets.clients(doc_id);
		// avoid unnecessary work if no clients are connected
		if (clientList.length === 0) {
			return;
		}
		// send updates to clients
		logger.log({doc_id, version: update.v, source: (update.meta != null ? update.meta.source : undefined), socketIoClients: (((() => {
			const result = [];
			for (client of Array.from(clientList)) { 				result.push(client.id);
			}
			return result;
		})()))}, "distributing updates to clients");
		const seen = {};
		// send messages only to unique clients (due to duplicate entries in io.sockets.clients)
		for (client of Array.from(clientList)) {
			if (!seen[client.id]) {
				seen[client.id] = true;
				if (client.publicId === update.meta.source) {
					logger.log({doc_id, version: update.v, source: (update.meta != null ? update.meta.source : undefined)}, "distributing update to sender");
					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.log({doc_id, version: update.v, source: (update.meta != null ? update.meta.source : undefined), client_id: 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);
			return logger.log({doc_id, socketIoClients: (((() => {
				const result1 = [];
				for (client of Array.from(clientList)) { 					result1.push(client.id);
				}
				return result1;
			})()))}, "discarded duplicate clients");
		}
	},

	_processErrorFromDocumentUpdater(io, doc_id, error, message) {
		return (() => {
			const result = [];
			for (const client of Array.from(io.sockets.clients(doc_id))) {
				logger.warn({err: error, doc_id, client_id: client.id}, "error from document updater, disconnecting client");
				client.emit("otUpdateError", error, message);
				result.push(client.disconnect());
			}
			return result;
		})();
	}
});