<commit_before>/**
 * Copyright 2016 Netsend.
 *
 * This file is part of PersDB.
 *
 * PersDB is free software: you can redistribute it and/or modify it under the
 * terms of the GNU Affero General Public License as published by the Free Software
 * Foundation, either version 3 of the License, or (at your option) any later
 * version.
 *
 * PersDB is distributed in the hope that it will be useful, but WITHOUT ANY
 * WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS FOR A
 * PARTICULAR PURPOSE. See the GNU Affero General Public License for more details.
 *
 * You should have received a copy of the GNU Affero General Public License along
 * with PersDB. If not, see <https://www.gnu.org/licenses/>.
 */

/* jshint -W116 */

'use strict';

var programName = require('path').basename(__filename, '.js');

if (process.getuid() !== 0) {
  console.error('%s: execute as root', programName);
  process.exit(1);
}

if (typeof process.send !== 'function') {
  console.error('%s: use child_process.spawn to invoke this module', programName);
  process.exit(2);
}

var net = require('net');
var url = require('url');

var async = require('async');
var chroot = require('chroot');
var BSONStream = require('bson-stream');
var mongodb = require('mongodb');
var posix = require('posix');
var xtend = require('xtend');

var logger = require('../../lib/logger');
var noop = require('../../lib/noop');
var OplogTransform = require('./oplog_transform');
var filterSecrets = require('../../lib/filter_secrets');

var MongoClient = mongodb.MongoClient;

/**
 * Instantiate a connection to a mongo database and read changes of a collection
 * via the oplog. Furthermore accept new versions from a db_exec instance and try
 * to write them to the mongo collection.
 *
 * Full FSM: init --> listen
 *
 * The first message emitted is "init" which signals that this process is ready to
 * receive the configuration data. File desciptors for the log can optionally be
 * set at fd 4 and 5. A version control channel is expected at fd 7 and a data
 * exchange channel is expected at fd 6.
 *
 * {
 *   log:            {Object}      // log configuration
 *   [url]:          {String}      // connection string
 *                                 // defaults to 127.0.0.1:27017
 *   [oplogTransformOpts]: {Object} // any oplog transform options
 *   [master]:       {Boolean}     // whether or not to enable writes
 *                                 // defaults to false
 *   [chroot]:       {String}      // defaults to /var/empty
 *   [user]:         {String}      // defaults to "_pdbnull"
 *   [group]:        {String}      // defaults to "_pdbnull"
 * }
 *
 * Only a kill message can be sent to signal the end of the process.
 *
 * As soon as the connection to mongodb is setup, and version and data channels are
 * ready, a "listen" message is emitted.
 */

// globals
var ot, coll, db, log; // used after receiving the log configuration

/**
 * Start oplog transformer:
 *  * pass new versions and confirms to data channel
 *  * receive new versions with parent from data channel
 *
 * @param {mongodb.Db} oplogDb  connection to the oplog database
 * @param {String} oplogCollName  oplog collection name, defaults to oplog.$main
 * @param {String} ns  namespace of the database.collection to follow
 * @param {Object} versionControl  version control channel
 * @param {Object} dataChannel  data channel
 * @param {Object} [opts]
 *
 * Options:
 *   any OplogTransform options
 */
function start(oplogDb, oplogCollName, ns, dataChannel, versionControl, opts) {
  // track newly written versions so that new items can be recognized correctly
  var expected = {};

  log.debug('start oplog transform...');
  ot = new OplogTransform(oplogDb, oplogCollName, ns, versionControl, versionControl, xtend(opts, {
    log: log,
    bson: true,
    expected: expected
  }));
  ot.pipe(dataChannel);
  ot.startStream();

  var bs = new BSONStream();
  dataChannel.pipe(bs).on('readable', function() {
    var obj = bs.read();

    if (!obj) {
      log.notice('local data channel closed, expecting shutdown');
      return;
    }

    // remove parent so that this item can be used later as a local confirmation
    delete obj.n.h.pa;

    expected[obj.n.h.id] = obj.n;

    if (obj.o == null || obj.o.b == null) { // insert
      if (obj.n == null) { throw new Error('new object expected'); }
      // put either mongo id or h.id back on the document
      obj.n.b._id = obj.n.m && obj.n.m._id || obj.n.h.id;
      coll.insertOne(obj.n.b, function(err, r) {
        if (err) { throw err; }
        if (!r.insertedCount) {
          log.notice('item not inserted %j', obj.n.h);
        } else {
          log.debug('item inserted %j', obj.n.h);
        }
      });
    } else if (obj.n.h.d === true) { // delete
      if (obj.o == null) { throw new Error('old object expected'); }
      coll.deleteOne({ _id: obj.n.h.id }, function(err, r) {
        if (err) { throw err; }
        if (!r.deletedCount) {
          log.notice('item not deleted %j', obj.o.h);
        } else {
          log.debug('item deleted %j', obj.o.h);
        }
      });
    } else { // update
      // put mongo id back on the document
      obj.n.b._id = obj.n.m && obj.n.m._id || obj.n.h.id;
      coll.findOneAndReplace(obj.o.b, obj.n.b, function(err, r) {
        if (err) { throw err; }
        if (!r.lastErrorObject.n) {
          log.notice('item not updated %j', obj.n.h);
        } else {
          log.debug('item updated %j', obj.n.h);
        }
      });
    }
  });
}

