// Copyright Abridged, Inc. 2021,2024. All Rights Reserved.
// Node module: @collabland/aws
// This file is licensed under the MIT License.
// License text available at https://opensource.org/licenses/MIT

import {
  AttributeDefinition,
  AttributeValue,
  BatchExecuteStatementCommandInput,
  BatchWriteItemInput,
  CreateTableInput,
  DeleteItemInput,
  DeleteItemOutput,
  DeleteRequest,
  DeleteTableInput,
  DescribeTableInput,
  DynamoDB,
  DynamoDBClientConfig,
  ExecuteStatementCommandInput,
  ExecuteStatementInput,
  ExecuteStatementOutput,
  ExecuteTransactionCommandInput,
  GetItemInput,
  GetItemOutput,
  GlobalSecondaryIndex,
  KeySchemaElement,
  KeyType,
  ListTablesCommandInput,
  Projection,
  PutItemInput,
  PutItemOutput,
  PutRequest,
  QueryCommandInput,
  QueryInput,
  ScalarAttributeType,
  ScanCommandInput,
  ScanInput,
  TransactWriteItemsInput,
  UpdateItemInput,
  UpdateItemOutput,
  waitUntilTableExists,
  waitUntilTableNotExists,
  WriteRequest,
} from '@aws-sdk/client-dynamodb';
import {
  DescribeStreamCommand,
  DynamoDBStreamsClient,
  GetRecordsCommand,
  GetShardIteratorCommand,
  ListStreamsCommand,
} from '@aws-sdk/client-dynamodb-streams';
import {
  convertToAttr,
  convertToNative,
  marshall,
  marshallOptions,
  NativeAttributeValue,
  unmarshall,
  unmarshallOptions,
} from '@aws-sdk/util-dynamodb';
import {
  AnyError,
  fetchIteratorByPage,
  getResourceName,
  isProduction,
  loggers,
  pMap,
  pMapByPage,
  PMapByPageOptions,
  ResourceType,
} from '@collabland/common';
import {WaiterResult} from '@smithy/util-waiter';
import lodash, {CloneDeepWithCustomizer} from 'lodash';
import {getAWSDynamoDBRegion, getDefaultConfig} from '../common/types.js';
const {cloneDeepWith} = lodash;

const {trace, info, error, debug} = loggers('collabland:aws:dynamodb');

/**
 * Index name for (classifier, sk)
 */
export const GSI_CLASSIFIER_SK = 'GSI-CLASSIFIER-SK';

/**
 * Index name for (classifier, pk)
 */
export const GSI_CLASSIFIER_PK = 'GSI-CLASSIFIER-PK';

/**
 * Index name for (classifier, slug)
 */
export const GSI_CLASSIFIER_SLUG = 'GSI-CLASSIFIER-SLUG';

/**
 * Index name for (sk, pk)
 */
export const GSI_SK = 'GSI-SK';

/**
 * Index name for (source, bucket)
 */
export const GSI_SOURCE_BUCKET = 'GSI-SOURCE-BUCKET';

/**
 * Index name for (platform, userName)
 */
export const GSI_PLATFORM_USER_NAME = 'GSI-PLATFORM-USER_NAME';

/**
 * Index name for (address, pk) for wallet addresses
 */
export const GSI_ADDRESS_PK = 'GSI-ADDRESS-PK';

/**
 * Index name of (classifier, status) for jobs
 */
export const GSI_CLASSIFIER_STATUS = 'GSI-CLASSIFIER-STATUS';

/**
 * Index name of (flowAccountId, pk) for Dapper wallets
 */
export const GSI_FLOW_ACCOUNT_ID_PK = 'GSI-FLOW_ACC_ID-PK';

/**
 * Index name of (classifierGroup, pk)
 */
export const GSI_CLASSIFIER_GROUP_PK = 'GSI-CLASSIFIER_GROUP-PK';

/**
 * Index name of (classifierGroup, communityPk)
 */
export const GSI_CLASSIFIER_GROUP_COMMUNITY_PK =
  'GSI-CLASSIFIER_GROUP-COMMUNITY_PK';

/**
 * Index name of (classifier, userName)
 */
