/**
 * Recurrence gates for the two ways this provider's upload has been wrong.
 *
 * ISS-0016 — a one-shot `createReadStream` passed straight to PutObject as
 * `Body` with no ContentLength. AWS SDK v3 fell back to aws-chunked streaming
 * and failed with "The request body terminated unexpectedly", and could not
 * replay the stream across S3's retries and redirects. The verify path used a
 * string body (length known) and so never exercised the broken path, which is
 * why every real S3 backup silently failed.
 *
 * celilo#685 — the fix for ISS-0016 was `readFileSync` into one Buffer, which
 * is replayable and was fine for the system envelopes this provider carried at
 * the time. Module backups then started using the same provider at a thousand
 * times the size, and a ~1.9 GB Buffer on a 3784 MB management server was
 * OOM-killed every hour for a day.
 *
 * The two constrain opposite things — replayable versus not resident — so
 * neither test is meaningful alone, and satisfying one by breaking the other is
 * exactly the history here. `Upload` (multipart) satisfies both: each PART is a
 * replayable buffer, and only a bounded number of parts exist at once.
 *
 * No live S3 / MinIO harness exists, so we intercept S3Client.prototype.send
 * and inspect the commands the provider builds.
 */

import { type Mock, afterEach, describe, expect, it, spyOn } from 'bun:test';
import { mkdtempSync, readFileSync, rmSync, writeFileSync } from 'node:fs';
import { tmpdir } from 'node:os';
import { join } from 'node:path';
import { Readable } from 'node:stream';
import {
  CompleteMultipartUploadCommand,
  CreateMultipartUploadCommand,
  GetObjectCommand,
  PutObjectCommand,
  S3Client,
  UploadPartCommand,
} from '@aws-sdk/client-s3';
import {
  type S3StorageConfig,
  UPLOAD_PART_SIZE,
  UPLOAD_QUEUE_SIZE,
  createS3StorageProvider,
} from './s3';

const CONFIG: S3StorageConfig = {
  bucket: 'test-bucket',
  region: 'us-east-1',
  endpoint: 'http://localhost:9000',
  accessKeyId: 'test-key',
  secretAccessKey: 'test-secret',
};

