doc.js

var emitter = require('../emitter');
var ShareDBError = require('../error');
var types = require('../types');

/**
 * A Doc is a client's view on a sharejs document.
 *
 * It is is uniquely identified by its `id` and `collection`.  Documents
 * should not be created directly. Create them with connection.get()
 *
 *
 * Subscriptions
 * -------------
 *
 * We can subscribe a document to stay in sync with the server.
 *   doc.subscribe(function(error) {
 *     doc.subscribed // = true
 *   })
 * The server now sends us all changes concerning this document and these are
 * applied to our data. If the subscription was successful the initial
 * data and version sent by the server are loaded into the document.
 *
 * To stop listening to the changes we call `doc.unsubscribe()`.
 *
 * If we just want to load the data but not stay up-to-date, we call
 *   doc.fetch(function(error) {
 *     doc.data // sent by server
 *   })
 *
 *
 * Presence
 * --------
 *
 * We can associate transient "presence" data with a document, eg caret position, etc.
 * The presence data is synchronized on the best-effort basis between clients subscribed to the same document.
 * Each client has their own presence data which is read-write. Other clients' data is read-only.
 *
 *
 * Events
 * ------
 *
 * You can use doc.on(eventName, callback) to subscribe to the following events:
 * - `before op (op, source)` Fired before a partial operation is applied to the data.
 *   It may be used to read the old data just before applying an operation
 * - `op (op, source)` Fired after every partial operation with this operation as the
 *   first argument
 * - `create (source)` The document was created. That means its type was
 *   set and it has some initial data.
 * - `del (data, source)` Fired after the document is deleted, that is
 *   the data is null. It is passed the data before delteion as an
 *   arguments
 * - `load ()` Fired when a new snapshot is ingested from a fetch, subscribe, or query
 * - `presence ([src])` Fired after the presence data has changed.
 */

module.exports = Doc;
function Doc(connection, collection, id) {
  emitter.EventEmitter.call(this);

  this.connection = connection;

  this.collection = collection;
  this.id = id;

  this.version = null;
  this.type = null;
  this.data = undefined;

  // The current presence data
  // Map of src -> presence data
  // Local src === ''
  this.presence = Object.create(null);
  // The presence objects received from the server
  // Map of src -> presence
  this.receivedPresence = Object.create(null);
  // The minimum amount of time to wait before removing processed presence from this.receivedPresence.
  // The processed presence is removed to avoid leaking memory, in case peers keep connecting and disconnecting a lot.
  // The processed presence is not removed immediately to enable avoiding race conditions, where messages with lower
  // sequence number arrive after messages with higher sequence numbers.
  this.receivedPresenceTimeout = 60000;
  // If set to true, then the next time the local presence is sent,
  // all other clients will be asked to reply with their own presence data.
  this.requestReplyPresence = true;
  // A list of ops sent by the server. These are needed for transforming presence data,
  // if we get that presence data for an older version of the document.
  // The ops are cached for at least 1 minute by default, which should be lots, considering that the presence
  // data is supposed to be synced in real-time.
  this.cachedOps = [];
  this.cachedOpsTimeout = 60000;
  // The sequence number of the inflight presence request.
  this.inflightPresenceSeq = 0;

  // Array of callbacks or nulls as placeholders
  this.inflightFetch = [];
  this.inflightSubscribe = [];
  this.inflightUnsubscribe = [];
  this.inflightPresence = null;
  this.pendingFetch = [];
  this.pendingPresence = null;

  // Whether we think we are subscribed on the server. Synchronously set to
  // false on calls to unsubscribe and disconnect. Should never be true when
  // this.wantSubscribe is false
  this.subscribed = false;
  // Whether to re-establish the subscription on reconnect
  this.wantSubscribe = false;

  // The op that is currently roundtripping to the server, or null.
  //
  // When the connection reconnects, the inflight op is resubmitted.
  //
  // This has the same format as an entry in pendingOps
  this.inflightOp = null;

  // All ops that are waiting for the server to acknowledge this.inflightOp
  // This used to just be a single operation, but creates & deletes can't be
  // composed with regular operations.
  //
  // This is a list of {[create:{...}], [del:true], [op:...], callbacks:[...]}
  this.pendingOps = [];

  // The OT type of this document. An uncreated document has type `null`
  this.type = null;

  // The applyStack enables us to track any ops submitted while we are
  // applying an op incrementally. This value is an array when we are
  // performing an incremental apply and null otherwise. When it is an array,
  // all submitted ops should be pushed onto it. The `_otApply` method will
  // reset it back to null when all incremental apply loops are complete.
  this.applyStack = null;

  // Disable the default behavior of composing submitted ops. This is read at
  // the time of op submit, so it may be toggled on before submitting a
  // specifc op and toggled off afterward
  this.preventCompose = false;
}
emitter.mixin(Doc);

Doc.prototype.destroy = function(callback) {
  var doc = this;
  doc.whenNothingPending(function() {
    if (doc.wantSubscribe) {
      doc.unsubscribe(function(err) {
        if (err) {
          if (callback) callback(err);
          else this.emit('error', err);
          return;
        }
        doc.receivedPresence = Object.create(null);
        doc.cachedOps.length = 0;
        doc.connection._destroyDoc(doc);
        doc.connection.onDocDestroy(doc);
        if (callback) callback();
      });
    } else {
      doc.receivedPresence = Object.create(null);
      doc.cachedOps.length = 0;
      doc.connection._destroyDoc(doc);
      doc.connection.onDocDestroy(doc);
      if (callback) callback();
    }
  });
};


// ****** Manipulating the document data, version and type.