export const GSI_CLASSIFIER_USER_NAME = 'GSI-CLASSIFIER-USER_NAME';

/**
 * Index name of (classifier, firstName)
 */
export const GSI_CLASSIFIER_FIRST_NAME = 'GSI-CLASSIFIER-FIRST_NAME';

/**
 * Index name of (classifier, _searchFirstName)
 */
export const GSI_CLASSIFIER_SEARCH_FIRST_NAME =
  'GSI-CLASSIFIER-SEARCH_FIRST_NAME';

/**
 * Index name of (classifier, _searchUserName)
 */
export const GSI_CLASSIFIER_SEARCH_USER_NAME =
  'GSI-CLASSIFIER-SEARCH_USER_NAME';

/**
 * Index name of (classifier, _searchLastName)
 */
export const GSI_CLASSIFIER_SEARCH_LAST_NAME =
  'GSI-CLASSIFIER-SEARCH_LAST_NAME';

/**
 * An Item that use TS/JS types/values
 */
export type Item = Record<string, NativeAttributeValue>;

/**
 * A DynamoDB item with extra typing
 */
export type DbItem = Record<string, AttributeValue>;

/**
 * A TS/JS type
 */
export type NativeType<T extends object, K extends keyof T> = T[K] extends
  | DbItem
  | undefined
  ? Omit<T, K> & {
      [P in K]?: Item;
    }
  : never;

/**
 * A DynamoDB type
 */
export type DbType<T extends object, K extends keyof T> = T[K] extends
  | Item
  | undefined
  ? Omit<T, K> & {
      [P in K]?: DbItem;
    }
  : never;

/**
 * Result set from DynamoDB table scan
 */
export type ResultSet = {
  // eslint-disable-next-line @typescript-eslint/no-explicit-any
  items: Record<string, any>[]; // Item[]
  lastEvaluatedKey?: Item;
  paginationToken?: string;
};

export type Pagination = {
  /**
   * Exclusive start key
   */
  paginationKey?: Item;
  /**
   * Maximum number of items to be fetched after the start key. Default to
   * unlimited (-1).
   */
  limit?: number;
};

/**
 * Encode pagination key into a string
 * @param paginationKey - Exclusive start key
 */
export function encodePaginationToken(paginationKey: Item) {
  return Buffer.from(JSON.stringify(paginationKey), 'utf-8').toString('base64');
}

/**
 * Encode pagination key into a string
 * @param paginationKey - Exclusive start key
 */
export function decodePaginationToken(paginationToken: string) {
  const json = Buffer.from(paginationToken, 'base64').toString('utf-8');
  return JSON.parse(json) as Item;
}

const defaultMarshallOptions: marshallOptions = {
  convertEmptyValues: false,
  removeUndefinedValues: false,
};

/**
 * Convert the item to DB item
 * @param item - Item value
 */
export function toDb(item?: Item) {
  if (item == null) return item;
  return marshall(item, defaultMarshallOptions);
}

export function toDbSafe(item?: Item) {
  if (item == null) return item;
  // Workaround for https://github.com/aws/aws-sdk-js-v3/blob/v3.3.0/packages/util-dynamodb/src/convertToAttr.ts#L17
  // Convert the object into a plain json
  const object = JSON.parse(JSON.stringify(item)) as Item;
  return marshall(object, defaultMarshallOptions);
}

export function toDbAttribute(val?: NativeAttributeValue) {
  return convertToAttr(val, defaultMarshallOptions);
}

export function toDbList(items?: Item[]) {
  if (items == null) return items;
  return items.map(item => toDbSafe(item));
}

/**
 * Copied from https://github.com/aws/aws-sdk-js-v3/blob/v3.3.0/packages/util-dynamodb/src/convertToNative.ts#L59
 * @param numString - Number string value
 * @param options - Unmarshall options
 */