/**
 * Expect one init request (request that contains config)
 * {
 *   log:            {Object}      // log configuration
 *   url:            {String}      // mongodb connection string
 *   coll:           {String}      // collection name
 *   [dbUser]:       {String}      // user to read the collection
 *   [oplogDbUser]:  {String}      // user to read the oplog database collection
 *   [secrets]:      {Object}      // object containing the passwords for dbUser
 *                                 // and oplogDbUser
 *   [authDb]:       {String}      // authDb database, defaults to db from url<commit_msg>mongo: be more clever about creating an _id<commit_after>/**
 * Copyright 2016 Netsend.
 *
 * This file is part of PersDB.
 *
 * PersDB is free software: you can redistribute it and/or modify it under the
 * terms of the GNU Affero General Public License as published by the Free Software
 * Foundation, either version 3 of the License, or (at your option) any later
 * version.
 *
 * PersDB is distributed in the hope that it will be useful, but WITHOUT ANY
 * WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS FOR A
 * PARTICULAR PURPOSE. See the GNU Affero General Public License for more details.
 *
 * You should have received a copy of the GNU Affero General Public License along
 * with PersDB. If not, see <https://www.gnu.org/licenses/>.
 */

/* jshint -W116 */

'use strict';

var programName = require('path').basename(__filename, '.js');

if (process.getuid() !== 0) {
  console.error('%s: execute as root', programName);
  process.exit(1);
}

if (typeof process.send !== 'function') {
  console.error('%s: use child_process.spawn to invoke this module', programName);
  process.exit(2);
}

var net = require('net');
var url = require('url');

var async = require('async');
var chroot = require('chroot');
var BSONStream = require('bson-stream');
var mongodb = require('mongodb');
var posix = require('posix');
var xtend = require('xtend');

var logger = require('../../lib/logger');
var noop = require('../../lib/noop');
var OplogTransform = require('./oplog_transform');
var filterSecrets = require('../../lib/filter_secrets');

var MongoClient = mongodb.MongoClient;
var ObjectID = mongodb.ObjectID;

/**
 * Determine id for mongodb. Try to get m._id, split h.id on 0x01 or just get h.id
 * without processing as a last resort.
 *
 * @param {Object} dagItem  item from the pdb DAG
 * @return {mixed} id for use in a mongodb collection
 */
function getMongoId(obj) {
  var id;
  try {
    id = obj.n.m._id;
  } catch (err) {
    // fallback
    id = obj.n.h.id;
  }

  if (typeof id === 'string') {
    // expect zero or one 0x01 byte
    var nid = id.split('\x01', 2)[1];
    if (!nid || !nid.length) {
      nid = id;
    }
    if (!nid.length) { throw new Error('could not determine id'); }

    // auto convert 24 character strings to ObjectIDs
    if (nid.length === 24) {
      try {
        nid = new ObjectID(nid);
      } catch(err) {
        log.err('could not convert id to object id %s %s', id, err);
      }
    }
  }
  return nid;
}