// Set the document's type, and associated properties. Most of the logic in
// this function exists to update the document based on any added & removed API
// methods.
//
// @param newType OT type provided by the ottypes library or its name or uri
Doc.prototype._setType = function(newType) {
  if (typeof newType === 'string') {
    newType = types.map[newType];
  }

  if (newType) {
    this.type = newType;

  } else if (newType === null) {
    this.type = newType;
    // If we removed the type from the object, also remove its data
    this.data = undefined;

  } else {
    var err = new ShareDBError(4008, 'Missing type ' + newType);
    return this.emit('error', err);
  }
};

// Ingest snapshot data. This data must include a version, snapshot and type.
// This is used both to ingest data that was exported with a webpage and data
// that was received from the server during a fetch.
//
// @param snapshot.v    version
// @param snapshot.data
// @param snapshot.type
// @param callback
Doc.prototype.ingestSnapshot = function(snapshot, callback) {
  if (!snapshot) return callback && callback();

  if (typeof snapshot.v !== 'number') {
    var err = new ShareDBError(5008, 'Missing version in ingested snapshot. ' + this.collection + '.' + this.id);
    if (callback) return callback(err);
    return this.emit('error', err);
  }

  // If the doc is already created or there are ops pending, we cannot use the
  // ingested snapshot and need ops in order to update the document
  if (this.type || this.hasWritePending()) {
    // The version should only be null on a created document when it was
    // created locally without fetching
    if (this.version == null) {
      if (this.hasWritePending()) {
        // If we have pending ops and we get a snapshot for a locally created
        // document, we have to wait for the pending ops to complete, because
        // we don't know what version to fetch ops from. It is possible that
        // the snapshot came from our local op, but it is also possible that
        // the doc was created remotely (which would conflict and be an error)
        return callback && this.once('no write pending', callback);
      }
      // Otherwise, we've encounted an error state
      var err = new ShareDBError(5009, 'Cannot ingest snapshot in doc with null version. ' + this.collection + '.' + this.id);
      if (callback) return callback(err);
      return this.emit('error', err);
    }
    // If we got a snapshot for a version further along than the document is
    // currently, issue a fetch to get the latest ops and catch us up
    if (snapshot.v > this.version) return this.fetch(callback);
    return callback && callback();
  }

  // Ignore the snapshot if we are already at a newer version. Under no
  // circumstance should we ever set the current version backward
  if (this.version > snapshot.v) return callback && callback();

  this.version = snapshot.v;
  this.cachedOps.length = 0;
  var type = (snapshot.type === undefined) ? types.defaultType : snapshot.type;
  this._setType(type);
  this.data = (this.type && this.type.deserialize) ?
    this.type.deserialize(snapshot.data) :
    snapshot.data;
  this.connection.onDocLoad(this);
  this.emit('load');
  this._processAllReceivedPresence();
  callback && callback();
};

Doc.prototype.whenNothingPending = function(callback) {
  if (this.hasPending()) {
    this.once('nothing pending', callback);
    return;
  }
  callback();
};

Doc.prototype.hasPending = function() {
  return !!(
    this.inflightOp ||
    this.pendingOps.length ||
    this.inflightFetch.length ||
    this.inflightSubscribe.length ||
    this.inflightUnsubscribe.length ||
    this.pendingFetch.length ||
    this.inflightPresence ||
    this.pendingPresence
  );
};

Doc.prototype.hasWritePending = function() {
  return !!(this.inflightOp || this.pendingOps.length);
};

Doc.prototype._emitNothingPending = function() {
  if (this.hasWritePending()) return;
  this.emit('no write pending');
  if (this.hasPending()) return;
  this.emit('nothing pending');
};

// **** Helpers for network messages

Doc.prototype._emitResponseError = function(err, callback) {
  if (callback) {
    callback(err);
    this._emitNothingPending();
    return;
  }
  this._emitNothingPending();
  this.emit('error', err);
};

Doc.prototype._handleFetch = function(err, snapshot) {
  var callback = this.inflightFetch.shift();
  if (err) return this._emitResponseError(err, callback);
  this.ingestSnapshot(snapshot, callback);
  this._emitNothingPending();
};

Doc.prototype._handleSubscribe = function(err, snapshot) {
  var callback = this.inflightSubscribe.shift();
  if (err) return this._emitResponseError(err, callback);
  // Indicate we are subscribed only if the client still wants to be. In the
  // time since calling subscribe and receiving a response from the server,
  // unsubscribe could have been called and we might already be unsubscribed
  // but not have received the response. Also, because requests from the
  // client are not serialized and may take different async time to process,
  // it is possible that we could hear responses back in a different order
  // from the order originally sent
  if (this.wantSubscribe) this.subscribed = true;
  this.ingestSnapshot(snapshot, callback);
  this._emitNothingPending();
  this.flush();
};

Doc.prototype._handleUnsubscribe = function(err) {
  var callback = this.inflightUnsubscribe.shift();
  if (err) return this._emitResponseError(err, callback);
  if (callback) callback();
  this._emitNothingPending();
};

