// @pdsjs/blobs-s3 - S3-compatible blob storage (AWS S3, MinIO, R2 via S3 API)

export { createS3FetchBackupTarget } from './fetch-target.js';

/**
 * @typedef {import('@pdsjs/core/ports').BlobPort} BlobPort
 * @typedef {import('@pdsjs/core/ports').BackupTargetPort} BackupTargetPort
 */

/**
 * @typedef {Object} S3Client
 * @property {(command: any) => Promise<any>} send - Send a command to S3
 */

// Multipart part size. S3 requires at least 5 MiB for every part but the
// last.
const S3_PART_SIZE = 8 * 1024 * 1024;

/**
 * @param {Uint8Array[]} chunks
 * @param {number} size
 * @returns {Uint8Array}
 */
function joinChunks(chunks, size) {
  const out = new Uint8Array(size);
  let at = 0;
  for (const chunk of chunks) {
    out.set(chunk, at);
    at += chunk.length;
  }
  return out;
}

/**
 * Create S3-compatible blob adapter
 * Works with AWS S3, MinIO, Cloudflare R2 (via S3 API), and other S3-compatible storage
 *
 * The five multipart and copy constructors are optional as one group. With
 * them, the adapter offers `createWriter` and a streamed upload stages under
 * a temporary key, lands under its CID through a server-side copy, and holds
 * one part buffer in memory at most. Without them, core buffers streamed
 * uploads and calls `put`.
 * @param {Object} options
 * @param {S3Client} options.client - S3 client instance
 * @param {string} options.bucket - Bucket name
 * @param {new (params: any) => any} options.GetObjectCommand - GetObjectCommand constructor
 * @param {new (params: any) => any} options.PutObjectCommand - PutObjectCommand constructor
 * @param {new (params: any) => any} options.DeleteObjectCommand - DeleteObjectCommand constructor
 * @param {new (params: any) => any} [options.CreateMultipartUploadCommand]
 * @param {new (params: any) => any} [options.UploadPartCommand]
 * @param {new (params: any) => any} [options.CompleteMultipartUploadCommand]
 * @param {new (params: any) => any} [options.AbortMultipartUploadCommand]
 * @param {new (params: any) => any} [options.CopyObjectCommand]
 * @returns {BlobPort}
 */
export function createS3Blobs({
  client,
  bucket,
  GetObjectCommand,
  PutObjectCommand,
  DeleteObjectCommand,
  CreateMultipartUploadCommand,
  UploadPartCommand,
  CompleteMultipartUploadCommand,
  AbortMultipartUploadCommand,
  CopyObjectCommand,
}) {
  const multipartReady = Boolean(
    CreateMultipartUploadCommand &&
      UploadPartCommand &&
      CompleteMultipartUploadCommand &&
      AbortMultipartUploadCommand &&
      CopyObjectCommand,
  );

  /**
   * @param {string} did
   * @returns {Promise<import('@pdsjs/core/ports').BlobWriter>}
   */
  async function createWriter(did) {
    const tempKey = `${did}/tmp/${crypto.randomUUID()}`;
    /** @type {string|null} */
    let uploadId = null;
    /** @type {{ETag: string, PartNumber: number}[]} */
    const parts = [];
    /** @type {Uint8Array[]} */
    let pending = [];
    let pendingSize = 0;

    async function flushPart() {
      if (!uploadId) {
        const created = await client.send(
          new /** @type {new (params: any) => any} */ (
            CreateMultipartUploadCommand
          )({ Bucket: bucket, Key: tempKey }),
        );
        uploadId = created.UploadId;
      }
      const body = joinChunks(pending, pendingSize);
      const sent = await client.send(
        new /** @type {new (params: any) => any} */ (UploadPartCommand)({
          Bucket: bucket,
          Key: tempKey,
          UploadId: uploadId,
          PartNumber: parts.length + 1,
          Body: body,
        }),
      );
      parts.push({ ETag: sent.ETag, PartNumber: parts.length + 1 });
      pending = [];
      pendingSize = 0;
    }

    return {
      /** @param {Uint8Array} chunk */
      async write(chunk) {
        pending.push(chunk);
        pendingSize += chunk.length;
        if (pendingSize >= S3_PART_SIZE) await flushPart();
      },

      /**
       * @param {string} cid
       * @param {string} mimeType
       */
      async commit(cid, mimeType) {
        const finalKey = `${did}/${cid}`;
        if (!uploadId) {
          await client.send(
            new PutObjectCommand({
              Bucket: bucket,
              Key: finalKey,
              Body: joinChunks(pending, pendingSize),
              ContentType: mimeType,
              Metadata: { mime_type: mimeType },
            }),
          );
          return;
        }
        if (pendingSize > 0) await flushPart();
        await client.send(
          new /** @type {new (params: any) => any} */ (
            CompleteMultipartUploadCommand
          )({
            Bucket: bucket,
            Key: tempKey,
            UploadId: uploadId,
            MultipartUpload: { Parts: parts },
          }),
        );
        await client.send(
          new /** @type {new (params: any) => any} */ (CopyObjectCommand)({
            Bucket: bucket,
            Key: finalKey,
            CopySource: encodeURI(`${bucket}/${tempKey}`),
            MetadataDirective: 'REPLACE',
            ContentType: mimeType,
            Metadata: { mime_type: mimeType },
          }),
        );
        await client.send(
          new DeleteObjectCommand({ Bucket: bucket, Key: tempKey }),
        );
      },

      async abort() {
        if (uploadId) {
          await client.send(
            new /** @type {new (params: any) => any} */ (
              AbortMultipartUploadCommand
            )({ Bucket: bucket, Key: tempKey, UploadId: uploadId }),
          );
        }
        pending = [];
        pendingSize = 0;
      },
    };
  }

  return {
    ...(multipartReady ? { createWriter } : {}),

    /**
     * @param {string} did
     * @param {string} cid
     * @param {{start: number, end: number}} [range] - Inclusive byte positions
     */
    async getStream(did, cid, range) {
      try {
        const response = await client.send(
          new GetObjectCommand({
            Bucket: bucket,
            Key: `${did}/${cid}`,
            ...(range ? { Range: `bytes=${range.start}-${range.end}` } : {}),
          }),
        );
        return {
          body: /** @type {ReadableStream<Uint8Array>} */ (
            response.Body.transformToWebStream()
          ),
          mimeType:
            response.Metadata?.mime_type ||
            response.ContentType ||
            'application/octet-stream',
        };
      } catch (e) {
        if (e.name === 'NoSuchKey') return null;
        throw e;
      }
    },

    /**
     * @param {string} did
     * @param {string} cid
     */
    async get(did, cid) {
      try {
        const response = await client.send(
          new GetObjectCommand({
            Bucket: bucket,
            Key: `${did}/${cid}`,
          }),
        );
        const data = new Uint8Array(await response.Body.transformToByteArray());
        const mimeType =
          response.Metadata?.mime_type || 'application/octet-stream';
        return { data, mimeType };
      } catch (e) {
        if (e.name === 'NoSuchKey') return null;
        throw e;
      }
    },

    /**
     * @param {string} did
     * @param {string} cid
     * @param {Uint8Array} data
     * @param {string} mimeType
     */
    async put(did, cid, data, mimeType) {
      await client.send(
        new PutObjectCommand({
          Bucket: bucket,
          Key: `${did}/${cid}`,
          Body: data,
          ContentType: mimeType,
          Metadata: { mime_type: mimeType },
        }),
      );
    },

    /**
     * @param {string} did
     * @param {string} cid
     */
    async delete(did, cid) {
      await client.send(
        new DeleteObjectCommand({
          Bucket: bucket,
          Key: `${did}/${cid}`,
        }),
      );
    },
  };
}

