sync ris prefixes

var fetch=require("node-fetch"),sqlite3=require("sqlite3"),async=require("neo-async"),lf=require("localforage"),https=require('https');

var agent = new https.Agent({
    keepAlive: true,
   // keepAliveMsecs: 1500,
    maxSockets: 70
});


var q1=async.queue(geoasn1, 15);
var q2=async.queue(risprefix1, 15);
var db = new sqlite3.Database('ripe-geoloc.db3');
db.run("CREATE TABLE if not exists geoloc (asn int not null primary key,data text not null,ts DATETIME DEFAULT CURRENT_TIMESTAMP)");
db.run("CREATE TABLE if not exists risprefix (asn INTEGER NOT NULL, v4o INTEGER NOT NULL, v4t INTEGER NOT NULL, v6o INTEGER NOT NULL, v6t INTEGER NOT NULL,ts DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP,PRIMARY KEY(asn,ts));");

var i=0;
fetch("https://stat.ripe.net/data/ris-asns/data.json?list_asns=true").then(function(r){
    return r.ok?r.json():{}
})
.then(function(d){
    console.log(d.data.counts.total);

    //q1.push(d.data.asns);
    q2.push(d.data.asns);
})

function geoasn1(asn,callback){
    var t1=new Date();i++;
    fetch("https://stat.ripe.net/data/geoloc/data.json?resource="+asn,{agent:agent,compress: true,headers: {connection:'keepAlive'}}).then(function(r){
    return r.ok?r.json():{}
})
.then(function(d){
    if (!d.status) {callback();return false}
    console.log([i,d.data.resource,new Date()-t1,d.data.announced_prefixes_time,d.data.locations.length].join("\t"));

    //var stm = db.prepare("replace into geoloc (asn, data) values (?,?)");

    db.run("replace into geoloc (asn, data,ts) values ($asn,$data,$ts)", {
          $asn: d.data.resource,
          $data: JSON.stringify(d.data),
          $ts: d.data.announced_prefixes_time.replace("T"," ")
      });


    callback(d.data);

})

}

function risprefix1(asn,callback) {
    var t1=new Date();i++;
    fetch("https://stat.ripe.net/data/ris-prefixes/data.json?resource="+asn,{agent:agent,compress: true,headers: {connection:'keepAlive'}}).then(function(r){
    return r.ok?r.json():{}
})
.then(function(d){
    if (!d.status) {callback();return false}
    console.log([i,d.data.resource,new Date()-t1,d.data.query_time,d.data.counts.v4.originating,d.data.counts.v4.transitting,d.data.counts.v6.originating,d.data.counts.v6.transitting].join("\t"));
    callback(d.data);
    db.run("replace into risprefix (asn, v4o,v4t,v6o,v6t,ts) values ($asn,$v4o,$v4t,$v6o,$v6t,$ts)", {
          $asn: d.data.resource,
          $v4o: d.data.counts.v4.originating,
          $v4t: d.data.counts.v4.transitting,
          $v6o: d.data.counts.v6.originating,
          $v6t: d.data.counts.v6.transitting,
          $ts: d.data.query_time.replace("T"," ")
      });

    }
    )


}