Doc.prototype._handleOp = function(err, message) {
  if (err) {
    if (this.inflightOp) {
      // The server has rejected submission of the current operation. If we get
      // an error code 4002 "Op submit rejected", this was done intentionally
      // and we should roll back but not return an error to the user.
      if (err.code === 4002) err = null;
      return this._rollback(err);
    }
    return this.emit('error', err);
  }

  if (this.inflightOp &&
      message.src === this.inflightOp.src &&
      message.seq === this.inflightOp.seq) {
    // The op has already been applied locally. Just update the version
    // and pending state appropriately
    this._opAcknowledged(message);
    return;
  }

  if (this.version == null || message.v > this.version) {
    // This will happen in normal operation if we become subscribed to a
    // new document via a query. It can also happen if we get an op for
    // a future version beyond the version we are expecting next. This
    // could happen if the server doesn't publish an op for whatever reason
    // or because of a race condition. In any case, we can send a fetch
    // command to catch back up.
    //
    // Fetch only sends a new fetch command if no fetches are inflight, which
    // will act as a natural debouncing so we don't send multiple fetch
    // requests for many ops received at once.
    this.fetch();
    return;
  }

  if (message.v < this.version) {
    // We can safely ignore the old (duplicate) operation.
    return;
  }

  var serverOp = {
    src: message.src,
    time: Date.now(),
    create: !!message.create,
    op: message.op,
    del: !!message.del
  };

  if (this.inflightOp) {
    var transformErr = transformX(this.inflightOp, message);
    if (transformErr) return this._hardRollback(transformErr);
  }

  for (var i = 0; i < this.pendingOps.length; i++) {
    var transformErr = transformX(this.pendingOps[i], message);
    if (transformErr) return this._hardRollback(transformErr);
  }

  this.version++;
  this._cacheOp(serverOp);
  this._otApply(message);
  this._processAllReceivedPresence();
  return;
};

// Called whenever (you guessed it!) the connection state changes. This will
// happen when we get disconnected & reconnect.
Doc.prototype._onConnectionStateChanged = function() {
  if (this.connection.canSend) {
    this.flush();
    this._resubscribe();
  } else {
    if (this.inflightOp) {
      this.pendingOps.unshift(this.inflightOp);
      this.inflightOp = null;
    }
    this.subscribed = false;
    if (this.inflightFetch.length || this.inflightSubscribe.length) {
      this.pendingFetch = this.pendingFetch.concat(this.inflightFetch, this.inflightSubscribe);
      this.inflightFetch.length = 0;
      this.inflightSubscribe.length = 0;
    }
    if (this.inflightUnsubscribe.length) {
      var callbacks = this.inflightUnsubscribe;
      this.inflightUnsubscribe = [];
      this._pausePresence();
      callEach(callbacks);
    } else {
      this._pausePresence();
    }
  }
};

Doc.prototype._resubscribe = function() {
  var callbacks = this.pendingFetch;
  this.pendingFetch = [];

  if (this.wantSubscribe) {
    if (callbacks.length) {
      this.subscribe(function(err) {
        callEach(callbacks, err);
      });
      return;
    }
    this.subscribe();
    return;
  }

  if (callbacks.length) {
    this.fetch(function(err) {
      callEach(callbacks, err);
    });
  }
};

// Request the current document snapshot or ops that bring us up to date
Doc.prototype.fetch = function(callback) {
  if (this.connection.canSend) {
    var isDuplicate = this.connection.sendFetch(this);
    pushActionCallback(this.inflightFetch, isDuplicate, callback);
    return;
  }
  this.pendingFetch.push(callback);
};

// Fetch the initial document and keep receiving updates
Doc.prototype.subscribe = function(callback) {
  this.wantSubscribe = true;
  if (this.connection.canSend) {
    var isDuplicate = this.connection.sendSubscribe(this);
    pushActionCallback(this.inflightSubscribe, isDuplicate, callback);
    return;
  }
  this.pendingFetch.push(callback);
};

// Unsubscribe. The data will stay around in local memory, but we'll stop
// receiving updates
Doc.prototype.unsubscribe = function(callback) {
  this.wantSubscribe = false;
  // The subscribed state should be conservative in indicating when we are
  // subscribed on the server. We'll actually be unsubscribed some time
  // between sending the message and hearing back, but we cannot know exactly
  // when. Thus, immediately mark us as not subscribed
  this.subscribed = false;
  if (this.connection.canSend) {
    var isDuplicate = this.connection.sendUnsubscribe(this);
    pushActionCallback(this.inflightUnsubscribe, isDuplicate, callback);
    this._pausePresence();
    return;
  }
  this._pausePresence();
  if (callback) process.nextTick(callback);
};

function pushActionCallback(inflight, isDuplicate, callback) {
  if (isDuplicate) {
    var lastCallback = inflight.pop();
    inflight.push(function(err) {
      lastCallback && lastCallback(err);
      callback && callback(err);
    });
  } else {
    inflight.push(callback);
  }
}


// Operations //

// Send the next pending op to the server, if we can.
//
// Only one operation can be in-flight at a time. If an operation is already on
// its way, or we're not currently connected, this method does nothing.
//
// If there are no pending ops, this method sends the pending presence data, if possible.
Doc.prototype.flush = function() {
  if (this.paused) return;

  if (this.connection.canSend && !this.inflightOp && this.pendingOps.length) {
    this._sendOp();
  }

  if (this.subscribed && !this.inflightPresence && this.pendingPresence && !this.hasWritePending()) {
    this.inflightPresence = this.pendingPresence;
    this.inflightPresenceSeq = this.connection.seq;
    this.pendingPresence = null;
    this.connection.sendPresence(this, this.presence[''], this.requestReplyPresence);
    this.requestReplyPresence = false;
  }
};

// Helper function to set op to contain a no-op.
function setNoOp(op) {
  delete op.op;
  delete op.create;
  delete op.del;
}

