triggerjob.js

process.env.NODE_PATH = "/usr/lib/nodejs:/usr/lib/node_modules:/usr/share/javascript";
//__dirname;
require('module').Module._initPaths();

var os = require("os");
var url = require('url');
var http = require('http');
var https = require('https');
var request = require('request');
var follow = require('follow');
var triggers = {};
var sysclient = http;
var thishost = [os.hostname()];

var args = process.argv.slice(2);
var config = {};
var port = 0;
var sysopt = {};
var feeds = [];

var loadTrigger = function(type) {
	if (typeof triggers[type]!=="undefined") return triggers[type];
	try {
		triggers[type] = require('./trigger-'+type);
	}catch(ex){
		log("CANNOT FIND .JS FILE FOR TRIGGER TYPE: \""+type.toString()+"\"");
		triggers[type] = false;
	}
	return triggers[type];
};

var log = function(mesg) {
  if (args.length>0) console.log(mesg);
  else console.log(JSON.stringify(["log", mesg]));
};

var escapeRegExp = function(string) {
    return string.replace(/([.*+?^=!:${}()|\[\]\/\\])/g, "\\$1");
};

var subst_config = function(obj) {
	// replace config values: <!myvar!> will be replaced with value of <triggerjob.myvar>
	if (typeof obj!="object") return obj;
	var newtr = JSON.stringify(obj);
	if (typeof obj.allowed == "object") {
		for (var i in obj.allowed) {
			var k = obj.allowed[i];
			if (config[k]) {
				newtr = newtr.replace(new RegExp(escapeRegExp("<!"+k+"!>"),'gi'),config[k]);
			}
		}
	}
	newtr = newtr.replace(/<![^!]+!>/gi,"<!VAR-UNKNOWN-OR-NOT-ALLOWED!>");
	return JSON.parse(newtr);
};


var markTriggerAs = function(pathurl,tries, dockey, trkey, action, code, out, cb) {
	//log(tries.toString()+" Marking trigger "+trkey+"@"+dockey+" as "+action);

	var ropt = { 
		method: 'PUT'
	    , url:pathurl+"/trigger/"+dockey
	    , headers : sysopt.headers
	    //, json : true
	    , body : JSON.stringify({
			'action' : action,
      		'trkey' : trkey,
      		'code' : code,
      		'out' : out	    	
	    })
	};
	//log("PREP REQUEST: "+JSON.stringify(ropt));
	//log("Trigger url: "+pathurl+"/trigger/"+dockey);

	request(ropt , function (error, response, body) {
		var e = (error||"").toString();
	  	if (e.length>0) {
	  		cb("REQUEST EXCEPTION:"+e);
	  	} else {
		  	//log("RESPONSE("+(response.statusCode||"").toString()+"):"+(body||"").substring(0,200));
			try {
        		switch(parseInt(response.statusCode)) {
        			case 200:
        			case 201:
        				cb(false,body);
        				break;

        			case 409:
	        			if (tries < 3) {
		        			setTimeout(function(){
		        				markTriggerAs(pathurl,tries+1, dockey, trkey, action, code, out, cb);
		        			},Math.random()*4000);
	        			} else {
	        				cb("Conflict persists after 3 attempts");
	        			}
        				break;

        			default:
        				cb(body)
        				break;
        		}
        	} catch(ex){
        		cb(ex);
        	}
	    }
	});


};

var executeTrigger = function(pathurl,dockey,trkey,tr,executecb){
	var cbdone = function(code,out){
		var iserrmsg = typeof code != "number";
		markTriggerAs(pathurl,0,dockey,trkey,'done',iserrmsg?false:code,iserrmsg?code:out,function(error,data){
			if (!error) {
	        	log("DONE "+ trkey+"@"+dockey+": "+(data||""));
	        	executecb();
			} else {
				executecb(error);
			}
		});		
	};

	var fn = loadTrigger((tr.type||"http").toString());
	if (fn) {
		markTriggerAs(pathurl,0,dockey,trkey,'queued',null,null,function(error,data){
			if (!error) {
	        	log("QUEUED "+ trkey+"@"+dockey+": "+(data||""));

				fn(log,sysopt,pathurl,tr,dockey,trkey,cbdone);
			} else {
				executecb(error);
			}
		})
	} else {
		executecb("Cannot find trigger type \""+(tr.type||"http")+"\"");
	}
};

