overleaf/libraries/overleaf-editor-core/lib/ot_client.js

'use strict'

const _ = require('lodash')
const BPromise = require('bluebird')

const ChangeNote = require('./change_note')
const ChangeRequest = require('./change_request')
const Chunk = require('./chunk')
const Operation = require('./operation')

/**
 * @class
 * @classdesc
 * Operational Transformation client.
 *
 * See OT.md for explanation.
 */
function OtClient(_projectId, _editor, _blobStore, _socket) {
  const STATE_DISCONNECTED = 0
  const STATE_LOADING = 1
  const STATE_READY = 2
  const STATE_WAITING = 3

  let _version = null
  let _state = STATE_DISCONNECTED
  const _buffer = []
  let _ackVersion = null
  let _outstanding = []
  let _pending = []
  const _waiting = []

  this.connect = function otClientConnect() {
    switch (_state) {
      case STATE_DISCONNECTED:
        _state = STATE_LOADING
        _socket.emit('authenticate', {
          projectId: _projectId,
          token: 'letmein',
        })
        break
      default:
        throw new Error('connect in state ' + _state)
    }
  }

  /**
   * The latest project version number for which the client can construct the
   * project content.
   *
   * @return {number} non-negative
   */
  this.getVersion = function () {
    return _version
  }

  _socket.on('load', function otClientOnLoad(data) {
    switch (_state) {
      case STATE_LOADING: {
        const chunk = Chunk.fromRaw(data)
        const snapshot = chunk.getSnapshot()
        snapshot.applyAll(chunk.getChanges(), { strict: true })
        _version = chunk.getEndVersion()
        // TODO: we can get remote changes here, so it's not correct to wait for
        // the editor to load before transitioning to the READY state
        _editor.load(snapshot).then(function () {
          _state = STATE_READY
        })
        break
      }
      default:
        throw new Error('loaded in state ' + _state)
    }
  })

  //
  // Local Operations
  //

  function sendOutstandingChange() {
    const changeRequest = new ChangeRequest(_version, _outstanding)
    _socket.emit('change', changeRequest.toRaw())
    _state = STATE_WAITING
  }

  function sendLocalOperation(operation) {
    _outstanding.push(operation)
    sendOutstandingChange()
  }

  function queueLocalOperation(operation) {
    _pending.push(operation)
  }

  this.handleLocalOperation = function otClientHandleLocalOperation(operation) {
    switch (_state) {
      case STATE_READY:
        sendLocalOperation(operation)
        break
      case STATE_WAITING:
        queueLocalOperation(operation)
        break
      default:
        throw new Error('local operation in state ' + _state)
    }
  }

  /**
   * A promise that resolves when the project reaches the given version.
   *
   * @param {number} version non-negative
   * @return {Promise}
   */
  this.waitForVersion = function otClientWaitForVersion(version) {
    if (!_waiting[version]) _waiting[version] = []
    return new BPromise(function (resolve, reject) {
      _waiting[version].push(resolve)
    })
  }

  function resolveWaitingPromises() {
    for (const version in _waiting) {
      if (!Object.prototype.hasOwnProperty.call(_waiting, version)) continue
      if (version > _version) continue
      _waiting[version].forEach(function (resolve) {
        resolve()
      })
      delete _waiting[version]
    }
  }

  //
  // Messages from Server
  //

  function advanceIfReady() {
    if (_ackVersion !== null && _version === _ackVersion) {
      _version += 1
      _ackVersion = null
      handleAckReady()
      advanceIfReady()
      return
    }
    const changeNotes = _.remove(_buffer, function (changeNote) {
      return changeNote.getBaseVersion() === _version
    })
    if (changeNotes.length === 1) {
      handleRemoteChangeReady(changeNotes[0].getChange())
      _version += 1
      advanceIfReady()
      return
    }
    if (changeNotes.length !== 0) {
      throw new Error('multiple remote changes in client version ' + _version)
    }
  }

  function bufferRemoteChangeNote(changeNote) {
    const version = changeNote.getBaseVersion()
    if (_.find(_buffer, 'baseVersion', version)) {
      throw new Error('multiple changes in version ' + version)
    }
    if (version === _ackVersion) {
      throw new Error('received change that was acked in ' + _ackVersion)
    }
    _buffer.push(changeNote)
  }

  function handleAckReady() {
    // console.log('handleAckReady')
    if (_outstanding.length === 0) {
      throw new Error('ack complete without outstanding change')
    }
    if (_state !== STATE_WAITING) {
      throw new Error('ack complete in state ' + _state)
    }
    _editor.handleChangeAcknowledged()
    resolveWaitingPromises()
    if (_pending.length > 0) {
      _outstanding = _pending
      _pending = []
      sendOutstandingChange()
    } else {
      _outstanding = []
      _state = STATE_READY
    }
  }

  function handleRemoteChangeReady(change) {
    if (_pending.length > 0) {
      if (_outstanding.length === 0) {
        throw new Error('pending change without outstanding change')
      }
    }

    Operation.transformMultiple(_outstanding, change.getOperations())
    Operation.transformMultiple(_pending, change.getOperations())

    _editor.applyRemoteChange(change)
  }

  _socket.on('ack', function otClientOnAck(data) {
    switch (_state) {
      case STATE_WAITING: {
        const changeNote = ChangeNote.fromRaw(data)
        _ackVersion = changeNote.getBaseVersion()
        advanceIfReady()
        break
      }
      default:
        throw new Error('ack in state ' + _state)
    }
  })

  _socket.on('change', function otClientOnChange(data) {
    switch (_state) {
      case STATE_READY:
      case STATE_WAITING:
        bufferRemoteChangeNote(ChangeNote.fromRaw(data))
        advanceIfReady()
        break
      default:
        throw new Error('remote change in state ' + _state)
    }
  })

  //
  // Connection State
  // TODO: socket.io error handling
  //

  _socket.on('disconnect', function () {
    _state = STATE_DISCONNECTED
    // eslint-disable-next-line no-console
    console.log('disconnected') // TODO: how do we handle disconnect?
  })
}
module.exports = OtClient