// Transform server op data by a client op, and vice versa. Ops are edited in place.
function transformX(client, server) {
  // Order of statements in this function matters. Be especially careful if
  // refactoring this function

  // A client delete op should dominate if both the server and the client
  // delete the document. Thus, any ops following the client delete (such as a
  // subsequent create) will be maintained, since the server op is transformed
  // to a no-op
  if (client.del) return setNoOp(server);

  if (server.del) {
    return new ShareDBError(4017, 'Document was deleted');
  }
  if (server.create) {
    return new ShareDBError(4018, 'Document alredy created');
  }

  // Ignore no-op coming from server
  if (!server.op) return;

  // I believe that this should not occur, but check just in case
  if (client.create) {
    return new ShareDBError(4018, 'Document already created');
  }

  // They both edited the document. This is the normal case for this function -
  // as in, most of the time we'll end up down here.
  //
  // You should be wondering why I'm using client.type instead of this.type.
  // The reason is, if we get ops at an old version of the document, this.type
  // might be undefined or a totally different type. By pinning the type to the
  // op data, we make sure the right type has its transform function called.
  if (client.type.transformX) {
    var result = client.type.transformX(client.op, server.op);
    client.op = result[0];
    server.op = result[1];
  } else {
    var clientOp = client.type.transform(client.op, server.op, 'left');
    var serverOp = client.type.transform(server.op, client.op, 'right');
    client.op = clientOp;
    server.op = serverOp;
  }
};

/**
 * Applies the operation to the snapshot
 *
 * If the operation is create or delete it emits `create` or `del`. Then the
 * operation is applied to the snapshot and `op` and `after op` are emitted.
 * If the type supports incremental updates and `this.incremental` is true we
 * fire `op` after every small operation.
 *
 * This is the only function to fire the above mentioned events.
 *
 * @private
 */
Doc.prototype._otApply = function(op, options) {
  var source = options && options.source || false;
  if (op.op) {
    if (!this.type) {
      var err = new ShareDBError(4015, 'Cannot apply op to uncreated document. ' + this.collection + '.' + this.id);
      return this.emit('error', err);
    }
    var undoOp = options && options.undoOp || null;
    var undoable = options && options.undoable || false;
    var fixUp = options && options.fixUp || false;

    // Iteratively apply multi-component remote operations and rollback ops
    // (source === false) for the default JSON0 OT type. It could use
    // type.shatter(), but since this code is so specific to use cases for the
    // JSON0 type and ShareDB explicitly bundles the default type, we might as
    // well write it this way and save needing to iterate through the op
    // components twice.
    //
    // Ideally, we would not need this extra complexity. However, it is
    // helpful for implementing bindings that update DOM nodes and other
    // stateful objects by translating op events directly into corresponding
    // mutations. Such bindings are most easily written as responding to
    // individual op components one at a time in order, and it is important
    // that the snapshot only include updates from the particular op component
    // at the time of emission. Eliminating this would require rethinking how
    // such external bindings are implemented.
    if (!source && this.type === types.defaultType && op.op.length > 1) {
      if (!this.applyStack) this.applyStack = [];
      var stackLength = this.applyStack.length;
      for (var i = 0; i < op.op.length; i++) {
        var component = op.op[i];
        var componentOp = {op: [component]};
        // Transform componentOp against any ops that have been submitted
        // sychronously inside of an op event handler since we began apply of
        // our operation
        for (var j = stackLength; j < this.applyStack.length; j++) {
          var transformErr = transformX(this.applyStack[j], componentOp);
          if (transformErr) return this._hardRollback(transformErr);
        }
        // Apply the individual op component
        this.emit('before op', componentOp.op, source);
        this._applyOp(componentOp, undoOp, source, undoable, fixUp);
        this._transformAllPresence(componentOp);
        this.emit('op', componentOp.op, source);
      }
      // Pop whatever was submitted since we started applying this op
      this._popApplyStack(stackLength);
      return;
    }

    // The 'before op' event enables clients to pull any necessary data out of
    // the snapshot before it gets changed
    this.emit('before op', op.op, source);
    // Apply the operation to the local data, mutating it in place
    this._applyOp(op, undoOp, source, undoable, fixUp);
    this._transformAllPresence(op);
    // Emit an 'op' event once the local data includes the changes from the
    // op. For locally submitted ops, this will be synchronously with
    // submission and before the server or other clients have received the op.
    // For ops from other clients, this will be after the op has been
    // committed to the database and published
    this.emit('op', op.op, source);
    return;
  }

  if (op.create) {
    this._setType(op.create.type);
    this.data = (this.type.deserialize) ?
      (this.type.createDeserialized) ?
        this.type.createDeserialized(op.create.data) :
        this.type.deserialize(this.type.create(op.create.data)) :
      this.type.create(op.create.data);
    this._transformAllPresence(op);
    this.connection.onDocCreate(this);
    this.emit('create', source);
    return;
  }

  if (op.del) {
    var oldData = this.data;
    this._setType(null);
    this._transformAllPresence(op);
    this.connection.onDocDelete(this);
    this.emit('del', oldData, source);
    return;
  }
};

// Applies `op` to `this.data` and updates the undo/redo stacks.
Doc.prototype._applyOp = function(op, undoOp, source, undoable, fixUp) {
  if (undoOp == null && (undoable || fixUp || op.needsUndoOp)) {
    if (this.type.applyAndInvert) {
      var result = this.type.applyAndInvert(this.data, op.op);
      this.data = result[0];
      undoOp = { op: result[1] };
    } else {
      this.data = this.type.apply(this.data, op.op);
      undoOp = { op: this.type.invert(op.op) };
    }
  } else {
    this.data = this.type.apply(this.data, op.op);
  }

  this.connection.onDocOp(this, op, undoOp, source, undoable, fixUp);
};

// ***** Sending operations

