inter.js

var app = require('app')
var assert = require('assert')
var db  = app.db;                                                                           


var get_state = function(callback) {

    db.get('_state', function(err,doc) {

        if(err) {
            console.log("processed_seq does not exist");
            // -------- CHANGED
            callback(err);
            return;
        }
        else {
            var seq = doc.processed_seq;

            console.log("processed sequence = "+seq);
            // -------- CHANGED
            if(seq != null && seq != undefined) {
              callback(null, seq);
              return;
            }
            // -------- CHANGED
            else {
              callback(new Error("Something went wrong"), null);
            }

        }
    });
}

var set_state = function(seq) {

    db.get('state', function(err,doc) {

        if(err) {
            console.log("processed_seq does not exist");
            db.save('state',{processed_seq:0},function(e,res){console.log("saved state");});
        }
        else { 
            var rev = doc._rev;
            db.save('state',rev,{processed_seq:seq},function(e,res){console.log("saved state");});

        }
    });
}

// retrieve changes
var act_on_changes = function() {
    
    var seq = get_state();
    
    console.log("get state = " + seq);
    db.changes({since: seq }).on('response',function(res) {
        res.on('data',function(change) {
            console.log("changes:");
            console.log(change);
            console.log("save_seq = " + seq);
            seq++;
            // persist procecced changes
            set_state(seq);
        });
        res.on('end', function() {
            console.log("oops, connection closed");
        });
    });
};

// loop
act_on_changes();