feat(inbox): wire CL-7238 mail id read-back through the real adapter

packages/inbox/src/delivery.ts

@@ -3,13 +3,20 @@
 // group) and status so list filters and the three-column UI work without
 // re-deriving on every read.
 
+import { and, eq, or } from "drizzle-orm";
 import {
   deliverInboxItems,
+  mailboxKey,
+  principalMail,
   type DeliverInboxItemsOpts,
   type MailboxDb,
   type MailboxEventBus,
 } from "@corbits/mailbox";
-import type { MailboxDelivery, NotifyInboxItem } from "@corbits/notify/mailbox";
+import type {
+  MailboxDelivery,
+  NotifyInboxItem,
+  ResolveExistingMailIds,
+} from "@corbits/notify/mailbox";
 
 import { classificationFromRefs } from "./group";
 
@@ -74,3 +81,58 @@ export function createWorkbenchMailboxDelivery(
     return deliverInboxItems(db, stamped, writeOpts);
   };
 }
+
+/**
+ * CL-7238 read-back for `@corbits/notify`'s crash window: a redelivery
+ * reports an already-committed row as pre-existing (`id: null`), so the
+ * host resolves it to its row id here and the missing dispatch rows get
+ * repaired instead of lost. Served from the mailbox unique mail key —
+ * the same `mailboxKey.inbox(source, externalId)` the write dedupes on —
+ * so the lookup can never disagree with the delivery about which row an
+ * item belongs to. Positional in, positional out, `null` where no row
+ * exists; the dispatch store dedupes by (mail row, sink), so a benign
+ * redelivery's repair never double-queues.
+ */
+export function createResolveExistingMailIds(
+  db: MailboxDb,
+): ResolveExistingMailIds {
+  return async (items) => {
+    if (items.length === 0) return [];
+    const keyed = items.map((item) => ({
+      tenantId: item.tenantId,
+      principalId: item.principalId,
+      messageKey: mailboxKey.inbox(item.source, item.externalId),
+    }));
+    const rows = await db
+      .select({
+        id: principalMail.id,
+        tenantId: principalMail.tenantId,
+        principalId: principalMail.principalId,
+        messageKey: principalMail.messageKey,
+      })
+      .from(principalMail)
+      .where(
+        or(
+          ...keyed.map((key) =>
+            and(
+              eq(principalMail.tenantId, key.tenantId),
+              eq(principalMail.principalId, key.principalId),
+              eq(principalMail.messageKey, key.messageKey),
+            ),
+          ),
+        ),
+      );
+    const idsByKey = new Map(
+      rows.map((row) => [
+        `${row.tenantId}\n${row.principalId}\n${row.messageKey}`,
+        row.id,
+      ]),
+    );
+    return keyed.map(
+      (key) =>
+        idsByKey.get(
+          `${key.tenantId}\n${key.principalId}\n${key.messageKey}`,
+        ) ?? null,
+    );
+  };
+}

packages/inbox/src/index.ts

@@ -24,6 +24,7 @@ export {
   type WorkbenchInboxStatus,
 } from "./vocabulary";
 export {
+  createResolveExistingMailIds,
   createWorkbenchMailboxDelivery,
   type CreateWorkbenchMailboxDeliveryOpts,
 } from "./delivery";

packages/inbox/test/delivery.test.ts

@@ -20,12 +20,18 @@ import {
   deliverCredentialMail,
   deliverMentionMail,
   deliverRunFailureMail,
+  type ApprovalNotification,
   type NotifyDeliveryDeps,
+  type NotifyDispatchStore,
+  type SinkDeliveryResult,
 } from "@corbits/notify";
 
 import { setupDatabase } from "../../../scripts/db-setup";
 import { e2eDatabaseUrl } from "../../../scripts/e2e/database-url";
-import { createWorkbenchMailboxDelivery } from "../src/delivery";
+import {
+  createResolveExistingMailIds,
+  createWorkbenchMailboxDelivery,
+} from "../src/delivery";
 import { inboxGroupOf } from "../src/group";
 import { WORKBENCH_INBOX_PRIORITIES } from "../src/vocabulary";
 import { dbGate } from "../../../scripts/e2e/db-gate";
@@ -213,5 +219,91 @@ describeIfDb(
         await mailboxDb.close();
       }
     }, 15000);
+
+    test("a crash between mail write and dispatch enqueue is repaired on redelivery without doubles (CL-7238)", async () => {
+      const mailboxDb = createMailboxDb(scratchUrl);
+      try {
+        const innerDispatch = createInMemoryNotifyDispatchStore();
+        let crashNext = true;
+        const dispatch: NotifyDispatchStore = {
+          ...innerDispatch,
+          enqueue: async (inputs) => {
+            if (crashNext) {
+              crashNext = false;
+              throw new Error(
+                "boom: crash between the mail write and the dispatch enqueue",
+              );
+            }
+            await innerDispatch.enqueue(inputs);
+          },
+        };
+        const delivered: SinkDeliveryResult = { status: "delivered" };
+        const sinks = createSinkRegistry();
+        sinks.register({
+          name: "always",
+          isEnabledFor: async () => true,
+          deliver: async () => delivered,
+        });
+        const deps: NotifyDeliveryDeps = {
+          mail: createWorkbenchMailboxDelivery({ db: mailboxDb.db }),
+          addressing: {
+            inbox: (recipient) => `${recipient.principalId}@inbox.test`,
+            from: (kind) => `${kind}@notify.test`,
+          },
+          dispatch,
+          resolveExistingMailIds: createResolveExistingMailIds(mailboxDb.db),
+          sinks,
+        };
+        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, event)).rejects.toThrow("boom");
+
+        // The redelivery dedupes on the mail key (no new row), resolves the
+        // committed row's id through the adapter read-back, and queues the
+        // dispatch the crash swallowed.
+        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
+        // both redeliveries 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);
   },
 );

