<commit_before>var  moment = require('moment'),
    sqlQueries = require('./sql');

/**
 * @param {Object} db The db client to use.
 * @param {int|null} jobId The ID  of the job to add a job snapshot to.
 *                         If null, then this is a new job and a fresh job ID
 *                         is assigned by the sequence on the database.
 * @param {int} processIn Number of milliseconds from now the job should
 *                        become eligible to obtain service.
 * @param {Object} data The job data.
 * @param {function} callback(err, jobId)
 */
exports.write = function(db, jobId, processIn, data, callback) {
  var processAt = (processIn === null) ?
      null :
      // parseInt for injection attack prevention.
      "NOW() + INTERVAL '" + parseInt(processIn, 10) + " milliseconds'";
  var sql;
  if (jobId !== null) {
    sql =
      "INSERT INTO job_snapshots (job_id, data, process_at) VALUES ($1, $2, " +
      processAt + ");";
    db.query(sql, [jobId, data], callback);
  } else {
    sql =
      "INSERT INTO job_snapshots (data, process_at) VALUES ($1, " +
      processAt + ") RETURNING job_id;";
    db.query(sql, [data], function (err, result) {
      if(err) return callback(err);
      return callback(null, result.rows[0].job_id);
    });
  }
};

exports.setProcessedNow = function(db, jobId) {
  db.query(sqlQueries.setProcessedNow, [jobId]);
};

exports.nextToProcess = function(db, callback) {
  db.query(sqlQueries.obtainNextUnlockedJob, returnResult);

  function returnResult(err, result) {
    if(err) {
      return callback(err);
    }
    if(!result.rows) {
      return callback();
    }
    callback(null, result.rows[0]);
  }
};

exports.unlock = function(db, jobId) {
  db.query(sqlQueries.unlockJob, [jobId]);
};

exports.obtainLock = function(db, jobId, callback) {
  db.query(sqlQueries.obtainLockForJob, [jobId], gotLock);

  function gotLock(err, result) {
    if (err) return callback(err);
    db.query(sqlQueries.getJobData, [jobId], gotData);
    function gotData(err, result) {
      if (err) return callback(err);
      callback(null, result.rows[0]);
    }
  }
};
<commit_msg>Address CR.<commit_after>var  moment = require('moment'),
    sqlQueries = require('./sql');

/**
 * @param {Object} db The db client to use.
 * @param {int|null} jobId The ID  of the job to add a job snapshot to.
 *                         If null, then this is a new job and a fresh job ID
 *                         is assigned by the sequence on the database.
 * @param {int} processIn Number of milliseconds from now the job should
 *                        become eligible to obtain service.
 * @param {Object} data The job data.
 * @param {function} callback(err, jobId)
 */
exports.write = function(db, jobId, processIn, data, callback) {
  var processAt = (processIn === null) ?
      null :
      // parseInt for injection attack prevention.
      "NOW() + INTERVAL '" + parseInt(processIn, 10) + " milliseconds'";
  var sql;
  if (jobId !== null) {
    sql =
      "INSERT INTO job_snapshots (job_id, data, process_at) VALUES ($1, $2, " +
      processAt + ");";
    db.query(sql, [jobId, data], callback);
  } else {
    sql =
      "INSERT INTO job_snapshots (data, process_at) VALUES ($1, " +
      processAt + ") RETURNING job_id;";
    db.query(sql, [data], function (err, result) {
      if(err) return callback(err);
      return callback(null, result.rows[0].job_id);
    });
  }
};

exports.setProcessedNow = function(db, jobId) {
  db.query(sqlQueries.setProcessedNow, [jobId]);
};

exports.nextToProcess = function(db, callback) {
  db.query(sqlQueries.obtainNextUnlockedJob, returnResult);

  function returnResult(err, result) {
    if(err) {
      return callback(err);
    }
    if(!result.rows) {
      return callback();
    }
    callback(null, result.rows[0]);
  }
};

exports.unlock = function(db, jobId) {
  db.query(sqlQueries.unlockJob, [jobId]);
};

exports.obtainLock = function(db, jobId, callback) {
  db.query(sqlQueries.obtainLockForJob, [jobId], gotLock);

  function gotLock(err, result) {
    if (err) return callback(err);
    db.query(sqlQueries.getJobData, [jobId], gotData);
  }

  function gotData(err, result) {
    if (err) return callback(err);
    callback(null, result.rows[0]);
  }
};

{'fraction_non_alphanumeric': 0.14019127595054817, 'fraction_numerical': 0.0037322136692325637, 'mean_word_length': 2.9777365491651206, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '5766112', 'n_tokens_mistral': 1399, 'n_tokens_neox': 1373, 'n_words': 381}