couchjs

#!/usr/bin/env node

/*
    Tend to be ugly!!!
 */
var fs   = require('fs');
var util = require('util');
var vm   = require('vm');

var reader = (function() {
  var SharedCouchAPI = null;
  var InitSharedCouchAPI = function () {
    this.sender = [];
    this.start = { headers: {} };
    this.stop = undefined;
    this.events = {};
    this.rows = [];
  };

  var CouchAPI = function() { }
  CouchAPI.prototype = {
      log: function(str) { process.stdout.write(JSON.stringify(["log", util.inspect(str)])+"\n", 'utf-8'); },
      sum: function(arr) { var ret=0; for(var i in arr) { ret += arr[i] }; return ret; },
      isArray: Array.isArray,
      start: function(obj) { SharedCouchAPI.start = obj },
      stop: function(obj) { SharedCouchAPI.stop = true; SharedCouchAPI.sender.push(obj); },
      send: function(obj) { SharedCouchAPI.sender.push(obj); },
      on: function(name, fn)  { 
        if (!SharedCouchAPI.events[name]) { SharedCouchAPI.events[name] = [] }
        SharedCouchAPI.events[name].push(fn);
      }
  }

  var map_fn = {
                  require: require,
                  log: CouchAPI.prototype.log,
                  isArray: Array.isArray,
                  doIts: [],
                  emit: function(key, val) { out_data.push([key, val]); }
               };
  var reduce_fn = null;
  var reduce_fns = {};
  var out_data = [];
  var ddocs = {};
  var state = {};
  return function(line) {
    try {
       var op = JSON.parse(line);
    } catch(e) {
      return null;
    }
//console.error("OP:"+line);
    var factory = {
      reset: function(opts) { 
        map_fn.doIts = [];
        state = opts || {};
        SharedCouchAPI = new InitSharedCouchAPI();
        return true; 
      },    
      add_fun: function(fn) { 
//console.error("add_fun:"+fn);
        vm.runInNewContext("doIts.push((function(){ return "+fn+";})());", map_fn);
        return true; 
      },    
      ddoc: function() {
        var opts = Array.prototype.slice.call(arguments);
        var key = opts.shift();
        if (key == 'new') {
          key = opts.shift();
          ddocs[key] = opts.shift();
          return true;
        } else  {
          var base = ddocs[key];
          var path = opts.shift();
          for(var i in path) { base = base[path[i]]; }
          if (typeof(base) != 'function') {
            try {
              var fn_str = base;
              base = new CouchAPI();
              vm.runInNewContext("doIt = "+fn_str, base);
            } catch(e) {
              base = { doIt: function() { return e+"=>"+fn_str; } };
            }
          } 
          var style = path[0];
          try {
            var styler = {
              "shows": function(args) {
//console.error("*******************:"+base.doIt);
                var resp = base.doIt.apply(base,args);
                resp = (typeof(resp) == 'object' && resp) || {body: resp};
                return ["resp", resp];
              },
              "lists": function(args) {
                SharedCouchAPI = new InitSharedCouchAPI();
                base.listsArgs = args;
                base.doIt.apply(base, args);
                SharedCouchAPI.sender = []; // reset send output
                for(var i in SharedCouchAPI.events.head) {
                  SharedCouchAPI.events.head[i]();
                }
                SharedCouchAPI.enteredLists = true;
                return ["start", SharedCouchAPI.sender, SharedCouchAPI.start];
              },
              "updates": function(args) {
                var resp = base.doIt.apply(base,args);
                var doc = resp.shift();
                var resp = resp.shift();
                resp = (typeof(resp) == 'object' && resp) || {body: resp};
                return ["up", doc, resp ];
              },
              "filters": function(args) {
                var ret = [];
                var docs = args.shift();
                var req = args.shift();
                for(var i in docs) {
                  ret.push(!!base.doIt(docs[i], req));
                }
                return [true, ret];
              },
              "validate_doc_update": function(args) {
                 var ret = base.doIt.apply(base, args);
                 return (typeof(ret) == 'undefined' && 1) || ret;
              }
            };
            if (SharedCouchAPI && SharedCouchAPI.enteredLists) {
              return ["error","list_error","not a row '"+style+"'"]
            }
            return styler[style].apply(styler, opts);
          } catch(e) {
            return e;
          }
        }
      },
      rereduce: function(func_str, data) {
        return this.reduce(func_str, data, true);
      },
      list_row: function(data) {
        if (SharedCouchAPI.stop) {
          return ["error", "unknown_command", "list_row"]
        }
        SharedCouchAPI.sender = [];
        for(var i in SharedCouchAPI.events.row) {
          SharedCouchAPI.events.row[i](data);
          if (SharedCouchAPI.stop) {
            SharedCouchAPI.enteredLists = false;
            return ["end", SharedCouchAPI.sender];
          }
        }
        return ["chunks", SharedCouchAPI.sender];
      },
      list_end: function() {
        SharedCouchAPI.sender = [];
        var ret = [];
        for(var i in SharedCouchAPI.events.tail) {
          ret.push(SharedCouchAPI.events.tail[i]());
        }
        SharedCouchAPI.enteredLists = false;
        return ["end", ret];
      },
      reduce:  function(fns, data, rereduce) { 
        reduce_fn = new CouchAPI();
        var keys = [];
        var values = [];
        if (data.length > 0 && data[0].shift) {
          for(var j = 0, l = data.length; j < l; ++j) {
            var d = data[j];
            keys.push(d[0]);
            values.push(d[1]);
          }
        } else {
          for(var j = 0, l = data.length; j < l; ++j) {
            var d = data[j];
            values.push(d);
          }
        }
        var func = reduce_fns[fns[0]];
        if (!func) {
          vm.runInNewContext("doIt = "+fns[0], reduce_fn);
          func = reduce_fns[func] = reduce_fn;
        }
        /* 
        if (state.query_config && state.query_config.reduce_limit &&
            reduce_length > 200 && ((reduce_length * 2) > State.line_length)) {
          var reduce_preview = "Current output: '"+(reduce_line.substring(0,100) + "'... (first 100 of "+reduce_length+" bytes)");
          throw(["error",
            "reduce_overflow_error",
            "Reduce output must shrink more rapidly: "+reduce_preview]);
        }
        */
        return [true, [func.doIt(keys, values, rereduce)]];
      },
      map_doc: function(data) { 
        try {
          var ret = [];
          for(var i in map_fn.doIts) {
            out_data = [];
            map_fn.doIts[i](data);
            ret.push(out_data);
          }
          return ret;
        } catch(e) {
          console.error('MAP_DOC:ERROR:'+e+" with "+data._id);
          // UGLY.....
          process.stdout.write(JSON.stringify(["log", e +" with " + data._id])+"\n", 'utf-8');
          return [[], [[data._id, null]]];
        }
      }
    };
    var opt = op.shift();
    if (SharedCouchAPI && SharedCouchAPI.enteredLists && !(opt in {list_row:0,list_end:0})) {
      var out = ["error","list_error","not a row '"+opt+"'"]
    } else {
//console.error("FACTORY:"+opt);
      var out = factory[opt].apply(factory, op);
    }
    if (out !== null) {
      var need_exit = false;
      if (out[0] == "fatal") {
        out[0] = "error";
        need_exit = true;
      }
//console.error("OUT:"+util.inspect(out));
      process.stdout.write(JSON.stringify(out)+"\n", 'utf-8');
      if (need_exit || out[0] == "error" && (out[1] in {fatal:0,unknown_command:1,list_error:2})) {
        process.exit(1);
      }
    }
  }
})();


process.stdin.resume();
process.stdin.setEncoding('utf-8');

(function(stream, fn) {
  var queue = [];
  stream.on('data', function(chunk) {
    var str = chunk.toString('utf-8');
    var nl_idx = str.indexOf("\n");
    if (nl_idx >= 0) {
      queue.push(str.slice(0,nl_idx));  
      fn(queue.join(''));      
      queue = [str.slice(nl_idx+1, str.length)];
    } else {
      queue.push(str);  
    }
  });
})(process.stdin, reader);