Ekklesia/core/tests/module-storage.integration.ts

import { randomBytes } from "node:crypto";
import { createServer, type Server, type Socket } from "node:net";
import {
  type ModuleStorage,
  type ModuleStorageBucket,
  ModuleStorageError,
} from "@ekklesia/contracts";
import { afterAll, describe, expect, it } from "vitest";
import { physicalBucketName } from "../src/storage/binding.js";
import { createStorageProviderConfig } from "../src/storage/config.js";
import { createModuleStorage } from "../src/storage/module-storage.js";
import type { StorageProvider } from "../src/storage/provider.js";
import { createS3StorageProvider } from "../src/storage/s3-provider.js";

/**
 * The storage contract against a **real S3-compatible server**.
 *
 * The unit suite proves scoping, binding, and validation with an in-memory provider; this proves the
 * things only a real server can — and the reason it must exist is concrete. Probing the pinned
 * SeaweedFS 4.28 image directly revealed that `CreateBucket` on an existing bucket reports
 * `BucketAlreadyExists`, **not** AWS's `BucketAlreadyOwnedByYou`. A provider that only treated the
 * AWS spelling as success would have failed on every second boot against this repository's own
 * development storage service, and no amount of mocking would have caught it.
 *
 * SeaweedFS is used rather than a generic mock because it is the intended deployment target
 * (ADR 0004) and because the repository's Compose file already pins and provisions it, so this adds
 * no new CI infrastructure. Nothing here is SeaweedFS-specific in the *contract* — the same suite
 * would pass against MinIO or AWS S3, which is the portability claim being tested.
 *
 * ## Why this suite is frugal with buckets
 *
 * **A bucket is a scarce resource on a real server, so tests share one.** On the pinned SeaweedFS
 * image every bucket becomes its own storage *collection*, and each collection immediately claims a
 * block of volumes out of a budget the server derives from free disk. A suite that provisioned a
 * fresh bucket per test therefore consumed that budget in proportion to its own size, and once the
 * budget was gone the server did not fail the write — it accepted the connection and retried
 * allocation internally, forever. That is environment-dependent by construction: it passed on a
 * roomy disk and hung on a smaller one, which is exactly how it reached CI.
 *
 * So bucket identity is provisioned only where bucket identity is the thing under test — module
 * isolation, the longest representable name, and the never-delete invariant. Everything else shares
 * one bucket and isolates by **key namespace**, which tests prefix scoping honestly at the same
 * time. Keep it that way: a new test should reach for `sharedBucket()` and a `keyNamespace()`
 * unless it genuinely needs a bucket of its own.
 */
const endpointEnvironment = "EKKLESIA_TEST_STORAGE_ENDPOINT";
const accessKeyEnvironment = "EKKLESIA_TEST_STORAGE_ACCESS_KEY";
const secretKeyEnvironment = "EKKLESIA_TEST_STORAGE_SECRET_KEY";

class IntegrationHarnessError extends Error {
  constructor(message: string) {
    super(message);
    this.name = "IntegrationHarnessError";
    this.stack = `${this.name}: ${this.message}`;
  }
}

function providerConfig(endpoint = process.env[endpointEnvironment]) {
  if (typeof endpoint !== "string" || endpoint.trim() === "") {
    throw new IntegrationHarnessError("The storage integration endpoint is not configured.");
  }
  const config = createStorageProviderConfig(
    {
      accessKey: process.env[accessKeyEnvironment],
      endpoint,
      secretKey: process.env[secretKeyEnvironment],
    },
    // The disposable service speaks plaintext on loopback, exactly like the disposable PostgreSQL
    // this suite already uses; production configuration would refuse it.
    { environment: "development" },
  );
  if (!config) {
    throw new IntegrationHarnessError("The storage integration configuration is invalid.");
  }
  return config;
}

/** A unique prefix per run, so repeated runs and parallel jobs never collide and no test depends on
 * another's leftovers. */
function disposablePrefix(): string {
  return `ekktest${randomBytes(3).toString("hex")}`;
}

const openProviders: StorageProvider[] = [];

function provider(): StorageProvider {
  const created = createS3StorageProvider(providerConfig());
  openProviders.push(created);
  return created;
}

/** The prefix every shared-bucket test binds through. */
const runPrefix = disposablePrefix();

let cachedProvider: StorageProvider | undefined;

/** One client for the whole file. Each `S3Client` carries its own connection pool, so building one
 * per test spent sockets to prove nothing. */