// Actually send op to the server.
Doc.prototype._sendOp = function() {
  // Wait until we have a src id from the server
  var src = this.connection.id;
  if (!src) return;

  // When there is no inflightOp, send the first item in pendingOps. If
  // there is inflightOp, try sending it again
  if (!this.inflightOp) {
    // Send first pending op
    this.inflightOp = this.pendingOps.shift();
  }
  var op = this.inflightOp;
  if (!op) {
    var err = new ShareDBError(5010, 'No op to send on call to _sendOp');
    return this.emit('error', err);
  }

  // Track data for retrying ops
  op.sentAt = Date.now();
  op.retries = (op.retries == null) ? 0 : op.retries + 1;

  // The src + seq number is a unique ID representing this operation. This tuple
  // is used on the server to detect when ops have been sent multiple times and
  // on the client to match acknowledgement of an op back to the inflightOp.
  // Note that the src could be different from this.connection.id after a
  // reconnect, since an op may still be pending after the reconnection and
  // this.connection.id will change. In case an op is sent multiple times, we
  // also need to be careful not to override the original seq value.
  if (op.seq == null) op.seq = this.connection.seq++;

  this.connection.sendOp(this, op);

  // src isn't needed on the first try, since the server session will have the
  // same id, but it must be set on the inflightOp in case it is sent again
  // after a reconnect and the connection's id has changed by then
  if (op.src == null) op.src = src;
};


// Queues the operation for submission to the server and applies it locally.
//
// Internal method called to do the actual work for submit(), create() and del().
// @private
//
// @param op
// @param [op.op]
// @param [op.del]
// @param [op.create]
// @param options { source, skipNoop, undoable, undoOp, fixUp }
// @param [callback] called when operation is submitted
Doc.prototype._submit = function(op, options, callback) {
  if (!options) options = {};

  // Locally submitted ops must always have a truthy source
  if (!options.source) options.source = true;

  // The op contains either op, create, delete, or none of the above (a no-op).
  if (op.op) {
    if (!this.type) {
      var err = new ShareDBError(4015, 'Cannot submit op. Document has not been created. ' + this.collection + '.' + this.id);
      if (callback) return callback(err);
      return this.emit('error', err);
    }
    var needsUndoOp = options.undoable || options.fixUp || op.needsUndoOp;
    if (needsUndoOp && !this.type.invert && !this.type.applyAndInvert) {
      var err = new ShareDBError(4028, 'Cannot submit op. OT type does not support invert not applyAndInvert. ' + this.collection + '.' + this.id);
      if (callback) return callback(err);
      return this.emit('error', err);
    }
    // Try to normalize the op. This removes trailing skip:0's and things like that.
    if (this.type.normalize) op.op = this.type.normalize(op.op);
    // Try to skip processing no-ops.
    if (options.skipNoop && this.type.isNoop && this.type.isNoop(op.op)) {
      if (callback) process.nextTick(callback);
      return;
    }
  }

  this._pushOp(op, callback);
  this._otApply(op, options);

  // The call to flush is delayed so if submit() is called multiple times
  // synchronously, all the ops are combined before being sent to the server.
  var doc = this;
  process.nextTick(function() {
    doc.flush();
  });
};

Doc.prototype._pushOp = function(op, callback) {
  if (this.applyStack) {
    // If we are in the process of incrementally applying an operation, don't
    // compose the op and push it onto the applyStack so it can be transformed
    // against other components from the op or ops being applied
    this.applyStack.push(op);
  } else {
    // If the type supports composes, try to compose the operation onto the
    // end of the last pending operation.
    var composed = this._tryCompose(op);
    if (composed) {
      composed.callbacks.push(callback);
      return;
    }
  }
  // Push on to the pendingOps queue of ops to submit if we didn't compose
  op.type = this.type;
  op.callbacks = [callback];
  this.pendingOps.push(op);
};

Doc.prototype._popApplyStack = function(to) {
  if (to > 0) {
    this.applyStack.length = to;
    return;
  }
  // Once we have completed the outermost apply loop, reset to null and no
  // longer add ops to the applyStack as they are submitted
  var op = this.applyStack[0];
  this.applyStack = null;
  if (!op) return;
  // Compose the ops added since the beginning of the apply stack, since we
  // had to skip compose when they were originally pushed
  var i = this.pendingOps.indexOf(op);
  if (i === -1) return;
  var ops = this.pendingOps.splice(i);
  for (var i = 0; i < ops.length; i++) {
    var op = ops[i];
    var composed = this._tryCompose(op);
    if (composed) {
      composed.callbacks = composed.callbacks.concat(op.callbacks);
    } else {
      this.pendingOps.push(op);
    }
  }
};

// Try to compose a submitted op into the last pending op. Returns the
// composed op if it succeeds, undefined otherwise
Doc.prototype._tryCompose = function(op) {
  if (this.preventCompose) return;

  // We can only compose into the last pending op. Inflight ops have already
  // been sent to the server, so we can't modify them
  var last = this.pendingOps[this.pendingOps.length - 1];
  if (!last) return;

  // Compose an op into a create by applying it. This effectively makes the op
  // invisible, as if the document were created including the op originally
  if (last.create && op.op) {
    last.create.data = this.type.apply(last.create.data, op.op);
    return last;
  }

  // Compose two ops into a single op if supported by the type. Types that
  // support compose must be able to compose any two ops together
  if (last.op && op.op && this.type.compose) {
    last.op = this.type.compose(last.op, op.op);
    return last;
  }
};

// *** Client OT entrypoints.

