diff --git a/ARCHITECTURE.md b/ARCHITECTURE.md index 2b41d04..265f584 100644 --- a/ARCHITECTURE.md +++ b/ARCHITECTURE.md @@ -58,15 +58,16 @@ tables foreign-key to both. Nothing changed in Interchange to make that work — the coupling lives entirely on this side. One further seam lives outside `mountMailbox`, on the write side: -`createMailboxPersist(db, { upstream, authorizeSender, bus?, onRow? })` wraps a -host's own mail-persist function so every addressed principal also gets a -durable row. `authorizeSender(address) => { tenantId, domain } | null` is the -host's decision — whether a sender address belongs to a *live* agent instance -is not a schema fact. Returning `null` skips the mailbox write entirely while -the frame still goes upstream. On the recipient side the package does consult -the control plane: an address whose local part matches no known principal in -the authorized tenant is skipped with a warning rather than minting a phantom -mailbox row, and never costs the frame's real recipients their durable copy. +`createMailboxPersist(db, { upstream, authorizeSender, bus?, onRow?, resolveRefs? })` +wraps a host's own mail-persist function so every addressed principal also +gets a durable row. `authorizeSender(address) => { tenantId, domain } | null` +is the host's decision — whether a sender address belongs to a *live* agent +instance is not a schema fact. Returning `null` skips the mailbox write +entirely while the frame still goes upstream. On the recipient side the +package does consult the control plane: an address whose local part matches +no known principal in the authorized tenant is skipped with a warning rather +than minting a phantom mailbox row, and never costs the frame's real +recipients their durable copy. The wrapper's contract is **dual-write independence in both directions**: an `upstream` throw still attempts the mailbox write and then re-throws the @@ -75,6 +76,31 @@ persist that upstream already completed. Under retry, the mailbox side is idempotent: package-owned transport `messageKey`s plus `onConflictDoNothing` collapse duplicate frames without failing the call or re-announcing. +`resolveRefs(args)` — `args` is `MailboxPersistArgs` plus the resolved +`senderAuthorization` and the `decoded` frame (or `null` if the parser +rejected it) — is called ONCE per frame, before the transaction opens, not +once per recipient: a host pointing every row at the same upstream entity +does one lookup, not N. It runs AFTER `upstream` resolves and serially with +it, so its latency adds to the call rather than overlapping. Its result is +validated with `MailboxRefArraySchema` and capped at `MAX_MAILBOX_REFS` the +same way `writeMailboxMessage`'s `refs` argument is (excess entries +truncated from the end of the list, logged with `messageId` and +`senderAddress`, never a throw) — so a resolver must return a small set with +the load-bearing ref FIRST, since anything past the cap is silently dropped. +Refs are then stored on every recipient row of that frame INSIDE the same +transaction — so the post-commit bus `create` event and any SSE subscriber +already see refs on the row once it's readable. + +Refs are frozen at the FIRST successful insert for a given frame: a retry +(same idempotency key) still calls `resolveRefs` — it is not skipped — but +because `onConflictDoNothing` writes no row on a retry, a different result +from that second call is simply discarded; only the first call's refs ever +land. A throwing `resolveRefs` is handled exactly like any other +pre-transaction failure: it falls under the same dual-write contract as the +rest of the wrapper — logged (naming `resolveRefs` as the failing stage), +upstream unaffected (already ran, or still will, independently of this), and +no mailbox row for that frame. + `resolvePrincipal`'s signature is identical across the Corbits cores, so a host mounting more than one passes the same function to each. diff --git a/CHANGELOG.md b/CHANGELOG.md index 24dd77c..77a70fc 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -11,6 +11,15 @@ always called out under their own heading. ### Added +- **`createMailboxPersist` accepts `resolveRefs` at insert time.** The new + `resolveRefs(args)` option — `args` is the persist args plus the resolved + `senderAuthorization` and the `decoded` frame (or `null`) — is called once + per frame, before the transaction opens, and its result (validated with + `MailboxRefArraySchema`, capped at `MAX_MAILBOX_REFS`) is stored on every + recipient row inside that same transaction, so the post-commit bus event + already carries refs. A throwing `resolveRefs` follows the existing + dual-write contract: logged, upstream unaffected, no mailbox row for that + frame. - **Threading headers on the frame and in the list projection.** `buildMailFrame` accepts `references` — the thread's ancestry, oldest first — and emits it as a folded `References:` header; `In-Reply-To` diff --git a/README.md b/README.md index 357ddc4..392222d 100644 --- a/README.md +++ b/README.md @@ -10,6 +10,30 @@ Requires `@intx` 0.2.2 or newer. See [ARCHITECTURE.md](./ARCHITECTURE.md) for the data model. +## Dual-write persist + +```ts +import { createMailboxPersist } from "@corbits/mailbox"; + +const persist = createMailboxPersist(db, { + upstream: hostMailTransport.persist, + authorizeSender: (address) => resolveActiveInstance(address), + // Called once per frame, before the transaction — every recipient row + // gets the same refs, so a bus subscriber sees them on the `create` event. + resolveRefs: ({ decoded }) => + decoded ? [{ kind: "workbench", id: decoded.messageId ?? "" }] : undefined, +}); +``` + +`resolveRefs` runs after `upstream` resolves and serially with it, and its +refs are frozen at the frame's first successful insert — a retry still runs +the resolver but a different result on that later call is discarded. Return +a small set with the load-bearing ref first: the list is capped at +`MAX_MAILBOX_REFS` by truncating from the end. + +See ARCHITECTURE.md's persist section for the full contract, including +`resolveRefs`'s dual-write-failure semantics. + ## Install ```sh diff --git a/src/persist.test.ts b/src/persist.test.ts index fdf5455..a5ef006 100644 --- a/src/persist.test.ts +++ b/src/persist.test.ts @@ -9,6 +9,7 @@ import { type MailboxPersistArgs, type SenderAuthorization, type PersistedMailboxRow, + type ResolveMailboxRefs, } from "./persist.js"; import { createInMemoryMailboxEventBus } from "./bus.js"; import { buildMailFrame } from "./frame.js"; @@ -17,8 +18,10 @@ import { withTestDb, seedScope } from "./test-helpers.js"; import type { MailboxDb } from "./db.js"; import { MAX_MAILBOX_FRAME_BYTES, + MAX_MAILBOX_REFS, assertMailboxFrameBytes, } from "./write.js"; +import { getMailboxMessage, type MailboxRef } from "./read.js"; let db: MailboxDb; @@ -520,6 +523,223 @@ describe("transport insert idempotency", () => { }); }); +describe("resolveRefs", () => { + test("refs are visible to a bus subscriber on the create event", async () => { + const bus = createInMemoryMailboxEventBus(); + const seenIds: string[] = []; + bus.subscribe({ tenantId: "acme", principalId: "user-1" }, (e) => + seenIds.push(e.id), + ); + const refs: MailboxRef[] = [{ kind: "workbench", id: "thread-1" }]; + const { upstream } = recordingUpstream(); + const persist = createMailboxPersist(db, { + upstream, + authorizeSender: () => ACTIVE, + bus, + resolveRefs: () => refs, + }); + + await persist(args()); + + // By the time the bus fires, the insert has already committed — a + // subscriber reading the row by the announced id sees refs already + // stamped, not a later-arriving update. + expect(seenIds).toHaveLength(1); + const message = await getMailboxMessage(db, { + tenantId: "acme", + principalId: "user-1", + id: seenIds[0]!, + }); + expect(message?.refs).toEqual(refs); + }); + + test("resolveRefs is called once for three recipients, and every row gets its refs", async () => { + await seedScope(db, "acme", "user-3"); + const refs: MailboxRef[] = [{ kind: "workbench", id: "thread-1" }]; + const calls: unknown[] = []; + const resolveRefs: ResolveMailboxRefs = (a) => { + calls.push(a); + return refs; + }; + const { upstream } = recordingUpstream(); + const persist = createMailboxPersist(db, { + upstream, + authorizeSender: () => ACTIVE, + resolveRefs, + }); + + await persist( + args({ + recipients: [ + "usr_user-1@acme.example", + "usr_user-2@acme.example", + "usr_user-3@acme.example", + ], + }), + ); + + expect(calls).toHaveLength(1); + for (const principalId of ["user-1", "user-2", "user-3"]) { + const [row] = await rowsFor("acme", principalId); + const message = await getMailboxMessage(db, { + tenantId: "acme", + principalId, + id: row!.id, + }); + expect(message?.refs).toEqual(refs); + } + }); + + test("a throwing resolveRefs writes zero rows; upstream still completes", async () => { + const { upstream, result } = recordingUpstream(); + const persist = createMailboxPersist(db, { + upstream, + authorizeSender: () => ACTIVE, + resolveRefs: () => { + throw new Error("resolver exploded"); + }, + }); + + expect(await persist(args())).toBe(result); + expect(await rowsFor("acme", "user-1")).toHaveLength(0); + }); + + test("over-cap refs from resolveRefs are truncated to MAX_MAILBOX_REFS", async () => { + const refs: MailboxRef[] = Array.from({ length: MAX_MAILBOX_REFS + 5 }, (_, i) => ({ + kind: "workbench", + id: `thread-${i}`, + })); + const { upstream } = recordingUpstream(); + const persist = createMailboxPersist(db, { + upstream, + authorizeSender: () => ACTIVE, + resolveRefs: () => refs, + }); + + await persist(args()); + + const [row] = await rowsFor("acme", "user-1"); + const message = await getMailboxMessage(db, { + tenantId: "acme", + principalId: "user-1", + id: row!.id, + }); + expect(message?.refs?.length).toBe(MAX_MAILBOX_REFS); + }); + + test("retried frame with a different resolver result keeps the FIRST refs", async () => { + let call = 0; + const { upstream } = recordingUpstream(); + const persist = createMailboxPersist(db, { + upstream, + authorizeSender: () => ACTIVE, + resolveRefs: () => [{ kind: "workbench", id: `attempt-${++call}` }], + }); + + await persist(args()); + await persist(args()); + + const all = await rowsFor("acme", "user-1"); + expect(all).toHaveLength(1); + expect(call).toBe(2); + const message = await getMailboxMessage(db, { + tenantId: "acme", + principalId: "user-1", + id: all[0]!.id, + }); + expect(message?.refs).toEqual([{ kind: "workbench", id: "attempt-1" }]); + }); + + test("resolver returning undefined stores SQL NULL, not []", async () => { + const { upstream } = recordingUpstream(); + const persist = createMailboxPersist(db, { + upstream, + authorizeSender: () => ACTIVE, + resolveRefs: () => undefined, + }); + + await persist(args()); + + const [row] = await rowsFor("acme", "user-1"); + expect(row!.refs).toBeNull(); + }); + + test("resolver returning [] stores SQL NULL, not []", async () => { + const { upstream } = recordingUpstream(); + const persist = createMailboxPersist(db, { + upstream, + authorizeSender: () => ACTIVE, + resolveRefs: () => [], + }); + + await persist(args()); + + const [row] = await rowsFor("acme", "user-1"); + expect(row!.refs).toBeNull(); + }); + + test("schema-invalid resolver output writes zero rows; upstream result still returned", async () => { + const { upstream, result } = recordingUpstream(); + const persist = createMailboxPersist(db, { + upstream, + authorizeSender: () => ACTIVE, + resolveRefs: () => [{ kind: 42, id: "x" }] as unknown as MailboxRef[], + }); + + expect(await persist(args())).toBe(result); + expect(await rowsFor("acme", "user-1")).toHaveLength(0); + }); + + test("resolver runs after upstream resolves (serial), adding its latency to the call", async () => { + const order: string[] = []; + const persist = createMailboxPersist(db, { + upstream: async () => { + await Bun.sleep(150); + order.push("upstream"); + return [{ delivered: true }]; + }, + authorizeSender: () => ACTIVE, + resolveRefs: async () => { + order.push("resolver-start"); + await Bun.sleep(150); + order.push("resolver-end"); + return undefined; + }, + }); + + const t0 = performance.now(); + await persist(args()); + const elapsed = performance.now() - t0; + + expect(order).toEqual(["upstream", "resolver-start", "resolver-end"]); + expect(elapsed).toBeGreaterThanOrEqual(290); + }); + + test("the 21st ref (a workbench ref appended last) is silently dropped", async () => { + const refs: MailboxRef[] = Array.from({ length: MAX_MAILBOX_REFS }, (_, i) => ({ + kind: "thread", + id: `t-${i}`, + })); + refs.push({ kind: "workbench", id: "must-be-present" }); + const { upstream } = recordingUpstream(); + const persist = createMailboxPersist(db, { + upstream, + authorizeSender: () => ACTIVE, + resolveRefs: () => refs, + }); + + await persist(args()); + + const [row] = await rowsFor("acme", "user-1"); + const message = await getMailboxMessage(db, { + tenantId: "acme", + principalId: "user-1", + id: row!.id, + }); + expect(message?.refs?.some((r) => r.kind === "workbench")).toBe(false); + }); +}); + describe("frame size and recipient hard caps", () => { test("assertMailboxFrameBytes accepts at-cap and throws RangeError one byte over", () => { // Dual-write swallows the RangeError inside attemptMailboxWrite; the pure diff --git a/src/persist.ts b/src/persist.ts index 7a85395..12f172c 100644 --- a/src/persist.ts +++ b/src/persist.ts @@ -17,19 +17,37 @@ import { createHash } from "node:crypto"; import { and, eq, inArray, sql } from "drizzle-orm"; +import { type } from "arktype"; import { getLogger } from "@intx/log"; import { hostPrincipal, mailbox, principalMail } from "./schema.js"; import type { MailboxDb } from "./db.js"; import { publishMailboxEvent, type MailboxEventBus } from "./bus.js"; -import { decodeMailFrame, parseMsgIdList } from "./frame.js"; +import { decodeMailFrame, parseMsgIdList, type DecodedFrame } from "./frame.js"; import { resolveMailboxRecipients } from "./recipients.js"; +import { MailboxRefArraySchema, type MailboxRef } from "./read.js"; import { assertMailboxScope, assertMailboxFrameBytes, + boundRefs, } from "./write.js"; const logger = getLogger(["corbits-mailbox", "persist"]); +/** + * Tags a thrown error with the persist stage that produced it, so the + * dual-write failure log can name `resolveRefs` specifically instead of a + * generic "mailbox write failed". The wrapped error is what's logged and + * (never here) rethrown — see `attemptMailboxWrite`. + */ +class MailboxPersistStageError extends Error { + readonly stage: string; + constructor(stage: string, cause: unknown) { + super(cause instanceof Error ? cause.message : String(cause), { cause }); + this.name = "MailboxPersistStageError"; + this.stage = stage; + } +} + // Cap the sender-controlled recipient list before resolve / inArray / multi-row // insert. Matches MAX_BULK_MAILBOX_IDS posture: hard refuse, never clamp. export const MAX_MAILBOX_RECIPIENTS = 50; @@ -91,6 +109,43 @@ export type PersistedMailboxRow = { senderAddress: string; }; +/** + * Host seam for stamping every recipient row of one frame with the same + * `refs`. Called once per frame, before the transaction opens — NOT once per + * recipient — so a host pointing every row at the same upstream entity + * (`{ kind: "workbench", id }`) does one lookup, not N. + * + * Runs AFTER `upstream` resolves, and serially with it — not concurrently — + * so its latency adds to the call. This keeps refs available before the + * mailbox transaction opens without racing `upstream`'s own effects. + * + * Refs are frozen at the FIRST successful insert for a frame: a retried + * frame (same idempotency key) that reaches `resolveRefs` again still runs + * the resolver — it is not skipped — but a different result is discarded, + * since `onConflictDoNothing` means no row is written for the retry. Do not + * rely on a resolver's return value being applied on any call after the + * first that actually inserts. + * + * The result is validated with `MailboxRefArraySchema` and capped at + * `MAX_MAILBOX_REFS` the same way `writeMailboxMessage`'s `refs` argument is; + * see `boundRefs`. Returning `undefined` (or an empty array) stores no refs. + * Because excess entries are truncated rather than rejected, a resolver + * MUST return a small set with the load-bearing ref FIRST — anything past + * `MAX_MAILBOX_REFS` is silently dropped from the end of the list. + * + * A throwing `resolveRefs` is handled exactly like a mailbox-write failure + * under the dual-write contract: logged (naming `resolveRefs` as the failing + * stage), upstream still runs (it already ran, or still will, independently + * of this), and no mailbox row is written for that frame. See + * ARCHITECTURE.md's persist section. + */ +export type ResolveMailboxRefs = ( + args: MailboxPersistArgs & { + senderAuthorization: SenderAuthorization; + decoded: DecodedFrame | null; + }, +) => Promise | MailboxRef[] | undefined; + export type CreateMailboxPersistOpts = { /** The host's own persist path. Always called, for every frame. */ upstream: (args: MailboxPersistArgs) => Promise; @@ -99,6 +154,12 @@ export type CreateMailboxPersistOpts = { bus?: MailboxEventBus; /** Best-effort hook per inserted row; a throw is logged, never propagated. */ onRow?: (row: PersistedMailboxRow) => void; + /** + * Resolve the `refs` every recipient row of one frame gets, INSIDE the + * existing single transaction — so the post-commit bus event and any SSE + * subscriber already see them. See `ResolveMailboxRefs`. + */ + resolveRefs?: ResolveMailboxRefs; }; /** @@ -232,6 +293,35 @@ export function createMailboxPersist( // an upgrade's backfill would have produced for the same frame. const inReplyTo = parseMsgIdList(decoded?.headers.get("in-reply-to"))[0] ?? null; + // Resolved ONCE per frame, before the transaction — every recipient row + // gets the same refs, and a resolver that hits an upstream entity does one + // lookup regardless of recipient count. A throw here propagates out of + // `writeMailboxRows` exactly like any other pre-transaction failure: + // `attemptMailboxWrite` catches and logs it, upstream still stands, and no + // mailbox row is written for this frame. + let refs: MailboxRef[] | undefined; + if (opts.resolveRefs) { + let resolvedRefs: MailboxRef[] | undefined; + try { + resolvedRefs = await opts.resolveRefs({ + senderAddress, + recipients, + raw, + senderAuthorization: auth, + decoded, + }); + } catch (err) { + throw new MailboxPersistStageError("resolveRefs", err); + } + if (resolvedRefs !== undefined && resolvedRefs.length > 0) { + const validated = MailboxRefArraySchema(resolvedRefs); + if (validated instanceof type.errors) { + throw new RangeError(`invalid mailbox refs: ${validated.summary}`); + } + refs = boundRefs(validated, messageId, { senderAddress }); + } + } + // Mail rows and their management rows commit together: the management row // is created eagerly with the message (see `writeMailboxMessage`), and a // message without one is unreachable by every mutation. messageKey makes @@ -252,6 +342,7 @@ export function createMailboxPersist( fromAddress, messageId, inReplyTo, + refs: refs ?? null, messageKey: transportMessageKey( messageId, raw, @@ -305,8 +396,14 @@ export function createMailboxPersist( try { await writeMailboxRows(args); } catch (err) { + // Decoded independently of `writeMailboxRows`'s own decode: the throw + // may have happened before that decode ran (e.g. authorizeSender), and + // this log line must still correlate to a messageId when one exists. + const messageId = decodeMailFrame(args.raw)?.messageId ?? null; logger.error("mailbox write failed for mail from {senderAddress}", { senderAddress: args.senderAddress, + messageId, + ...(err instanceof MailboxPersistStageError ? { stage: err.stage } : {}), error: err instanceof Error ? err : new Error(String(err)), }); } diff --git a/src/write.ts b/src/write.ts index f6c2a18..3e092fd 100644 --- a/src/write.ts +++ b/src/write.ts @@ -92,9 +92,11 @@ export function assertMailboxFrameBytes(raw: Uint8Array): void { // `messageKey` is the caller's own identifier and is absent for externally // delivered mail, which is never deduped — the warning below carries whatever // the caller actually supplied rather than minting an id nobody can correlate. -function boundRefs( +export function boundRefs( refs: MailboxRef[] | undefined, messageKey: string | null, + /** Extra correlation fields merged into the truncation log line, e.g. `senderAddress` on the persist path. */ + extra?: Record, ): MailboxRef[] | undefined { if (refs === undefined || refs.length === 0) return undefined; if (refs.length <= MAX_MAILBOX_REFS) return refs; @@ -102,6 +104,7 @@ function boundRefs( messageKey, received: refs.length, kept: MAX_MAILBOX_REFS, + ...extra, }); return refs.slice(0, MAX_MAILBOX_REFS); }