/**
 * Backups in an S3-compatible bucket, following the injected-client pattern
 * above so the AWS SDK stays out of this package's dependencies. Give
 * backups their own bucket: core deletes anything under `snapshots/` and
 * `blobs/` that its retention policy no longer wants.
 * @param {Object} options
 * @param {S3Client} options.client - S3 client instance
 * @param {string} options.bucket - Bucket name
 * @param {new (params: any) => any} options.GetObjectCommand
 * @param {new (params: any) => any} options.PutObjectCommand
 * @param {new (params: any) => any} options.DeleteObjectCommand
 * @param {new (params: any) => any} options.ListObjectsV2Command
 * @returns {BackupTargetPort}
 */
export function createS3BackupTarget({
  client,
  bucket,
  GetObjectCommand,
  PutObjectCommand,
  DeleteObjectCommand,
  ListObjectsV2Command,
}) {
  return {
    /**
     * @param {string} key
     * @param {Uint8Array} data
     * @param {string} [contentType]
     */
    async put(key, data, contentType) {
      await client.send(
        new PutObjectCommand({
          Bucket: bucket,
          Key: key,
          Body: data,
          ContentType: contentType || 'application/octet-stream',
        }),
      );
    },

    /** @param {string} key */
    async get(key) {
      try {
        const response = await client.send(
          new GetObjectCommand({ Bucket: bucket, Key: key }),
        );
        return new Uint8Array(await response.Body.transformToByteArray());
      } catch (e) {
        if (e.name === 'NoSuchKey' || e.name === 'NotFound') return null;
        throw e;
      }
    },

    /** @param {string} prefix */
    async list(prefix) {
      /** @type {string[]} */
      const keys = [];
      /** @type {string|undefined} */
      let continuationToken;
      do {
        const response = await client.send(
          new ListObjectsV2Command({
            Bucket: bucket,
            Prefix: prefix,
            ContinuationToken: continuationToken,
          }),
        );
        for (const object of response.Contents || []) keys.push(object.Key);
        continuationToken = response.IsTruncated
          ? response.NextContinuationToken
          : undefined;
      } while (continuationToken);
      return keys;
    },

    /** @param {string} key */
    async delete(key) {
      await client.send(new DeleteObjectCommand({ Bucket: bucket, Key: key }));
    },
  };
}