import * as cdk from "@aws-cdk/core";
import * as dynamodb from "@aws-cdk/aws-dynamodb";
import * as lambda from "@aws-cdk/aws-lambda";
import * as lambdaEventSources from "@aws-cdk/aws-lambda-event-sources";
import { App } from "./App";
import { Function as Fn, FunctionProps, FunctionDefinition } from "./Function";
import { KinesisStream } from "./KinesisStream";
import { Permissions } from "./util/permission";

export enum TableFieldType {
  BINARY = dynamodb.AttributeType.BINARY,
  NUMBER = dynamodb.AttributeType.NUMBER,
  STRING = dynamodb.AttributeType.STRING,
}

/////////////////////
// Interfaces
/////////////////////

export interface TableProps {
  readonly fields?: { [key: string]: TableFieldType };
  readonly primaryIndex?: TableIndexProps;
  readonly secondaryIndexes?: { [key: string]: TableIndexProps };
  readonly dynamodbTable?: dynamodb.ITable | TableCdkProps;
  readonly kinesisStream?: KinesisStream;
  readonly stream?: boolean | dynamodb.StreamViewType;
  readonly consumers?: {
    [consumerName: string]: FunctionDefinition | TableConsumerProps;
  };
  readonly defaultFunctionProps?: FunctionProps;
}

export interface TableConsumerProps {
  readonly function: FunctionDefinition;
  readonly consumerProps?: lambdaEventSources.DynamoEventSourceProps;
}

export interface TableIndexProps {
  readonly partitionKey: string;
  readonly sortKey?: string;
  readonly indexProps?: TableCdkIndexProps;
}

export type TableCdkProps = Omit<
  dynamodb.TableProps,
  "partitionKey" | "sortKey"
>;

export type TableCdkIndexProps = Omit<
  dynamodb.GlobalSecondaryIndexProps,
  "indexName" | "partitionKey" | "sortKey"
>;

/////////////////////
// Construct
/////////////////////

export class Table extends cdk.Construct {
  public readonly dynamodbTable: dynamodb.Table;
  private functions: { [consumerName: string]: Fn };
  private readonly permissionsAttachedForAllConsumers: Permissions[];
  private readonly defaultFunctionProps?: FunctionProps;
  private readonly stream?: dynamodb.StreamViewType;

