test: e2e for SSE live delivery, isolation and purge

tests/isolation-and-purge.test.ts

@@ -0,0 +1,82 @@
+import { afterAll, beforeAll, expect, test } from "bun:test";
+import type { Hono } from "hono";
+import type { TenantEnv } from "@intx/hub-api";
+import {
+  createInMemoryMailboxEventBus,
+  purgePrincipalMailbox,
+  writeMailboxMessage,
+} from "../src/index.js";
+import { seedScope } from "../src/test-helpers.js";
+import { as, createTestApp, createTestDb, type TestDb } from "./lib/db-harness.js";
+
+let testDb: TestDb | undefined;
+let app: Hono<TenantEnv>;
+let uid: number;
+
+beforeAll(async () => {
+  testDb = await createTestDb();
+  const { db } = testDb;
+  await seedScope(db, "tA", "alice");
+  await seedScope(db, "tB", "mallory");
+  app = createTestApp({
+    db,
+    bus: createInMemoryMailboxEventBus(),
+    senderAddressFor: ({ principalId }) => `${principalId}@example`,
+    deliver: () => {},
+  });
+  const written = await writeMailboxMessage(db, {
+    tenantId: "tA",
+    principalId: "alice",
+    address: "alice@tA.example",
+    fromAddress: "bob@tA.example",
+    subject: "Private",
+    body: "tenant A only",
+  });
+  uid = written!.uid;
+});
+
+afterAll(async () => {
+  await testDb?.close();
+});
+
+type ListBody = { messages: { uid: number }[] };
+
+async function list(tenantId: string, principalId: string): Promise<ListBody> {
+  const res = await app.request("/mailbox/me/inbox", {
+    headers: as(tenantId, principalId),
+  });
+  expect(res.status).toBe(200);
+  return (await res.json()) as ListBody;
+}
+
+test("a principal in another tenant cannot list or read tenant A's mail", async () => {
+  expect((await list("tA", "alice")).messages).toHaveLength(1);
+
+  expect((await list("tB", "mallory")).messages).toEqual([]);
+  expect((await list("tB", "alice")).messages).toEqual([]);
+
+  const thread = await app.request(`/mailbox/me/inbox/threads/${uid}`, {
+    headers: as("tB", "mallory"),
+  });
+  expect(thread.status).toBe(404);
+  const flag = await app.request(`/mailbox/me/inbox/${uid}/read`, {
+    method: "POST",
+    headers: as("tB", "mallory"),
+  });
+  expect(flag.status).toBe(404);
+
+  const ownFlag = await app.request(`/mailbox/me/inbox/${uid}/read`, {
+    method: "POST",
+    headers: as("tA", "alice"),
+  });
+  expect(ownFlag.status).toBe(200);
+});
+
+test("purging a principal removes their mail from the routes", async () => {
+  const purged = await purgePrincipalMailbox(testDb!.db, {
+    tenantId: "tA",
+    principalId: "alice",
+  });
+  expect(purged).toBe(1);
+  expect((await list("tA", "alice")).messages).toEqual([]);
+});

tests/lib/db-harness.ts

@@ -1,9 +1,8 @@
 import { drizzle } from "drizzle-orm/postgres-js";
 import { Hono } from "hono";
 import postgres from "postgres";
