// @pdsjs/blobs-s3/fetch-target - S3-compatible backup target over plain fetch
//
// Signature Version 4 implemented directly on WebCrypto, so this runs
// anywhere fetch does — Cloudflare Workers included, where the AWS SDK does
// not go. Path-style addressing, the four basic operations, and multipart
// upload: all a backup target needs, and the part every S3-compatible
// provider (Bunny, Backblaze B2, MinIO, R2, AWS itself) implements.

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

const encoder = new TextEncoder();
/** SHA-256 of an empty body, the constant AWS documents. */
const EMPTY_PAYLOAD_HASH =
  'e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855';
/**
 * Bytes per part of a multipart upload. Above S3's 5 MiB minimum, which
 * every part except the last one must meet, and 10,000 parts at this size
 * carry 80 GB. An upload holds the part it is sending and the one behind
 * it, never the whole object.
 */
export const PART_SIZE = 8 * 1024 * 1024;

/**
 * @param {Uint8Array} bytes
 * @returns {string}
 */
function hex(bytes) {
  let out = '';
  for (const b of bytes) out += b.toString(16).padStart(2, '0');
  return out;
}

/**
 * @param {Uint8Array} data
 * @returns {Promise<string>}
 */
async function sha256Hex(data) {
  const digest = await crypto.subtle.digest(
    'SHA-256',
    /** @type {BufferSource} */ (data),
  );
  return hex(new Uint8Array(digest));
}

/**
 * @param {Uint8Array} key
 * @param {string} data
 * @returns {Promise<Uint8Array>}
 */
async function hmac(key, data) {
  const cryptoKey = await crypto.subtle.importKey(
    'raw',
    /** @type {BufferSource} */ (key),
    { name: 'HMAC', hash: 'SHA-256' },
    false,
    ['sign'],
  );
  const mac = await crypto.subtle.sign(
    'HMAC',
    cryptoKey,
    /** @type {BufferSource} */ (encoder.encode(data)),
  );
  return new Uint8Array(mac);
}

/**
 * AWS uri-encoding: RFC 3986 strict — encodeURIComponent plus the characters
 * it leaves bare that AWS requires encoded.
 * @param {string} value
 * @returns {string}
 */