const convertNumber = (numString: string, options: unmarshallOptions) => {
  if (options?.wrapNumbers) {
    return {value: numString};
  }
  const num = Number(numString);
  const infinityValues = [Number.POSITIVE_INFINITY, Number.NEGATIVE_INFINITY];
  if (
    (num > Number.MAX_SAFE_INTEGER || num < Number.MIN_SAFE_INTEGER) &&
    !infinityValues.includes(num)
  ) {
    if (typeof BigInt === 'function') {
      return BigInt(numString);
    } else {
      throw new Error(
        `${numString} is outside SAFE_INTEGER bounds. Set options.wrapNumbers to get string value.`,
      );
    }
  }
  /*
  else if (num.toString() !== numString) {
      throw new Error(`Value ${numString} is outside IEEE 754 Floating-Point Arithmetic. Set options.wrapNumbers to get string value.`);
  }
  */
  return num;
};

/**
 * DynamoDB data type keys
 */
const DYNAMODB_TYPES = [
  'S',
  'N',
  'B',
  'SS',
  'NS',
  'BS',
  'M',
  'L',
  'NULL',
  'BOOL',
];

/**
 * Check if the value is a DynamoDB attribute value in `{[TYPE]: val}` form
 * @param val - Value to be checked
 */
function isAttributeValue(val: unknown): val is AttributeValue {
  if (val == null) return false;
  if (typeof val !== 'object') return false;
  const keys = Object.keys(val!);
  if (keys.length === 1 && DYNAMODB_TYPES.includes(keys[0])) return true;
  return false;
}

const attributeCustomizer: CloneDeepWithCustomizer<unknown> = (
  value: unknown,
  key,
) => {
  if (!isAttributeValue(value)) {
    // Return undefined so that clone recursion continues into its child items
    return undefined;
  }
  if (typeof value.N === 'string') {
    // Work around that `@aws-sdk/util-dynamodb` throws if a number value is
    // 0.0000000000000001
    value = convertNumber(value.N, {wrapNumbers: false});
  } else {
    value = convertToNative(value);
  }
  return value;
};

export function _fromDb(item?: DbItem) {
  if (item == null) return item;
  return unmarshall(item, {
    wrapNumbers: false,
  });
}

/**
 * Unmarshall DB item into a native item
 * @param item - DynamoDB item
 */
export function fromDb(item?: DbItem): Item | undefined {
  if (item == null) return item;
  const result = cloneDeepWith(item, attributeCustomizer);
  return result;
}

export function fromDbList(items?: DbItem[]) {
  if (items == null) return [];
  return items.map(item => fromDb(item)!);
}

function toDbInputWithKey<T extends {Key?: Item}>(input: T) {
  return {
    ...input,
    Key: toDb(input.Key),
  };
}

function fromDbOutput<T extends {Attributes?: DbItem}>(output: T) {
  return {
    ...output,
    Attributes: fromDb(output.Attributes),
  };
}

function fromDbOutputWithItem<T extends {Item?: DbItem}>(output: T) {
  return {
    ...output,
    Item: fromDb(output.Item),
  };
}

function toDbInputWithItem<T extends {Item?: Item}>(input: T) {
  return {
    ...input,
    Item: toDbSafe(input.Item),
  };
}

export function setPaginationToken(resultSet: ResultSet) {
  if (resultSet.lastEvaluatedKey != null) {
    resultSet.paginationToken = encodePaginationToken(
      resultSet.lastEvaluatedKey,
    );
  }
  return resultSet;
}

export function getDefaultDynamoDBConfig(): DynamoDBClientConfig {
  return {
    ...getDefaultConfig(),
    region: getAWSDynamoDBRegion(),
  };
}

/**
 * Configuration of AWS DynamoDB
 */
export type DynamoDbConfig = {
  /**
   * DynamoDB configuration
   */
  config?: DynamoDBClientConfig;
  /**
   * Tables
   */
  tables?: string[];

  consistentRead?: boolean;
};

export interface SourceBucket {
  source: string;
  bucket?: string;
}

export type IndexKeys = Record<string, {pk?: string; sk?: string}>;

export type PutItemInputSpec = NativeType<PutItemInput, 'Item'>;
export type PutItemOutputSpec = NativeType<PutItemOutput, 'Attributes'>;

export type GetItemInputSpec = NativeType<GetItemInput, 'Key'>;
export type GetItemOutputSpec = NativeType<GetItemOutput, 'Item'>;

export type DeleteItemInputSpec = NativeType<DeleteItemInput, 'Key'>;
export type DeleteItemOutputSpec = NativeType<DeleteItemOutput, 'Attributes'>;