var executeAt =function(pathurl,id,trkey,tr,cb) {
    var now = new Date().getTime();
	//tr.start = now+20000;
	if ((tr.delay||0)>0) {
		setTimeout(function(){
			markTriggerAs(pathurl,0,id,trkey,'delay',null,null,function(error,data){
				if (error) cb(error);
			});
		},2000+Math.random()*2000);
	} else {
		log("Should start at :"+(tr.start||0).toString()+" now:"+(now).toString());
		if ((tr.start||0) < now){
			setTimeout(function(){
				executeTrigger(pathurl,id,trkey,tr,cb);
			},Math.random()*4000);
		} else {
			var toelapse = tr.start-now;
			log("Postponing of :"+toelapse.toString()+" ms");
			setTimeout(function(){
				executeTrigger(pathurl,id,trkey,tr,cb);
			},toelapse);
		}
	}
};

var followChanges = function(path) {
	var cb=function(err){
		if (typeof err !== "undefined") {
			log("EXECUTEAT-CB: "+(err||"UNK-ERROR"));
		}
	}, opt=sysopt, pathurl=""; 
    opt.pathname = path;
	pathurl = url.format(opt);

	log("Following path: "+path);
	log("Following url: "+pathurl);
	return follow({db:pathurl+"/follow",headers:opt.headers}, function(error, change) {
	    //log("follow changes return a doc");
		  if(!error) {
		    log("Got change number " + change.seq + ": " + change.id);
		    for (var trkey in change.doc&&change.doc.triggers?change.doc.triggers:{}){
		    	var tr = change.doc.triggers[trkey];
				if (typeof tr.queued==="undefined" && (!tr.h||thishost.indexOf(tr.h||"")>=0)) {
					log("CALLING ExecuteAt "+trkey)
					executeAt(pathurl,change.id,trkey,subst_config(tr),cb);
				}
		    }
		  } else {
		  	log(error.toString());
		  }
	});
};

var start = function(paths){
	if (config.job_path) delete config.job_path;
	if (config.job_authorization) delete config.job_authorization;
	if (typeof paths == "string" && paths.length>0) {
		paths = paths.split(',');
		for (var i in paths) {
			if (paths[i].length>0) {
				feeds[paths[i]] = followChanges(paths[i]);
			}
		}
	} else {
		log("Invalid triggerjob.job_path");
	}
};

var cmdlineinit = function(){
	sysopt = url.parse(args[0]||"");
	sysopt.method = "GET";
	var credentials = "Basic "+(new Buffer(args[1]||"").toString('base64'));
    sysopt.headers = {authorization:credentials?credentials:""}

	var schema = ["http","https"].indexOf((sysopt.protocol||"").toLowerCase().split(":")[0]);
	if (schema === 1) {
		sysclient = https;
		sysopt.protocol = "https:";
		//sysopt.hostname = sysopt.host;
		//sysopt.port = sysopt.port||443;
	}

    var opt=sysopt; 
    opt.path = "/_config/triggerjob"; opt.method = "GET";
	var req = sysclient.request(opt, function(response) {
        var body = '';
        response.on('data', function(d) {
            body += d;
        });
        response.on('end', function(d) {
            try {
            	if (response.statusCode!=200) throw(body);
            	config = JSON.parse(body);
            	start(config.job_path);
            } catch(ex) {
            	log("init exception: "+ ex.toString());
            }
        });
    }).on('error', function(ex) {
    	log(ex);
	}).end();
};


// We use stdin in a couple ways. First, we
// listen for data that will be the requested
// port information. We also listen for it
// to close which indicates that CouchDB has
// exited and that means its time for us to
// exit as well.
var stdin = process.openStdin();

stdin.on('data', function(msg) {
	var parsed = {};
	//log("Receiving stdin message: " + msg);
	try {
		parsed = JSON.parse(msg);
		if (parseInt(parsed)==parsed) {
			port = parseInt(parsed);
			console.log(JSON.stringify(["register", "triggerjob"]));
			console.log(JSON.stringify(["get", "triggerjob"]));
		} else if (parsed.couchdb) { // welcome message
			//log("welcome message: " + msg);
		} else if (parsed.job_path)  { 
			//log("triggerjob config: " + msg);
			config = parsed;
			var credentials = "Basic "+(new Buffer(config.job_authorization||"").toString('base64'));

			sysclient = http;
			sysopt = url.parse("http://127.0.0.1:"+port.toString());
			//sysopt.host = "127.0.0.1:"+port;
			//sysopt.hostname = "127.0.0.1";
			delete sysopt.host;
			//sysopt.port = port;
			sysopt.method = "GET";
		    sysopt.headers = {authorization:credentials?credentials:""}

			start(config.job_path);
		} else {
			log("Discarding unknown message: " + msg);
		}
	} catch(ex){
		log("Not a JSON string: " + msg);
	}
});

stdin.on('end', function () {
	for (var i in feeds){
		log("Stopping feed "+feeds[i]);
		feeds[i].stop();
		delete feeds[i];
	}
	process.exit(0);
});

if (args.length>0) {
	// port and config from command line
	cmdlineinit();
} else {
	// port and config from stdin
	console.log(JSON.stringify(["get", "httpd", "port"]));
}