import now = require('performance-now');
import { SpadeEvent } from './models/SpadeEvent';
import { SpadeDataPoint } from './models/SpadeDataPoint';
import { DataPoints } from './types/influx';
import { InfluxClient as influx } from './influx';
import { ArcanaClient } from "@twitch/arcana";

const arcana = new ArcanaClient();
let password: string;

async function getSecret() {
  password = password || await arcana.decryptAndTrim('influx_password');
}

export default async function processRecords(event: any, context: any, callback: any) {
  console.info(`Received ${event.Records.length} events`);

  await getSecret();

  if (!password) {
    throw new Error("Unable to get password")
  }

  const startTime = now();

  // Parse raw data into objects.
  const spadeEvents: SpadeEvent[] = event.Records.map((record: any) => new SpadeEvent(record.kinesis.data));
  const doneParsing = now();

  // const points = dataPoints.map((dataPoint: SpadeDataPoint) => dataPoint.toInflux());

  // Bucket our data points by database.
  const dataPoints: SpadeDataPoint[] = [].concat.apply([], spadeEvents.map((event: SpadeEvent) => event.dataPoints));
  const dataPointsByDB: DataPoints = {};

  for (let point of dataPoints) {
    let database = point.database;

    if (dataPointsByDB.hasOwnProperty(database)) {
      dataPointsByDB[database].push(point.toInflux());
    } else {
      dataPointsByDB[database] = [point.toInflux()];
    }
  }

  // Accounting for batched writing.
  const batchSize = 10000;
  let batchCount = 0;
  let pointCount = 0;

  let databases = Object.keys(dataPointsByDB);

  for (let database of databases) {
    let points = dataPointsByDB[database].length;
    batchCount += Math.ceil(points / batchSize);
    pointCount += points;
    // console.log(`CREATE DATABASE "${database}" WITH DURATION 1d SHARD DURATION 1h`)
    // console.log(`ALTER RETENTION POLICY "raw_statistics" ON "${database}" DURATION 90d SHARD DURATION 1h DEFAULT`)
  }

  console.info(`Discovered ${pointCount} data points across ${databases.length} databases.`);
  console.info(`Submitting up to ${batchSize} at a time across ${batchCount} batches.`)

  // There are multiple shards in flight at the same time that could have overlapped timings.
  // This will give each shard between 1 and 30 microsecond randomized offsets from each other to help ensure uniqueness.
  const bucketOffset: number = Math.max(Math.floor(Math.random() * 30) + 1, 30);

  let successfulBatches: number = 0;

  // Submitting batches to influx.
  for (let database of databases) {
    let points = dataPointsByDB[database];

    // Add 1 microsecond and bucket offset to each point for this database.
    for (let index = 0; index < points.length; index++) {
      (points[index].timestamp as number) += index + bucketOffset;
    }

    console.info(`Processing ${points.length} data points to ${database}...`);
    if (!process.env.IS_LOCAL) {
      for (let startIndex = 0; startIndex < points.length; startIndex += batchSize) {

        console.info(`Submitting points ${startIndex} to ${startIndex + batchSize} for ${database}`);
        influx(password).writePoints(points.slice(startIndex, startIndex + batchSize), { database: database, precision: 'u' }).then(() => {

          successfulBatches++;
          console.info(`Success: batch ${successfulBatches} of ${batchCount} complete.`)

          // If all batches are successful tell lambda we're done with this event set.
          if (successfulBatches >= batchCount) {
            const completeTime = now();
            console.info(`Successfully saved ${pointCount}. Parsing: ${(doneParsing - startTime).toFixed(3)}ms. Writing: ${(completeTime - doneParsing).toFixed(3)}ms. Total: ${(completeTime - startTime).toFixed(3)}ms.`);
            callback(null, `Processed ${pointCount} total data points across ${databases.length} and ${batchCount} write batches.`);
          }
        }).catch(err => {
          console.error(`Error saving data to InfluxDB! ${err.stack}`);
          callback(err, null);
        });
      }
    } else {
      console.info(`Skipping InfluxDB Write due to local mode`)
    }
  }
}
