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);
  });
}
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
Either insert update or delete the given records from the fhsync - <datasetId > - records collection .