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

@@ -156,6 +156,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:
@@ -303,6 +304,16 @@ Get a read-only snapshot of a document at the requested version.
   }
   ```
 
+`connection.getPresence(collection, id, presenceId): LocalPresence;`
+Get a [`LocalPresence`](#class-sharedblocalpresence) instance that can be used to send presence information to other clients. Note that the `Doc` must use a type that supports presence.
+
+* `collection` _(String)_
+  Collection name of the `Doc`
+* `id` _(String)_
+  ID of the `Doc`
+* `presenceId` _(String)_
+  A unique presence ID that will be sent to remote clients along with the presence data
+
 ### Class: `ShareDB.Doc`
 
 `doc.type` _(String_)
@@ -349,6 +360,9 @@ The document was deleted. Document contents before deletion are passed in as an
 `doc.on('error', function(err) {...})`
 There was an error fetching the document or applying an operation.
 
+`doc.on('presence', function(id, presence) {...})`
+A remote client has sent presence information. `id` is an ID provided by the remote client, and `presence` is the presence data, whose structure will depend on document's OT type.
+
 `doc.removeListener(eventName, listener)`
 Removes any listener you added with `doc.on`. `eventName` should be one of `'load'`, `'create'`, `'before op'`, `'op'`, `'del'`, or `'error'`. `listener` should be the function you passed in as the second argument to `on`. Note that both `on` and `removeListener` are inherited from [EventEmitter](https://nodejs.org/api/events.html#events_class_eventemitter).
 
@@ -379,6 +393,12 @@ Invokes the given callback function after
 
 Note that `whenNothingPending` does NOT wait for pending `model.query()` calls.
 
+`doc.subscribeToPresence([function(err) {...}])`
+Subscribes to presence updates sent by other clients, emitting `presence` events (see above).
+
+`doc.unsubscribeFromPresence(function(err) {...})`
+Unsubscribe from presence updates sent by other clients, and stop receiving `presence` events (see above).
+
 ### Class: `ShareDB.Query`
 
 `query.ready` _(Boolean)_
@@ -629,6 +649,49 @@ var connectionInfo = getUserPermissions();
 var connection = backend.connect(null, connectionInfo);
 ```
 
