Add Presence functionality

This change adds the ability for clients to broadcast information about
"Presence" - the notion of a client's position or state in a particular
document. This might be represent a cursor in a text document, or a
highlighted field in a more complex JSON document, or any other
transient, current information about a client that shouldn't necessarily
be stored in the document's chain of ops.

The main complication that this feature solves is the issue of keeping
presence correctly associated with the version of a `Doc` it was created
at. For example, in a "naive" implementation of presence, presence
information can arrive ahead of or behind ops, which - in a text-based
example - can cause the cursor to "jitter" around the change. Using the
ShareDB implementation will ensure that the presence is correctly
transformed against any ops, and will ensure that presence information
is always consistent with the version of the document. We also locally
transform existing presence, which should help to keep (static) remote
presence correctly positioned, independent of latency.

In order to facilitate this, the feature must be used with an OT type
that supports presence. The only requirement for supporting presence is
the support of a `transformPresence` method:

```javascript
type.transformPresence(presence, op, isOwnOperation): presence;
```

* `presence` _Object_: the presence data being transformed. The type
  will define this shape to be whatever is appropriate for the type.
* `op` _Op_: the operation against which to transform the presence
* `isOwnOperation`: _boolean_: whether the presence and the op have the
  same "owner". This information can be useful for some types to break
  ties when transforming a presence, for example as used in
  [`rich-text`][1]

This work is based on the [work][2] by @gkubisa and @curran, but with
the following aims:

  - avoid modifying the existing `Doc` class as much as possible, and
    instead use lifecycle hooks
  - keep presence separate as its own conceptual entity
  - break the presence subscriptions apart from `Doc` subscriptions
    (although in practice, the two are obviously tightly coupled)
  - allow multiple presences on a single `Doc` on the same `Connection`

[1]: https://github.com/quilljs/delta#tranformposition
[2]: #288

Author: Alec Gibson

README.md

@@ -158,6 +158,7 @@ Register a new middleware.
     the database.
   * `'receive'`: Received a message from a client
   * `'reply'`: About to send a non-error reply to a client message
+  * `'sendPresence'`: About to send presence information to a client
 * `fn` _(Function(context, callback))_
   Call this function at the time specified by `action`.
   * `context` will always have the following properties:
@@ -307,6 +308,20 @@ Get a read-only snapshot of a document at the requested version.
   }
   ```
 
+`connection.getPresence(channel): Presence;`
+Get a [`Presence`](#class-sharedbpresence) instance that can be used to subscribe to presence information to other clients, and create instances of local presence.
+
+* `channel` _(String)_
+  Presence channel to subscribe to
+
+`connection.getDocPresence(collection, id): DocPresence;`
+Get a special [`DocPresence`](#class-sharedbdocpresence) instance that can be used to subscribe to presence information to other clients, and create instances of local presence. This is tied to a `Doc`, and all presence will be automatically transformed against ops to keep presence current. Note that the `Doc` must be of a type that supports presence.
+
+* `collection` _(String)_
+  Document collection
+* `id` _(String)_
+  Document ID
+
 ### Class: `ShareDB.Doc`
 
 `doc.type` _(String_)
@@ -640,6 +655,109 @@ const connectionInfo = getUserPermissions();
 const connection = backend.connect(null, connectionInfo);
 ```
 