// Submit an operation to the document.
//
// @param component operation handled by the OT type
// @param options.source passed into 'op' event handler
// @param options.skipNoop should processing be skipped entirely, if `component` is a no-op.
// @param options.undoable should the operation be undoable
// @param options.fixUp If true, this operation is meant to fix the current invalid state of the snapshot.
//   It also updates UndoManagers accordingly. This feature requires the OT type to implement `compose`.
// @param [callback] called after operation submitted
//
// @fires before op, op
Doc.prototype.submitOp = function(component, options, callback) {
  if (typeof options === 'function') {
    callback = options;
    options = null;
  }
  var op = {op: component};
  var submitOptions = {
    source: options && options.source,
    skipNoop: options && options.skipNoop,
    undoable: options && options.undoable,
    fixUp: options && options.fixUp
  };
  this._submit(op, submitOptions, callback);
};

// Submits new content for the document.
//
// This function works only if the type supports `diff` or `diffX`.
// It diffs the current and new snapshot to generate an operation,
// which is then submitted as usual.
//
// @param snapshot new snapshot data
// @param options.source passed into 'op' event handler
// @param options.skipNoop should processing be skipped entirely, if the generated operation is a no-op.
// @param options.undoable should the operation be undoable
// @param options.fixUp If true, this operation is meant to fix the current invalid state of the snapshot.
//   It also updates UndoManagers accordingly. This feature requires the OT type to implement `compose`.
// @param options.diffHint a hint passed into diff/diffX
// @param [callback] called after operation submitted

// @fires before op, op
Doc.prototype.submitSnapshot = function(snapshot, options, callback) {
  if (typeof options === 'function') {
    callback = options;
    options = null;
  }
  if (!this.type) {
    var err = new ShareDBError(4015, 'Cannot submit snapshot. Document has not been created. ' + this.collection + '.' + this.id);
    if (callback) return callback(err);
    return this.emit('error', err);
  }
  if (!this.type.diff && !this.type.diffX) {
    var err = new ShareDBError(4027, 'Cannot submit snapshot. Document type does not support diff nor diffX. ' + this.collection + '.' + this.id);
    if (callback) return callback(err);
    return this.emit('error', err);
  }

  var undoable = !!(options && options.undoable);
  var fixUp = options && options.fixUp;
  var diffHint = options && options.diffHint;
  var needsUndoOp = undoable || fixUp;
  var op, undoOp;

  if ((needsUndoOp && this.type.diffX) || !this.type.diff) {
    var diffs = this.type.diffX(this.data, snapshot, diffHint);
    undoOp = { op: diffs[0] };
    op = { op: diffs[1] };
  } else {
    undoOp = null;
    op = { op: this.type.diff(this.data, snapshot, diffHint) };
  }

  var submitOptions = {
    source: options && options.source,
    skipNoop: options && options.skipNoop,
    undoable: undoable,
    undoOp: undoOp,
    fixUp: fixUp
  };
  this._submit(op, submitOptions, callback);
};

// Create the document, which in ShareJS semantics means to set its type. Every
// object implicitly exists in the database but has no data and no type. Create
// sets the type of the object and can optionally set some initial data on the
// object, depending on the type.
//
// @param data  initial
// @param type  OT type
// @param options  {source: ...}
// @param callback  called when operation submitted
Doc.prototype.create = function(data, type, options, callback) {
  if (typeof type === 'function') {
    callback = type;
    options = null;
    type = null;
  } else if (typeof options === 'function') {
    callback = options;
    options = null;
  }
  if (!type) {
    type = types.defaultType.uri;
  }
  if (this.type) {
    var err = new ShareDBError(4016, 'Document already exists');
    if (callback) return callback(err);
    return this.emit('error', err);
  }
  var op = {create: {type: type, data: data}};
  var source = options && options.source;
  this._submit(op, { source: source }, callback);
};

// Delete the document. This creates and submits a delete operation to the
// server. Deleting resets the object's type to null and deletes its data. The
// document still exists, and still has the version it used to have before you
// deleted it (well, old version +1).
//
// @param options  {source: ...}
// @param callback  called when operation submitted
Doc.prototype.del = function(options, callback) {
  if (typeof options === 'function') {
    callback = options;
    options = null;
  }
  if (!this.type) {
    var err = new ShareDBError(4015, 'Document does not exist');
    if (callback) return callback(err);
    return this.emit('error', err);
  }
  var op = {del: true};
  var source = options && options.source;
  this._submit(op, { source: source }, callback);
};


// Stops the document from sending any operations to the server.
Doc.prototype.pause = function() {
  this.paused = true;
};

// Continue sending operations to the server
Doc.prototype.resume = function() {
  this.paused = false;
  this.flush();
};


// *** Receiving operations

// This is called when the server acknowledges an operation from the client.
Doc.prototype._opAcknowledged = function(message) {
  if (this.inflightOp.create) {
    this.version = message.v;
    this.cachedOps.length = 0;

  } else if (message.v !== this.version) {
    // We should already be at the same version, because the server should
    // have sent all the ops that have happened before acknowledging our op
    console.warn('Invalid version from server. Expected: ' + this.version + ' Received: ' + message.v, message);

    // Fetching should get us back to a working document state
    return this.fetch();
  }

  // The op was committed successfully. Increment the version number
  this.version++;
  this._cacheOp({
    src: this.inflightOp.src,
    time: Date.now(),
    create: !!this.inflightOp.create,
    op: this.inflightOp.op,
    del: !!this.inflightOp.del
  });

  this._clearInflightOp();
  this._processAllReceivedPresence();
};

