'use strict';

const get = require('lodash.get');
const csv = require('fast-csv');
const AWS = require('aws-sdk');
const queue = require('async.queue');
const es = require('./es');

let esClient;

function pushToEs(records, callback) {
  //callback(null);
  es.saveRecords(esClient, records).then(() => callback()).catch(e => callback(e));
}

function processCsvFile(bucket, key, transform) {
  // batch processing config

  const q = queue(pushToEs, 2);
  const allScenes = [];
  const batchSize = 400;
  let bufferArray = [];

  // download the csv file
  const s3 = new AWS.S3();
  const params = {
    Bucket: bucket,
    Key: key
  };
  console.log(`starting processing ${key}`);

  const csvStream = csv.parse({ headers: true, objectMode: true });
  s3.getObject(params).createReadStream().pipe(csvStream);

  return new Promise((resolve, reject) => {
    csvStream.on('data', (data) => {
      transform(data, (e, record) => {
        if (e) {
          console.log(e);
          return;
        }
        if (bufferArray.length >= batchSize) {
          q.push([bufferArray.slice()], e => console.log(e));
          bufferArray = [];
        }

        allScenes.push(record.scene_id);
        bufferArray.push(record);
      });
    });

    csvStream.on('end', () => {
      // push remaning records to kinesis
      if (bufferArray.length > 0) {
        q.push([bufferArray.slice()], e => console.log(e));
      }
    });

    q.drain = () => resolve(allScenes.length);
    csvStream.on('error', e => reject(e));
  });
}

function prepare(event, transform, cb) {
  const bucket = get(event, 'bucket');
  const key = get(event, 'key');
  const direction = get(event, 'direction', 'asc');
  const satellite = get(event, 'satellite', 'landsat');
  const currentFileNum = get(event, 'currentFileNum', 0);
  const lastFileNum = get(event, 'lastFileNum', 0);
  const arn = get(event, 'arn');

  let nextKey;
  let nextFileNum;
  const currentKey = `${key}/${satellite}_${currentFileNum}.csv`;

  if (direction === 'desc') {
    if (currentFileNum > lastFileNum) {
      nextKey = `${key}/${satellite}_${currentFileNum - 1}.csv`;
      nextFileNum = currentFileNum - 1;
    }
  }
  else {
    if (currentFileNum < lastFileNum) {
      nextKey = `${key}/${satellite}_${currentFileNum + 1}.csv`;
      nextFileNum = currentFileNum + 1;
    }
  }

  processCsvFile(
    bucket, currentKey, transform
  ).then((scenes) => {
    if (nextKey) {
      const stepfunctions = new AWS.StepFunctions();
      const params = {
        stateMachineArn: arn,
        input: JSON.stringify({ bucket, key, satellite, currentFileNum: nextFileNum, lastFileNum, arn, direction }),
        name: `csv_${satellite}_${nextFileNum}_${Date.now()}`
      };
      stepfunctions.startExecution(params, function(err, data) {
        cb(err, `launched ${JSON.stringify(params)}`);
      });
    }
    else {
      cb(null, scenes);
    }
  }).catch(e => cb(e));
}

function update(event, transform, cb) {
  if (!esClient) {
    es.connect().then((client) => {
      esClient = client;
      prepare(event, transform, cb);
    });
  }
  else {
    prepare(event, transform, cb);
  }
}

module.exports.update = update;

@@ -249,16 +249,74 @@
 (null);%0A
+  console.log(%60pushing $%7Brecords.length%7D records to ES%60);%0A
   es.sav
@@ -387,24 +387,25 @@
 llback(e));%0A
+%0A
 %7D%0A%0Afunction 
@@ -505,24 +505,26 @@
 ToEs, 2);%0A  
+//
 const allSce
@@ -581,16 +581,33 @@
 ay = %5B%5D;
+%0A  var count = 0;
 %0A%0A  // d
@@ -730,17 +730,8 @@
 og(%60
-starting 
 proc
@@ -953,24 +953,50 @@
 (data) =%3E %7B%0A
+      //console.log(data)%0A
       transf
@@ -1249,16 +1249,18 @@
         
+//
 allScene
@@ -1314,24 +1314,41 @@
 sh(record);%0A
+        count++;%0A
       %7D);%0A  
@@ -1428,16 +1428,86 @@
 kinesis%0A
+      console.log(%60finished processing $%7Bcount%7D records from $%7Bkey%7D%60)%0A
       if
@@ -1639,25 +1639,27 @@
 resolve(
-allScenes
+bufferArray
 .length)
@@ -2077,16 +2077,53 @@
 arn');%0A%0A
+  // figure out next file to process%0A
   let ne
@@ -2214,17 +2214,16 @@
 %7D.csv%60;%0A
-%0A
   if (di
@@ -2288,73 +2288,8 @@
 ) %7B%0A
-      nextKey = %60$%7Bkey%7D/$%7Bsatellite%7D_$%7BcurrentFileNum - 1%7D.csv%60;%0A
     
@@ -2397,207 +2397,337 @@
 next
-Key = %60$%7Bkey%7D/$%7Bsatellite%7D_$%7BcurrentFileNum + 1%7D.csv%60;%0A      nextFileNum = currentFileNum + 1;%0A    %7D%0A  %7D%0A%0A  processCsvFile(%0A    bucket, currentKey, transform%0A  ).then((scenes) =%3E %7B%0A    if (nextKey) %7B
+FileNum = currentFileNum + 1;%0A    %7D%0A  %7D%0A  if (nextFileNum) %7B%0A    nextKey = %60$%7Bkey%7D/$%7Bsatellite%7D_$%7BnextFileNum%7D.csv%60;%0A  %7D%0A%0A  processCsvFile(%0A    bucket, currentKey, transform%0A  ).then((scenes) =%3E %7B%0A    // kick off processing next CSV file%0A    //if (nextKey) %7B%0A    if (false) %7B%0A      console.log('start execution of next CSV file')
 %0A   
@@ -3276,24 +3276,60 @@
 esClient) %7B%0A
+    console.log('connecting to ES')%0A
     es.conne

add some logging