Merge pull request #581 from share/server-fixup

✨ Allow middleware to mutate submitted ops

Author: alecgibson

docs/middleware/op-submission.md

@@ -140,12 +140,20 @@ Since `'submitRequestEnd'` is an event -- not a middleware hook -- it provides n
 
 ## Mutating ops
 
-{: .warn :}
-Mutating ops in middleware is generally a **bad idea**, and should be avoided.
+Ops may be amended in the `apply` middleware using the special `request.$fixup()` method:
 
-The main reason for avoiding op mutation is that the client who submitted the op will not be informed of the mutation, so the client's doc will never receive the mutation.
+```js
+backend.use('apply', (request, next) => {
+  let error;
+  try {
+    request.$fixup([{p: ['meta'], oi: {timestamp: Date.now()}}]);
+  } catch (e) {
+    error = e;
+  }
 
-The general workaround is to trigger a second op submission, rather than mutate the provided op. This obviously has the downside of op submissions being unatomic, but is the safest way to get the original client to receive the update.
+  next(error);
+});
+```
 
 {: .warn :}
-When submitting ops from the middleware, set careful conditions under which you submit ops in order to avoid infinite op submission loops, where submitting an op recursively triggers infinite op submissions.
+The `request.$fixup()` method may throw an error, which should be handled appropriately, usually by passing directly to the `next()` callback.

lib/agent.js