+### 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.
+
+#### `update`
+
+```javascript
+localPresence.update(presence, options, callback);
+```
+
+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
+* `options` _Object (optional)_
+* `options.subscribe` _boolean (default `true`)_: if set to `true`, will subscribe to presence updates on the document. Setting to `false` will unsubscribe.
+* `callback` _Function_: a callback with the signature `function (error: Error): void;`
+
+#### `clear`
+
+```javascript
+localPresence.clear(callback);
+```
+
+Tell remote subscribers that this presence is no longer present in the `Doc`. For example, this might prompt remote clients to remove a cursor from a text document, or to clear highlighting around a selected field, etc.
+
+The `LocalPresence` object will not be destroyed (see below), and can still be used to send further updates.
+
+This is shorthand for `localPresence.update(null, callback)`.
+
+* `callback` _Function_: a callback with the signature `function (error: Error): void;`
+
+#### `destroy`
+
+```javascript
+localPresence.destroy(callback);
+```
+
+Updates all remote clients with a `null` presence. Then destroys the local instance of the presence by de-registering all the `Doc` hooks it listens to, 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.
+
+This method is automatically called when calling `doc.destroy`.
+
+* `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

@@ -26,6 +26,10 @@ 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 collection -> id -> stream
+  this.subscribedPresences = {};
+
   // We need to track this manually to make sure we don't reply to messages
   // after the stream was closed.
   this.closed = false;
@@ -74,6 +78,15 @@ Agent.prototype._cleanup = function() {
   }
   this.subscribedDocs = {};
 
+  for (var collection in this.subscribedPresences) {
+    var streams = this.subscribedPresences[collection];
+    for (var id in streams) {
+      var stream = streams[id];
+      stream.destroy();
+    }
+  }
+  this.subscribedPresences = {};
+
   // Clean up query subscription streams
   for (var id in this.subscribedQueries) {
     var emitter = this.subscribedQueries[id];
@@ -117,6 +130,25 @@ Agent.prototype._subscribeToStream = function(collection, id, stream) {
   });
 };
 
+Agent.prototype._subscribeToPresenceStream = function(collection, id, stream) {
+  if (this.closed) return stream.destroy();
+
+  stream.on('data', function(data) {
+    if (data.error) {
+      logger.error('Presence subscription stream error', collection, id, data.error);
+    }
+    this._handlePresenceData(data);
+  }.bind(this));
+
+  stream.on('end', function() {
+    var streams = this.subscribedPresences[collection];
+    if (!streams || !streams[id] !== stream) return;
+    delete streams[id];
+    if (util.hasKeys(streams)) return;
+    delete agent.subscribedPresences[collection];
+  }.bind(this));
+};
+
 Agent.prototype._subscribeToQuery = function(emitter, queryId, collection, query) {
   var previous = this.subscribedQueries[queryId];
   if (previous) previous.destroy();
@@ -311,14 +343,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';
@@ -360,6 +396,13 @@ 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 (!util.supportsPresence(types.map[presence.t])) {
+          return callback({code: 9999, message: 'Type does not support presence: ' + presence.t});
+        }
+        if (!presence) return callback({code: 4000, message: 'Invalid presence message'});
+        return this._broadcastPresence(request.c, request.d, presence, callback);
       default:
         callback({code: 4000, message: 'Invalid or unknown message'});
     }
@@ -628,6 +671,77 @@ Agent.prototype._fetchSnapshotByTimestamp = function(collection, id, timestamp,
   this.backend.fetchSnapshotByTimestamp(this, collection, id, timestamp, callback);
 };
 
+Agent.prototype._broadcastPresence = function(collection, id, presence, callback) {
+  var wantsSubscribe = presence.s;
+  this._handlePresenceSubscription(collection, id, wantsSubscribe, function(error) {
+    if (error) return callback(error);
+    var channel = this.backend.getPresenceChannel(collection, id);
+    this.backend.pubsub.publish([channel], presence, function(error) {
+      if (error) return callback(error);
+      callback(null, presence);
+    });
+  }.bind(this));
+};
+
+Agent.prototype._handlePresenceSubscription = function(collection, id, wantsSubscribe, callback) {
+  var streams = this.subscribedPresences[collection] || (this.subscribedPresences[collection] = {});
+  var stream = streams[id];
+
+  if (stream) {
+    if (wantsSubscribe) {
+      return callback();
+    }
+    stream.destroy();
+    return callback();
+  }
+
+  if (!wantsSubscribe) return callback();
+
+  var channel = this.backend.getPresenceChannel(collection, id);
+  this.backend.pubsub.subscribe(channel, function(error, stream) {
+    if (error) return callback(error);
+    streams[id] = stream;
+    this._subscribeToPresenceStream(collection, id, stream);
+    callback();
+  }.bind(this));
+};
+
+Agent.prototype._createPresence = function(request) {
+  // src can be provided if it is not the same as the current agent,
+  // such as a resubmission after a reconnect, but it usually isn't needed
+  var src = request.src || this.clientId;
+  return {
+    a: 'p',
+    src: src,
+    seq: request.seq,
+    c: request.c,
+    d: request.d,
+    id: request.id,
+    v: request.v,
+    p: request.p,
+    t: request.t,
+    r: !!request.r,
+    s: !!request.s
+  };
+};
+
+Agent.prototype._handlePresenceData = function(data) {
+  if (data.src !== this.clientId) {
+    var backend = this.backend;
+    var agent = this;
+    backend.transformPresenceToLatestVersion(agent, data, function(error, presence) {
+      if (error) return agent.send({error: error});
+      var context = {
+        collection: presence.c,
+        presence: presence
+      };
+      backend.trigger(backend.MIDDLEWARE_ACTIONS.sendPresence, agent, context, function(error) {
+        if (error) return agent.send({error: error});
+        agent.send(presence);
+      });
+    });
+  }
+};
 
 function createClientOp(request, clientId) {
   // src can be provided if it is not the same as the current agent,

lib/backend.js

@@ -87,6 +87,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'
 };
@@ -680,6 +682,10 @@ Backend.prototype.getDocChannel = function(collection, id) {
   return collection + '.' + id;
 };
 
+Backend.prototype.getPresenceChannel = function(collection, id) {
+  return this.getDocChannel(collection, id) + '.presence';
+};
+
 Backend.prototype.getChannels = function(collection, id) {
   return [
     this.getCollectionChannel(collection),
@@ -800,6 +806,21 @@ Backend.prototype._buildSnapshotFromOps = function(id, startingSnapshot, ops, ca
   callback(error, snapshot);
 };
 
+Backend.prototype.transformPresenceToLatestVersion = function(agent, presence, callback) {
+  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++) {