export type UpdateItemInputSpec = NativeType<
  UpdateItemInput,
  'Key' | 'ExpressionAttributeValues'
>;
export type UpdateItemOutputSpec = NativeType<UpdateItemOutput, 'Attributes'>;

export type DeleteRequestSpec = NativeType<DeleteRequest, 'Key'>;
export type PutRequestSpec = NativeType<PutRequest, 'Item'>;

export type QueryInputSpec = NativeType<
  QueryInput,
  'ExpressionAttributeValues' | 'ExclusiveStartKey'
>;

export type ScanInputSpec = NativeType<
  ScanInput,
  'ExpressionAttributeValues' | 'ExclusiveStartKey'
>;

export type ExecuteStatementInputSpec = Omit<
  ExecuteStatementInput,
  'Parameters'
> & {
  Parameters?: NativeAttributeValue[];
};

type ExecuteStatementOutputSpec = Omit<ExecuteStatementOutput, 'Items'> & {
  Items?: NativeAttributeValue[];
};
export class DynamoDbServiceClient {
  readonly dynamodb: DynamoDB;
  readonly tables: string[];
  indexKeys: Record<string, IndexKeys> = {};
  readonly streamsClient: DynamoDBStreamsClient;

  constructor(private dynamoDbConfig: DynamoDbConfig = {}) {
    this.dynamoDbConfig = {
      tables: [
        getResourceName(ResourceType.DynamoDBTable),
        getResourceName(ResourceType.DynamoDBAPITable),
      ],
      consistentRead: !isProduction(),
      ...dynamoDbConfig,
      config: {
        ...getDefaultDynamoDBConfig(),
        ...dynamoDbConfig.config,
      },
    };

    debug('DynamoDB config: %O', this.dynamoDbConfig);
    this.tables = this.dynamoDbConfig.tables!;
    this.dynamodb = new DynamoDB(this.dynamoDbConfig.config ?? {});
    this.streamsClient = new DynamoDBStreamsClient(
      this.dynamoDbConfig.config ?? {},
    );
  }

  get apiTable() {
    return this.tables[1];
  }

  get botTable() {
    return this.tables[0];
  }

  async init() {
    info('Describing index keys for %s', this.tables);
    await pMap(this.tables, async table => {
      this.indexKeys[table] = await this.describeIndexKeys(table);
    });

    info('Index keys for %s: %O', this.tables, this.indexKeys);
  }

  async create(input: PutItemInputSpec): Promise<PutItemOutputSpec> {
    try {
      const dbInput = toDbInputWithItem(input);
      const output = await this.dynamodb.putItem(dbInput);
      return fromDbOutput(output);
    } catch (err: AnyError) {
      error('Fail to create item %O: %O', input, err);
      throw err;
    }
  }

  private consistentRead(spec: {ConsistentRead?: boolean}) {
    spec.ConsistentRead = this.dynamoDbConfig.consistentRead;
  }

  async get(input: GetItemInputSpec): Promise<GetItemOutputSpec> {
    try {
      const dbInput = toDbInputWithKey(input);
      this.consistentRead(dbInput);
      const output = await this.dynamodb.getItem(dbInput);
      return fromDbOutputWithItem(output);
    } catch (err: AnyError) {
      error('Fail to get item %O: %O', input, err);
      throw err;
    }
  }

  async delete(input: DeleteItemInputSpec): Promise<DeleteItemOutputSpec> {
    try {
      const dbInput = toDbInputWithKey(input);
      const output = await this.dynamodb.deleteItem(dbInput);
      return fromDbOutput(output);
    } catch (err: AnyError) {
      error('Fail to delete item %O: %O', input, err);
      throw err;
    }
  }

  async update(input: UpdateItemInputSpec): Promise<UpdateItemOutputSpec> {
    try {
      const dbInput = toDbInputWithKey(input);
      const output = await this.dynamodb.updateItem({
        ...dbInput,
        ExpressionAttributeValues: toDb(dbInput.ExpressionAttributeValues),
      });
      return fromDbOutput(output);
    } catch (err: AnyError) {
      error('Fail to update item %O: %O', input, err);
      throw err;
    }
  }