@@ -718,9 +718,10 @@ Agent.prototype._unsubscribeBulk = function(collection, ids, callback) {
 
 Agent.prototype._submit = function(collection, id, op, callback) {
   var agent = this;
-  this.backend.submit(this, collection, id, op, null, function(err, ops) {
+  this.backend.submit(this, collection, id, op, null, function(err, ops, request) {
     // Message to acknowledge the op was successfully submitted
     var ack = {src: op.src, seq: op.seq, v: op.v};
+    if (request._fixupOps.length) ack[ACTIONS.fixup] = request._fixupOps;
     if (err) {
       // Occasional 'Op already submitted' errors are expected to happen as
       // part of normal operation, since inflight ops need to be resent after

lib/backend.js

@@ -229,7 +229,7 @@ Backend.prototype.submit = function(agent, index, id, op, options, originalCallb
 
   var callback = function(error, ops) {
     backend.emit('submitRequestEnd', error, request);
-    originalCallback(error, ops);
+    originalCallback(error, ops, request);
   };
 
   var err = ot.checkOp(op);

lib/client/doc.js

@@ -5,6 +5,7 @@ var types = require('../types');
 var util = require('../util');
 var clone = util.clone;
 var deepEqual = require('fast-deep-equal');
+var ACTIONS = require('../message-actions').ACTIONS;
 
 var ERROR_CODE = ShareDBError.CODES;
 
@@ -960,6 +961,23 @@ Doc.prototype._opAcknowledged = function(message) {
     return this.fetch();
   }
 
+  if (message[ACTIONS.fixup]) {
+    for (var i = 0; i < message[ACTIONS.fixup].length; i++) {
+      var fixupOp = message[ACTIONS.fixup][i];
+
+      for (var j = 0; j < this.pendingOps.length; j++) {
+        var transformErr = transformX(this.pendingOps[i], fixupOp);
+        if (transformErr) return this._hardRollback(transformErr);
+      }
+
+      try {
+        this._otApply(fixupOp, false);
+      } catch (error) {
+        return this._hardRollback(error);
+      }
+    }
+  }
+
   // The op was committed successfully. Increment the version number
   this.version++;
 

lib/error.js

@@ -15,6 +15,8 @@ ShareDBError.prototype.name = 'ShareDBError';
 ShareDBError.CODES = {
   ERR_APPLY_OP_VERSION_DOES_NOT_MATCH_SNAPSHOT: 'ERR_APPLY_OP_VERSION_DOES_NOT_MATCH_SNAPSHOT',
   ERR_APPLY_SNAPSHOT_NOT_PROVIDED: 'ERR_APPLY_SNAPSHOT_NOT_PROVIDED',
+  ERR_FIXUP_IS_ONLY_VALID_ON_APPLY: 'ERR_FIXUP_IS_ONLY_VALID_ON_APPLY',
+  ERR_CANNOT_FIXUP_DELETION: 'ERR_CANNOT_FIXUP_DELETION',
   ERR_CLIENT_ID_BADLY_FORMED: 'ERR_CLIENT_ID_BADLY_FORMED',
   ERR_CANNOT_PING_OFFLINE: 'ERR_CANNOT_PING_OFFLINE',
   ERR_CONNECTION_SEQ_INTEGER_OVERFLOW: 'ERR_CONNECTION_SEQ_INTEGER_OVERFLOW',
@@ -68,6 +70,7 @@ ShareDBError.CODES = {
   ERR_SNAPSHOT_READS_REJECTED: 'ERR_SNAPSHOT_READS_REJECTED',
   ERR_SUBMIT_TRANSFORM_OPS_NOT_FOUND: 'ERR_SUBMIT_TRANSFORM_OPS_NOT_FOUND',
   ERR_TYPE_CANNOT_BE_PROJECTED: 'ERR_TYPE_CANNOT_BE_PROJECTED',
+  ERR_TYPE_DOES_NOT_SUPPORT_COMPOSE: 'ERR_TYPE_DOES_NOT_SUPPORT_COMPOSE',
   ERR_TYPE_DOES_NOT_SUPPORT_PRESENCE: 'ERR_TYPE_DOES_NOT_SUPPORT_PRESENCE',
   ERR_UNKNOWN_ERROR: 'ERR_UNKNOWN_ERROR'
 };

lib/message-actions.js

@@ -9,6 +9,7 @@ exports.ACTIONS = {
   bulkSubscribe: 'bs',
   bulkUnsubscribe: 'bu',
   fetch: 'f',
+  fixup: 'fixup',
   subscribe: 's',
   unsubscribe: 'u',
   op: 'op',

lib/submit-request.js

@@ -1,6 +1,7 @@
 var ot = require('./ot');
 var projections = require('./projections');
 var ShareDBError = require('./error');
+var types = require('./types');
 
 var ERROR_CODE = ShareDBError.CODES;
 
@@ -40,9 +41,47 @@ function SubmitRequest(backend, agent, index, id, op, options) {
   this.snapshot = null;
   this.ops = [];
   this.channels = null;
+  this._fixupOps = [];
 }
 module.exports = SubmitRequest;
 
+SubmitRequest.prototype.$fixup = function(op) {
+  if (this.action !== this.backend.MIDDLEWARE_ACTIONS.apply) {
+    throw new ShareDBError(
+      ERROR_CODE.ERR_FIXUP_IS_ONLY_VALID_ON_APPLY,
+      'fixup can only be called during the apply middleware'
+    );
+  }
+
+  if (this.op.del) {
+    throw new ShareDBError(
+      ERROR_CODE.ERR_CANNOT_FIXUP_DELETION,
+      'fixup cannot be applied on deletion ops'
+    );
+  }
+
+  var typeId = this.op.create ? this.op.create.type : this.snapshot.type;
+  var type = types.map[typeId];
+  if (typeof type.compose !== 'function') {
+    throw new ShareDBError(
+      ERROR_CODE.ERR_TYPE_DOES_NOT_SUPPORT_COMPOSE,
+      typeId + ' does not support compose'
+    );
+  }
+
+  if (this.op.create) this.op.create.data = type.apply(this.op.create.data, op);
+  else this.op.op = type.compose(this.op.op, op);
+
+  var fixupOp = {
+    src: this.op.src,
+    seq: this.op.seq,
+    v: this.op.v,
+    op: op
+  };
+
+  this._fixupOps.push(fixupOp);
+};
+
 SubmitRequest.prototype.submit = function(callback) {
   var request = this;
   var backend = this.backend;
@@ -103,7 +142,7 @@ SubmitRequest.prototype.submit = function(callback) {
 
     // Transform the op up to the current snapshot version, then apply
     var from = op.v;
-    backend.db.getOpsToSnapshot(collection, id, from, snapshot, null, function(err, ops) {
+    backend.db.getOpsToSnapshot(collection, id, from, snapshot, {metadata: true}, function(err, ops) {
       if (err) return callback(err);
 
       if (ops.length !== snapshot.v - from) {
@@ -134,6 +173,8 @@ SubmitRequest.prototype.apply = function(callback) {
   // Always set the channels before each attempt to apply. If the channels are
   // modified in a middleware and we retry, we want to reset to a new array
   this.channels = this.backend.getChannels(this.collection, this.id);
+  this._fixupOps = [];
+  delete this.op.m.fixup;
 
   var request = this;
   this.backend.trigger(this.backend.MIDDLEWARE_ACTIONS.apply, this.agent, this, function(err) {
@@ -152,6 +193,7 @@ SubmitRequest.prototype.commit = function(callback) {
   var backend = this.backend;
   backend.trigger(backend.MIDDLEWARE_ACTIONS.commit, this.agent, this, function(err) {
     if (err) return callback(err);
+    if (request._fixupOps.length) request.op.m.fixup = request._fixupOps;
 
     // Try committing the operation and snapshot to the database atomically
     backend.db.commit(
@@ -204,6 +246,7 @@ SubmitRequest.prototype._transformOp = function(ops) {
       // can happen in normal operation, such as a client resending an
       // unacknowledged operation at reconnect. It's important we don't apply
       // the same op twice
+      if (op.m.fixup) this._fixupOps = op.m.fixup;
       return this.alreadySubmittedError();
     }
 
@@ -213,6 +256,7 @@ SubmitRequest.prototype._transformOp = function(ops) {
 
     var err = ot.transform(type, this.op, op);
     if (err) return err;
+    delete op.m;
     this.ops.push(op);
   }
 };

test/client/submit.js

@@ -4,6 +4,7 @@ var sinon = require('sinon');
 var types = require('../../lib/types');
 var deserializedType = require('./deserialized-type');
 var numberType = require('./number-type');
+var errorHandler = require('../util').errorHandler;
 types.register(deserializedType.type);
 types.register(deserializedType.type2);
 types.register(numberType.type);
@@ -1210,5 +1211,52 @@ module.exports = function() {
         });
       });
     });
+
+    describe('submitting when behind the server', function() {
+      var doc;
+      var remoteDoc;
+
+      beforeEach(function(done) {
+        var connection = this.backend.connect();
+        doc = connection.get('dogs', 'fido');
+        var remoteConnection = this.backend.connect();
+        remoteDoc = remoteConnection.get('dogs', 'fido');
+
+        async.series([
+          doc.create.bind(doc, {name: 'fido'}),
+          remoteDoc.fetch.bind(remoteDoc),
+          remoteDoc.submitOp.bind(remoteDoc, [{p: ['tricks'], oi: ['fetch']}]),
+          function(next) {
+            expect(doc.data).to.eql({name: 'fido'});
+            expect(remoteDoc.data).to.eql({name: 'fido', tricks: ['fetch']});
+            next();
+          }
+        ], done);
+      });
+
+      it('is sent ops it has missed when submitting, without calling fetch', function(done) {
+        sinon.spy(doc, 'fetch');
+
+        doc.submitOp([{p: ['age'], oi: 2}], errorHandler(done));
+
+        doc.once('op', function() {
+          expect(doc.data).to.eql({name: 'fido', tricks: ['fetch'], age: 2});
+          expect(doc.fetch.called).to.be.false;
+          done();
+        });
+      });
+
+      it('does not expose op metadata in the middleware when sending missing ops', function(done) {
+        this.backend.use('apply', function(request) {
+          expect(request.ops).to.have.length(1);
+          var op = request.ops[0];
+          expect(op.op).to.eql([{p: ['tricks'], oi: ['fetch']}]);
+          expect(op.m).to.be.undefined;
+          done();
+        });
+
+        doc.submitOp([{p: ['age'], oi: 2}], errorHandler(done));
+      });
+    });
   });
 };