lib/api-sync.js
function sync(datasetId, params, cb) {
  debug('[%s] process sync request for the dataset', datasetId);
  var queryParams = params.query_params || {};
  var metaData = params.meta_data || {};
  var datasetClient = new DatasetClient(datasetId, {queryParams: queryParams, metaData: metaData});
  debug('[%s] processing sync API request :: query_params = %j:: meta_data = %j', datasetId, queryParams, metaData);
  async.series({
    requestInterceptor: async.apply(interceptors.requestInterceptor, datasetId, params),
    readDatasetClient: function checkDatasetclientStopped(callback) {
      syncStorage.readDatasetClient(datasetClient.getId(), function(err, datasetClientJson) {
        if (err) {
          return callback(err);
        }
        if (datasetClientJson && datasetClientJson.stopped === true) {
          return callback(new Error('sync stopped for dataset ' + datasetId));
        } else {
          return callback(null, datasetClientJson);
        }
      });
    }
  }, function(err, results) {
    if (err) {
      debugError('[%s] sync request returns error = %s %j', datasetId, err, params);
      return cb(err);
    }
    return processSyncAPI(datasetId, params, results.readDatasetClient, cb);
  });
}
function sync(datasetId, params, cb) {
  debug('[%s] process sync request for the dataset', datasetId);
  var queryParams = params.query_params || {};
  var metaData = params.meta_data || {};
  var datasetClient = new DatasetClient(datasetId, {queryParams: queryParams, metaData: metaData});
  debug('[%s] processing sync API request :: query_params = %j:: meta_data = %j', datasetId, queryParams, metaData);
  async.series({
    requestInterceptor: async.apply(interceptors.requestInterceptor, datasetId, params),
    readDatasetClient: function checkDatasetclientStopped(callback) {
      syncStorage.readDatasetClient(datasetClient.getId(), function(err, datasetClientJson) {
        if (err) {
          return callback(err);
        }
        if (datasetClientJson && datasetClientJson.stopped === true) {
          return callback(new Error('sync stopped for dataset ' + datasetId));
        } else {
          return callback(null, datasetClientJson);
        }
      });
    }
  }, function(err, results) {
    if (err) {
      debugError('[%s] sync request returns error = %s %j', datasetId, err, params);
      return cb(err);
    }
    return processSyncAPI(datasetId, params, results.readDatasetClient, cb);
  });
}
Process the sync request. It will
- validate the request body via the requestInterceptor
- check if the dataset client is stopped for sync
- create or update the dataset client
- process acknowledgements, push each of them to the ackQueue
- process pending changes, push each of them to the pendingQueue
- list any processed updates for the given client
@param {String} datasetId the id of the dataset
@param {Object} params the request body, it normally contain those fields:
@param {Object} params.query_params the query parameter for the dataset from the client
@param {Object} params.meta_data the meta data for the dataset from the client
@param {Object} params.__fh an object added by the client sdk. it could have the `cuid` of the client
@param {Array} params.pending the pending changes array
@param {Array} params.acknowledgements the acknowledgements from the client
@param {Function} cb the callback function