loopback/common/models/change.js

819 lines
22 KiB
JavaScript

// Copyright IBM Corp. 2014,2019. All Rights Reserved.
// Node module: loopback
// This file is licensed under the MIT License.
// License text available at https://opensource.org/licenses/MIT
/*!
* Module Dependencies.
*/
'use strict';
const g = require('../../lib/globalize');
const PersistedModel = require('../../lib/loopback').PersistedModel;
const loopback = require('../../lib/loopback');
const utils = require('../../lib/utils');
const crypto = require('crypto');
const CJSON = {stringify: require('canonical-json')};
const async = require('async');
const assert = require('assert');
const debug = require('debug')('loopback:change');
/**
* Change list entry.
*
* @property {String} id Hash of the modelName and ID.
* @property {String} rev The current model revision.
* @property {String} prev The previous model revision.
* @property {Number} checkpoint The current checkpoint at time of the change.
* @property {String} modelName Model name.
* @property {String} modelId Model ID.
* @property {Object} settings Extends the `Model.settings` object.
* @property {String} settings.hashAlgorithm Algorithm used to create cryptographic hash, used as argument
* to [crypto.createHash](http://nodejs.org/api/crypto.html#crypto_crypto_createhash_algorithm). Default is sha1.
* @property {Boolean} settings.ignoreErrors By default, when changes are rectified, an error will throw an exception.
* However, if this setting is true, then errors will not throw exceptions.
* @class Change
* @inherits {PersistedModel}
*/
module.exports = function(Change) {
/*!
* Constants
*/
Change.UPDATE = 'update';
Change.CREATE = 'create';
Change.DELETE = 'delete';
Change.UNKNOWN = 'unknown';
/*!
* Conflict Class
*/
Change.Conflict = Conflict;
/*!
* Setup the extended model.
*/
Change.setup = function() {
PersistedModel.setup.call(this);
const Change = this;
Change.getter.id = function() {
const hasModel = this.modelName && this.modelId;
if (!hasModel) return null;
return Change.idForModel(this.modelName, this.modelId);
};
};
Change.setup();
/**
* Track the recent change of the given modelIds.
*
* @param {String} modelName
* @param {Array} modelIds
* @callback {Function} callback
* @param {Error} err
* @param {Array} changes Changes that were tracked
*/
Change.rectifyModelChanges = function(modelName, modelIds, callback) {
const Change = this;
const errors = [];
callback = callback || utils.createPromiseCallback();
const tasks = modelIds.map(function(id) {
return function(cb) {
Change.findOrCreateChange(modelName, id, function(err, change) {
if (err) return next(err);
change.rectify(next);
});
function next(err) {
if (err) {
err.modelName = modelName;
err.modelId = id;
errors.push(err);
}
cb();
}
};
});
async.parallel(tasks, function(err) {
if (err) return callback(err);
if (errors.length) {
const desc = errors
.map(function(e) {
return '#' + e.modelId + ' - ' + e.toString();
})
.join('\n');
const msg = g.f('Cannot rectify %s changes:\n%s', modelName, desc);
err = new Error(msg);
err.details = {errors: errors};
return callback(err);
}
callback();
});
return callback.promise;
};
/**
* Get an identifier for a given model.
*
* @param {String} modelName
* @param {String} modelId
* @return {String}
*/
Change.idForModel = function(modelName, modelId) {
return this.hash([modelName, modelId].join('-'));
};
/**
* Find or create a change for the given model.
*
* @param {String} modelName
* @param {String} modelId
* @callback {Function} callback
* @param {Error} err
* @param {Change} change
* @end
*/
Change.findOrCreateChange = function(modelName, modelId, callback) {
assert(this.registry.findModel(modelName), modelName + ' does not exist');
callback = callback || utils.createPromiseCallback();
const id = this.idForModel(modelName, modelId);
const Change = this;
this.findById(id, function(err, change) {
if (err) return callback(err);
if (change) {
callback(null, change);
} else {
const ch = new Change({
id: id,
modelName: modelName,
modelId: modelId,
});
ch.debug('creating change');
Change.updateOrCreate(ch, callback);
}
});
return callback.promise;
};
/**
* Update (or create) the change with the current revision.
*
* @callback {Function} callback
* @param {Error} err
* @param {Change} change
*/
Change.prototype.rectify = function(cb) {
const change = this;
const currentRev = this.rev;
change.debug('rectify change');
cb = cb || utils.createPromiseCallback();
const model = this.getModelCtor();
const id = this.getModelId();
model.findById(id, function(err, inst) {
if (err) return cb(err);
if (inst) {
inst.fillCustomChangeProperties(change, function() {
const rev = Change.revisionForInst(inst);
prepareAndDoRectify(rev);
});
} else {
prepareAndDoRectify(null);
}
});
return cb.promise;
function prepareAndDoRectify(rev) {
// avoid setting rev and prev to the same value
if (currentRev === rev) {
change.debug('rev and prev are equal (not updating anything)');
return cb(null, change);
}
// FIXME(@bajtos) Allow callers to pass in the checkpoint value
// (or even better - a memoized async function to get the cp value)
// That will enable `rectifyAll` to cache the checkpoint value
change.constructor.getCheckpointModel().current(
function(err, checkpoint) {
if (err) return cb(err);
doRectify(checkpoint, rev);
},
);
}
function doRectify(checkpoint, rev) {
if (rev) {
if (currentRev === rev) {
change.debug('ASSERTION FAILED: Change currentRev==rev ' +
'should have been already handled');
return cb(null, change);
} else {
change.rev = rev;
change.debug('updated revision (was ' + currentRev + ')');
if (change.checkpoint !== checkpoint) {
// previous revision is updated only across checkpoints
change.prev = currentRev;
change.debug('updated prev');
}
}
} else {
change.rev = null;
change.debug('updated revision (was ' + currentRev + ')');
if (change.checkpoint !== checkpoint) {
// previous revision is updated only across checkpoints
if (currentRev) {
change.prev = currentRev;
} else if (!change.prev) {
change.debug('ERROR - could not determine prev');
change.prev = Change.UNKNOWN;
}
change.debug('updated prev');
}
}
if (change.checkpoint != checkpoint) {
debug('update checkpoint to', checkpoint);
change.checkpoint = checkpoint;
}
if (change.prev === Change.UNKNOWN) {
// this occurs when a record of a change doesn't exist
// and its current revision is null (not found)
change.remove(cb);
} else {
change.save(cb);
}
}
};
/**
* Get a change's current revision based on current data.
* @callback {Function} callback
* @param {Error} err
* @param {String} rev The current revision
*/
Change.prototype.currentRevision = function(cb) {
cb = cb || utils.createPromiseCallback();
const model = this.getModelCtor();
const id = this.getModelId();
model.findById(id, function(err, inst) {
if (err) return cb(err);
if (inst) {
cb(null, Change.revisionForInst(inst));
} else {
cb(null, null);
}
});
return cb.promise;
};
/**
* Create a hash of the given `string` with the `options.hashAlgorithm`.
* **Default: `sha1`**
*
* @param {String} str The string to be hashed
* @return {String} The hashed string
*/
Change.hash = function(str) {
return crypto
.createHash(Change.settings.hashAlgorithm || 'sha1')
.update(str)
.digest('hex');
};
/**
* Get the revision string for the given object
* @param {Object} inst The data to get the revision string for
* @return {String} The revision string
*/
Change.revisionForInst = function(inst) {
assert(inst, 'Change.revisionForInst() requires an instance object.');
return this.hash(CJSON.stringify(inst));
};
/**
* Get a change's type. Returns one of:
*
* - `Change.UPDATE`
* - `Change.CREATE`
* - `Change.DELETE`
* - `Change.UNKNOWN`
*
* @return {String} the type of change
*/
Change.prototype.type = function() {
if (this.rev && this.prev) {
return Change.UPDATE;
}
if (this.rev && !this.prev) {
return Change.CREATE;
}
if (!this.rev && this.prev) {
return Change.DELETE;
}
return Change.UNKNOWN;
};
/**
* Compare two changes.
* @param {Change} change
* @return {Boolean}
*/
Change.prototype.equals = function(change) {
if (!change) return false;
const thisRev = this.rev || null;
const thatRev = change.rev || null;
return thisRev === thatRev;
};
/**
* Does this change conflict with the given change.
* @param {Change} change
* @return {Boolean}
*/
Change.prototype.conflictsWith = function(change) {
if (!change) return false;
if (this.equals(change)) return false;
if (Change.bothDeleted(this, change)) return false;
if (this.isBasedOn(change)) return false;
return true;
};
/**
* Are both changes deletes?
* @param {Change} a
* @param {Change} b
* @return {Boolean}
*/
Change.bothDeleted = function(a, b) {
return a.type() === Change.DELETE &&
b.type() === Change.DELETE;
};
/**
* Determine if the change is based on the given change.
* @param {Change} change
* @return {Boolean}
*/
Change.prototype.isBasedOn = function(change) {
return this.prev === change.rev;
};
/**
* Determine the differences for a given model since a given checkpoint.
*
* The callback will contain an error or `result`.
*
* **result**
*
* ```js
* {
* deltas: Array,
* conflicts: Array
* }
* ```
*
* **deltas**
*
* An array of changes that differ from `remoteChanges`.
*
* **conflicts**
*
* An array of changes that conflict with `remoteChanges`.
*
* @param {String} modelName
* @param {Number} since Compare changes after this checkpoint
* @param {Change[]} remoteChanges A set of changes to compare
* @callback {Function} callback
* @param {Error} err
* @param {Object} result See above.
*/
Change.diff = function(modelName, since, remoteChanges, callback) {
callback = callback || utils.createPromiseCallback();
if (!Array.isArray(remoteChanges) || remoteChanges.length === 0) {
callback(null, {deltas: [], conflicts: []});
return callback.promise;
}
const remoteChangeIndex = {};
const modelIds = [];
remoteChanges.forEach(function(ch) {
modelIds.push(ch.modelId);
remoteChangeIndex[ch.modelId] = new Change(ch);
});
// normalize `since`
since = Number(since) || 0;
this.find({
where: {
modelName: modelName,
modelId: {inq: modelIds},
},
}, function(err, allLocalChanges) {
if (err) return callback(err);
const deltas = [];
const conflicts = [];
const localModelIds = [];
const localChanges = allLocalChanges.filter(function(c) {
return c.checkpoint >= since;
});
localChanges.forEach(function(localChange) {
localChange = new Change(localChange);
localModelIds.push(localChange.modelId);
const remoteChange = remoteChangeIndex[localChange.modelId];
if (remoteChange && !localChange.equals(remoteChange)) {
if (remoteChange.conflictsWith(localChange)) {
remoteChange.debug('remote conflict');
localChange.debug('local conflict');
conflicts.push(localChange);
} else {
remoteChange.debug('remote delta');
deltas.push(remoteChange);
}
}
});
modelIds.forEach(function(id) {
if (localModelIds.indexOf(id) !== -1) return;
const d = remoteChangeIndex[id];
const oldChange = allLocalChanges.filter(function(c) {
return c.modelId === id;
})[0];
if (oldChange) {
d.prev = oldChange.rev;
} else {
d.prev = null;
}
deltas.push(d);
});
callback(null, {
deltas: deltas,
conflicts: conflicts,
});
});
return callback.promise;
};
/**
* Correct all change list entries.
* @param {Function} cb
*/
Change.rectifyAll = function(cb) {
debug('rectify all');
const Change = this;
// this should be optimized
this.find(function(err, changes) {
if (err) return cb(err);
async.each(
changes,
function(c, next) { c.rectify(next); },
cb,
);
});
};
/**
* Get the checkpoint model.
* @return {Checkpoint}
*/
Change.getCheckpointModel = function() {
let checkpointModel = this.Checkpoint;
if (checkpointModel) return checkpointModel;
// FIXME(bajtos) This code creates multiple different models with the same
// model name, which is not a valid supported usage of juggler's API.
this.Checkpoint = checkpointModel = loopback.Checkpoint.extend('checkpoint');
assert(this.dataSource, 'Cannot getCheckpointModel(): ' + this.modelName +
' is not attached to a dataSource');
checkpointModel.attachTo(this.dataSource);
return checkpointModel;
};
Change.prototype.debug = function() {
if (debug.enabled) {
const args = Array.prototype.slice.call(arguments);
args[0] = args[0] + ' %s';
args.push(this.modelName);
debug.apply(this, args);
debug('\tid', this.id);
debug('\trev', this.rev);
debug('\tprev', this.prev);
debug('\tcheckpoint', this.checkpoint);
debug('\tmodelName', this.modelName);
debug('\tmodelId', this.modelId);
debug('\ttype', this.type());
}
};
/**
* Get the `Model` class for `change.modelName`.
* @return {Model}
*/
Change.prototype.getModelCtor = function() {
return this.constructor.settings.trackModel;
};
Change.prototype.getModelId = function() {
// TODO(ritch) get rid of the need to create an instance
const Model = this.getModelCtor();
const id = this.modelId;
const m = new Model();
m.setId(id);
return m.getId();
};
Change.prototype.getModel = function(callback) {
const Model = this.constructor.settings.trackModel;
const id = this.getModelId();
Model.findById(id, callback);
};
/**
* When two changes conflict a conflict is created.
*
* **Note**: call `conflict.fetch()` to get the `target` and `source` models.
*
* @param {*} modelId
* @param {PersistedModel} SourceModel
* @param {PersistedModel} TargetModel
* @property {ModelClass} source The source model instance
* @property {ModelClass} target The target model instance
* @class Change.Conflict
*/
function Conflict(modelId, SourceModel, TargetModel) {
this.SourceModel = SourceModel;
this.TargetModel = TargetModel;
this.SourceChange = SourceModel.getChangeModel();
this.TargetChange = TargetModel.getChangeModel();
this.modelId = modelId;
}
/**
* Fetch the conflicting models.
*
* @callback {Function} callback
* @param {Error} err
* @param {PersistedModel} source
* @param {PersistedModel} target
*/
Conflict.prototype.models = function(cb) {
const conflict = this;
const SourceModel = this.SourceModel;
const TargetModel = this.TargetModel;
let source, target;
async.parallel([
getSourceModel,
getTargetModel,
], done);
function getSourceModel(cb) {
SourceModel.findById(conflict.modelId, function(err, model) {
if (err) return cb(err);
source = model;
cb();
});
}
function getTargetModel(cb) {
TargetModel.findById(conflict.modelId, function(err, model) {
if (err) return cb(err);
target = model;
cb();
});
}
function done(err) {
if (err) return cb(err);
cb(null, source, target);
}
};
/**
* Get the conflicting changes.
*
* @callback {Function} callback
* @param {Error} err
* @param {Change} sourceChange
* @param {Change} targetChange
*/
Conflict.prototype.changes = function(cb) {
const conflict = this;
let sourceChange, targetChange;
async.parallel([
getSourceChange,
getTargetChange,
], done);
function getSourceChange(cb) {
const SourceModel = conflict.SourceModel;
SourceModel.findLastChange(conflict.modelId, function(err, change) {
if (err) return cb(err);
sourceChange = change;
cb();
});
}
function getTargetChange(cb) {
const TargetModel = conflict.TargetModel;
TargetModel.findLastChange(conflict.modelId, function(err, change) {
if (err) return cb(err);
targetChange = change;
cb();
});
}
function done(err) {
if (err) return cb(err);
cb(null, sourceChange, targetChange);
}
};
/**
* Resolve the conflict.
*
* Set the source change's previous revision to the current revision of the
* (conflicting) target change. Since the changes are no longer conflicting
* and appear as if the source change was based on the target, they will be
* replicated normally as part of the next replicate() call.
*
* This is effectively resolving the conflict using the source version.
*
* @callback {Function} callback
* @param {Error} err
*/
Conflict.prototype.resolve = function(cb) {
const conflict = this;
conflict.TargetModel.findLastChange(
this.modelId,
function(err, targetChange) {
if (err) return cb(err);
conflict.SourceModel.updateLastChange(
conflict.modelId,
{prev: targetChange.rev},
cb,
);
},
);
};
/**
* Resolve the conflict using the instance data in the source model.
*
* @callback {Function} callback
* @param {Error} err
*/
Conflict.prototype.resolveUsingSource = function(cb) {
this.resolve(function(err) {
// don't forward any cb arguments from resolve()
cb(err);
});
};
/**
* Resolve the conflict using the instance data in the target model.
*
* @callback {Function} callback
* @param {Error} err
*/
Conflict.prototype.resolveUsingTarget = function(cb) {
const conflict = this;
conflict.models(function(err, source, target) {
if (err) return done(err);
if (target === null) {
return conflict.SourceModel.deleteById(conflict.modelId, done);
}
const inst = new conflict.SourceModel(
target.toObject(),
{persisted: true},
);
inst.save(done);
});
function done(err) {
// don't forward any cb arguments from internal calls
cb(err);
}
};
/**
* Return a new Conflict instance with swapped Source and Target models.
*
* This is useful when resolving a conflict in one-way
* replication, where the source data must not be changed:
*
* ```js
* conflict.swapParties().resolveUsingTarget(cb);
* ```
*
* @returns {Conflict} A new Conflict instance.
*/
Conflict.prototype.swapParties = function() {
const Ctor = this.constructor;
return new Ctor(this.modelId, this.TargetModel, this.SourceModel);
};
/**
* Resolve the conflict using the supplied instance data.
*
* @param {Object} data The set of changes to apply on the model
* instance. Use `null` value to delete the source instance instead.
* @callback {Function} callback
* @param {Error} err
*/
Conflict.prototype.resolveManually = function(data, cb) {
const conflict = this;
if (!data) {
return conflict.SourceModel.deleteById(conflict.modelId, done);
}
conflict.models(function(err, source, target) {
if (err) return done(err);
const inst = source || new conflict.SourceModel(target);
inst.setAttributes(data);
inst.save(function(err) {
if (err) return done(err);
conflict.resolve(done);
});
});
function done(err) {
// don't forward any cb arguments from internal calls
cb(err);
}
};
/**
* Determine the conflict type.
*
* Possible results are
*
* - `Change.UPDATE`: Source and target models were updated.
* - `Change.DELETE`: Source and or target model was deleted.
* - `Change.UNKNOWN`: the conflict type is uknown or due to an error.
*
* @callback {Function} callback
* @param {Error} err
* @param {String} type The conflict type.
*/
Conflict.prototype.type = function(cb) {
const conflict = this;
this.changes(function(err, sourceChange, targetChange) {
if (err) return cb(err);
const sourceChangeType = sourceChange.type();
const targetChangeType = targetChange.type();
if (sourceChangeType === Change.UPDATE && targetChangeType === Change.UPDATE) {
return cb(null, Change.UPDATE);
}
if (sourceChangeType === Change.DELETE || targetChangeType === Change.DELETE) {
return cb(null, Change.DELETE);
}
return cb(null, Change.UNKNOWN);
});
};
};