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; 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(" { // 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(); 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((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((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" }); }); });