describe('s3 storage provider (ISS-0016)', () => {
  let dir: string | undefined;
  // biome-ignore lint/suspicious/noExplicitAny: send() is heavily overloaded; the test only reads .input.
  let sendSpy: Mock<(command: any) => Promise<unknown>> | undefined;

  afterEach(() => {
    sendSpy?.mockRestore();
    sendSpy = undefined;
    if (dir) {
      rmSync(dir, { recursive: true, force: true });
      dir = undefined;
    }
  });

  it('uploads a small file as a self-describing Buffer body, not a stream (ISS-0016)', async () => {
    dir = mkdtempSync(join(tmpdir(), 's3-upload-'));
    const file = join(dir, 'envelope.tar.gz');
    // A multi-KB real file — the case that broke against live S3.
    const bytes = Buffer.from('celilo backup envelope payload — '.repeat(2000));
    writeFileSync(file, bytes);

    // biome-ignore lint/suspicious/noExplicitAny: capturing the built command.
    const sent: any[] = [];
    sendSpy = spyOn(S3Client.prototype, 'send').mockImplementation(async (command: unknown) => {
      sent.push(command);
      return {};
    });

    const provider = createS3StorageProvider(CONFIG);
    await provider.upload(file, 'celilo-mgmt/2026/envelope.tar.gz');

    // Under one part, so no multipart ceremony — a single PutObject, exactly
    // as before. The bound below is what changed, not the small-file path.
    expect(sent).toHaveLength(1);
    const command = sent[0];
    expect(command).toBeInstanceOf(PutObjectCommand);

    const body = command.input.Body;
    // The regression gate: a Node read stream (the original bug) is not a Buffer.
    expect(Buffer.isBuffer(body)).toBe(true);
    expect(body instanceof Readable).toBe(false);
    // Self-describing length == the whole file: the SDK derives ContentLength
    // and can replay the body across retries/redirects.
    expect(body.length).toBe(bytes.length);
    expect(Buffer.compare(body, bytes)).toBe(0);
    // Key is prefixed with the backup namespace.
    expect(command.input.Key).toBe('celilo-backups/celilo-mgmt/2026/envelope.tar.gz');
    expect(command.input.Bucket).toBe('test-bucket');
  });

  it('never holds more than one part per queue slot, however large the file (celilo#685)', async () => {
    dir = mkdtempSync(join(tmpdir(), 's3-upload-large-'));
    const file = join(dir, 'backup.tar.enc');

    // Deliberately larger than one part, so the multipart path runs for real.
    // It does not need to approach forgejo's 1.87 GB: the property under test
    // is that peak residency is set by the part size rather than by the file,
    // and a file that spans several parts demonstrates that at any scale. A
    // test that had to allocate the failing size to prove the fix would be
    // reproducing the bug rather than gating it.
    const parts = 3;
    const fileSize = UPLOAD_PART_SIZE * parts;
    writeFileSync(file, Buffer.alloc(fileSize, 0x7a));

    // Deliberately NOT a running total. Parts are uploaded concurrently and can
    // arrive in any order, so the only way to show the artifact survives is to
    // keep each part against its number and reassemble.
    const received = new Map<number, Buffer>();
    let inFlight = 0;
    let peakInFlight = 0;
    sendSpy = spyOn(S3Client.prototype, 'send').mockImplementation(async (command: unknown) => {
      if (command instanceof CreateMultipartUploadCommand) return { UploadId: 'upload-1' };
      if (command instanceof CompleteMultipartUploadCommand) return {};

      expect(command).toBeInstanceOf(UploadPartCommand);
      const part = command as UploadPartCommand;
      const body = part.input.Body as Buffer;
      const partNumber = part.input.PartNumber as number;

      // Each part is still a replayable Buffer — ISS-0016 holds per part.
      expect(Buffer.isBuffer(body)).toBe(true);
      // A part number reused would silently lose data on reassembly.
      expect(received.has(partNumber)).toBe(false);
      received.set(partNumber, Buffer.from(body));

      inFlight += 1;
      peakInFlight = Math.max(peakInFlight, inFlight);
      await new Promise((resolve) => setTimeout(resolve, 1));
      inFlight -= 1;
      return { ETag: `"etag-${partNumber}"` };
    });

    const provider = createS3StorageProvider(CONFIG);
    await provider.upload(file, 'forgejo/2026-08-13/backup.tar.enc');

    // The bound. No single request ever carries the whole artifact, and no more
    // than the queue depth are resident at once, so peak bytes is
    // UPLOAD_PART_SIZE * UPLOAD_QUEUE_SIZE no matter how big the file gets.
    expect(received.size).toBe(parts);
    for (const body of received.values()) expect(body.length).toBeLessThanOrEqual(UPLOAD_PART_SIZE);
    expect(peakInFlight).toBeLessThanOrEqual(UPLOAD_QUEUE_SIZE);

    // A bound is only worth having if the artifact still arrives. Reassembled
    // in part order, the bytes must be the file — sizes summing correctly would
    // not catch a swapped or duplicated part, and a backup that restores to
    // scrambled bytes is worse than one that fails loudly.
    const ordered = [...received.entries()].sort(([a], [b]) => a - b).map(([, body]) => body);
    expect(Buffer.concat(ordered).equals(readFileSync(file))).toBe(true);
  });

  it('downloads a multi-chunk response body to disk intact (short-read safe)', async () => {
    dir = mkdtempSync(join(tmpdir(), 's3-download-'));
    const out = join(dir, 'restored.bin');
    const chunks = [Buffer.from('chunk-1-'), Buffer.from('chunk-2-'), Buffer.from('chunk-3')];
    const expected = Buffer.concat(chunks);

    sendSpy = spyOn(S3Client.prototype, 'send').mockImplementation(async (command: unknown) => {
      expect(command).toBeInstanceOf(GetObjectCommand);
      const body = new Readable({ read() {} });
      // Push several chunks separately to model short reads over the wire.
      for (const chunk of chunks) body.push(chunk);
      body.push(null);
      return { Body: body };
    });

    const provider = createS3StorageProvider(CONFIG);
    await provider.download('celilo-mgmt/2026/envelope.tar.gz', out);

    expect(readFileSync(out)).toEqual(expected);
  });
});