feat(hub): compose notify delivery deps with mailbox read-back

apps/hub/src/index.ts

@@ -96,7 +96,6 @@ import type { FinalizedTurnToolCall } from "@corbits/turn-artifacts";
 import { decodedOrNull } from "@corbits/url-path";
 import {
   createInboxRoutes,
-  createWorkbenchMailboxDelivery,
   WORKBENCH_MAILBOX_VOCABULARY,
 } from "@corbits/inbox";
 import {
@@ -244,10 +243,7 @@ import {
   createProviderHealthPort,
   createProviderHealthStore,
 } from "@corbits/connections/provider-health";
-import {
-  createInMemoryNotifyDispatchStore,
-  createSinkRegistry,
-} from "@corbits/notify";
+import { createHubNotifyDeliveryDeps } from "./notify-delivery";
 import {
   createWorkflowAuthorRegistry,
   createWorkflowAuthorRoutes,
@@ -553,15 +549,6 @@ export async function createHub(config: HubConfig) {
     config.databaseUrl,
   );
   const mailboxBus = createInMemoryMailboxEventBus();
-  // Delivery adapter for `@corbits/notify` — kept at the composition root so
-  // routine / approval / mention writers can inject it without the hub
-  // re-implementing mailbox writes. The credential-expiry sweep below is
-  // its first live caller; approval/run-failure/mention still have no
-  // writer wired to this adapter.
-  const mailboxDelivery = createWorkbenchMailboxDelivery({
-    db: mailboxDb,
-    bus: mailboxBus,
-  });
   const log = getLogger(["hub", "auth"]);
   // Built once, tagged, and shared by every secret-at-rest seam in this
   // composition root — see `hubCredentialCipher`.
@@ -2426,8 +2413,8 @@ export async function createHub(config: HubConfig) {
   // Notify-to-reconnect for an OAuth-connected credential whose token
   // expired (Hugging Face today — see docs/onboarding-huggingface-connect.md):
   // a light periodic sweep over `@corbits/notify`'s pure
-  // `findDueCredentialExpiries`, mailing through the same delivery
-  // adapter above. `createInMemoryNotifyDispatchStore`/`createSinkRegistry()`
+  // `findDueCredentialExpiries`, mailing through the hub's notify delivery
+  // deps (`createHubNotifyDeliveryDeps`). `createInMemoryNotifyDispatchStore`/`createSinkRegistry()`
   // mean external sink fan-out (Slack, email) is a no-op until a sink is
   // registered — the mailbox row itself is what a person sees in their
   // inbox. Requires `@corbits/mailbox`'s and `@corbits/notify`'s own
@@ -2441,15 +2428,11 @@ export async function createHub(config: HubConfig) {
       sidecarRouter,
     ),
     hubUrl: config.baseUrl,
-    notify: {
-      mail: mailboxDelivery,
-      addressing: {
-        inbox: (recipient) => `${recipient.principalId}@inbox.${notifyHost}`,
-        from: (kind) => `${kind}@notify.${notifyHost}`,
-      },
-      dispatch: createInMemoryNotifyDispatchStore(),
-      sinks: createSinkRegistry(),
-    },
+    notify: createHubNotifyDeliveryDeps({
+      mailboxDb,
+      bus: mailboxBus,
+      host: notifyHost,
+    }),
   });
 
   // Reopen a snoozed inbox item once its `until` has passed (CL-7208) — a

apps/hub/src/notify-delivery.ts

@@ -0,0 +1,47 @@
+// The hub's `@corbits/notify` delivery deps, composed once at the root so
+// every mail writer (the credential-expiry sweep today) mails through the
+// same adapter the inbox read-back repairs. Kept here instead of inline in
+// `index.ts` so the composed path is a single construction site the
+// crash-window suite can exercise directly.
+import {
+  createResolveExistingMailIds,
+  createWorkbenchMailboxDelivery,
+} from "@corbits/inbox";
+import type { MailboxDb, MailboxEventBus } from "@corbits/mailbox";
+import {
+  createInMemoryNotifyDispatchStore,
+  createSinkRegistry,
+  type NotifyDeliveryDeps,
+  type NotifyDispatchStore,
+  type SinkRegistry,
+} from "@corbits/notify";
+
+export interface CreateHubNotifyDeliveryDepsOpts {
+  mailboxDb: MailboxDb;
+  bus?: MailboxEventBus;
+  host: string;
+  dispatch?: NotifyDispatchStore;
+  sinks?: SinkRegistry;
+}
+
+export function createHubNotifyDeliveryDeps(
+  opts: CreateHubNotifyDeliveryDepsOpts,
+): NotifyDeliveryDeps {
+  return {
+    mail: createWorkbenchMailboxDelivery({
+      db: opts.mailboxDb,
+      ...(opts.bus !== undefined ? { bus: opts.bus } : {}),
+    }),
+    addressing: {
+      inbox: (recipient) => `${recipient.principalId}@inbox.${opts.host}`,
+      from: (kind) => `${kind}@notify.${opts.host}`,
+    },
+    dispatch: opts.dispatch ?? createInMemoryNotifyDispatchStore(),
+    sinks: opts.sinks ?? createSinkRegistry(),
+    // CL-7238: without this read-back a redelivery after a crash between
+    // the mail write and the dispatch enqueue still loses the dispatch.
+    // Resolved from the mailbox unique mail key — the same key the write
+    // dedupes on — so the lookup can never disagree with the delivery.
+    resolveExistingMailIds: createResolveExistingMailIds(opts.mailboxDb),
+  };
+}

apps/hub/test/notify-delivery.test.ts

@@ -0,0 +1,215 @@
+// CL-7238 prod closure through the hub's own composed notify deps: the
+// credential-expiry sweep mails through `createHubNotifyDeliveryDeps`, so a
+// crash between the mail write and the dispatch enqueue must be repaired on
+// redelivery — the same crash-window pattern PR #734 proves for the bare
+// adapters, here through the exact factory `apps/hub/src/index.ts` composes.
+// Runs against its own scratch database, never the developer's or the
+// walking-skeleton suite's.
+import { afterAll, beforeAll, expect, test } from "bun:test";
+import postgres from "postgres";
+
+import { createDB, schema } from "@intx/db";
+import { generateId } from "@intx/hub-common";
+import {
+  createInMemoryMailboxEventBus,
+  createMailboxDb,
+  listUserMailbox,
+} from "@corbits/mailbox";
+import {
+  deliverApprovalMail,
+  type ApprovalNotification,
+  type NotifyDispatchStore,
+  type SinkDeliveryResult,
+} from "@corbits/notify";
+import { WORKBENCH_INBOX_PRIORITIES } from "@corbits/inbox";
+
+import { setupDatabase } from "../../../scripts/db-setup";
+import { e2eDatabaseUrl } from "../../../scripts/e2e/database-url";
+import { dbGate } from "../../../scripts/e2e/db-gate";
+import { createHubNotifyDeliveryDeps } from "../src/notify-delivery";
+
+function scratchUrlFor(e2eUrl: string): string {
+  const url = new URL(e2eUrl);
+  const database = url.pathname.replace(/^\//, "");
+  url.pathname = `/${database}_hub_notify_delivery_test`;
+  return url.toString();
+}
+
+// Parse DATABASE_URL the same way the hub does (apps/hub/src/index.ts): an
+// empty user falls through to the postgres client's OS-username default.
+function dbConfigFromUrl(databaseUrl: string) {
+  const url = new URL(databaseUrl);
+  return {
+    host: url.hostname,
+    port: url.port === "" ? 5432 : Number(url.port),
+    user: decodeURIComponent(url.username),
+    password: decodeURIComponent(url.password),
+    database: url.pathname.replace(/^\//, ""),
+  };
+}
+
+const databaseUrl = e2eDatabaseUrl();
+const describeIfDb = dbGate(databaseUrl, import.meta.path);
+
+describeIfDb(
+  "hub notify delivery repairs the CL-7238 crash window on redelivery",
+  () => {
+    const scratchUrl = scratchUrlFor(
+      databaseUrl ?? "postgres://localhost:5432/unused",
+    );
+    const scratchDatabase = new URL(scratchUrl).pathname.replace(/^\//, "");
+
+    const tenantId = generateId("tenant");
+    const principalId = generateId("principal");
+
+    beforeAll(async () => {
+      const maintenanceUrl = new URL(scratchUrl);
+      maintenanceUrl.pathname = "/postgres";
+      const maintenance = postgres(maintenanceUrl.toString(), {
+        max: 1,
+        onnotice: () => undefined,
+      });
+      try {
+        await maintenance.unsafe(
+          `DROP DATABASE IF EXISTS "${scratchDatabase}"`,
+        );
+        await maintenance.unsafe(`CREATE DATABASE "${scratchDatabase}"`);
+      } finally {
+        await maintenance.end();
+      }
+
+      // The same path the hub boots with: the platform's own migrations,
+      // then every installed package's, including `@corbits/notify` and
+      // `@corbits/mailbox`.
+      await setupDatabase(scratchUrl);
+
+      const { db, close } = createDB(dbConfigFromUrl(scratchUrl));
+      try {
+        await db.insert(schema.tenant).values({
+          id: tenantId,
+          name: "Hub Notify Test Bench",
+          slug: `hub-notify-${tenantId}`,
+          domain: `hub-notify-${tenantId}.localhost`,
+        });
+        await db.insert(schema.principal).values({
+          id: principalId,
+          tenantId,
+          kind: "agent",
+          refId: "not-a-real-agent-instance",
+          status: "active",
+        });
+      } finally {
+        await close();
+      }
+      // `setupDatabase` applies the platform's migrations plus every
+      // installed package's; under `bun run test`'s cross-package
+      // concurrency that comfortably exceeds bun:test's default 5s hook
+      // timeout, so it gets an explicit one here rather than a flaky suite.
+    }, 30000);
+
+    afterAll(async () => {
+      const maintenanceUrl = new URL(scratchUrl);
+      maintenanceUrl.pathname = "/postgres";
+      const maintenance = postgres(maintenanceUrl.toString(), {
+        max: 1,
+        onnotice: () => undefined,
+      });
+      try {
+        await maintenance.unsafe(
+          `DROP DATABASE IF EXISTS "${scratchDatabase}"`,
+        );
+      } finally {
+        await maintenance.end();
+      }
+    });
+
+    test("a crash-window redelivery through the composed hub deps queues its dispatch", async () => {
+      const mailboxDb = createMailboxDb(scratchUrl);
+      try {
+        // Exactly what the hub composes: the factory's own dispatch store
+        // and sink registry are the prod path (no sink registered means no
+        // fan-out, only the mailbox row the sweep's reader sees). The test
+        // registers one sink so the repaired dispatch has somewhere to go,
+        // and faults only the dispatch enqueue to open the crash window.
+        const deps = createHubNotifyDeliveryDeps({
+          mailboxDb: mailboxDb.db,
+          bus: createInMemoryMailboxEventBus(),
+          host: "hub.test",
+        });
+        const delivered: SinkDeliveryResult = { status: "delivered" };
+        deps.sinks.register({
+          name: "always",
+          isEnabledFor: async () => true,
+          deliver: async () => delivered,
+        });
+        const dispatch = deps.dispatch;
+        let crashNext = true;
+        const crashingDispatch: NotifyDispatchStore = {
+          ...dispatch,
+          enqueue: async (inputs) => {
+            if (crashNext) {
+              crashNext = false;
+              throw new Error(
+                "boom: crash between the mail write and the dispatch enqueue",
+              );
+            }
+            await dispatch.enqueue(inputs);
+          },
+        };
+        const event: ApprovalNotification = {
+          kind: "approval",
+          approvalId: generateId("approval"),
+          tenantId,
+          runId: generateId("workflowRun"),
+          deploymentId: generateId("workflowRun"),
+          toolName: "quarantine_token",
+          toolArguments: { repo: "corbitsdev/workbench" },
+          recipients: [{ tenantId, principalId }],
+          createdAt: new Date().toISOString(),
+        };
+
+        // The first attempt commits its mail row through the real adapter,
+        // then dies before the dispatch enqueue — the crash window.
+        await expect(
+          deliverApprovalMail({ ...deps, dispatch: crashingDispatch }, event),
+        ).rejects.toThrow("boom");
+
+        // The redelivery runs through the pristine composed deps: it
+        // dedupes on the mail key (no new row), resolves the committed
+        // row's id through the hub's read-back wiring, and queues the
+        // dispatch the crash swallowed. Without `resolveExistingMailIds`
+        // in the composed deps this queues nothing.
+        const redelivery = await deliverApprovalMail(deps, event);
+        expect(redelivery.deliveredMailboxRowIds).toEqual([]);
+        expect(redelivery.queuedDispatchCount).toBe(1);
+
+        const page = await listUserMailbox(mailboxDb.db, {
+          tenantId,
+          principalId,
+          limit: 10,
+          view: "all",
+          priorities: WORKBENCH_INBOX_PRIORITIES,
+        });
+        // Exactly one mail row for the redelivered event — the crash and
+        // the redelivery deduped onto the first attempt's row.
+        const matching = page.items.filter((item) =>
+          (item.subject ?? "").includes("quarantine_token"),
+        );
+        expect(matching).toHaveLength(1);
+        const mailId = matching[0]?.id;
+        if (mailId === undefined)
+          throw new Error("expected the redelivered mail row to list");
+        expect(await dispatch.listFor(mailId)).toHaveLength(1);
+
+        // A further benign redelivery repairs the same row again, and the
+        // dispatch store's (mail row, sink) dedupe holds it to one row.
+        const third = await deliverApprovalMail(deps, event);
+        expect(third.deliveredMailboxRowIds).toEqual([]);
+        expect(third.queuedDispatchCount).toBe(1);
+        expect(await dispatch.listFor(mailId)).toHaveLength(1);
+      } finally {
+        await mailboxDb.close();
+      }
+    }, 15000);
+  },
+);