diff --git a/packages/inbox/src/delivery.ts b/packages/inbox/src/delivery.ts index a43290cff6..8141c14502 100644 --- a/packages/inbox/src/delivery.ts +++ b/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, + ); + }; +} diff --git a/packages/inbox/src/index.ts b/packages/inbox/src/index.ts index 75816bc8ef..976ff600ef 100644 --- a/packages/inbox/src/index.ts +++ b/packages/inbox/src/index.ts @@ -24,6 +24,7 @@ export { type WorkbenchInboxStatus, } from "./vocabulary"; export { + createResolveExistingMailIds, createWorkbenchMailboxDelivery, type CreateWorkbenchMailboxDeliveryOpts, } from "./delivery"; diff --git a/packages/inbox/test/delivery.test.ts b/packages/inbox/test/delivery.test.ts index 54522447de..1d32ced1d7 100644 --- a/packages/inbox/test/delivery.test.ts +++ b/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); }, );