Sync presence and undo redo

.editorconfig

@@ -0,0 +1,9 @@
+root = true
+
+[*]
+indent_style = space
+indent_size = 2
+end_of_line = LF
+charset = utf-8
+trim_trailing_whitespace = true
+insert_final_newline = true

.travis.yml

@@ -1,10 +1,8 @@
 language: node_js
 node_js:
   - "10"
-  - "9"
   - "8"
   - "6"
-  - "4"
-script: "npm run jshint && npm run test-cover"
+script: "ln -s .. node_modules/sharedb; npm run jshint && npm run test-cover"
 # Send coverage data to Coveralls
 after_script: "cat ./coverage/lcov.info | ./node_modules/coveralls/bin/coveralls.js"

README.md

@@ -19,6 +19,8 @@ tracker](https://github.com/share/sharedb/issues).
 
 - Realtime synchronization of any JSON document
 - Concurrent multi-user collaboration
+- Realtime synchronization of any ephemeral "presence" data
+- Local undo and redo
 - Synchronous editing API with asynchronous eventual consistency
 - Realtime query subscriptions
 - Simple integration with any database - [MongoDB](https://github.com/share/sharedb-mongo), [PostgresQL](https://github.com/share/sharedb-postgres) (experimental)
@@ -38,7 +40,7 @@ var socket = new WebSocket('ws://' + window.location.host);
 var connection = new sharedb.Connection(socket);
 ```
 
-The native Websocket object that you feed to ShareDB's `Connection` constructor **does not** handle reconnections. 
+The native Websocket object that you feed to ShareDB's `Connection` constructor **does not** handle reconnections.
 
 The easiest way is to give it a WebSocket object that does reconnect. There are plenty of example on the web. The most important thing is that the custom reconnecting websocket, must have the same API as the native rfc6455 version.
 
@@ -73,6 +75,10 @@ initial data. Then you can submit editing operations on the document (using
 OT). Finally you can delete the document with a delete operation. By
 default, ShareDB stores all operations forever - nothing is truly deleted.
 
+## User presence synchronization
+
+Presence data represents a user and is automatically synchronized between all clients subscribed to the same document. Its format is defined by the document's [OT Type](https://github.com/ottypes/docs), for example it may contain a user ID and a cursor position in a text document. All clients can modify their own presence data and receive a read-only version of other client's data. Presence data is automatically cleared when a client unsubscribes from the document or disconnects. It is also automatically transformed against applied operations, so that it still makes sense in the context of a modified document, for example a cursor position may be automatically advanced when a user types at the beginning of a text document.
+
 ## Server API
 
 ### Initialization
@@ -228,9 +234,15 @@ changes. Returns a [`ShareDB.Query`](#class-sharedbquery) instance.
 * `options.*`
   All other options are passed through to the database adapter.
 
+`connection.createUndoManager(options)` creates a new `UndoManager`.
+
+* `options.source` if specified, only the operations from that `source` will be undo-able. If `null` or `undefined`, the `source` filter is disabled.
+* `options.limit` the max number of operations to keep on the undo stack.
+* `options.composeInterval` the max time difference between operations in milliseconds, which still allows the operations to be composed on the undo stack.
+
 ### Class: `ShareDB.Doc`
 
-`doc.type` _(String_)
+`doc.type` _(String)_
 The [OT type](https://github.com/ottypes/docs) of this document
 
 `doc.id` _(String)_
@@ -239,6 +251,9 @@ Unique document ID
 `doc.data` _(Object)_
 Document contents. Available after document is fetched or subscribed to.
 
+`doc.presence` _(Object)_
+Each property under `doc.presence` contains presence data shared by a client subscribed to this document. The property name is an empty string for this client's data and connection IDs for other clients' data.
+
 `doc.fetch(function(err) {...})`
 Populate the fields on `doc` with a snapshot of the document from the server.
 
@@ -268,6 +283,9 @@ An operation was applied to the data. `source` will be `false` for ops received
 `doc.on('del', function(data, source) {...})`
 The document was deleted. Document contents before deletion are passed in as an argument. `source` will be `false` for ops received from the server and defaults to `true` for ops generated locally.
 
+`doc.on('presence', function(srcList, submitted) {...})`
+Presence data has changed. `srcList` is an Array of `doc.presence` property names for which values have changed. `submitted` is `true`, if the event is the result of new presence data being submitted by the local or remote user, otherwise it is `false` - eg if the presence data was transformed against an operation or was cleared on unsubscribe, disconnect or roll-back.
+
 `doc.on('error', function(err) {...})`
 There was an error fetching the document or applying an operation.
 
@@ -287,6 +305,19 @@ Apply operation to document and send it to the server.
 [operations for the default `'ot-json0'` type](https://github.com/ottypes/json0#summary-of-operations).
 Call this after you've either fetched or subscribed to the document.
 * `options.source` Argument passed to the `'op'` event locally. This is not sent to the server or other clients. Defaults to `true`.
+* `options.skipNoop` Should processing be skipped entirely, if `op` is a no-op. Defaults to `false`.
+* `options.undoable` Should it be possible to undo this operation. Defaults to `false`.
+* `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`.
+
+`doc.submitSnapshot(snapshot[, options][, function(err) {...}])`
+Diff the current and the provided snapshots to generate an operation, apply the operation to the document and send it to the server.
+`snapshot` structure depends on the document type.
+Call this after you've either fetched or subscribed to the document.
+* `options.source` Argument passed to the `'op'` event locally. This is not sent to the server or other clients. Defaults to `true`.
+* `options.skipNoop` Should processing be skipped entirely, if `op` is a no-op. Defaults to `false`.
+* `options.undoable` Should it be possible to undo this operation. Defaults to `false`.
+* `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`.
+* `options.diffHint` A hint passed into the `diff`/`diffX` functions defined by the document type.
 
 `doc.del([options][, function(err) {...}])`
 Delete the document locally and send delete operation to the server.
@@ -301,6 +332,12 @@ Invokes the given callback function after
 
 Note that `whenNothingPending` does NOT wait for pending `model.query()` calls.
 
+`doc.submitPresence(presenceData[, function(err) {...}])`
+Set local presence data and publish it for other clients.
+
+`presenceData` structure depends on the document type.
+Presence is synchronized only when subscribed to the document.
+
 ### Class: `ShareDB.Query`
 
 `query.ready` _(Boolean)_
@@ -338,6 +375,28 @@ after a sequence of diffs are handled.
 `query.on('extra', function() {...}))`
 (Only fires on subscription queries) `query.extra` changed.
 
+### Class: `ShareDB.UndoManager`
+
+`undoManager.canUndo()`
+Return `true`, if there's an operation on the undo stack that can be undone, otherwise `false`.
+
+`undoManager.undo([options][, function(err) {...}])`
+Undo a previously applied undoable or redo operation.
+* `options.source` Argument passed to the `'op'` event locally. This is not sent to the server or other clients. Defaults to `true`.
+
+`undoManager.canRedo()`
+Return `true`, if there's an operation on the redo stack that can be undone, otherwise `false`.
+
+`undoManager.redo([options][, function(err) {...}])`
+Redo a previously applied undo operation.
+* `options.source` Argument passed to the `'op'` event locally. This is not sent to the server or other clients. Defaults to `true`.
+
+`undoManager.clear(doc)`
+Remove operations from the undo and redo stacks.
+* `doc` if specified, only the operations on that doc are removed, otherwise all operations are removed.
+
+`undoManager.destroy()`
+Remove all operations from the undo and redo stacks, and stop recording new operations.
 
 ## Error codes
 
@@ -376,6 +435,11 @@ Additional fields may be added to the error object for debugging context dependi
 * 4021 - Invalid client id
 * 4022 - Database adapter does not support queries
 * 4023 - Cannot project snapshots of this type
+* 4024 - OT Type does not support presence
+* 4025 - Not subscribed to document
+* 4026 - Presence data superseded
+* 4027 - OT Type does not support `diff` nor `diffX`
+* 4028 - OT Type does not support `invert` nor `applyAndInvert`
 
 ### 5000 - Internal error
 

lib/agent.js

@@ -1,6 +1,7 @@
 var hat = require('hat');
 var util = require('./util');
 var types = require('./types');
+var ShareDBError = require('./error');
 
 /**
  * Agent deserializes the wire protocol messages received from the stream and
@@ -25,6 +26,9 @@ function Agent(backend, stream) {
   // Map from queryId -> emitter
   this.subscribedQueries = {};
 
+  // The max presence sequence number received from the client.
+  this.maxPresenceSeq = 0;
+
   // We need to track this manually to make sure we don't reply to messages
   // after the stream was closed.
   this.closed = false;
@@ -98,10 +102,17 @@ Agent.prototype._subscribeToStream = function(collection, id, stream) {
       console.error('Doc subscription stream error', collection, id, data.error);
       return;
     }
+    if (data.a === 'p') {
+      // Send other clients' presence data
+      if (data.src !== agent.clientId) agent.send(data);
+      return;
+    }
     if (agent._isOwnOp(collection, data)) return;
     agent._sendOp(collection, id, data);
   });
   stream.on('end', function() {
+    var presence = agent._createPresence(collection, id);
+    agent.backend.sendPresence(presence);
     // The op stream is done sending, so release its reference
     var streams = agent.subscribedDocs[collection];
     if (!streams) return;
@@ -268,6 +279,13 @@ Agent.prototype._checkRequest = function(request) {
     // Bulk request
     if (request.c != null && typeof request.c !== 'string') return 'Invalid collection';
     if (typeof request.b !== 'object') return 'Invalid bulk subscribe data';
+  } else if (request.a === 'p') {
+    // Presence
+    if (typeof request.c !== 'string') return 'Invalid collection';
+    if (typeof request.d !== 'string') return 'Invalid id';
+    if (typeof request.v !== 'number' || request.v < 0) return 'Invalid version';
+    if (typeof request.seq !== 'number' || request.seq <= 0) return 'Invalid seq';
+    if (typeof request.r !== 'undefined' && typeof request.r !== 'boolean') return 'Invalid "request reply" value';
   }
 };
 
@@ -300,6 +318,9 @@ Agent.prototype._handleMessage = function(request, callback) {
         var op = this._createOp(request);
         if (!op) return callback({code: 4000, message: 'Invalid op message'});
         return this._submit(request.c, request.d, op, callback);
+      case 'p':
+        var presence = this._createPresence(request.c, request.d, request.p, request.v, request.r, request.seq);
+        return this._presence(presence, callback);
       default:
         callback({code: 4000, message: 'Invalid or unknown message'});
     }
@@ -582,3 +603,34 @@ Agent.prototype._createOp = function(request) {
     return new DeleteOp(src, request.seq, request.v, request.del);
   }
 };
+
+Agent.prototype._presence = function(presence, callback) {
+  if (presence.seq <= this.maxPresenceSeq) {
+    return process.nextTick(function() {
+      callback(new ShareDBError(4026, 'Presence data superseded'));
+    });
+  }
+  this.maxPresenceSeq = presence.seq;
+  if (!this.subscribedDocs[presence.c] || !this.subscribedDocs[presence.c][presence.d]) {
+    return process.nextTick(function() {
+      callback(new ShareDBError(4025, 'Cannot send presence. Not subscribed to document: ' + presence.c + ' ' + presence.d));
+    });
+  }
+  this.backend.sendPresence(presence, function(err) {
+    if (err) return callback(err);
+    callback(null, { seq: presence.seq });
+  });
+};
+
+Agent.prototype._createPresence = function(collection, id, data, version, requestReply, seq) {
+  return {
+    a: 'p',
+    src: this.clientId,
+    seq: seq != null ? seq : this.maxPresenceSeq,
+    c: collection,
+    d: id,
+    p: data,
+    v: version,
+    r: requestReply
+  };
+};

lib/backend.js

@@ -568,6 +568,11 @@ Backend.prototype.getChannels = function(collection, id) {
   ];
 };
 
+Backend.prototype.sendPresence = function(presence, callback) {
+  var channels = [ this.getDocChannel(presence.c, presence.d) ];
+  this.pubsub.publish(channels, presence, callback);
+};
+
 function pluckIds(snapshots) {
   var ids = [];
   for (var i = 0; i < snapshots.length; i++) {

lib/client/connection.js

@@ -1,5 +1,6 @@
 var Doc = require('./doc');
 var Query = require('./query');
+var UndoManager = require('./undoManager');
 var emitter = require('../emitter');
 var ShareDBError = require('../error');
 var types = require('../types');
@@ -33,6 +34,9 @@ function Connection(socket) {
   // (created documents MUST BE UNIQUE)
   this.collections = {};
 
+  // A list of active UndoManagers.
+  this.undoManagers = [];
+
   // Each query is created with an id that the server uses when it sends us
   // info about the query (updates, etc)
   this.nextQueryId = 1;
@@ -243,6 +247,11 @@ Connection.prototype.handleMessage = function(message) {
       if (doc) doc._handleOp(err, message);
       return;
 
+    case 'p':
+      var doc = this.getExisting(message.c, message.d);
+      if (doc) doc._handlePresence(err, message);
+      return;
+
     default:
       console.warn('Ignoring unrecognized message', message);
   }
@@ -408,6 +417,23 @@ Connection.prototype.sendOp = function(doc, op) {
   this.send(message);
 };
 
+Connection.prototype.sendPresence = function(doc, data, requestReply) {
+  // Ensure the doc is registered so that it receives the reply message
+  this._addDoc(doc);
+  var message = {
+    a: 'p',
+    c: doc.collection,
+    d: doc.id,
+    p: data,
+    v: doc.version || 0,
+    seq: this.seq++
+  };
+  if (requestReply) {
+    message.r = true;
+  }
+  this.send(message);
+};
+
 
 /**
  * Sends a message down the socket
@@ -584,3 +610,46 @@ Connection.prototype._firstQuery = function(fn) {
     }
   }
 };
+
+Connection.prototype.createUndoManager = function(options) {
+  var undoManager = new UndoManager(this, options);
+  this.undoManagers.push(undoManager);
+  return undoManager;
+};
+
+Connection.prototype.removeUndoManager = function(undoManager) {
+  var index = this.undoManagers.indexOf(undoManager);
+  if (index >= 0) {
+    this.undoManagers.splice(index, 1);
+  }
+};
+
+Connection.prototype.onDocLoad = function(doc) {
+  for (var i = 0; i < this.undoManagers.length; i++) {
+    this.undoManagers[i].onDocLoad(doc);
+  }
+};
+
+Connection.prototype.onDocDestroy = function(doc) {
+  for (var i = 0; i < this.undoManagers.length; i++) {
+    this.undoManagers[i].onDocDestroy(doc);
+  }
+};
+
+Connection.prototype.onDocCreate = function(doc) {
+  for (var i = 0; i < this.undoManagers.length; i++) {
+    this.undoManagers[i].onDocCreate(doc);
+  }
+};
+
+Connection.prototype.onDocDelete = function(doc) {
+  for (var i = 0; i < this.undoManagers.length; i++) {
+    this.undoManagers[i].onDocDelete(doc);
+  }
+};
+
+Connection.prototype.onDocOp = function(doc, op, undoOp, source, undoable, fixUp) {
+  for (var i = 0; i < this.undoManagers.length; i++) {
+    this.undoManagers[i].onDocOp(doc, op, undoOp, source, undoable, fixUp);
+  }
+};