+### Class: `ShareDB.Presence`
+
+Representation of the presence data associated with a given channel.
+
+#### `subscribe`
+
+```javascript
+presence.subscribe(callback): void;
+```
+
+Subscribe to presence updates from other clients. Note that presence can be submitted without subscribing, but remote clients will not be able to re-request presence from you if you are not subscribed.
+
+* `callback` _Function_: a callback with the signature `function (error: Error): void;`
+
+#### `unsubscribe`
+
+```javascript
+presence.unsubscribe(callback): void;
+```
+
+Unsubscribe from presence updates from remote clients.
+
+* `callback` _Function_: a callback with the signature `function (error: Error): void;`
+
+#### `on`
+
+```javascript
+presence.on('receive', callback): void;
+```
+
+An update from a remote presence client has been received.
+
+* `callback` _Function_: callback for handling the received presence: `function (presenceId, presenceValue): void;`
+
+```javascript
+presence.on('error', callback): void;
+```
+
+A presence-related error has occurred.
+
+* `callback` _Function_: a callback with the signature `function (error: Error): void;`
+
+#### `create`
+
+```javascript
+presence.create(presenceId): LocalPresence;
+```
+
+Create an instance of [`LocalPresence`](#class-sharedblocalpresence), which can be used to represent local presence. Many or none such local presences may exist on a `Presence` instance.
+
+* `presenceId` _string_: a unique ID representing the local presence. Remember - depending on use-case - the same client might have multiple presences, so this might not necessarily be a user or client ID.
+
+#### `destroy`
+
+```javascript
+presence.destroy(callback);
+```
+
+Updates all remote clients with a `null` presence, and removes it from the `Connection` cache, so that it can be garbage-collected. This should be called when you are done with a presence, and no longer need to use it to fire updates.
+
+* `callback` _Function_: a callback with the signature `function (error: Error): void;`
+
+### Class: `ShareDB.DocPresence`
+
+Specialised case of [`Presence`](#class-sharedbpresence), which is tied to a specific [`Doc`](#class-sharedbdoc). When using presence with an associated `Doc`, any ops applied to the `Doc` will automatically be used to transform associated presence. On destroy, the `DocPresence` will unregister its listeners from the `Doc`.
+
+See [`Presence`](#class-sharedbpresence) for available methods.
+
+### Class: `ShareDB.LocalPresence`
+
+`LocalPresence` represents the presence of the local client in a given `Doc`. For example, this might be the position of a caret in a text document; which field has been highlighted in a complex JSON object; etc. Multiple presences may exist per `Doc` even on the same client.
+
+#### `submit`
+
+```javascript
+localPresence.submit(presence, callback): void;
+```
+
+Update the local representation of presence, and broadcast that presence to any other document presence subscribers.
+
+* `presence` _Object_: the presence object to broadcast. The structure of this will depend on the OT type
+* `callback` _Function_: a callback with the signature `function (error: Error): void;`
+
+#### `send`
+
+```javascript
+localPresence.send(callback): void;
+```
+
+Send presence like `submit`, but without updating the value. Can be useful if local presences expire periodically.
+
+* `callback` _Function_: a callback with the signature `function (error: Error): void;`
+
+#### `destroy`
+
+```javascript
+localPresence.destroy(callback): void;
+```
+
+Informs all remote clients that this presence is now `null`, and deletes itself for garbage collection.
+
+* `callback` _Function_: a callback with the signature `function (error: Error): void;`
+
 ### Logging
 
 By default, ShareDB logs to `console`. This can be overridden if you wish to silence logs, or to log to your own logging driver or alert service.

lib/agent.js

@@ -29,6 +29,19 @@ function Agent(backend, stream) {
   // Map from queryId -> emitter
   this.subscribedQueries = {};
 
+  // Track which documents are subscribed to presence by the client. This is a
+  // map of channel -> stream
+  this.subscribedPresences = {};
+  // Highest seq received for a subscription request. Any seq lower than this
+  // value is stale, and should be ignored. Used for keeping the subscription
+  // state in sync with the client's desired state
+  this.presenceSubscriptionSeq = 0;
+  // Keep track of the last request that has been sent by each local presence
+  // belonging to this agent. This is used to generate a new disconnection
+  // request if the client disconnects ungracefully. This is a
+  // map of channel -> id -> request
+  this.presenceRequests = {};
+
   // We need to track this manually to make sure we don't reply to messages
   // after the stream was closed.
   this.closed = false;
@@ -77,6 +90,11 @@ Agent.prototype._cleanup = function() {
   }
   this.subscribedDocs = {};
 
+  for (var channel in this.subscribedPresences) {
+    this.subscribedPresences[channel].destroy();
+  }
+  this.subscribedPresences = {};
+
   // Clean up query subscription streams
   for (var id in this.subscribedQueries) {
     var emitter = this.subscribedQueries[id];
@@ -120,6 +138,31 @@ Agent.prototype._subscribeToStream = function(collection, id, stream) {
   });
 };
 
+Agent.prototype._subscribeToPresenceStream = function(channel, stream) {
+  if (this.closed) return stream.destroy();
+
+  stream.on('data', function(data) {
+    if (data.error) {
+      logger.error('Presence subscription stream error', channel, data.error);
+    }
+    this._handlePresenceData(data);
+  }.bind(this));
+
+  stream.on('end', function() {
+    var requests = this.presenceRequests[channel] || {};
+    for (var id in requests) {
+      var request = this.presenceRequests[channel][id];
+      request.seq++;
+      request.p = null;
+      this._broadcastPresence(request, function(error) {
+        if (error) logger.error('Error broadcasting disconnect presence', channel, error);
+      });
+    }
+    delete this.subscribedPresences[channel];
+    delete this.presenceRequests[channel];
+  }.bind(this));
+};
+
 Agent.prototype._subscribeToQuery = function(emitter, queryId, collection, query) {
   var previous = this.subscribedQueries[queryId];
   if (previous) previous.destroy();
@@ -310,14 +353,18 @@ Agent.prototype._checkRequest = function(request) {
   if (request.a === 'qf' || request.a === 'qs' || request.a === 'qu') {
     // Query messages need an ID property.
     if (typeof request.id !== 'number') return 'Missing query ID';
-  } else if (request.a === 'op' || request.a === 'f' || request.a === 's' || request.a === 'u') {
+  } else if (request.a === 'op' || request.a === 'f' || request.a === 's' || request.a === 'u' || request.a === 'p') {
     // Doc-based request.
     if (request.c != null && typeof request.c !== 'string') return 'Invalid collection';
     if (request.d != null && typeof request.d !== 'string') return 'Invalid id';
 
-    if (request.a === 'op') {
+    if (request.a === 'op' || request.a === 'p') {
       if (request.v != null && (typeof request.v !== 'number' || request.v < 0)) return 'Invalid version';
     }
+
+    if (request.a === 'p') {
+      if (typeof request.id !== 'string') return 'Missing presence ID';
+    }
   } else if (request.a === 'bf' || request.a === 'bs' || request.a === 'bu') {
     // Bulk request
     if (request.c != null && typeof request.c !== 'string') return 'Invalid collection';
@@ -359,6 +406,19 @@ Agent.prototype._handleMessage = function(request, callback) {
         return this._fetchSnapshot(request.c, request.d, request.v, callback);
       case 'nt':
         return this._fetchSnapshotByTimestamp(request.c, request.d, request.ts, callback);
+      case 'p':
+        var presence = this._createPresence(request);
+        if (presence.t && !util.supportsPresence(types.map[presence.t])) {
+          return callback({
+            code: ERROR_CODE.ERR_TYPE_DOES_NOT_SUPPORT_PRESENCE,
+            message: 'Type does not support presence: ' + presence.t
+          });
+        }
+        return this._broadcastPresence(presence, callback);
+      case 'ps':
+        return this._subscribePresence(request.ch, request.seq, callback);
+      case 'pu':
+        return this._unsubscribePresence(request.ch, request.seq, callback);
       default:
         callback(new ShareDBError(ERROR_CODE.ERR_MESSAGE_BADLY_FORMED, 'Invalid or unknown message'));
     }
@@ -645,6 +705,85 @@ Agent.prototype._fetchSnapshotByTimestamp = function(collection, id, timestamp,
   this.backend.fetchSnapshotByTimestamp(this, collection, id, timestamp, callback);
 };
 
+Agent.prototype._broadcastPresence = function(presence, callback) {
+  var requests = this.presenceRequests[presence.ch] || (this.presenceRequests[presence.ch] = {});
+  var previousRequest = requests[presence.id];
+  if (!previousRequest || previousRequest.seq < presence.seq) {
+    this.presenceRequests[presence.ch][presence.id] = presence;
+  }
+  this.backend.transformPresenceToLatestVersion(this, presence, function(error, presence) {
+    if (error) return callback(error);
+    var channel = this._getPresenceChannel(presence.ch);
+    this.backend.pubsub.publish([channel], presence, function(error) {
+      if (error) return callback(error);
+      callback(null, presence);
+    });
+  }.bind(this));
+};
+
+Agent.prototype._createPresence = function(request) {
+  return {
+    a: 'p',
+    ch: request.ch,
+    src: this.clientId,
+    seq: request.seq,
+    id: request.id,
+    p: request.p,
+    c: request.c,
+    d: request.d,
+    v: request.v,
+    t: request.t
+  };
+};
+
+Agent.prototype._subscribePresence = function(channel, seq, callback) {
+  var presenceChannel = this._getPresenceChannel(channel);
+  this.backend.pubsub.subscribe(presenceChannel, function(error, stream) {
+    if (error) return callback(error);
+    if (seq < this.presenceSubscriptionSeq) return callback(null, {ch: channel, seq: seq});
+    this.presenceSubscriptionSeq = seq;
+    this.subscribedPresences[channel] = stream;
+    this._subscribeToPresenceStream(channel, stream);
+    this._requestPresence(channel, function(error) {
+      callback(error, {ch: channel, seq: seq});
+    });
+  }.bind(this));
+};
+
+Agent.prototype._unsubscribePresence = function(channel, seq, callback) {
+  if (seq < this.presenceSubscriptionSeq) return;
+  this.presenceSubscriptionSeq = seq;
+  var stream = this.subscribedPresences[channel];
+  if (stream) stream.destroy();
+  callback(null, {ch: channel, seq: seq});
+};
+
+Agent.prototype._getPresenceChannel = function(channel) {
+  return '$presence.' + channel;
+};
+
+Agent.prototype._requestPresence = function(channel, callback) {
+  var presenceChannel = this._getPresenceChannel(channel);
+  this.backend.pubsub.publish([presenceChannel], {ch: channel, r: true, src: this.clientId}, callback);
+};
+
+Agent.prototype._handlePresenceData = function(presence) {
+  if (presence.src === this.clientId) return;
+
+  if (presence.r) return this.send({a: 'pr', ch: presence.ch});
+
+  var backend = this.backend;
+  var context = {
+    collection: presence.c,
+    presence: presence
+  };
+  backend.trigger(backend.MIDDLEWARE_ACTIONS.sendPresence, this, context, function(error) {
+    if (error) {
+      return this.send({a: 'p', ch: presence.ch, id: presence.id, error: getReplyErrorObject(error)});
+    }
+    this.send(presence);
+  }.bind(this));
+};
 
 function createClientOp(request, clientId) {
   // src can be provided if it is not the same as the current agent,

lib/backend.js

@@ -66,6 +66,8 @@ Backend.prototype.MIDDLEWARE_ACTIONS = {
   // by design, changing existing reply properties can cause weird bugs, since
   // the rest of ShareDB would be unaware of those changes.
   reply: 'reply',
+  // About to send presence information to a client
+  sendPresence: 'sendPresence',
   // An operation is about to be submitted to the database
   submit: 'submit'
 };
@@ -815,6 +817,22 @@ Backend.prototype._buildSnapshotFromOps = function(id, startingSnapshot, ops, ca
   callback(error, snapshot);
 };
 
+Backend.prototype.transformPresenceToLatestVersion = function(agent, presence, callback) {
+  if (!presence.c || !presence.d) return callback(null, presence);
+  this.getOps(agent, presence.c, presence.d, presence.v, null, function(error, ops) {
+    if (error) return callback(error);
+    for (var i = 0; i < ops.length; i++) {
+      var op = ops[i];
+      var isOwnOp = op.src === presence.src;
+      var transformError = ot.transformPresence(presence, op, isOwnOp);
+      if (transformError) {
+        return callback(transformError);
+      }
+    }
+    callback(null, presence);
+  });
+};
+
 function pluckIds(snapshots) {
   var ids = [];
   for (var i = 0; i < snapshots.length; i++) {