Doc.prototype._rollback = function(err) {
  // The server has rejected submission of the current operation. Invert by
  // just the inflight op if possible. If not possible to invert, cancel all
  // pending ops and fetch the latest from the server to get us back into a
  // working state, then call back
  var op = this.inflightOp;

  if (op.op && op.type.invert) {
    op.op = op.type.invert(op.op);

    // Transform the undo operation by any pending ops.
    for (var i = 0; i < this.pendingOps.length; i++) {
      var transformErr = transformX(this.pendingOps[i], op);
      if (transformErr) return this._hardRollback(transformErr);
    }

    // ... and apply it locally, reverting the changes.
    //
    // This operation is applied to look like it comes from a remote source.
    // I'm still not 100% sure about this functionality, because its really a
    // local op. Basically, the problem is that if the client's op is rejected
    // by the server, the editor window should update to reflect the undo.
    this._otApply(op);

    this._clearInflightOp(err);
    return;
  }

  this._hardRollback(err);
};

Doc.prototype._hardRollback = function(err) {
  var callbacks = [];
  if (this.inflightPresence) {
    callbacks.push.apply(callbacks, this.inflightPresence);
    this.inflightPresence = null;
    this.inflightPresenceSeq = 0;
  }
  if (this.pendingPresence) {
    callbacks.push.apply(callbacks, this.pendingPresence);
    this.pendingPresence = null;
  }
  if (this.inflightOp) {
    callbacks.push.apply(callbacks, this.inflightOp.callbacks);
  }
  for (var i = 0; i < this.pendingOps.length; i++) {
    callbacks.push.apply(callbacks, this.pendingOps[i].callbacks);
  }

  this._setType(null);
  this.version = null;
  this.inflightOp = null;
  this.pendingOps = [];
  this.cachedOps.length = 0;
  this.receivedPresence = Object.create(null);
  this.requestReplyPresence = true;

  var srcList = Object.keys(this.presence);
  var changedSrcList = [];
  for (var i = 0; i < srcList.length; i++) {
    var src = srcList[i];
    if (this._setPresence(src, null)) {
      changedSrcList.push(src);
    }
  }
  this._emitPresence(changedSrcList, false);

  // Fetch the latest from the server to get us back into a working state
  var doc = this;
  this.fetch(function() {
    var called = callEach(callbacks, err);
    if (err && !called) return doc.emit('error', err);
  });
};

Doc.prototype._clearInflightOp = function(err) {
  var callbacks = this.inflightOp && this.inflightOp.callbacks;
  this.inflightOp = null;
  var called = callbacks && callEach(callbacks, err);

  this.flush();
  this._emitNothingPending();

  if (err && !called) return this.emit('error', err);
};

function callEach(callbacks, err) {
  var called = false;
  for (var i = 0; i < callbacks.length; i++) {
    var callback = callbacks[i];
    if (callback) {
      callback(err);
      called = true;
    }
  }
  return called;
}

// *** Presence

Doc.prototype.submitPresence = function (data, callback) {
  if (data != null) {
    if (!this.type) {
      var doc = this;
      return process.nextTick(function() {
        var err = new ShareDBError(4015, 'Cannot submit presence. Document has not been created. ' + doc.collection + '.' + doc.id);
        if (callback) return callback(err);
        doc.emit('error', err);
      });
    }

    if (!this.type.createPresence || !this.type.transformPresence) {
      var doc = this;
      return process.nextTick(function() {
        var err = new ShareDBError(4024, 'Cannot submit presence. Document\'s type does not support presence. ' + doc.collection + '.' + doc.id);
        if (callback) return callback(err);
        doc.emit('error', err);
      });
    }

    data = this.type.createPresence(data);
  }

  if (this._setPresence('', data, true) || this.pendingPresence || this.inflightPresence) {
    if (!this.pendingPresence) {
      this.pendingPresence = [];
    }
    if (callback) {
      this.pendingPresence.push(callback);
    }

  } else if (callback) {
    process.nextTick(callback);
  }

  var doc = this;
  process.nextTick(function() {
    doc.flush();
  });
};

Doc.prototype._handlePresence = function(err, presence) {
  if (!this.subscribed) return;

  var src = presence.src;
  if (!src) {
    // Handle the ACK for the presence data we submitted.
    // this.inflightPresenceSeq would not equal presence.seq after a hard rollback,
    // when all callbacks are flushed with an error.
    if (this.inflightPresenceSeq === presence.seq) {
      var callbacks = this.inflightPresence;
      this.inflightPresence = null;
      this.inflightPresenceSeq = 0;
      var called = callbacks && callEach(callbacks, err);
      if (err && !called) this.emit('error', err);
      this.flush();
      this._emitNothingPending();
    }
    return;
  }

  // This shouldn't happen but check just in case.
  if (err) return this.emit('error', err);

  if (presence.r && !this.pendingPresence) {
    // Another client requested us to share our current presence data
    this.pendingPresence = [];
    this.flush();
  }

  // Ignore older messages which arrived out of order
  if (
    this.receivedPresence[src] && (
      this.receivedPresence[src].seq > presence.seq ||
      (this.receivedPresence[src].seq === presence.seq && presence.v != null)
    )
  ) return;

  this.receivedPresence[src] = presence;

  if (presence.v == null) {
      // null version should happen only when the server automatically sends
      // null presence for an unsubscribed client
      presence.processedAt = Date.now();
      return this._setPresence(src, null, true);
  }

  // Get missing ops first, if necessary
  if (this.version == null || this.version < presence.v) return this.fetch();

  this._processReceivedPresence(src, true);
};

