generate-edit-links-murals.js

var getDb = require('./mongo-getdb2.js'),
    getSourceDb = require('./mongo-getdb3.js'),
    humanize = require('humanize'),
    async = require('async'),
    util = require('util'),
    url = require('url');

var fetched = 0, step = 100, skip = 0;

console.log("[%s] Started re-index from editLinks", new Date().toString());

async.doWhilst(function(allDone){
    getSourceDb(function(source) {
      getDb(function(db) {
        var criteria = {};
        var fields = { id: 1, ownerId: 1, editLink: 1, hasEditPass: 1, editHash: 1, viewLink: 1, hasViewPass: 1, viewHash: 1 };
        var cursor = source.collection('muralcontents').find(criteria, fields);

        cursor.limit(step).skip(skip).toArray(function(err, murals) {
          if(err) return onFetchError(allDone);

          async.eachLimit(murals, 50, logAndContinue(deliver(db)), function(err) {
            skip += (fetched = murals.length);
            console.log("[%s] Processed %s murals so far", new Date(), skip);
            allDone(err);
          });
        });
      });
    });
}, function(){ return fetched != 0; }, function(err) {
  console.log("[%s] Completed re-index from backup", new Date().toString());
  process.exit(0);
});

function deliver(db) {
  return function(mural, done) {
    async.series([
      function(callback) {
        if(!mural.editLink) {
          console.log("skipping edit for %s mural", mural.id);
          return callback();
        }

        var editLink = util.format("https://mural.ly/api/murals/%s/share/edit%s",
          mural.id, (mural.hasEditPass ? "" : "/" + mural.editHash));

        var hash = url.parse(mural.editLink).pathname.replace(/^\//, '');

        console.log("%s, %s", editLink, hash);
        console.log("processing edit link for %s mural from %s", mural.id, mural.ownerId);

        db.collection('links').update({ _id: hash }, { _id: hash, url: editLink }, { upsert: true }, callback);
      },
      function(callback) {
        if(!mural.viewLink) {
          console.log("skipping view for %s mural", mural.id);
          return callback();
        }

        var viewLink = util.format("https://mural.ly/api/murals/%s/share/view%s",
          mural.id, (mural.hasViewPass ? "" : "/" + mural.viewHash));

        var hash = url.parse(mural.viewLink).pathname.replace(/^\//, '');

        console.log("%s, %s", viewLink, hash);
        console.log("processing view link for %s mural from %s", mural.id, mural.ownerId);

        db.collection('links').update({ _id: hash }, { _id: hash, url: viewLink }, { upsert: true }, callback);
      }
    ], done);
  };
}

function onFetchError(callback) {
  console.log("[%s] There's an error while retrieving murals", new Date().toString());
  callback();
}

function logAndContinue(fn) {
  return function(elem, done) {
    fn(elem, function(err) {
        if(err) console.log("[%s] There's an error while processing: %s", new Date().toString(), err);
        done();
    });
  };
}

mongo-getdb2.js

var MongoClient = require('mongodb').MongoClient;

var SERVER_OPTIONS = { auto_reconnect: true, poolSize: 4 };
var REPLSET_OPTIONS = { socketOptions: {  keepAlive: 1, connectTimeoutMS: 0, socketTimeoutMS: 0 }, ha: true, haInterval: 1000 };
var OPTIONS = { db: { native_parser: false, numberOfRetries: 10 }, server: SERVER_OPTIONS, replSet: REPLSET_OPTIONS };

var current = null;
var retryCount = 0;
var await = [];

module.exports = function(callback) {
	if(current) {
		var poolConnected = (current.serverConfig.connectionPool && current.serverConfig.isConnected());

		if(current.state == 'connected' || poolConnected) {
			return callback(current);
		}

		current = null;
	}

	await.push(callback); // create a wait queue
	if(await.length > 1) return;

	connectToDB();
};

function connectToDB() {
	MongoClient.connect(process.env.MONGOLAB_URI, OPTIONS, function(err, db) {
		if(err) return onConnectionError(err);

		// re-open on close
		db.on('close', function() { db.open(function(err2){ throw err2; }); });

		// return through the whole chain
		await.forEach(function(cb) { cb(current = db); });
		retryCount = 0;
		await = [];
	});
}

function onConnectionError(err) {
	if(err && retryCount > 3) throw err;
	retryCount += 1;

	// wait a couple of seconds a stepDown
	// may be happening right now. Although the switch
	// is almost immediate there's some "Not master found" time.
	setTimeout(connectToDB, 2 * 1000);
}

mongo-getdb3.js

var MongoClient = require('mongodb').MongoClient;

var SERVER_OPTIONS = { auto_reconnect: true, poolSize: 4 };
var REPLSET_OPTIONS = { socketOptions: {  keepAlive: 1, connectTimeoutMS: 0, socketTimeoutMS: 0 }, ha: true, haInterval: 1000 };
var OPTIONS = { db: { native_parser: false, numberOfRetries: 10 }, server: SERVER_OPTIONS, replSet: REPLSET_OPTIONS };

var current = null;
var retryCount = 0;
var await = [];

module.exports = function(callback) {
	if(current) {
		var poolConnected = (current.serverConfig.connectionPool && current.serverConfig.isConnected());

		if(current.state == 'connected' || poolConnected) {
			return callback(current);
		}

		current = null;
	}

	await.push(callback); // create a wait queue
	if(await.length > 1) return;

	connectToDB();
};

function connectToDB() {
	MongoClient.connect(process.env.MONGOLAB_URI_SOURCE, OPTIONS, function(err, db) {
		if(err) return onConnectionError(err);

		// re-open on close
		db.on('close', function() { db.open(function(err2){ throw err2; }); });

		// return through the whole chain
		await.forEach(function(cb) { cb(current = db); });
		retryCount = 0;
		await = [];
	});
}

function onConnectionError(err) {
	if(err && retryCount > 3) throw err;
	retryCount += 1;

	// wait a couple of seconds a stepDown
	// may be happening right now. Although the switch
	// is almost immediate there's some "Not master found" time.
	setTimeout(connectToDB, 2 * 1000);
}

package.json

{
  "name": "murally-webjobs",
  "version": "0.0.1",
  "private": true,
  "dependencies": {
    "request": "~2.34.0",
    "azure": "~0.8.1",
    "async": "~0.2.10",
    "underscore": "~1.6.0",
    "aws-sdk": "~2.0.0-rc10",
    "tmp": "0.0.23",
    "mime-of": "~0.1.0",
    "mongodb": "~1.3.23",
    "humanize": "0.0.9"
  },
  "engines": {
    "node": "0.10.x",
    "npm": ">=1.0.0"
  }
}