  async batchWrite(
    tableName: string,
    deleteItems: DeleteRequestSpec[],
    putItems: PutRequestSpec[],
  ) {
    const writeRequests: WriteRequest[] = [];
    const requests: BatchWriteItemInput = {
      RequestItems: {
        [tableName]: writeRequests,
      },
    };
    deleteItems.forEach(i =>
      writeRequests.push({
        DeleteRequest: {
          ...i,
          Key: toDb(i.Key),
        },
      }),
    );
    putItems.forEach(i =>
      writeRequests.push({
        PutRequest: {
          ...i,
          Item: toDbSafe(i.Item),
        },
      }),
    );
    try {
      const output = await this.dynamodb.batchWriteItem(requests);
      return output;
    } catch (err: AnyError) {
      error('Fail to batch write items %O: %O', requests, err);
      throw err;
    }
  }

  /**
   * Query the table by page
   * @param params - Query parameters
   * @param limit - Page size
   */
  private async queryByPage(
    params: QueryInputSpec,
    limit = -1,
  ): Promise<ResultSet> {
    trace(`Running query`, params);
    const allData: Item[] = [];
    const resultSet: ResultSet = {
      items: allData,
    };

    for (;;) {
      trace('Fetching data: %O', params);
      const input: QueryCommandInput = {
        ...params,
        ExpressionAttributeValues: toDb(params.ExpressionAttributeValues),
        ExclusiveStartKey: toDb(params.ExclusiveStartKey),
      };
      if (input.IndexName == null) {
        this.consistentRead(input);
      }
      const data = await this.dynamodb.query(input);

      trace('Items from query: %O', data);
      if (data?.Items?.length == null) return resultSet;

      if (data?.Items?.length > 0) {
        const items = fromDbList(data?.Items);
        allData.push(...items);
      }

      if (limit !== -1 && allData.length >= (limit ?? 0)) {
        resultSet.lastEvaluatedKey = fromDb(data.LastEvaluatedKey);
        return setPaginationToken(resultSet);
      }

      if (data.LastEvaluatedKey) {
        resultSet.lastEvaluatedKey = fromDb(data.LastEvaluatedKey);
        params.ExclusiveStartKey = fromDb(data.LastEvaluatedKey);
      } else {
        return resultSet;
      }
    }
  }

  fetchByPage(params: QueryInputSpec, pageSize = 1000) {
    trace(`Running query by page`, params);

    return fetchIteratorByPage({
      initialState: params.ExclusiveStartKey,
      next: async state => {
        trace('Fetching data: %O', params);
        const input: QueryCommandInput = {
          ...params,
          ExpressionAttributeValues: toDb(params.ExpressionAttributeValues),
          ExclusiveStartKey: toDb(state),
          Limit: pageSize,
        };
        if (input.IndexName == null) {
          this.consistentRead(input);
        }
        const data = await this.dynamodb.query(input);
        trace('Items from query: %O', data);
        const items = fromDbList(data?.Items);
        if (data.LastEvaluatedKey == null) return {done: true, value: items};
        return {
          value: items,
          done: false,
          state: fromDb(data.LastEvaluatedKey),
        };
      },
    });
  }

  /**
   * Query the table by page
   * @param params - Query parameters
   * @param pageSize - Page size
   */
  private async scanByPage(
    params: ScanInputSpec,
    pageSize = -1,
  ): Promise<ResultSet> {
    trace(`Running query`, params);
    const allData: Item[] = [];
    const resultSet: ResultSet = {
      items: allData,
    };

    for (;;) {
      trace('Fetching data: %O', params);
      const input: ScanCommandInput = {
        ...params,
        ExpressionAttributeValues: toDb(params.ExpressionAttributeValues),
        ExclusiveStartKey: toDb(params.ExclusiveStartKey),
      };
      if (input.IndexName == null) {
        this.consistentRead(input);
      }
      const data = await this.dynamodb.scan(input);

      trace('Items from query: %O', data);
      if (data?.Items?.length == null) return resultSet;

      if (data?.Items?.length > 0) {
        const items = fromDbList(data?.Items);
        allData.push(...items);
      }

      if (pageSize !== -1 && allData.length >= (pageSize ?? 0)) {
        resultSet.lastEvaluatedKey = fromDb(data.LastEvaluatedKey);
        return setPaginationToken(resultSet);
      }

      if (data.LastEvaluatedKey) {
        resultSet.lastEvaluatedKey = fromDb(data.LastEvaluatedKey);
        params.ExclusiveStartKey = fromDb(data.LastEvaluatedKey);
      } else {
        return resultSet;
      }
    }
  }