// If emit is true and presence has changed, emits a presence event.
// Returns true, if presence has changed for src. Otherwise false.
Doc.prototype._processReceivedPresence = function(src, emit) {
  if (!src) return false;
  var presence = this.receivedPresence[src];
  if (!presence) return false;

  if (presence.processedAt != null) {
    if (Date.now() >= presence.processedAt + this.receivedPresenceTimeout) {
        // Remove old received and processed presence
        delete this.receivedPresence[src];
    }
    return false;
  }

  if (this.version == null || this.version < presence.v) return false; // keep waiting for the missing snapshot or ops

  if (presence.p == null) {
    // Remove presence data as requested
    presence.processedAt = Date.now();
    return this._setPresence(src, null, emit);
  }

  if (!this.type || !this.type.createPresence || !this.type.transformPresence) {
    // Remove presence data because the document is not created or its type does not support presence
    presence.processedAt = Date.now();
    return this._setPresence(src, null, emit);
  }

  if (this.inflightOp && this.inflightOp.op == null) {
    // Remove presence data because receivedPresence can be transformed only against "op", not "create" nor "del"
    presence.processedAt = Date.now();
    return this._setPresence(src, null, emit);
  }

  for (var i = 0; i < this.pendingOps.length; i++) {
    if (this.pendingOps[i].op == null) {
      // Remove presence data because receivedPresence can be transformed only against "op", not "create" nor "del"
      presence.processedAt = Date.now();
      return this._setPresence(src, null, emit);
    }
  }

  var startIndex = this.cachedOps.length - (this.version - presence.v);
  if (startIndex < 0) {
    // Remove presence data because we can't transform receivedPresence
    presence.processedAt = Date.now();
    return this._setPresence(src, null, emit);
  }

  for (var i = startIndex; i < this.cachedOps.length; i++) {
    if (this.cachedOps[i].op == null) {
      // Remove presence data because receivedPresence can be transformed only against "op", not "create" nor "del"
      presence.processedAt = Date.now();
      return this._setPresence(src, null, emit);
    }
  }

  // Make sure the format of the data is correct
  var data = this.type.createPresence(presence.p);

  // Transform against past ops
  for (var i = startIndex; i < this.cachedOps.length; i++) {
    var op = this.cachedOps[i];
    data = this.type.transformPresence(data, op.op, presence.src === op.src);
  }

  // Transform against pending ops
  if (this.inflightOp) {
    data = this.type.transformPresence(data, this.inflightOp.op, false);
  }

  for (var i = 0; i < this.pendingOps.length; i++) {
    data = this.type.transformPresence(data, this.pendingOps[i].op, false);
  }

  // Set presence data
  presence.processedAt = Date.now();
  return this._setPresence(src, data, emit);
};

Doc.prototype._processAllReceivedPresence = function() {
  var srcList = Object.keys(this.receivedPresence);
  var changedSrcList = [];
  for (var i = 0; i < srcList.length; i++) {
    var src = srcList[i];
    if (this._processReceivedPresence(src)) {
        changedSrcList.push(src);
    }
  }
  this._emitPresence(changedSrcList, true);
};

Doc.prototype._transformPresence = function(src, op) {
  var presenceData = this.presence[src];
  if (op.op != null) {
    var isOwnOperation = src === (op.src || '');
    presenceData = this.type.transformPresence(presenceData, op.op, isOwnOperation);
  } else {
    presenceData = null;
  }
  return this._setPresence(src, presenceData);
};

Doc.prototype._transformAllPresence = function(op) {
  var srcList = Object.keys(this.presence);
  var changedSrcList = [];
  for (var i = 0; i < srcList.length; i++) {
    var src = srcList[i];
    if (this._transformPresence(src, op)) {
      changedSrcList.push(src);
    }
  }
  this._emitPresence(changedSrcList, false);
};

Doc.prototype._pausePresence = function() {
  if (this.inflightPresence) {
    this.pendingPresence =
      this.pendingPresence ?
        this.inflightPresence.concat(this.pendingPresence) :
        this.inflightPresence;
    this.inflightPresence = null;
    this.inflightPresenceSeq = 0;
  } else if (!this.pendingPresence && this.presence[''] != null) {
    this.pendingPresence = [];
  }
  this.receivedPresence = Object.create(null);
  this.requestReplyPresence = true;
  var srcList = Object.keys(this.presence);
  var changedSrcList = [];
  for (var i = 0; i < srcList.length; i++) {
    var src = srcList[i];
    if (src && this._setPresence(src, null)) {
      changedSrcList.push(src);
    }
  }
  this._emitPresence(changedSrcList, false);
};

// If emit is true and presence has changed, emits a presence event.
// Returns true, if presence has changed. Otherwise false.
Doc.prototype._setPresence = function(src, data, emit) {
  if (data == null) {
    if (this.presence[src] == null) return false;
    delete this.presence[src];
  } else {
    var isPresenceEqual =
      this.presence[src] === data ||
      (this.type.comparePresence && this.type.comparePresence(this.presence[src], data));
    if (isPresenceEqual) return false;
    this.presence[src] = data;
  }
  if (emit) this._emitPresence([ src ], true);
  return true;
};

Doc.prototype._emitPresence = function(srcList, submitted) {
  if (srcList && srcList.length > 0) {
    var doc = this;
    process.nextTick(function() {
      doc.emit('presence', srcList, submitted);
    });
  }
};

Doc.prototype._cacheOp = function(op) {
  // Remove the old ops.
  var oldOpTime = Date.now() - this.cachedOpsTimeout;
  var i;
  for (i = 0; i < this.cachedOps.length; i++) {
    if (this.cachedOps[i].time >= oldOpTime) {
      break;
    }
  }
  if (i > 0) {
    this.cachedOps.splice(0, i);
  }

  // Cache the new op.
  this.cachedOps.push(op);
};