import child_process from "child_process";
import fs from "fs";
import fsp from "fs/promises";
import util from "util";
import { pipeline } from "stream/promises";
import type { Readable } from "stream";

import os from "os";
import { createHash } from "crypto";

import { S3Client, GetObjectCommand } from "@aws-sdk/client-s3";
import { Upload } from "@aws-sdk/lib-storage";

import { request } from "undici";

import { initRedis } from "./redis.js";
import { logger } from "./logger.js";

import getFolderSize from "get-folder-size";

import { WACZ } from "./wacz.js";

import { DEFAULT_MAX_RETRIES } from "./constants.js";
import { CrawlerArgs } from "./argParser.js";

const DEFAULT_REGION = "us-east-1";

const PART_SIZE = 1024 * 1024 * 100;

// ===========================================================================
export type UploadResult = {
  path: string;
  hash: string;
  size: number;
  bytes: number;
};

// ===========================================================================
export class S3StorageSync {
  fullPrefix: string;
  client: S3Client;

  bucketName: string;
  objectPrefix: string;
  resources: object[] = [];

  userId: string;
  crawlId: string;
  webhookUrl?: string;

  constructor(
    urlOrData:
      | string
      | {
          endpointUrl: string;
          accessKey: string;
          secretKey: string;
        },
    {
      webhookUrl,
      userId,
      crawlId,
    }: { webhookUrl?: string; userId: string; crawlId: string },
  ) {
    let url: URL;
    let accessKey: string;
    let secretKey: string;

    if (typeof urlOrData === "string") {
      url = new URL(urlOrData);
      accessKey = url.username;
      secretKey = url.password;
      url.username = "";
      url.password = "";
      this.fullPrefix = url.href;
    } else {
      url = new URL(urlOrData.endpointUrl);
      accessKey = urlOrData.accessKey;
      secretKey = urlOrData.secretKey;
      this.fullPrefix = url.href;
    }

    const region = process.env.STORE_REGION || DEFAULT_REGION;

    this.client = new S3Client({
      credentials: {
        accessKeyId: accessKey,
        secretAccessKey: secretKey,
      },
      requestChecksumCalculation: "WHEN_REQUIRED",
      maxAttempts: DEFAULT_MAX_RETRIES,
      endpoint: url.origin,
      region,
      forcePathStyle: true,
    });

    this.bucketName = url.pathname.slice(1).split("/")[0];

    this.objectPrefix = url.pathname.slice(this.bucketName.length + 2);

    this.resources = [];

    this.userId = userId;
    this.crawlId = crawlId;
    this.webhookUrl = webhookUrl;
  }

  async uploadStreamingWACZ(
    wacz: WACZ,
    targetFilename: string,
  ): Promise<UploadResult> {
    await this.doUpload(targetFilename, wacz.generate());

    const hash = wacz.getHash();
    const path = targetFilename;

    const size = wacz.getSize();

    // for backwards compatibility, keep 'bytes'
    return { path, size, hash, bytes: size };
  }

  async uploadFile(
    srcFilename: string,
    targetFilename: string,
  ): Promise<UploadResult> {
    await this.doUpload(targetFilename, fs.createReadStream(srcFilename));

    const hash = await checksumFile("sha256", srcFilename);
    const path = targetFilename;

    const size = await getFileSize(srcFilename);

    // for backwards compatibility, keep 'bytes'
    return { path, size, hash, bytes: size };
  }

  private async doUpload(targetFilename: string, Body: Readable) {
    const Bucket = this.bucketName;

    const Key = this.objectPrefix + targetFilename;

    const fileUploadInfo = {
      bucket: Bucket,
      key: Key,
      endpoint: this.client.config.endpoint,
      crawlId: this.crawlId,
      prefix: this.objectPrefix,
      targetFilename,
    };

    logger.info("S3 file upload information", fileUploadInfo, "storage");

    const uploader = new Upload({
      client: this.client,
      params: { Bucket, Key, Body },

      // (optional) size of each part, in bytes, at least 5MB
      partSize: PART_SIZE,

      leavePartsOnError: false,
    });

    await uploader.done();
  }

  async downloadFile(srcFilename: string, destFilename: string) {
    logger.debug("Downloading profile", { srcFilename }, "storage");

    const res = await this.client.send(
      new GetObjectCommand({
        Bucket: this.bucketName,
        Key: this.objectPrefix + srcFilename,
      }),
    );
    await pipeline(res.Body as Readable, fs.createWriteStream(destFilename));
  }

  async uploadCollWACZ(
    srcOrWACZ: string | WACZ,
    targetFilename: string,
    completed = true,
  ) {
    const resource =
      typeof srcOrWACZ === "string"
        ? await this.uploadFile(srcOrWACZ, targetFilename)
        : await this.uploadStreamingWACZ(srcOrWACZ, targetFilename);

    logger.info(
      "WACZ S3 file upload resource",
      { targetFilename, resource },
      "storage",
    );

    if (this.webhookUrl) {
      const body = {
        id: this.crawlId,
        user: this.userId,

        //filename: `s3://${this.bucketName}/${this.objectPrefix}${this.waczFilename}`,
        filename: this.fullPrefix + targetFilename,

        ...resource,
        completed,
      };

      logger.info(`Pinging Webhook: ${this.webhookUrl}`);

      if (
        this.webhookUrl.startsWith("http://") ||
        this.webhookUrl.startsWith("https://")
      ) {
        await request(this.webhookUrl, {
          method: "POST",
          body: JSON.stringify(body),
        });
      } else if (this.webhookUrl.startsWith("redis://")) {
        const parts = this.webhookUrl.split("/");
        if (parts.length !== 5) {
          logger.fatal(
            "redis webhook url must be in format: redis://<host>:<port>/<db>/<key>",
            {},
            "redis",
          );
        }
        const redis = await initRedis(parts.slice(0, 4).join("/"));
        await redis.rpush(parts[4], JSON.stringify(body));
      }
    }
  }
}

