var later       = require('later');
var express     = require('express');
var Redis = require('ioredis');
var Promise     = require('bluebird');
var querystring = require('querystring');
var co          = require('co');
var moment      = require('moment');
var debug       = require('debug')('witty');
var request     = Promise.promisify(require('request'));

var app = express();

app.get('/', (req, res) => {
  res.send('Hello World!');
});

var server = app.listen(3000, () => {

  var host = server.address().address;
  var port = server.address().port;

  console.log('Witty listening at http://%s:%s', host, port);
});

var redis = new Redis(process.env.REDIS_CONNECTION);

var retrieveSched = later.parse.recur().every(1).minute();

var retrieveRecentThreads = function(argBlob) {

  var token = process.env.QUIP_TOKEN;
  let qs = querystring.stringify(argBlob);

  return request({
    uri: `https://platform.quip.com/1/threads/recent?${ qs }`,
    headers: {
      Authorization: `Bearer ${ token }`
    },
    json: true
  })
};

var retrieveSample = function*(max_updated_usec) {

  max_updated_usec = Date.now() * 1000 || max_updated_usec;

  let [response, data] = yield retrieveRecentThreads({
    max_updated_usec: max_updated_usec,
    count: 5
  });

  if (response.statusCode < 200 || response.statusCode >= 300) {
    debug(`error occurred retrieving sample from ${ max_updated_usec }`);
    debug(data);
  }

  let sampleThreads = Object.keys(data);
  let nearestSecond = Math.floor(max_updated_usec / 1000 / 1000);
  let nearestMinuteBucket = Math.floor(nearestSecond / 60) % 60;
  let bucketKey = `sample:${ nearestMinuteBucket }`;

  debug(`retrieved sample at ${ moment.unix(nearestSecond).format() }, bucketed at ${ nearestMinuteBucket } (${ sampleThreads.join(', ') })`);

  // filter out threads threads in the sample that are not documents
  sampleThreads = sampleThreads.filter((thread) => {
    return data[thread].thread.thread_class === 'document';
  });

  // save thread keys as a sorted set - basde on position (most recently changed)
  let scoredThreads = sampleThreads.map((thread, index) => {
    return [index, thread];
  }).reduce((a, b) => {
    return a.concat(b);
  });

  yield redis.pipeline().del(bucketKey).zadd(bucketKey, scoredThreads).exec();

  // write out individual threads by bucket
  for (let thread of sampleThreads) {
    let threadData = data[thread].thread;

    let sampleThreadKey = `sample:${ nearestMinuteBucket }:thread:${ threadData.id }`;
    yield redis.pipeline().del(sampleThreadKey).hset(sampleThreadKey, ['updated_usec', threadData.updated_usec ]).exec();

    let threadKey = `thread:${ threadData.id }`;
    yield redis.pipeline().del(threadKey).hmset(threadKey, threadData).exec();
  }
};

var analyzeSamples = function*() {

  let threads = {};
  for (let minute = 0; minute < 60; minute++) {

    // zrange will pull in the weighted order
    let sample = yield redis.zrange(`sample:${ minute }`, 0, -1);

    for (let thread of sample) {
      let threadData = yield redis.hgetall(`sample:${ minute }:thread:${ thread }`);

      if (!threads[thread]) threads[thread] = { updates: new Set() };
      threads[thread].updates.add(threadData.updated_usec);
    }
  }

  let threadHistogram = Object.keys(threads).reduce((map, threadKey) => {
    map[threadKey] = threads[threadKey].updates.size;
    return map;
  }, {});

  debug(threadHistogram);
};

var refreshData = function() {

  debug('refreshing data');

  co(function*() {
    yield retrieveSample();
    yield analyzeSamples();
  }).then((map) => {

  }).catch((err) => {
    console.log(err.stack);
  });

};

later.setInterval(refreshData, retrieveSched);
refreshData();

@@ -2324,54 +2324,85 @@
 eads
- by bucket%0A  for (let thread of sampleThreads)
+%0A  let redisPipeline = redis.pipeline();%0A%0A  sampleThreads.forEach((thread) =%3E
  %7B%0A 
@@ -2443,16 +2443,56 @@
 hread;%0A%0A
+    // write the updated_usec by bucket%0A
     let 
@@ -2574,38 +2574,45 @@
 %7D%60;%0A    
-yield
+redisPipeline =
  redis
-.p
+P
 ipeline
-()
 .del(sam
@@ -2690,25 +2690,72 @@
 _usec %5D)
-.exec();%0A
+;%0A%0A    // write the total thread blob agnostic of bucket
 %0A    let
@@ -2800,38 +2800,45 @@
 %7D%60;%0A    
-yield
+redisPipeline =
  redis
-.p
+P
 ipeline
-()
 .del(thr
@@ -2873,16 +2873,46 @@
 eadData)
+;%0A  %7D);%0A%0A  yield redisPipeline
 .exec();
@@ -2912,20 +2912,16 @@
 exec();%0A
-  %7D%0A
 %7D;%0A%0Avar 

Refactor and pipeline