  async query(
    params: QueryInputSpec,
    pagination?: Pagination,
  ): Promise<ResultSet> {
    try {
      this.setPagination(params, pagination);
      const result = await this.queryByPage(params, pagination?.limit ?? -1);
      return result;
    } catch (err: AnyError) {
      error('Fail to query items %O: %O', params, err);
      throw err;
    }
  }

  async scan(params: ScanInputSpec, pageSize = -1): Promise<ResultSet> {
    try {
      const result = await this.scanByPage(params, pageSize);
      return result;
    } catch (err: AnyError) {
      error('Fail to scan items %O: %O', params, err);
      throw err;
    }
  }

  async createAll(
    table: string,
    resultSet: ResultSet,
    options: PMapByPageOptions = {pageSize: 25, concurrency: 5},
  ) {
    const result = await pMapByPage(
      resultSet.items,
      page => {
        return this.batchWrite(
          table,
          [],
          page.map(i => {
            const req: PutRequestSpec = {Item: i};
            return req;
          }),
        );
      },
      options,
    );
    return result;
  }

  /**
   * Build query parameters based on the index name
   * @param tableName - Table name
   * @param indexName - Index name, use table name or '' for the main one
   * @param key - Partition and Sort key value for the index
   * @param pagination - Pagination
   */
  buildQueryByIndex(
    tableName: string,
    indexName = '',
    key: {
      pk: NativeAttributeValue;
      sk?: NativeAttributeValue;
    },
    pagination?: Pagination,
  ) {
    const index = this.indexKeys[tableName] ?? {};
    const keys = index[indexName];
    if (keys == null) {
      throw new Error(`Invalid index name: ${tableName}/${indexName}`);
    }
    const params: QueryInputSpec = {
      TableName: tableName,
      ConsistentRead: false,
      IndexName: indexName === tableName || !indexName ? undefined : indexName,
    };

    this.setPagination(params, pagination);

    const {pk, sk} = key;
    if (pk != null) {
      params.KeyConditionExpression =
        sk != null
          ? `#${keys.pk} = :${keys.pk} and #${keys.sk} = :${keys.sk}`
          : `#${keys.pk} = :${keys.pk}`;
    }

    if (pk != null) {
      params.ExpressionAttributeNames =
        sk != null
          ? {
              [`#${keys.pk}`]: keys.pk!,
              [`#${keys.sk}`]: keys.sk!,
            }
          : {
              [`#${keys.pk}`]: keys.pk!,
            };
    }

    if (pk != null) {
      params.ExpressionAttributeValues =
        sk != null
          ? {
              [`:${keys.pk}`]: pk,
              [`:${keys.sk}`]: sk,
            }
          : {[`:${keys.pk}`]: pk};
    }
    return params;
  }

  setPagination(params: QueryInputSpec, pagination?: Pagination) {
    if (pagination?.paginationKey != null) {
      params.ExclusiveStartKey = pagination.paginationKey;
    }
    if (pagination?.limit !== -1 && pagination?.limit) {
      params.Limit = pagination?.limit;
    }
  }

  /**
   * Build KeySchema for an index
   * @param pk - Partition key name
   * @param sk - Sort key name
   */
  buildKeySchema(pk: string, sk?: string): KeySchemaElement[] {
    const keys: KeySchemaElement[] = [
      {
        AttributeName: pk,
        KeyType: KeyType.HASH,
      },
    ];
    if (sk != null) {
      keys.push({
        AttributeName: sk,
        KeyType: KeyType.RANGE,
      });
    }
    return keys;
  }

