import AWS, { S3 } from "aws-sdk";
import { Options } from "csv-parser";
import { Readable } from "stream";
import zlib from "zlib";
import { config } from "../utils/config";
const parseCsv = require("csv-parser");

export default class S3Service {
  protected s3: S3;

  constructor() {
    this.s3 =
      config.OGURY_ENV === "local"
        ? new AWS.S3({
            accessKeyId: config.AWS_ACCESS_KEY_ID,
            secretAccessKey: config.AWS_SECRET_ACCESS_KEY,
            endpoint: config.S3_ENDPOINT,
            s3ForcePathStyle: true,
            signatureVersion: "v4",
          })
        : new AWS.S3();
  }

  /**
   * Create a line reader from a CSV file.
   * @param bucket Bucket of the file
   * @param key Key of the file
   * @param csvOptions If specified, provide custom options for parsing CSV
   */
  public readLinesCsv(bucket: string, key: string, csvOptions?: Options): Readable {
    let reader = this.s3.getObject({ Bucket: bucket, Key: key }).createReadStream();
    if (key.endsWith(".gz")) reader = reader.pipe(zlib.createGunzip());
    reader = reader.pipe(parseCsv(csvOptions));
    return reader;
  }

  /**
   * List all objects under a given path.
   * @param bucket Bucket of the path.
   * @param prefix Prefix of the path.
   * @param assertKeysExist If specified, assert that a given list of keys exist under the path.
   *                        This is useful, for example, to check if there is a `_SUCCESS`
   *                        file generated by Spark before reading the path contents.
   * @throws In case of S3 error, or if `assertKeysExist` check fails.
   */
  public async listKeys(
    bucket: string,
    prefix: string,
    assertKeysExist?: string[]
  ): Promise<string[]> {
    const data = await this.s3.listObjectsV2({ Bucket: bucket, Prefix: prefix }).promise();
    const keys = (data.Contents === undefined ? [] : data.Contents)
      .filter((c) => c.Key !== undefined)
      .map((c) => c.Key as string);
    // Check assertKeysExist
    if (assertKeysExist !== undefined) {
      for (const key of assertKeysExist) {
        const keyToSearchFor = prefix + "/" + key;
        if (!keys.includes(keyToSearchFor))
          throw Error("Missing file: s3://" + bucket + "/" + keyToSearchFor);
      }
    }
    return keys;
  }

  async loadFile(bucket: string, key: string) {
    const data = await this.s3.getObject({ Bucket: bucket, Key: key }).promise();

    return {
      file: data.Body!,
      lastModified: data.LastModified,
    };
  }

  /**
   * Get last modification date of file located in S3.
   * @param bucket Bucket of the file
   * @param key Key of the file
   * @return last modification of the file
   */
  async getLastModified(bucket: string, key: string): Promise<Date | undefined> {
    const data = await this.s3.getObject({ Bucket: bucket, Key: key }).promise();
    return data.LastModified;
  }
}