function sharedProvider(): StorageProvider {
  cachedProvider ??= provider();
  return cachedProvider;
}

let cachedStorage: ModuleStorage | undefined;

function sharedBucket(): ModuleStorageBucket {
  cachedStorage ??= createModuleStorage({
    bucketPrefix: runPrefix,
    logicalBucketIds: ["announcement-assets"],
    moduleId: "announcements",
    provider: sharedProvider(),
  });
  return cachedStorage.bucket("announcement-assets");
}

let namespaceCounter = 0;

/** A key namespace private to one test. `describe.sequential` means these never interleave, so a
 * counter is both sufficient and readable in a failure message. */
function keyNamespace(): string {
  namespaceCounter += 1;
  return `case-${namespaceCounter}`;
}

afterAll(async () => {
  for (const open of openProviders.splice(0)) {
    await open.close().catch(() => undefined);
  }
});

describe.sequential("live S3-compatible module storage", () => {
  it("creates a module's bucket idempotently, across both provider dialects", async () => {
    const bucket = physicalBucketName(runPrefix, "announcements", "announcement-assets");

    await sharedProvider().ensureBucket(bucket);
    // The assertion this whole suite exists for: the pinned SeaweedFS image answers a repeat
    // `CreateBucket` with `BucketAlreadyExists` rather than AWS's `BucketAlreadyOwnedByYou`, and the
    // provider must treat both as success. Every restart depends on this.
    await expect(sharedProvider().ensureBucket(bucket)).resolves.toBeUndefined();
    await expect(sharedProvider().ensureBucket(bucket)).resolves.toBeUndefined();
  });

  it("round-trips an object through the scoped facade with its content type preserved", async () => {
    const bucket = sharedBucket();
    const namespace = keyNamespace();
    const key = `${namespace}/slides/welcome.png`;
    const png = new Uint8Array([0x89, 0x50, 0x4e, 0x47, 0x0d, 0x0a, 0x1a, 0x0a]);

    const written = await bucket.put(key, png, { contentType: "image/png" });
    expect(written.key).toBe(key);

    const object = await bucket.get(key);
    expect(object.contentType).toBe("image/png");
    expect(object.contentLength).toBe(png.byteLength);
    // Unquoted for the caller — providers return a quoted ETag, and every module learning to strip
    // quotes would be a contract failure rather than a module failure.
    expect(object.etag).toMatch(/^[0-9a-f]{32}$/u);
    expect(new Uint8Array(await new Response(object.body).arrayBuffer())).toEqual(png);
  });

  it("streams a body without buffering, when the length is supplied", async () => {
    const bucket = sharedBucket();
    const key = `${keyNamespace()}/large.bin`;
    const payload = "x".repeat(64 * 1024);
    const stream = new Blob([payload]).stream() as ReadableStream<Uint8Array>;

    await bucket.put(key, stream, {
      contentLength: payload.length,
      contentType: "application/octet-stream",
    });

    const object = await bucket.get(key);
    expect(object.contentLength).toBe(payload.length);
    expect(await new Response(object.body).text()).toBe(payload);
  });

  it("returns a Web stream a module can hand straight to a Response", async () => {
    const bucket = sharedBucket();
    const key = `${keyNamespace()}/welcome.png`;
    await bucket.put(key, new Uint8Array([1, 2, 3]), { contentType: "image/png" });

    const object = await bucket.get(key);
    expect(object.body).toBeInstanceOf(ReadableStream);
    // The browser-delivery shape this pass settles on: browser → module route → scoped storage,
    // with no S3 knowledge or credential anywhere near the browser.
    const response = new Response(object.body, {
      headers: { "content-type": object.contentType ?? "application/octet-stream" },
    });
    expect(response.headers.get("content-type")).toBe("image/png");
    expect(new Uint8Array(await response.arrayBuffer())).toEqual(new Uint8Array([1, 2, 3]));
  });

  it("heads, lists with a prefix, and deletes", async () => {
    const bucket = sharedBucket();
    const namespace = keyNamespace();

    await bucket.put(`${namespace}/slides/a.png`, "a", { contentType: "image/png" });
    await bucket.put(`${namespace}/slides/b.png`, "b", { contentType: "image/png" });
    await bucket.put(`${namespace}/other/c.txt`, "c");

    const metadata = await bucket.head(`${namespace}/slides/a.png`);
    expect(metadata.contentType).toBe("image/png");
    expect(metadata.contentLength).toBe(1);
    expect(metadata.lastModified).toBeInstanceOf(Date);

    // The sibling `other/` key proves the prefix genuinely filters rather than the bucket happening
    // to hold only matching keys.
    const listing = await bucket.list({ prefix: `${namespace}/slides/` });
    expect(listing.objects.map((object) => object.key)).toEqual([
      `${namespace}/slides/a.png`,
      `${namespace}/slides/b.png`,
    ]);
    expect(listing.objects[0]?.contentLength).toBe(1);
    // A complete listing carries no cursor, so an absent cursor always means "done".
    expect(listing.cursor).toBeUndefined();

    await bucket.delete(`${namespace}/slides/a.png`);
    expect(
      (await bucket.list({ prefix: `${namespace}/slides/` })).objects.map((o) => o.key),
    ).toEqual([`${namespace}/slides/b.png`]);
  });

  it("paginates a listing rather than silently truncating it", async () => {
    const bucket = sharedBucket();
    const namespace = keyNamespace();
    for (const index of [0, 1, 2, 3, 4]) {
      await bucket.put(`${namespace}/object-${index}.txt`, String(index));
    }

    const first = await bucket.list({ limit: 2, prefix: `${namespace}/` });
    expect(first.objects).toHaveLength(2);
    expect(first.cursor).toBeDefined();

    const keys = [...first.objects.map((object) => object.key)];
    let cursor = first.cursor;
    while (cursor !== undefined) {
      const page = await bucket.list({ cursor, limit: 2, prefix: `${namespace}/` });
      keys.push(...page.objects.map((object) => object.key));
      cursor = page.cursor;
    }

    expect(keys.sort()).toEqual([
      `${namespace}/object-0.txt`,
      `${namespace}/object-1.txt`,
      `${namespace}/object-2.txt`,
      `${namespace}/object-3.txt`,
      `${namespace}/object-4.txt`,
    ]);
  });

  it("maps a missing object to a structured not-found error, never a provider exception", async () => {
    const bucket = sharedBucket();
    const namespace = keyNamespace();
    await bucket.put(`${namespace}/present.txt`, "x");

    let caught: unknown;
    try {
      await bucket.get(`${namespace}/absent.txt`);
    } catch (error) {
      caught = error;
    }

    expect(caught).toBeInstanceOf(ModuleStorageError);
    expect((caught as ModuleStorageError).code).toBe("STORAGE_OBJECT_NOT_FOUND");
    expect((caught as ModuleStorageError).key).toBe(`${namespace}/absent.txt`);
    // Nothing from the provider survives translation: no endpoint, no bucket, no request id, no XML.
    const rendered = `${String(caught)} ${JSON.stringify(caught)}`;
    expect(rendered).not.toContain("http");
    expect(rendered).not.toContain(runPrefix);
    expect(rendered).not.toContain("<?xml");
    expect(rendered).not.toContain(process.env[accessKeyEnvironment] ?? " ");
  });

  it("keeps two modules declaring the same logical bucket completely isolated", async () => {
    // Genuinely needs buckets of its own: the claim is that two modules with the same logical name
    // land in two different physical buckets, which a shared bucket could not express.
    const prefix = disposablePrefix();
    const announcements = createModuleStorage({
      bucketPrefix: prefix,
      logicalBucketIds: ["assets"],
      moduleId: "announcements",
      provider: sharedProvider(),
    });
    const video = createModuleStorage({
      bucketPrefix: prefix,
      logicalBucketIds: ["assets"],
      moduleId: "video",
      provider: sharedProvider(),
    });

    await announcements.bucket("assets").put("same-key.txt", "from-announcements");
    await video.bucket("assets").put("same-key.txt", "from-video");

    // Same logical name, same key, one real storage service — and two different objects.
    expect(
      await new Response((await announcements.bucket("assets").get("same-key.txt")).body).text(),
    ).toBe("from-announcements");
    expect(await new Response((await video.bucket("assets").get("same-key.txt")).body).text()).toBe(
      "from-video",
    );

    // Neither module can see the other's object, even by listing.
    expect((await announcements.bucket("assets").list()).objects.map((o) => o.key)).toEqual([
      "same-key.txt",
    ]);
    // And there is no way to ask for the other module's bucket at all.
    expect(() => announcements.bucket("video-assets")).toThrow(ModuleStorageError);
  });

  it("accepts the longest bucket name the mapping can produce", async () => {
    const moduleId = `m${"o".repeat(62)}`;
    const bucket = physicalBucketName(runPrefix, moduleId, `b${"u".repeat(46)}`);

    expect(bucket.length).toBeLessThanOrEqual(63);
    // Proven against the real server rather than against the documented rule.
    await expect(sharedProvider().ensureBucket(bucket)).resolves.toBeUndefined();
  });

  it("supports nested object keys, which modules genuinely use", async () => {
    const bucket = sharedBucket();
    const namespace = keyNamespace();

    await bucket.put(`${namespace}/2026/easter/slides/welcome.png`, "x", {
      contentType: "image/png",
    });
    expect(
      (await bucket.list({ prefix: `${namespace}/2026/easter/` })).objects.map((o) => o.key),
    ).toEqual([`${namespace}/2026/easter/slides/welcome.png`]);
  });

  it("probes a reachable provider, and reports an unreachable one as provider-unavailable", async () => {
    await expect(sharedProvider().probe()).resolves.toBeUndefined();

    const unreachable = createS3StorageProvider(
      // A port nothing listens on: a real connection failure rather than a simulated one.
      providerConfig("http://127.0.0.1:1"),
    );
    try {
      let caught: unknown;
      try {
        await unreachable.probe();
      } catch (error) {
        caught = error;
      }
      expect(caught).toBeInstanceOf(ModuleStorageError);
      // Transient by nature: a module retries through this, and no lifecycle state changes
      // (ADR 0030).
      expect((caught as ModuleStorageError).code).toBe("STORAGE_PROVIDER_UNAVAILABLE");
    } finally {
      await unreachable.close().catch(() => undefined);
    }
  });

  it("gives up on a provider that accepts the connection and then goes silent", async () => {
    // A refused connection fails fast on its own; a provider that *answers* and then stalls is the
    // failure with no natural end, and the one that has to be bounded deliberately. A server which
    // accepts sockets and never writes a byte reproduces it exactly.
    const accepted = new Set<Socket>();
    const blackHole: Server = createServer((socket) => {
      // Deliberately never responds and never closes; only remembered so teardown can cut it off.
      accepted.add(socket);
      socket.on("close", () => accepted.delete(socket));
    });
    await new Promise<void>((resolve) => blackHole.listen(0, "127.0.0.1", resolve));
    const address = blackHole.address();
    if (address === null || typeof address === "string") {
      throw new IntegrationHarnessError("The stalled-provider server did not bind a port.");
    }

    const stalled = createS3StorageProvider(providerConfig(`http://127.0.0.1:${address.port}`), {
      // The production defaults are deliberately patient; this proves the mechanism, not the number.
      maxAttempts: 2,
      socketIdleTimeoutMs: 250,
    });
    try {
      const startedAt = Date.now();
      let caught: unknown;
      try {
        await stalled.probe();
      } catch (error) {
        caught = error;
      }
      const elapsed = Date.now() - startedAt;

      expect(caught).toBeInstanceOf(ModuleStorageError);
      expect((caught as ModuleStorageError).code).toBe("STORAGE_PROVIDER_UNAVAILABLE");
      // The point of the assertion is that it *ended*. Without a socket deadline the SDK waits
      // forever and only the test runner's own timeout stops it, which is what a module's request
      // handler would do in production with no runner to rescue it.
      expect(elapsed).toBeLessThan(5_000);
    } finally {
      await stalled.close().catch(() => undefined);
      // `close()` alone only stops new connections and then waits for established ones to end,
      // and the SDK's keep-alive agent holds this one open indefinitely — the never-finished socket
      // is the whole point of the server. Cut them off so teardown cannot become its own hang.
      for (const socket of accepted) {
        socket.destroy();
      }
      await new Promise<void>((resolve) => blackHole.close(() => resolve()));
    }
  });

  it("never deletes a bucket, even when every object in it is removed", async () => {
    // Needs a bucket of its own: the claim is about a bucket whose *every* object is gone, which
    // cannot be asserted inside a bucket other tests are also writing to.
    const bucket = createModuleStorage({
      bucketPrefix: disposablePrefix(),
      logicalBucketIds: ["assets"],
      moduleId: "announcements",
      provider: sharedProvider(),
    }).bucket("assets");

    await bucket.put("only.txt", "x");
    await bucket.delete("only.txt");

    // The bucket still exists and still accepts writes: emptying is not removal, and nothing in this
    // platform destroys a bucket (ADR 0030).
    expect((await bucket.list()).objects).toEqual([]);
    await expect(bucket.put("again.txt", "y")).resolves.toMatchObject({ key: "again.txt" });
  });
});