  constructor(scope: cdk.Construct, id: string, props: TableProps) {
    super(scope, id);

    const root = scope.node.root as App;
    const {
      fields,
      primaryIndex,
      secondaryIndexes,
      dynamodbTable,
      kinesisStream,
      stream,
      consumers,
      defaultFunctionProps,
    } = props;
    this.functions = {};
    this.permissionsAttachedForAllConsumers = [];
    this.defaultFunctionProps = defaultFunctionProps;

    ////////////////////
    // Create Table
    ////////////////////

    this.validateFieldsAndIndexes(id, props);

    if (cdk.Construct.isConstruct(dynamodbTable)) {
      // Validate "fields" is not configured
      if (fields !== undefined) {
        throw new Error(
          `Cannot configure the "fields" when "dynamodbTable" is a construct in the "${id}" Table`
        );
      }

      // Validate "stream" is not configured
      if (stream !== undefined) {
        throw new Error(
          `Cannot configure the "stream" when "dynamodbTable" is a construct in the "${id}" Table`
        );
      }

      this.dynamodbTable = dynamodbTable as dynamodb.Table;
    } else {
      let dynamodbTableProps = (dynamodbTable || {}) as dynamodb.TableProps;

      // Validate "fields" is configured
      if (fields === undefined) {
        throw new Error(`Missing "fields" in the "${id}" Table`);
      }

      // Validate dynamodbTableProps does not contain "partitionKey", "sortKey" and "stream"
      if (dynamodbTableProps.partitionKey) {
        throw new Error(
          `Cannot configure the "dynamodbTableProps.partitionKey" in the "${id}" Table`
        );
      }
      if (dynamodbTableProps.sortKey) {
        throw new Error(
          `Cannot configure the "dynamodbTableProps.sortKey" in the "${id}" Table`
        );
      }
      if (dynamodbTableProps.stream) {
        throw new Error(
          `Cannot configure the "dynamodbTableProps.stream" in the "${id}" Table`
        );
      }

      if (fields && primaryIndex) {
        dynamodbTableProps = {
          ...dynamodbTableProps,
          partitionKey: this.buildAttribute(fields, primaryIndex.partitionKey),
          sortKey: primaryIndex.sortKey
            ? this.buildAttribute(fields, primaryIndex.sortKey)
            : undefined,
        };
      }

      this.dynamodbTable = new dynamodb.Table(this, "Table", {
        tableName: root.logicalPrefixedName(id),
        pointInTimeRecovery: true,
        billingMode: dynamodb.BillingMode.PAY_PER_REQUEST,
        stream: this.buildStreamConfig(stream),
        ...(dynamodbTableProps as dynamodb.TableProps),
      });
    }

    //////////////////////////////
    // Create Secondary Indexes
    //////////////////////////////

    if (fields && secondaryIndexes) {
      Object.keys(secondaryIndexes).forEach((indexName) => {
        const { partitionKey, sortKey, indexProps = {} } = secondaryIndexes[
          indexName
        ];

        // Validate indexProps does not contain "indexName", "partitionKey" and "sortKey"
        if ((indexProps as dynamodb.GlobalSecondaryIndexProps).indexName) {
          throw new Error(
            `Cannot configure the "indexProps.indexName" in the "${indexName}" index of the "${id}" Table`
          );
        }
        if ((indexProps as dynamodb.GlobalSecondaryIndexProps).partitionKey) {
          throw new Error(
            `Cannot configure the "indexProps.partitionKey" in the "${indexName}" index of the "${id}" Table`
          );
        }
        if ((indexProps as dynamodb.GlobalSecondaryIndexProps).sortKey) {
          throw new Error(
            `Cannot configure the "indexProps.sortKey" in the "${indexName}" index of the "${id}" Table`
          );
        }

        this.dynamodbTable.addGlobalSecondaryIndex({
          indexName,
          partitionKey: this.buildAttribute(fields, partitionKey),
          sortKey: sortKey ? this.buildAttribute(fields, sortKey) : undefined,
          ...indexProps,
        });
      });
    }

    ///////////////////////////
    // Create Consumers
    ///////////////////////////

    if (consumers) {
      // Handle deprecated props
      this.checkDeprecatedConsumers(consumers);

      Object.keys(consumers).forEach((consumerName: string) =>
        this.addConsumer(this, consumerName, consumers[consumerName])
      );
    }

    // Create Kinesis Stream
    this.buildKinesisStreamSpec(kinesisStream);
  }

  public get tableArn(): string {
    return this.dynamodbTable.tableArn;
  }

  public get tableName(): string {
    return this.dynamodbTable.tableName;
  }

  public addConsumers(
    scope: cdk.Construct,
    consumers: {
      [consumerName: string]: FunctionDefinition | TableConsumerProps;
    }
  ): void {
    // Handle deprecated consumers
    this.checkDeprecatedConsumers(consumers);

    Object.keys(consumers).forEach((consumerName: string) => {
      this.addConsumer(scope, consumerName, consumers[consumerName]);
    });
  }

  public attachPermissions(permissions: Permissions): void {
    Object.values(this.functions).forEach((fn) =>
      fn.attachPermissions(permissions)
    );
    this.permissionsAttachedForAllConsumers.push(permissions);
  }

  public attachPermissionsToConsumer(
    consumerName: string,
    permissions: Permissions
  ): void {
    if (!this.functions[consumerName]) {
      throw new Error(
        `The "${consumerName}" consumer was not found in the "${this.node.id}" Table.`
      );
    }

    this.functions[consumerName].attachPermissions(permissions);
  }

  public getFunction(consumerName: string): Fn | undefined {
    return this.functions[consumerName];
  }

