inter.js

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


var get_state = function(cb) {
    
    db.get('_state', function(err,doc) {
        
        if(err) {
            console.log("processed_seq does not exist");
            cb(null, 0);
        }
        else {
            var seq = doc.processed_seq;

            console.log("processed sequence = "+seq);
            if(seq)
                cb(null, seq);
            else
                cb(new Error("uups"), 0);
        }
    });
}

var set_state = function(seq, cb) {
    
    db.get('_state', function(err,doc) {
        
        if(err) {
            console.log("processed_seq does not exist");
            db.save('_state', {processed_seq:seq}, function(e, res) {
              console.log("created state document");
              cb(null, seq)
            });
        }
        else {
            var rev = doc._rev;
            db.save('_state',rev, {processed_seq: seq}, function(e,res){
              console.log("saved state@" + rev + " = "+ seq);
              cb(null, seq)
            });
   
        }
    });
}

// retrieve changes
var act_on_changes = function() {

    get_state(function(e, seq) {
    
        console.log("get state = " + seq);

        var opts = {filter: 'documents/all'};
        if (seq) opts.since = seq;

        db.changes(opts).on('response',function(res) {
            res.on('data',function(change) {
                console.log("changes:");
                console.log(change);
                console.log("save_seq = " + seq);
                newseq = Math.max(change.seq, seq);
    
                // persist procecced changes
                set_state(newseq, function(err, s) {
                    console.log("saved seq = " + s);
                });

            });
            res.on('end', function() {
                console.log("oops, connection closed");
            });
        });
    });
};

// loop
act_on_changes();