  /**
   * Query the table by index
   * @param tableName - Table name
   * @param indexName - Index name, use table name or '' for the main one
   * @param key - Partition key/Sort key for the index
   * @param pagination - Pagination
   */
  queryByIndex(
    tableName: string,
    indexName = '',
    key: {
      pk: NativeAttributeValue;
      sk?: NativeAttributeValue;
    },
    pagination?: Pagination,
  ) {
    const params = this.buildQueryByIndex(
      tableName,
      indexName,
      key,
      pagination,
    );
    return this.query(params, pagination);
  }

  /**
   *Build a global secondary index (GSI)
   * @param name - Index name
   * @param pk - Partition key name
   * @param sk - Sort key name
   */
  buildGlobalSecondaryIndex(
    name: string,
    pk: string,
    sk?: string,
    projection: Projection = {ProjectionType: 'ALL'},
  ): GlobalSecondaryIndex {
    return {
      IndexName: name,
      KeySchema: this.buildKeySchema(pk, sk),
      Projection: projection,
    };
  }

  /**
   * Build attribute definitions
   * @param attrs - Attributes (name, type)
   */
  buildAttributeDefinitions(
    attrs: Record<string, ScalarAttributeType>,
  ): AttributeDefinition[] {
    return Object.entries(attrs).map(val => ({
      AttributeName: val[0],
      AttributeType: val[1],
    }));
  }

  /**
   * Create a table
   * @param tableName - Table name
   */
  createTable(tableName: string) {
    const params: CreateTableInput = {
      TableName: tableName,
      KeySchema: this.buildKeySchema('pk', 'sk'),
      AttributeDefinitions: this.buildAttributeDefinitions({
        pk: 'S',
        sk: 'S',
        classifier: 'S',
        airtableRecId: 'S',
        source: 'S',
        bucket: 'S',
        platform: 'S',
        userName: 'S',
      }),
      GlobalSecondaryIndexes: [
        this.buildGlobalSecondaryIndex(GSI_SK, 'sk', 'pk'),
        this.buildGlobalSecondaryIndex(GSI_CLASSIFIER_SK, 'classifier', 'sk'),
        this.buildGlobalSecondaryIndex(GSI_CLASSIFIER_PK, 'classifier', 'pk'),
        this.buildGlobalSecondaryIndex(
          GSI_PLATFORM_USER_NAME,
          'platform',
          'userName',
        ),
        this.buildGlobalSecondaryIndex(
          'GSI-AIRTABLE_RECID-PK',
          'airtableRecId',
          'pk',
        ),
        // Build a secondary index for data source/buckets to make it easy to clean
        // up data based on the bucket
        this.buildGlobalSecondaryIndex(GSI_SOURCE_BUCKET, 'source', 'bucket', {
          ProjectionType: 'KEYS_ONLY',
        }),
        this.buildGlobalSecondaryIndex(
          GSI_CLASSIFIER_STATUS,
          'classifier',
          'status',
        ),
        this.buildGlobalSecondaryIndex(
          GSI_CLASSIFIER_GROUP_COMMUNITY_PK,
          'classifierGroup',
          'communityPk',
        ),
        this.buildGlobalSecondaryIndex(
          GSI_CLASSIFIER_GROUP_PK,
          'classifierGroup',
          'pk',
        ),
      ],
      BillingMode: 'PAY_PER_REQUEST',
    };
    return this.dynamodb.createTable(params);
  }

  async listTables() {
    const params: ListTablesCommandInput = {};
    const output = await this.dynamodb.listTables(params);
    return output.TableNames!;
  }

  /**
   * Describe a table
   * @param tableName - Table name
   */
  describeTable(tableName: string) {
    const params: DescribeTableInput = {
      TableName: tableName,
    };
    return this.dynamodb.describeTable(params);
  }

  /**
   * Describe table/index keys
   * @param tableName - Table name
   */
  async describeIndexKeys(tableName: string) {
    const result = await this.describeTable(tableName);
    const keys: IndexKeys = {};
    const table = result.Table;
    if (table == null) return keys;
    if (table.TableName != null) {
      keys[table.TableName] = keys[''] = this.getKeys(table.KeySchema);
    }
    if (table.GlobalSecondaryIndexes) {
      table.GlobalSecondaryIndexes.forEach(gsi => {
        if (gsi.IndexName != null) {
          keys[gsi.IndexName] = this.getKeys(gsi.KeySchema);
        }
      });
    }
    if (table.LocalSecondaryIndexes) {
      table.LocalSecondaryIndexes.forEach(lsi => {
        if (lsi.IndexName != null) {
          keys[lsi.IndexName!] = this.getKeys(lsi.KeySchema);
        }
      });
    }
    return keys;
  }