  private addConsumer(
    scope: cdk.Construct,
    consumerName: string,
    consumer: FunctionDefinition | TableConsumerProps
  ): Fn {
    // validate stream enabled
    if (!this.dynamodbTable.tableStreamArn) {
      throw new Error(
        `Please enable the "stream" option to add consumers to the "${this.node.id}" Table.`
      );
    }

    // parse consumer
    let consumerFunction, consumerProps;
    if ((consumer as TableConsumerProps).function) {
      consumer = consumer as TableConsumerProps;
      consumerFunction = consumer.function;
      consumerProps = consumer.consumerProps;
    } else {
      consumerFunction = consumer as FunctionDefinition;
    }
    consumerProps = {
      startingPosition: lambda.StartingPosition.LATEST,
      ...(consumerProps || {}),
    };

    // create function
    const fn = Fn.fromDefinition(
      scope,
      consumerName,
      consumerFunction,
      this.defaultFunctionProps,
      `The "defaultFunctionProps" cannot be applied if an instance of a Function construct is passed in. Make sure to define all the consumers using FunctionProps, so the Table construct can apply the "defaultFunctionProps" to them.`
    );
    this.functions[consumerName] = fn;

    // create event source
    const eventSource = new lambdaEventSources.DynamoEventSource(
      this.dynamodbTable,
      consumerProps
    );
    fn.addEventSource(eventSource);

    // attach permissions
    this.permissionsAttachedForAllConsumers.forEach((permissions) => {
      fn.attachPermissions(permissions);
    });

    return fn;
  }

  validateFieldsAndIndexes(id: string, props: TableProps): void {
    const { fields, primaryIndex, secondaryIndexes } = props;

    // Validate "fields"
    if (fields && Object.keys(fields).length === 0) {
      throw new Error(`No fields defined for the "${id}" Table`);
    }

    // Validate "primaryIndex"
    if (primaryIndex && !primaryIndex.partitionKey) {
      throw new Error(
        `Missing "partitionKey" in primary index for the "${id}" Table`
      );
    }

    // Validate "fields" and "primaryIndex" co-exists
    if (fields) {
      if (!primaryIndex) {
        throw new Error(`Missing "primaryIndex" in "${id}" Table`);
      }
    } else {
      if (primaryIndex) {
        throw new Error(
          `Cannot configure the "primaryIndex" without setting the "fields" in "${id}" Table`
        );
      }
      if (secondaryIndexes) {
        throw new Error(
          `Cannot configure the "secondaryIndexes" without setting the "fields" in "${id}" Table`
        );
      }
    }
  }

  private buildAttribute(
    fields: { [key: string]: TableFieldType },
    name: string
  ): dynamodb.Attribute {
    return {
      name,
      type: this.convertTableFieldTypeToAttributeType(fields[name]),
    };
  }

  private buildStreamConfig(
    stream?: boolean | dynamodb.StreamViewType
  ): dynamodb.StreamViewType | undefined {
    if (stream === true) {
      return dynamodb.StreamViewType.NEW_AND_OLD_IMAGES;
    } else if (stream === false) {
      return undefined;
    }

    return stream;
  }

  private convertTableFieldTypeToAttributeType(
    fieldType: TableFieldType
  ): dynamodb.AttributeType {
    if (fieldType === TableFieldType.BINARY) {
      return dynamodb.AttributeType.BINARY;
    } else if (fieldType === TableFieldType.NUMBER) {
      return dynamodb.AttributeType.NUMBER;
    } else {
      return dynamodb.AttributeType.STRING;
    }
  }

  private buildKinesisStreamSpec(kinesisStream?: KinesisStream): void {
    if (!kinesisStream) {
      return;
    }

    const cfTable = this.dynamodbTable.node.defaultChild as dynamodb.CfnTable;
    cfTable.addPropertyOverride(
      "KinesisStreamSpecification.StreamArn",
      kinesisStream.streamArn
    );
  }

  private checkDeprecatedConsumers(consumers: {
    [consumerName: string]: FunctionDefinition | TableConsumerProps;
  }): void {
    if (Array.isArray(consumers)) {
      throw new Error(
        `The "consumers" property no longer takes an array. It nows takes an associative array with the consumer name being the index key. More details on upgrading - https://docs.serverless-stack.com/constructs/Table#upgrading-to-v0210`
      );
    }
  }
}