lib/storage/dataset-clients.js
function upsertOrDeleteDatasetRecords(datasetId, datasetClientId, records, cb) {
  var datasetRecordsCol = mongoClient.collection(getDatasetRecordsCollectionName(datasetId));
  debug('upsertOrDeleteDatasetRecords diff records = %j', records);
  async.mapLimit(records, RECORDS_UPDATE_CONCURRENCY, function(record, callback) {
    var op = record.op;
    var fields = {
      uid: record.uid,
      hash: record.hash,
      data: record.data
    };
    var update = {'$set': fields};
    if (op === 'update') {
      //$addToSet only adds the value to the array if it doesn't exist
      update['$addToSet'] = {'refs': datasetClientId};
    } else if (op === 'delete') {
      //remove the ref
      update['$pull'] = {'refs': datasetClientId};
    }
    datasetRecordsCol.findOneAndUpdate({uid: record.uid}, update, {upsert: true, returnOriginal: false}, function(err, updated) {
      if (err) {
        return callback(err);
      } else {
        return callback(null, updated.value);
      }
    });
  }, function(err, updates) {
    if (err) {
      debugError('Failed to upsertOrDeleteDatasetRecords due to error %s :: datasetId = %s', err, datasetId);
      return cb(err);
    }
    var returned = _.compact(updates);
    return cb(null, returned);
  });
}
function upsertOrDeleteDatasetRecords(datasetId, datasetClientId, records, cb) {
  var datasetRecordsCol = mongoClient.collection(getDatasetRecordsCollectionName(datasetId));
  debug('upsertOrDeleteDatasetRecords diff records = %j', records);
  async.mapLimit(records, RECORDS_UPDATE_CONCURRENCY, function(record, callback) {
    var op = record.op;
    var fields = {
      uid: record.uid,
      hash: record.hash,
      data: record.data
    };
    var update = {'$set': fields};
    if (op === 'update') {
      //$addToSet only adds the value to the array if it doesn't exist
      update['$addToSet'] = {'refs': datasetClientId};
    } else if (op === 'delete') {
      //remove the ref
      update['$pull'] = {'refs': datasetClientId};
    }
    datasetRecordsCol.findOneAndUpdate({uid: record.uid}, update, {upsert: true, returnOriginal: false}, function(err, updated) {
      if (err) {
        return callback(err);
      } else {
        return callback(null, updated.value);
      }
    });
  }, function(err, updates) {
    if (err) {
      debugError('Failed to upsertOrDeleteDatasetRecords due to error %s :: datasetId = %s', err, datasetId);
      return cb(err);
    }
    var returned = _.compact(updates);
    return cb(null, returned);
  });
}
Either insert, update or delete the given records from the `fhsync-<datasetId>-records` collection.
@param {String} datasetId
@param {storage~DatasetClientRecord[]} records the array of records that need to be updated
@param {Function} cb