/*
 * This is a stream which is parametrized on a source. The streams behavior depends on the transform variable which is set on module.export
 */

//TODO: Need to make transform a class variable or transform could be shared between many lambdas

import { Transform, TransformOptions, TransformCallback } from "stream";
const schema = require("../schema.json");
const pkg = require("../package.json");

import bugsnag from "./sources/bugsnag";
import canny from "./sources/canny";
import code from "./sources/code";
import customer_io from "./sources/customer_io";
import diag from "./sources/diag";
import freshpaint from "./sources/freshpaint";
import hubspot from "./sources/hubspot";
import intercom from "./sources/intercom";
import mixpanel from "./sources/mixpanel";
import segment from "./sources/segment";
import sendgrid from "./sources/sendgrid";
import sentry from "./sources/sentry";
import stripe from "./sources/stripe";
import twilio from "./sources/twilio";

import {
  Source,
  Normalizer,
  RawEvent,
  Stream,
  NormalizedEvent,
  NormalizerWithMeta
} from "./types";

type SingerRecord = {
  type: "RECORD";
  stream: string;
  record: RawEvent;
};

const transformers: {
  bugsnag: Normalizer<"bugsnag">;
  canny: Normalizer<"canny">;
  code: Normalizer<"code">;
  customer_io: Normalizer<"customer_io">;
  diag: Normalizer<"diag">;
  freshpaint: Normalizer<"freshpaint">;
  hubspot: Normalizer<"hubspot">;
  intercom: Normalizer<"intercom">;
  mixpanel: Normalizer<"mixpanel">;
  segment: Normalizer<"segment">;
  sendgrid: Normalizer<"sendgrid">;
  sentry: Normalizer<"sentry">;
  stripe: Normalizer<"stripe">;
  twilio: Normalizer<"twilio">;
} = {
  bugsnag,
  canny,
  code,
  customer_io,
  diag,
  freshpaint,
  hubspot,
  intercom,
  mixpanel,
  segment,
  sendgrid,
  sentry,
  stripe,
  twilio
};

//Takes a normalized event and adds metadata to it.
//WARNING: THIS SHOULD ONLY BE DONE FOR FIELDS THAT ARE **ALWAYS** THE SAME REGARDLESS OF THE SOURCE OR STREAM
// (THIS DOES NOT INCLUDE THE SOURCE/STREAM VALUES THEMSELVES)
// AS a convention, anything you set here should begin with _ to mark it as metadata
const getMeta = (event: RawEvent) => {
  return {
    _version: pkg.version,
    _raw: JSON.stringify(event)
  };
};

const SCHEMA_LINE = `{"type": "SCHEMA", "stream": "events", "key_properties": ["id"], "schema": ${JSON.stringify(
  schema
)}}`;

class Normalize<T extends Source> extends Transform {
  transform: Normalizer<T>;

  constructor(options: TransformOptions | undefined, transform: Normalizer<T>) {
    super(options);
    this.transform = transform;
    this.push(SCHEMA_LINE + "\n");
  }
  _transform(chunk: any, enc: string, cb: TransformCallback) {
    let line;
    try {
      line = JSON.parse(chunk.toString());
    } catch (e) {
      console.error(`Could not parse line
	    ============================
	    |${chunk.toString()}|
	    ============================`);
      cb();
      return;
    }
    if (line.type === "RECORD") {
      const data: SingerRecord = line;
      try {
        // Since we don't know if the stream is a valid one
        const streamTransform:
          | ((event: RawEvent) => NormalizedEvent<T>)
          | null = this.transform[data.stream as Stream<T>] as any;
        if (streamTransform) {
          data.record = Object.assign(
            streamTransform(data.record),
            getMeta(data.record)
          );
          data.stream = "events";
          this.push(JSON.stringify(data) + "\n");
        } else {
          console.error(
            `WARNING: There is no transform for stream ${data.stream}. Skipping line`
          );
        }
      } catch (e) {
        console.error(
          `CRITICAL: There was an error normalizing a record of type (${data.stream}). Skipping line. The record and error are logged right after this line.`
        );
        console.error(data.record);
        console.error(e);
      }
      cb();
    } else {
      // We drop all other message types. We can whitelist `STATE` here if we care, or blacklist `SCHEMA`
      cb();
    }
  }
}

export default {
  stream: async <T extends Source>(source: T) => {
    const transform: Normalizer<T> = await import(`./sources/${source}`);
    return new Normalize(undefined, transform);
  },
  transform: <T extends Source>(source: T): NormalizerWithMeta<T> => {
    // @ts-ignore
    const fnMap: Normalizer<T> = transformers[source];
    const rtnMap: NormalizerWithMeta<T> = {} as NormalizerWithMeta<T>;
    Object.keys(fnMap).forEach(key => {
      const k = key as keyof Normalizer<T>;
      rtnMap[k] = (event: RawEvent) =>
        Object.assign(fnMap[k](event), getMeta(event));
    });
    return rtnMap;
  }
};