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

/* eslint-disable
    camelcase,
*/
const OError = require('@overleaf/o-error')
const logger = require('logger-sharelatex')
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,
} = require('./Errors')

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, project_id, callback) {
    if (client.disconnected) {
      metrics.inc('editor.join-project.disconnected', 1, {
        status: 'immediately',
      })
      return callback()
    }

    const user_id = user._id
    logger.log(
      { user_id, project_id, client_id: client.id },
      'user joining project'
    )
    metrics.inc('editor.join-project', 1, { status: client.transport })
    WebApiManager.joinProject(
      project_id,
      user,
      function (error, project, privilegeLevel, isRestrictedUser) {
        if (error) {
          return callback(error)
        }
        if (client.disconnected) {
          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 = user_id
        client.ol_context.project_id = project_id
        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 = !!isRestrictedUser

        RoomManager.joinProject(client, project_id, function (err) {
          if (err) {
            return callback(err)
          }
          logger.log(
            { user_id, project_id, client_id: client.id },
            '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(
          project_id,
          client.publicId,
          user,
          null,
          function (err) {
            if (err) {
              logger.warn(
                { err, project_id, user_id, client_id: 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, user_id } = client.ol_context
    if (!project_id) {
      return callback()
    } // client did not join project

    metrics.inc('editor.leave-project', 1, { status: client.transport })
    logger.log(
      { project_id, user_id, client_id: client.id },
      'client leaving project'
    )
    WebsocketLoadBalancer.emitToRoom(
      project_id,
      'clientTracking.clientDisconnected',
      client.publicId
    )

    // We can do this in the background
    ConnectedUsersManager.markUserAsDisconnected(
      project_id,
      client.publicId,
      function (err) {
        if (err) {
          logger.error(
            { err, project_id, user_id, client_id: client.id },
            'error marking client as disconnected'
          )
        }
      }
    )

    RoomManager.leaveProjectAndDocs(client)
    setTimeout(function () {
      const remainingClients = io.sockets.clients(project_id)
      if (remainingClients.length === 0) {
        // Flush project in the background
        DocumentUpdaterManager.flushProjectToMongoAndDelete(
          project_id,
          function (err) {
            if (err) {
              logger.error(
                { err, project_id, user_id, client_id: client.id },
                'error flushing to doc updater after leaving project'
              )
            }
          }
        )
      }
      callback()
    }, WebsocketController.FLUSH_IF_EMPTY_DELAY)
  },

  joinDoc(client, doc_id, 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, user_id, is_restricted_user } = client.ol_context
    if (!project_id) {
      return callback(new NotJoinedError())
    }
    logger.log(
      { user_id, project_id, doc_id, fromVersion, client_id: client.id },
      'client joining doc'
    )

    WebsocketController._assertClientAuthorization(
      client,
      doc_id,
      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, doc_id, 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(
            project_id,
            doc_id,
            fromVersion,
            function (error, lines, version, ranges, ops) {
              if (error) {
                return callback(error)
              }
              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 (is_restricted_user && 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, doc_id, () => {})
              logger.log(
                {
                  user_id,
                  project_id,
                  doc_id,
                  fromVersion,
                  client_id: client.id,
                },
                'client joined doc'
              )
              callback(null, escapedLines, version, ops, ranges)
            }
          )
        })
      }
    )
  },

  _assertClientAuthorization(client, doc_id, 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,
        doc_id,
        function (error) {
          if (error) {
            // No cached access, check docupdater
            const { project_id } = client.ol_context
            DocumentUpdaterManager.checkDocument(
              project_id,
              doc_id,
              function (error) {
                if (error) {
                  return callback(error)
                } else {
                  // Success
                  AuthorizationManager.addAccessToDoc(client, doc_id, callback)
                }
              }
            )
          } else {
            // Access already cached
            callback()
          }
        }
      )
    })
  },

  leaveDoc(client, doc_id, 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, user_id } = client.ol_context
    logger.log(
      { user_id, project_id, doc_id, client_id: client.id },
      'client leaving doc'
    )
    RoomManager.leaveDoc(client, doc_id)
    // 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, first_name, last_name, email, user_id } =
      client.ol_context
    logger.log(
      { user_id, project_id, client_id: client.id, cursorData },
      'updating client position'
    )

    AuthorizationManager.assertClientCanViewProjectAndDoc(
      client,
      cursorData.doc_id,
      function (error) {
        if (error) {
          logger.info(
            { err: error, client_id: client.id, project_id, user_id },
            "silently ignoring unauthorized updateClientPosition. Client likely hasn't called joinProject yet."
          )
          return callback()
        }
        cursorData.id = client.publicId
        if (user_id) {
          cursorData.user_id = user_id
        }
        if (email) {
          cursorData.email = email
        }
        // Don't store anonymous users in redis to avoid influx
        if (!user_id || user_id === 'anonymous-user') {
          cursorData.name = ''
          // consistent async behaviour
          setTimeout(callback)
        } else {
          cursorData.name =
            first_name && last_name
              ? `${first_name} ${last_name}`
              : first_name || last_name || ''
          ConnectedUsersManager.updateUserPosition(
            project_id,
            client.publicId,
            {
              first_name,
              last_name,
              email,
              _id: user_id,
            },
            {
              row: cursorData.row,
              column: cursorData.column,
              doc_id: cursorData.doc_id,
            },
            callback
          )
        }
        WebsocketLoadBalancer.emitToRoom(
          project_id,
          '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, user_id, is_restricted_user } = client.ol_context
    if (is_restricted_user) {
      return callback(null, [])
    }
    if (!project_id) {
      return callback(new NotJoinedError())
    }
    logger.log(
      { user_id, project_id, client_id: client.id },
      'getting connected users'
    )
    AuthorizationManager.assertClientCanViewProject(client, function (error) {
      if (error) {
        return callback(error)
      }
      WebsocketLoadBalancer.emitToRoom(project_id, 'clientTracking.refresh')
      setTimeout(
        () =>
          ConnectedUsersManager.getConnectedUsers(
            project_id,
            function (error, users) {
              if (error) {
                return callback(error)
              }
              logger.log(
                { user_id, project_id, client_id: client.id },
                'got connected users'
              )
              callback(null, users)
            }
          ),
        WebsocketController.CLIENT_REFRESH_DELAY
      )
    })
  },

  applyOtUpdate(client, doc_id, update, callback) {
    // client may have disconnected, but we can submit their update to doc-updater anyways.
    const { user_id, project_id } = client.ol_context
    if (!project_id) {
      return callback(new NotJoinedError())
    }

    WebsocketController._assertClientCanApplyUpdate(
      client,
      doc_id,
      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 = user_id
        metrics.inc('editor.doc-update', 0.3, { status: client.transport })

        logger.log(
          {
            user_id,
            doc_id,
            project_id,
            client_id: client.id,
            version: update.v,
          },
          'sending update to doc updater'
        )

        DocumentUpdaterManager.queueChange(
          project_id,
          doc_id,
          update,
          function (error) {
            if ((error && error.message) === 'update is too large') {
              metrics.inc('update_too_large')
              const { updateSize } = error.info
              logger.warn(
                { user_id, project_id, doc_id, 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,
                doc_id,
                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, doc_id, update, callback) {
    AuthorizationManager.assertClientCanEditProjectAndDoc(
      client,
      doc_id,
      function (error) {
        if (
          error &&
          error.message === 'not authorized' &&
          WebsocketController._isCommentUpdate(update)
        ) {
          // This might be a comment op, which we only need read-only priveleges for
          AuthorizationManager.assertClientCanViewProjectAndDoc(
            client,
            doc_id,
            callback
          )
          return
        }
        callback(error)
      }
    )
  },

  _isCommentUpdate(update) {
    if (!(update && update.op instanceof Array)) {
      return false
    }
    for (const op of update.op) {
      if (!op.c) {
        return false
      }
    }
    return true
  },
}