  /**
   * Extract pk/sk attribute names from the key schema
   * @param schema - Key schema
   */
  private getKeys(schema: KeySchemaElement[] = []) {
    const pk = schema.find(e => e.KeyType === 'HASH')?.AttributeName;
    const sk = schema.find(e => e.KeyType === 'RANGE')?.AttributeName;
    return {pk, sk};
  }

  /**
   * Delete a table
   * @param tableName - Table name
   */
  deleteTable(tableName: string) {
    const params: DeleteTableInput = {
      TableName: tableName,
    };
    return this.dynamodb.deleteTable(params);
  }

  /**
   * Wait for the table to exist
   * @param tableName - Table name
   */
  waitUntilTableExists(
    tableName: string,
    maxWaitTime = 120,
  ): Promise<WaiterResult> {
    return waitUntilTableExists(
      {
        client: this.dynamodb,
        maxWaitTime,
      },
      {
        TableName: tableName,
      },
    );
  }

  /**
   * Wait for the table to not exist
   * @param tableName - Table name
   */
  async waitUntilTableNotExists(
    tableName: string,
    maxWaitTime = 120,
  ): Promise<WaiterResult> {
    return waitUntilTableNotExists(
      {
        client: this.dynamodb,
        maxWaitTime,
      },
      {
        TableName: tableName,
      },
    );
  }

  transactWrite(params: TransactWriteItemsInput) {
    return this.dynamodb.transactWriteItems(params);
  }

  async executeStatement(
    params: ExecuteStatementInputSpec,
  ): Promise<ExecuteStatementOutputSpec> {
    const input: ExecuteStatementCommandInput = {
      ...params,
      Parameters: params.Parameters?.map(p => convertToAttr(p)),
    };
    const output = await this.dynamodb.executeStatement(input);
    return {
      ...output,
      Items: fromDbList(output.Items),
    };
  }

  executeBatchStatement(params: BatchExecuteStatementCommandInput) {
    return this.dynamodb.batchExecuteStatement(params);
  }

  executeTransaction(params: ExecuteTransactionCommandInput) {
    return this.dynamodb.executeTransaction(params);
  }

  async listStreams(tableName: string) {
    const command = new ListStreamsCommand({TableName: tableName, Limit: 100});
    const response = await this.streamsClient.send(command);
    return response.Streams ?? [];
  }

  async getStreamShards(
    streamArn: string,
    limit = 100,
    exclusiveStartShardId?: string,
  ) {
    const command = new DescribeStreamCommand({
      StreamArn: streamArn,
      Limit: limit,
      ExclusiveStartShardId: exclusiveStartShardId,
    });
    const response = await this.streamsClient.send(command);
    return {
      shards: response.StreamDescription?.Shards ?? [],
      lastEvaluatedShardId: response.StreamDescription?.LastEvaluatedShardId,
    };
  }

  async getSharedIterator(
    streamArn: string,
    shardId: string,
    iteratorType:
      | 'TRIM_HORIZON'
      | 'LATEST'
      | 'AT_SEQUENCE_NUMBER'
      | 'AFTER_SEQUENCE_NUMBER',
    sequenceNumber?: string,
  ) {
    const command = new GetShardIteratorCommand({
      StreamArn: streamArn,
      ShardIteratorType: iteratorType,
      ShardId: shardId,
      SequenceNumber: sequenceNumber,
    });
    const response = await this.streamsClient.send(command);
    return response.ShardIterator;
  }

  async getStreamRecords(iterator: string, limit = 1000) {
    const command = new GetRecordsCommand({
      ShardIterator: iterator,
      Limit: limit,
    });
    const response = await this.streamsClient.send(command);
    return {records: response.Records, next: response.NextShardIterator};
  }
}