export function initStorage() {
  if (!process.env.STORE_ENDPOINT_URL) {
    return null;
  }

  const endpointUrl =
    process.env.STORE_ENDPOINT_URL + (process.env.STORE_PATH || "");
  const storeInfo = {
    endpointUrl,
    accessKey: process.env.STORE_ACCESS_KEY || "",
    secretKey: process.env.STORE_SECRET_KEY || "",
  };

  const opts = {
    crawlId: process.env.CRAWL_ID || os.hostname(),
    webhookUrl: process.env.WEBHOOK_URL || "",
    userId: process.env.STORE_USER || "",
  };

  logger.info("Initing Storage...");
  return new S3StorageSync(storeInfo, opts);
}

export async function getFileSize(filename: string) {
  const stats = await fsp.stat(filename);
  return stats.size;
}

export async function getDirSize(dir: string): Promise<number> {
  const { size, errors } = await getFolderSize(dir);
  if (errors && errors.length) {
    logger.warn("Size check errors", { errors }, "storage");
  }
  return size;
}

export async function isDiskFull(collDir: string) {
  const diskUsage: Record<string, string> = await getDiskUsage(collDir);
  const usedPercentage = parseInt(diskUsage["Use%"].slice(0, -1));
  return usedPercentage >= 99;
}

export async function checkDiskUtilization(
  collDir: string,
  params: CrawlerArgs,
  archiveDirSize: number,
  dfOutput = null,
  doLog = true,
) {
  const diskUsage: Record<string, string> = await getDiskUsage(
    collDir,
    dfOutput,
  );
  const usedPercentage = parseInt(diskUsage["Use%"].slice(0, -1));

  // Check that disk usage isn't already above threshold
  if (usedPercentage >= params.diskUtilization) {
    if (doLog) {
      logger.info(
        `Disk utilization threshold reached ${usedPercentage}% > ${params.diskUtilization}%, stopping`,
      );
    }
    return {
      stop: true,
      used: usedPercentage,
      projected: null,
      threshold: params.diskUtilization,
    };
  }

  // Check that disk usage isn't likely to cross threshold
  const kbUsed = parseInt(diskUsage["Used"]);
  const kbTotal = parseInt(diskUsage["1K-blocks"]);

  let kbArchiveDirSize = Math.round(archiveDirSize / 1024);

  // assume if has STORE_ENDPOINT_URL, will be uploading to remote
  // and not storing local copy of either WACZ or WARC
  if (!process.env.STORE_ENDPOINT_URL) {
    if (params.combineWARC && params.generateWACZ) {
      kbArchiveDirSize *= 4;
    } else if (params.combineWARC || params.generateWACZ) {
      kbArchiveDirSize *= 2;
    }
  }

  const projectedTotal = kbUsed + kbArchiveDirSize;
  const projectedUsedPercentage = calculatePercentageUsed(
    projectedTotal,
    kbTotal,
  );

  if (projectedUsedPercentage >= params.diskUtilization) {
    if (doLog) {
      logger.info(
        `Disk utilization projected to reach threshold ${projectedUsedPercentage}% > ${params.diskUtilization}%, stopping`,
      );
    }
    return {
      stop: true,
      used: usedPercentage,
      projected: projectedUsedPercentage,
      threshold: params.diskUtilization,
    };
  }

  return {
    stop: false,
    used: usedPercentage,
    projected: projectedUsedPercentage,
    threshold: params.diskUtilization,
  };
}

export async function getDFOutput(path: string) {
  const exec = util.promisify(child_process.exec);
  const res = await exec(`df ${path}`);
  return res.stdout;
}

export async function getDiskUsage(path = "/crawls", dfOutput = null) {
  const result = dfOutput || (await getDFOutput(path));
  const lines = result.split("\n");
  const keys = lines[0].split(/\s+/gi);
  const rows = lines.slice(1).map((line) => {
    const values = line.split(/\s+/gi);
    return keys.reduce((o: Record<string, string>, k, index) => {
      o[k] = values[index];
      return o;
    }, {});
  });
  return rows[0];
}

export function calculatePercentageUsed(used: number, total: number) {
  return Math.round((used / total) * 100);
}

function checksumFile(hashName: string, path: string): Promise<string> {
  return new Promise((resolve, reject) => {
    const hash = createHash(hashName);

    const stream = fs.createReadStream(path);
    stream.on("error", (err) => reject(err));
    stream.on("data", (chunk) => {
      hash.update(chunk);
    });
    stream.on("end", () => resolve(hash.digest("hex")));
  });
}

export function interpolateFilename(filename: string, crawlId: string) {
  filename = filename.replace(
    "@ts",
    new Date().toISOString().replace(/[:TZz.-]/g, ""),
  );
  filename = filename.replace("@hostname", os.hostname());
  filename = filename.replace("@hostsuffix", os.hostname().slice(-14));
  filename = filename.replace("@id", crawlId);
  return filename;
}
