/**
 * S3-compatible storage provider.
 * Works with AWS S3, MinIO, Backblaze B2, Wasabi, and any S3-compatible service.
 */

import { createReadStream, createWriteStream } from 'node:fs';
import { mkdir } from 'node:fs/promises';
import { dirname } from 'node:path';
import { Readable } from 'node:stream';
import { pipeline } from 'node:stream/promises';
import {
  DeleteObjectCommand,
  GetObjectCommand,
  ListObjectsV2Command,
  PutObjectCommand,
  S3Client,
} from '@aws-sdk/client-s3';
import { Upload } from '@aws-sdk/lib-storage';
import type { StorageProvider, StorageVerifyResult } from './types';

const BACKUP_PREFIX = 'celilo-backups';

/**
 * Multipart upload sizing. Peak resident bytes for an upload is
 * `UPLOAD_PART_SIZE * UPLOAD_QUEUE_SIZE` — 64 MB — and does NOT grow with the
 * artifact. That bound is the whole point of these two constants.
 *
 * `upload` used to `readFileSync` the artifact into one Buffer and hand it to
 * PutObject. The comment justifying that read:
 *
 *   > Upload a Buffer, NOT a read stream (ISS-0016). A streamed Body fails with
 *   > "The request body terminated unexpectedly" [...] Backup envelopes are
 *   > small (state + key, not provider binaries — ISS-0015), so buffering is
 *   > fine.
 *
 * The first half is true and still is: a one-shot Node stream as a PutObject
 * `Body` cannot be replayed across the retries and redirects S3 issues, and
 * with no ContentLength the SDK falls back to aws-chunked encoding on top.
 * Passing `createReadStream` straight to PutObject would reintroduce exactly
 * that bug.
 *
 * The second half stopped being true without anyone revisiting it. It was
 * written when this provider only carried SYSTEM backups (celilo state plus a
 * key, a few MB). MODULE backups now use the same provider and are three orders
 * of magnitude larger — forgejo's envelope reached 1.87 GB — so "buffering is
 * fine" became a ~1.9 GB Buffer on a 3784 MB management server, and the OOM
 * killer took the backup every hour for a day while the on_backup hook itself
 * reported success (celilo#685).
 *
 * `Upload` resolves the two halves rather than trading one for the other: it
 * reads the stream a part at a time and each PART is a replayable buffer, so
 * retries work without the whole object ever being resident. 16 MB parts keep
 * a 1.87 GB artifact at ~117 requests, well inside S3's 10,000-part limit,
 * which leaves headroom to ~160 GB.
 */
export const UPLOAD_PART_SIZE = 16 * 1024 * 1024;
export const UPLOAD_QUEUE_SIZE = 4;

export interface S3StorageConfig {
  bucket: string;
  region: string;
  endpoint: string;
  accessKeyId: string;
  secretAccessKey: string;
}

function createS3Client(config: S3StorageConfig): S3Client {
  return new S3Client({
    region: config.region,
    endpoint: config.endpoint,
    credentials: {
      accessKeyId: config.accessKeyId,
      secretAccessKey: config.secretAccessKey,
    },
    forcePathStyle: true, // Required for MinIO and most S3-compatible services
  });
}

function prefixedKey(remotePath: string): string {
  return `${BACKUP_PREFIX}/${remotePath}`;
}

export function createS3StorageProvider(config: S3StorageConfig): StorageProvider {
  const client = createS3Client(config);
  const bucket = config.bucket;

  return {
    async upload(localPath: string, remotePath: string): Promise<void> {
      // Multipart, so peak memory is UPLOAD_PART_SIZE * UPLOAD_QUEUE_SIZE
      // regardless of how large the artifact is.
      await new Upload({
        client,
        partSize: UPLOAD_PART_SIZE,
        queueSize: UPLOAD_QUEUE_SIZE,
        params: {
          Bucket: bucket,
          Key: prefixedKey(remotePath),
          Body: createReadStream(localPath),
        },
      }).done();
    },

    async download(remotePath: string, localPath: string): Promise<void> {
      const response = await client.send(
        new GetObjectCommand({
          Bucket: bucket,
          Key: prefixedKey(remotePath),
        }),
      );

      if (!response.Body) {
        throw new Error(`Empty response for key: ${remotePath}`);
      }

      await mkdir(dirname(localPath), { recursive: true });

      const readable =
        response.Body instanceof Readable
          ? response.Body
          : Readable.fromWeb(response.Body as ReadableStream);

      const writable = createWriteStream(localPath);
      await pipeline(readable, writable);
    },

    async delete(remotePath: string): Promise<void> {
      await client.send(
        new DeleteObjectCommand({
          Bucket: bucket,
          Key: prefixedKey(remotePath),
        }),
      );
    },

    async list(prefix: string): Promise<string[]> {
      const fullPrefix = prefixedKey(prefix);
      const results: string[] = [];
      let continuationToken: string | undefined;

      do {
        const response = await client.send(
          new ListObjectsV2Command({
            Bucket: bucket,
            Prefix: fullPrefix,
            ContinuationToken: continuationToken,
          }),
        );

        for (const obj of response.Contents ?? []) {
          if (obj.Key) {
            // Strip the backup prefix to return relative paths
            const relative = obj.Key.startsWith(`${BACKUP_PREFIX}/`)
              ? obj.Key.slice(BACKUP_PREFIX.length + 1)
              : obj.Key;
            results.push(relative);
          }
        }

        continuationToken = response.IsTruncated ? response.NextContinuationToken : undefined;
      } while (continuationToken);

      return results;
    },

    async verify(): Promise<StorageVerifyResult> {
      const testKey = prefixedKey('.celilo-verify-test');

      try {
        // Write test object
        await client.send(
          new PutObjectCommand({
            Bucket: bucket,
            Key: testKey,
            Body: 'verify',
          }),
        );

        // Read it back
        const response = await client.send(
          new GetObjectCommand({
            Bucket: bucket,
            Key: testKey,
          }),
        );

        const body = await response.Body?.transformToString();
        if (body !== 'verify') {
          return { success: false, message: 'Read-back verification failed' };
        }

        // Delete test object
        await client.send(
          new DeleteObjectCommand({
            Bucket: bucket,
            Key: testKey,
          }),
        );

        return {
          success: true,
          message: `Connected to bucket '${bucket}', write test passed`,
        };
      } catch (error) {
        return {
          success: false,
          message: `S3 verification failed: ${error instanceof Error ? error.message : String(error)}`,
        };
      }
    },

    async initialize(): Promise<void> {
      // S3 doesn't need directory creation — objects are created on upload.
      // Just verify the bucket is accessible.
      await client.send(
        new ListObjectsV2Command({
          Bucket: bucket,
          Prefix: BACKUP_PREFIX,
          MaxKeys: 1,
        }),
      );
    },
  };
}