function uriEncode(value) {
  return encodeURIComponent(value).replace(
    /[!'()*]/g,
    (c) => `%${c.charCodeAt(0).toString(16).toUpperCase()}`,
  );
}

/**
 * A path for the canonical request and the URL: each segment encoded, the
 * slashes kept.
 * @param {string} path
 * @returns {string}
 */
function encodePath(path) {
  return path.split('/').map(uriEncode).join('/');
}

/** Decode the five XML entities a ListObjectsV2 key can carry. */
/**
 * @param {string} value
 * @returns {string}
 */
function decodeXmlEntities(value) {
  return value
    .replaceAll('&lt;', '<')
    .replaceAll('&gt;', '>')
    .replaceAll('&quot;', '"')
    .replaceAll('&apos;', "'")
    .replaceAll('&amp;', '&');
}

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

/**
 * Cut a byte stream into parts. The last part is whatever is left, which is
 * the one place S3 accepts a part under the minimum size.
 * @param {ReadableStream<Uint8Array>} stream
 * @returns {AsyncGenerator<Uint8Array>}
 */
async function* streamParts(stream) {
  const reader = stream.getReader();
  /** @type {Uint8Array[]} */
  let pending = [];
  let size = 0;
  try {
    while (true) {
      const { done, value } = await reader.read();
      if (done) break;
      if (!value.byteLength) continue;
      pending.push(value);
      size += value.byteLength;
      while (size >= PART_SIZE) {
        const joined = concat(pending, size);
        yield joined.subarray(0, PART_SIZE);
        const remainder = joined.subarray(PART_SIZE);
        pending = remainder.byteLength ? [remainder] : [];
        size = remainder.byteLength;
      }
    }
  } finally {
    reader.releaseLock();
  }
  if (size) yield concat(pending, size);
}

/**
 * Cut a buffer into parts, as views rather than copies.
 * @param {Uint8Array} data
 * @returns {AsyncGenerator<Uint8Array>}
 */
async function* bufferParts(data) {
  for (let at = 0; at < data.byteLength; at += PART_SIZE) {
    yield data.subarray(at, Math.min(at + PART_SIZE, data.byteLength));
  }
}

/**
 * Backups in any S3-compatible bucket, over nothing but fetch and WebCrypto —
 * the target to use on Cloudflare Workers, and the simplest one anywhere
 * else. Uses path-style addressing (`endpoint/bucket/key`), which every
 * compatible provider accepts. Give backups their own bucket: core deletes
 * anything under `snapshots/` and `blobs/` that its retention policy no
 * longer wants.
 *
 * @param {Object} options
 * @param {string} options.endpoint - Base URL of the S3 API, e.g.
 *   `https://ny.storage.bunnycdn.com` or `https://s3.us-west-004.backblazeb2.com`
 * @param {string} options.bucket - Bucket (or storage zone) name
 * @param {string} options.region - Signing region; many providers accept
 *   anything here, AWS and B2 want the real one
 * @param {string} options.accessKeyId
 * @param {string} options.secretAccessKey
 * @param {typeof fetch} [options.fetchImpl] - Injectable for tests
 * @returns {BackupTargetPort}
 */
export function createS3FetchBackupTarget({
  endpoint,
  bucket,
  region,
  accessKeyId,
  secretAccessKey,
  fetchImpl,
}) {
  const url = new URL(endpoint);
  const basePath = url.pathname.replace(/\/+$/, '');
  const doFetch = fetchImpl ?? fetch;

  /**
   * Sign and send one request.
   * @param {string} method
   * @param {string} key - Object key, or '' for bucket-level requests
   * @param {Object} [options]
   * @param {Record<string, string>} [options.query]
   * @param {Uint8Array} [options.body]
   * @param {string} [options.contentType]
   * @returns {Promise<Response>}
   */
  async function send(method, key, { query = {}, body, contentType } = {}) {
    const now = new Date();
    const amzDate = `${now.toISOString().slice(0, 19).replace(/[-:]/g, '')}Z`;
    const dateStamp = amzDate.slice(0, 8);
    const payloadHash = body ? await sha256Hex(body) : EMPTY_PAYLOAD_HASH;

    const canonicalUri = encodePath(
      key ? `${basePath}/${bucket}/${key}` : `${basePath}/${bucket}`,
    );
    const canonicalQuery = Object.keys(query)
      .sort()
      .map((k) => `${uriEncode(k)}=${uriEncode(query[k])}`)
      .join('&');
    const signedHeaders = 'host;x-amz-content-sha256;x-amz-date';
    const canonicalRequest = [
      method,
      canonicalUri,
      canonicalQuery,
      `host:${url.host}`,
      `x-amz-content-sha256:${payloadHash}`,
      `x-amz-date:${amzDate}`,
      '',
      signedHeaders,
      payloadHash,
    ].join('\n');

    const scope = `${dateStamp}/${region}/s3/aws4_request`;
    const stringToSign = [
      'AWS4-HMAC-SHA256',
      amzDate,
      scope,
      await sha256Hex(encoder.encode(canonicalRequest)),
    ].join('\n');

    let signingKey = await hmac(
      encoder.encode(`AWS4${secretAccessKey}`),
      dateStamp,
    );
    for (const step of [region, 's3', 'aws4_request']) {
      signingKey = await hmac(signingKey, step);
    }
    const signature = hex(await hmac(signingKey, stringToSign));

    /** @type {Record<string, string>} */
    const headers = {
      Authorization: `AWS4-HMAC-SHA256 Credential=${accessKeyId}/${scope}, SignedHeaders=${signedHeaders}, Signature=${signature}`,
      'x-amz-content-sha256': payloadHash,
      'x-amz-date': amzDate,
    };
    if (contentType) headers['Content-Type'] = contentType;

    const target = `${url.origin}${canonicalUri}${
      canonicalQuery ? `?${canonicalQuery}` : ''
    }`;
    return doFetch(target, {
      method,
      headers,
      body: /** @type {BodyInit|undefined} */ (body),
    });
  }

  /**
   * The error for a response that failed, for the caller to throw. Reading
   * the detail out of the body is what makes this asynchronous, so callers
   * write `throw await failure(...)`. A `return` of a rejected promise from
   * inside a `try` is not caught by that same try.
   * @param {Response} response
   * @param {string} doing
   * @returns {Promise<Error>}
   */
  async function failure(response, doing) {
    const detail = (await response.text().catch(() => '')).slice(0, 200);
    return new Error(
      `Backup target ${doing} failed: HTTP ${response.status}${
        detail ? ` — ${detail}` : ''
      }`,
    );
  }

  /**
   * One object in one request.
   * @param {string} key
   * @param {Uint8Array} data
   * @param {string} [contentType]
   * @returns {Promise<void>}
   */
  async function putBytes(key, data, contentType) {
    const response = await send('PUT', key, {
      body: data,
      contentType: contentType || 'application/octet-stream',
    });
    if (!response.ok) throw await failure(response, `put ${key}`);
    // Some providers only release the connection once the body is read.
    await response.arrayBuffer().catch(() => undefined);
  }

  /**
   * One object as a multipart upload, or as a plain PUT when it fits in a
   * single part. Providers answer HTTP 413 to a body past their single
   * request limit, which every video-sized blob is.
   * @param {string} key
   * @param {AsyncGenerator<Uint8Array>} parts
   * @param {string} [contentType]
   * @returns {Promise<void>}
   */
  async function putParts(key, parts, contentType) {
    const first = await parts.next();
    const head = first.done ? new Uint8Array(0) : first.value;
    let next = await parts.next();
    if (next.done) return putBytes(key, head, contentType);

    const created = await send('POST', key, {
      query: { uploads: '' },
      contentType: contentType || 'application/octet-stream',
    });
    if (!created.ok) throw await failure(created, `start upload ${key}`);
    const uploadId = decodeXmlEntities(
      /<UploadId>([^<]*)<\/UploadId>/.exec(await created.text())?.[1] ?? '',
    );
    if (!uploadId) {
      throw new Error(`Backup target start upload ${key} failed: no upload id`);
    }

    /** @type {string[]} */
    const etags = [];
    try {
      let part = head;
      while (true) {
        const number = etags.length + 1;
        const response = await send('PUT', key, {
          query: { partNumber: String(number), uploadId },
          body: part,
        });
        if (!response.ok)
          throw await failure(response, `put part ${number} of ${key}`);
        await response.arrayBuffer().catch(() => undefined);
        const etag = response.headers.get('etag');
        if (!etag) {
          throw new Error(
            `Backup target put part ${number} of ${key} failed: no etag`,
          );
        }
        etags.push(etag);
        if (next.done) break;
        part = next.value;
        next = await parts.next();
      }

      const body = [
        '<CompleteMultipartUpload>',
        ...etags.map(
          (etag, index) =>
            `<Part><PartNumber>${index + 1}</PartNumber><ETag>${etag}</ETag></Part>`,
        ),
        '</CompleteMultipartUpload>',
      ].join('');
      const completed = await send('POST', key, {
        query: { uploadId },
        body: encoder.encode(body),
        contentType: 'application/xml',
      });
      if (!completed.ok)
        throw await failure(completed, `complete upload ${key}`);
      // S3 answers 200 and then reports the failure in the body.
      const text = await completed.text();
      if (text.includes('<Error')) {
        throw new Error(
          `Backup target complete upload ${key} failed: ${text.slice(0, 200)}`,
        );
      }
    } catch (err) {
      // The bucket charges for the parts of an upload nobody completes.
      await send('DELETE', key, { query: { uploadId } })
        .then((response) => response.arrayBuffer())
        .catch(() => undefined);
      throw err;
    }
  }

  return {
    /**
     * @param {string} key
     * @param {Uint8Array} data
     * @param {string} [contentType]
     */
    async put(key, data, contentType) {
      if (data.byteLength <= PART_SIZE) {
        return putBytes(key, data, contentType);
      }
      return putParts(key, bufferParts(data), contentType);
    },

    /**
     * @param {string} key
     * @param {ReadableStream<Uint8Array>} body
     * @param {string} [contentType]
     */
    async putStream(key, body, contentType) {
      return putParts(key, streamParts(body), contentType);
    },

    /** @param {string} key */
    async get(key) {
      const response = await send('GET', key);
      if (response.status === 404) return null;
      if (!response.ok) throw await failure(response, `get ${key}`);
      return new Uint8Array(await response.arrayBuffer());
    },

    /** @param {string} prefix */
    async list(prefix) {
      /** @type {string[]} */
      const keys = [];
      /** @type {string|undefined} */
      let continuationToken;
      do {
        /** @type {Record<string, string>} */
        const query = { 'list-type': '2', prefix };
        if (continuationToken) query['continuation-token'] = continuationToken;
        const response = await send('GET', '', { query });
        if (!response.ok) throw await failure(response, `list ${prefix}`);
        const xml = await response.text();
        for (const match of xml.matchAll(/<Key>([^<]*)<\/Key>/g)) {
          keys.push(decodeXmlEntities(match[1]));
        }
        continuationToken = xml.includes('<IsTruncated>true</IsTruncated>')
          ? decodeXmlEntities(
              /<NextContinuationToken>([^<]*)<\/NextContinuationToken>/.exec(
                xml,
              )?.[1] ?? '',
            ) || undefined
          : undefined;
      } while (continuationToken);
      return keys;
    },

    /** @param {string} key */
    async delete(key) {
      const response = await send('DELETE', key);
      // Gone already is the outcome we wanted.
      if (!response.ok && response.status !== 404) {
        throw await failure(response, `delete ${key}`);
      }
      await response.arrayBuffer().catch(() => undefined);
    },
  };
}