-import type { TenantEnv } from "@intx/hub-api";
+import type { TenantEnv, TenantRow } from "@intx/hub-api";
 import {
-  createInMemoryMailboxEventBus,
   createMailboxRoutes,
   runMailboxMigrations,
   type CreateMailboxRoutesDeps,
@@ -57,36 +56,54 @@ export async function createTestDb(): Promise<TestDb> {
   return { db, close };
 }
 
-export const TEST_TENANT = {
-  id: "t1",
-  name: "t1",
-  slug: "t1",
-  domain: "t1.example",
-  parentId: null,
-  config: null,
-  createdAt: new Date(0),
-  updatedAt: new Date(0),
-};
-
-/** Request header naming the principal the test host authenticates the request as. */
+/** Request headers naming the tenant and principal the test host authenticates the request as. */
+export const TENANT_HEADER = "x-test-tenant";
 export const PRINCIPAL_HEADER = "x-test-principal";
 
+export function testTenant(id: string): TenantRow {
+  return {
+    id,
+    name: id,
+    slug: id,
+    domain: `${id}.example`,
+    parentId: null,
+    config: null,
+    createdAt: new Date(0),
+    updatedAt: new Date(0),
+  };
+}
+
+/** Headers for a request made as `principalId` in `tenantId`. */
+export function as(tenantId: string, principalId: string): Headers {
+  return new Headers({ [TENANT_HEADER]: tenantId, [PRINCIPAL_HEADER]: principalId });
+}
+
+/** `as`, for a JSON request body. */
+export function jsonAs(tenantId: string, principalId: string): Headers {
+  const headers = as(tenantId, principalId);
+  headers.set("content-type", "application/json");
+  return headers;
+}
+
 /**
- * A host app that plays the tenant middleware: every request runs as
- * `TEST_TENANT` and the principal named by `PRINCIPAL_HEADER`, and the
- * mailbox routes are mounted under `/mailbox`.
+ * A host app that plays the tenant middleware: every request runs as the
+ * tenant and principal named by `TENANT_HEADER` and `PRINCIPAL_HEADER`, and
+ * the mailbox routes are mounted under `/mailbox`.
  */
 export function createTestApp(
-  deps: Pick<CreateMailboxRoutesDeps, "db" | "senderAddressFor" | "deliver">,
+  deps: Pick<CreateMailboxRoutesDeps, "db" | "bus" | "senderAddressFor" | "deliver">,
 ): Hono<TenantEnv> {
   const app = new Hono<TenantEnv>();
   app.use(async (c, next) => {
+    const tenantId = c.req.header(TENANT_HEADER);
     const principalId = c.req.header(PRINCIPAL_HEADER);
-    if (principalId === undefined) return c.json({ error: "unauthenticated" }, 401);
-    c.set("tenant", TEST_TENANT);
+    if (tenantId === undefined || principalId === undefined) {
+      return c.json({ error: "unauthenticated" }, 401);
+    }
+    c.set("tenant", testTenant(tenantId));
     c.set("principal", {
       id: principalId,
-      tenantId: TEST_TENANT.id,
+      tenantId,
       kind: "user",
       refId: principalId,
       status: "active",
@@ -99,7 +116,7 @@ export function createTestApp(
     "/mailbox",
     createMailboxRoutes({
       db: deps.db,
-      bus: createInMemoryMailboxEventBus(),
+      bus: deps.bus,
       requireGrant: allowAllGrants,
       senderAddressFor: deps.senderAddressFor,
       deliver: deps.deliver,

tests/send-and-read.test.ts

@@ -1,30 +1,37 @@
 import { afterAll, beforeAll, expect, test } from "bun:test";
 import type { Hono } from "hono";
 import type { TenantEnv } from "@intx/hub-api";
-import { createMailboxPersist } from "../src/index.js";
+import {
+  createInMemoryMailboxEventBus,
+  createMailboxPersist,
+} from "../src/index.js";
 import { seedScope } from "../src/test-helpers.js";
 import {
+  as,
   createTestApp,
   createTestDb,
-  PRINCIPAL_HEADER,
-  TEST_TENANT,
+  jsonAs,
+  testTenant,
   type TestDb,
 } from "./lib/db-harness.js";
 
+const TENANT = testTenant("t1");
+
 let testDb: TestDb | undefined;
 let app: Hono<TenantEnv>;
 
 beforeAll(async () => {
   const { db } = (testDb = await createTestDb());
-  await seedScope(db, TEST_TENANT.id, "alice", "bob");
+  await seedScope(db, TENANT.id, "alice", "bob");
   // The host's transport files each sent message into its recipients' inboxes.
   const persist = createMailboxPersist(db, {
     upstream: async () => {},
-    authorizeSender: () => ({ tenantId: TEST_TENANT.id, domain: TEST_TENANT.domain }),
+    authorizeSender: () => ({ tenantId: TENANT.id, domain: TENANT.domain }),
   });
   app = createTestApp({
     db,
-    senderAddressFor: ({ principalId }) => `${principalId}@${TEST_TENANT.domain}`,
+    bus: createInMemoryMailboxEventBus(),
+    senderAddressFor: ({ principalId }) => `${principalId}@${TENANT.domain}`,
     deliver: ({ from, to, raw }) => persist({ senderAddress: from, recipients: to, raw }),
   });
 });
@@ -44,13 +51,13 @@ type ListBody = {
 test("a message sent over HTTP is listed and readable in the recipient's inbox", async () => {
   const sent = await app.request("/mailbox/me/inbox/send", {
     method: "POST",
-    headers: { "content-type": "application/json", [PRINCIPAL_HEADER]: "alice" },
+    headers: jsonAs(TENANT.id, "alice"),
     body: JSON.stringify({ to: ["bob@t1.example"], subject: "Lunch", body: "Noon?" }),
   });
   expect(sent.status).toBe(200);
 
   const inbox = await app.request("/mailbox/me/inbox", {
-    headers: { [PRINCIPAL_HEADER]: "bob" },
+    headers: as(TENANT.id, "bob"),
   });
   expect(inbox.status).toBe(200);
   const { messages } = (await inbox.json()) as ListBody;
@@ -61,7 +68,7 @@ test("a message sent over HTTP is listed and readable in the recipient's inbox",
   expect(Buffer.from(message!.raw, "base64").toString()).toContain("Noon?");
 
   const thread = await app.request(`/mailbox/me/inbox/threads/${message!.uid}`, {
-    headers: { [PRINCIPAL_HEADER]: "bob" },
+    headers: as(TENANT.id, "bob"),
   });
   expect(thread.status).toBe(200);
   const { thread: root } = (await thread.json()) as {
@@ -71,7 +78,7 @@ test("a message sent over HTTP is listed and readable in the recipient's inbox",
   expect(root.envelope.subject).toBe("Lunch");
 
   const aliceInbox = await app.request("/mailbox/me/inbox", {
-    headers: { [PRINCIPAL_HEADER]: "alice" },
+    headers: as(TENANT.id, "alice"),
   });
   expect(((await aliceInbox.json()) as ListBody).messages).toHaveLength(0);
 });

tests/sse.test.ts

@@ -0,0 +1,106 @@
+import { afterAll, beforeAll, expect, test } from "bun:test";
+import {
+  createInMemoryMailboxEventBus,
+  writeMailboxMessage,
+  type MailboxEventBus,
+} from "../src/index.js";
+import { seedScope } from "../src/test-helpers.js";
+import { as, createTestApp, createTestDb, type TestDb } from "./lib/db-harness.js";
+
+let testDb: TestDb | undefined;
+
+beforeAll(async () => {
+  testDb = await createTestDb();
+  await seedScope(testDb.db, "t1", "bob");
+});
+
+afterAll(async () => {
+  await testDb?.close();
+});
+
+/** The in-memory bus, counting live subscriptions. */
+function countingBus(): { bus: MailboxEventBus; live: () => number } {
+  const inner = createInMemoryMailboxEventBus();
+  let live = 0;
+  return {
+    live: () => live,
+    bus: {
+      publish: (scope, event) => inner.publish(scope, event),
+      subscribe: (scope, listener) => {
+        live += 1;
+        const unsubscribe = inner.subscribe(scope, listener);
+        let done = false;
+        return () => {
+          if (done) return;
+          done = true;
+          live -= 1;
+          unsubscribe();
+        };
+      },
+    },
+  };
+}
+
+async function readUntil(
+  reader: ReadableStreamDefaultReader<Uint8Array>,
+  done: (text: string) => boolean,
+): Promise<string> {
+  const decoder = new TextDecoder();
+  let text = "";
+  const deadline = Date.now() + 5_000;
+  while (!done(text) && Date.now() < deadline) {
+    const chunk = await reader.read();
+    if (chunk.done) break;
+    text += decoder.decode(chunk.value);
+  }
+  return text;
+}
+
+async function waitFor(condition: () => boolean): Promise<void> {
+  const deadline = Date.now() + 5_000;
+  while (!condition() && Date.now() < deadline) {
+    await new Promise((resolve) => setTimeout(resolve, 10));
+  }
+}
+
+test("a connected client receives new mail live and unsubscribes on disconnect", async () => {
+  const { db } = testDb!;
+  const { bus, live } = countingBus();
+  const app = createTestApp({
+    db,
+    bus,
+    senderAddressFor: ({ principalId }) => `${principalId}@t1.example`,
+    deliver: () => {},
+  });
+  const abort = new AbortController();
+  const res = await app.request("/mailbox/me/inbox/events", {
+    headers: as("t1", "bob"),
+    signal: abort.signal,
+  });
+  expect(res.status).toBe(200);
+  await waitFor(() => live() === 1);
+  expect(live()).toBe(1);
+
+  await writeMailboxMessage(
+    db,
+    {
+      tenantId: "t1",
+      principalId: "bob",
+      address: "bob@t1.example",
+      fromAddress: "alice@t1.example",
+      subject: "Live",
+      body: "now",
+    },
+    bus,
+  );
+
+  const reader = res.body!.getReader();
+  const text = await readUntil(reader, (t) => t.includes("event: mailbox"));
+  expect(text).toContain("event: mailbox");
+  expect(text).toContain('"op":"create"');
+
+  abort.abort();
+  await reader.cancel();
+  await waitFor(() => live() === 0);
+  expect(live()).toBe(0);
+});