/**
 * Instantiate a connection to a mongo database and read changes of a collection
 * via the oplog. Furthermore accept new versions from a db_exec instance and try
 * to write them to the mongo collection.
 *
 * Full FSM: init --> listen
 *
 * The first message emitted is "init" which signals that this process is ready to
 * receive the configuration data. File desciptors for the log can optionally be
 * set at fd 4 and 5. A version control channel is expected at fd 7 and a data
 * exchange channel is expected at fd 6.
 *
 * {
 *   log:            {Object}      // log configuration
 *   [url]:          {String}      // connection string
 *                                 // defaults to 127.0.0.1:27017
 *   [oplogTransformOpts]: {Object} // any oplog transform options
 *   [master]:       {Boolean}     // whether or not to enable writes
 *                                 // defaults to false
 *   [chroot]:       {String}      // defaults to /var/empty
 *   [user]:         {String}      // defaults to "_pdbnull"
 *   [group]:        {String}      // defaults to "_pdbnull"
 * }
 *
 * Only a kill message can be sent to signal the end of the process.
 *
 * As soon as the connection to mongodb is setup, and version and data channels are
 * ready, a "listen" message is emitted.
 */

// globals
var ot, coll, db, log; // used after receiving the log configuration

/**
 * Start oplog transformer:
 *  * pass new versions and confirms to data channel
 *  * receive new versions with parent from data channel
 *
 * @param {mongodb.Db} oplogDb  connection to the oplog database
 * @param {String} oplogCollName  oplog collection name, defaults to oplog.$main
 * @param {String} ns  namespace of the database.collection to follow
 * @param {Object} versionControl  version control channel
 * @param {Object} dataChannel  data channel
 * @param {Object} [opts]
 *
 * Options:
 *   any OplogTransform options
 */
function start(oplogDb, oplogCollName, ns, dataChannel, versionControl, opts) {
  // track newly written versions so that new items can be recognized correctly
  var expected = {};

  log.debug('start oplog transform...');
  ot = new OplogTransform(oplogDb, oplogCollName, ns, versionControl, versionControl, xtend(opts, {
    log: log,
    bson: true,
    expected: expected
  }));
  ot.pipe(dataChannel);
  ot.startStream();

  var bs = new BSONStream();
  dataChannel.pipe(bs).on('readable', function() {
    var obj = bs.read();

    if (!obj) {
      log.notice('local data channel closed, expecting shutdown');
      return;
    }

    // remove parent so that this item can be used later as a local confirmation
    delete obj.n.h.pa;

    expected[obj.n.h.id] = obj.n;

    var mongoId = getMongoId(obj);
    if (obj.o == null || obj.o.b == null) { // insert
      if (obj.n == null) { throw new Error('new object expected'); }
      // put either mongo id or h.id back on the document
      obj.n.b._id = mongoId;
      coll.insertOne(obj.n.b, function(err, r) {
        if (err) { throw err; }
        if (!r.insertedCount) {
          log.notice('item not inserted %j', obj.n.h);
        } else {
          log.debug('item inserted %j', obj.n.h);
        }
      });
    } else if (obj.n.h.d === true) { // delete
      if (obj.o == null) { throw new Error('old object expected'); }
      coll.deleteOne({ _id: mongoId }, function(err, r) {
        if (err) { throw err; }
        if (!r.deletedCount) {
          log.notice('item not deleted %j', obj.o.h);
        } else {
          log.debug('item deleted %j', obj.o.h);
        }
      });
    } else { // update
      // put mongo id back on the document
      obj.n.b._id = mongoId;
      coll.findOneAndReplace(obj.o.b, obj.n.b, function(err, r) {
        if (err) { throw err; }
        if (!r.lastErrorObject.n) {
          log.notice('item not updated %j', obj.n.h);
        } else {
          log.debug('item updated %j', obj.n.h);
        }
      });
    }
  });
}

/**
 * Expect one init request (request that contains config)
 * {
 *   log:            {Object}      // log configuration
 *   url:            {String}      // mongodb connection string
 *   coll:           {String}      // collection name
 *   [dbUser]:       {String}      // user to read the collection
 *   [oplogDbUser]:  {String}      // user to read the oplog database collection
 *   [secrets]:      {Object}      // object containing the passwords for dbUser
 *                                 // and oplogDbUser
 *   [authDb]:       {String}      // authDb database, defaults to db from url
{'fraction_non_alphanumeric': 0.12599318955732122, 'fraction_numerical': 0.004994324631101022, 'mean_word_length': 2.487991554499868, 'pattern_counts': {'":': 0, '<': 5, '<?xml version=': 0, '>': 7, 'https://': 2, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 2, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 6, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '5488334', 'n_tokens_mistral': 4057, 'n_tokens_neox': 3949, 'n_words': 1444}