packages/notify/src/deliver.ts

@@ -7,12 +7,13 @@
 // dispatch row per (mail row, enabled sink). A sink is never called from this
 // path: the mail is the durable record, and a copy of it is the worker's job.
 //
-// That queuing step is NOT atomic with the mail write, and nothing
-// reconciles the two if a crash or a throw lands between them (CL-7238):
-// a mail row can end up committed with no dispatch row ever queued for it.
-// Closing that gap needs a seam `@corbits/mailbox` doesn't expose today
-// (an in-transaction hook, or a way to read back an existing row's id by
-// key) — see CL-7238 for why this can't be fixed from this package alone.
+// That queuing step is NOT atomic with the mail write: a crash or a
+// throw between them leaves a mail row committed with no dispatch row
+// queued (CL-7238). The mail delivery reports such rows with a `null`
+// id, so a redelivery resolves their ids through the host's
+// `resolveExistingMailIds` read-back and repairs the missing dispatch
+// rows; the dispatch store dedupes by (mail row, sink), so the repair
+// never double-queues.
 import {
   parseNotificationEvent,
   type ApprovalNotification,
@@ -25,6 +26,7 @@ import type {
   MailboxDelivery,
   NotifyAddressing,
   NotifyInboxItem,
+  ResolveExistingMailIds,
 } from "./mailbox";
 import { notificationExternalId, renderNotification } from "./render";
 import type { SinkRegistry } from "./sinks";
@@ -38,6 +40,15 @@ export interface NotifyDeliveryDeps {
   readonly addressing: NotifyAddressing;
   readonly dispatch: NotifyDispatchStore;
   readonly sinks: SinkRegistry;
+  /**
+   * Read-back for the CL-7238 crash window: when the mail delivery
+   * reports an item as pre-existing (`id: null`), the row was committed
+   * by an earlier attempt that died before its dispatch enqueue.
+   * Resolving those items to their row ids lets this call repair the
+   * missing dispatch rows; without it a redelivery after such a crash
+   * still loses the dispatch. The host owns this wiring.
+   */
+  readonly resolveExistingMailIds?: ResolveExistingMailIds;
 }
 
 export interface NotifyDeliveryReport {
@@ -75,8 +86,9 @@ export async function deliverNotification(
   input: unknown,
 ): Promise<NotifyDeliveryReport> {
   const event = parseNotificationEvent(input);
+  const items = toInboxItems(event, deps.addressing);
   const written: { id: string; tenantId: string; principalId: string }[] = [];
-  await deps.mail(toInboxItems(event, deps.addressing), {
+  const results = await deps.mail(items, {
     enqueue: ({ id, item }) => {
       written.push({
         id,
@@ -86,8 +98,30 @@ export async function deliverNotification(
     },
   });
 
+  // CL-7238: a `null` id marks a row committed by an earlier attempt
+  // that died before its dispatch enqueue (or a benign redelivery).
+  // Resolve those ids so the enqueue below repairs the missing rows;
+  // without the read-back there is nothing to enqueue for them.
+  const deduped = items.filter(
+    (_, index) => (results[index]?.id ?? null) === null,
+  );
+  const repaired: { id: string; tenantId: string; principalId: string }[] = [];
+  if (deduped.length > 0 && deps.resolveExistingMailIds !== undefined) {
+    const resolved = await deps.resolveExistingMailIds(deduped);
+    deduped.forEach((item, index) => {
+      const id = resolved[index];
+      if (id !== undefined && id !== null) {
+        repaired.push({
+          id,
+          tenantId: item.tenantId,
+          principalId: item.principalId,
+        });
+      }
+    });
+  }
+
   const queued: EnqueueDispatchInput[] = [];
-  for (const row of written) {
+  for (const row of [...written, ...repaired]) {
     const enabled = await deps.sinks.listEnabledFor({
       tenantId: row.tenantId,
       principalId: row.principalId,

packages/notify/src/index.ts

@@ -84,4 +84,5 @@ export type {
   NotifyDeliverOpts,
   NotifyInboxItem,
   NotifyMailRef,
+  ResolveExistingMailIds,
 } from "./mailbox";

packages/notify/src/mailbox.ts

@@ -37,11 +37,24 @@ export type NotifyDeliverOpts = {
   readonly enqueue?: (delivered: { id: string; item: NotifyInboxItem }) => void;
 };
 
+/** Results align positionally with `items`, mirroring the vendor delivery. */
 export type MailboxDelivery = (
   items: NotifyInboxItem[],
   opts?: NotifyDeliverOpts,
 ) => Promise<NotifyDeliveredItem[]>;
 
+/**
+ * Read-back for mail rows an earlier delivery committed before dying
+ * ahead of its dispatch enqueue (CL-7238). The input lists the items
+ * the mail delivery just reported as pre-existing (`id: null`); the
+ * result aligns positionally with the input, with `null` where no row
+ * exists. The host serves this from the mailbox's unique mail key — a
+ * lookup the seam doesn't expose today, so the host owns the wiring.
+ */
+export type ResolveExistingMailIds = (
+  items: readonly NotifyInboxItem[],
+) => Promise<readonly (string | null)[]>;
+
 /**
  * How a notification addresses mail. The host owns its own address space —
  * a mailbox address is a deployment fact, not a product rule — so both halves