From d0fd954a276047a3ed609a70220304d35544adca Mon Sep 17 00:00:00 2001 From: Sawyer Date: Fri, 4 Sep 2026 18:20:31 -0700 Subject: [PATCH 1/5] Add tests for caller-supplied Message-ID/direction/messageKey and batch write Covers WriteMailboxMessageArgs.messageId/direction, the messageKey default (and override), and the new one-transaction writeMailboxMessages batch: atomic commit, mid-batch rollback, dedupe-on-retry, and post-commit bus events. --- src/write.test.ts | 349 ++++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 349 insertions(+) diff --git a/src/write.test.ts b/src/write.test.ts index 6ddb89a..e2e14fa 100644 --- a/src/write.test.ts +++ b/src/write.test.ts @@ -1,11 +1,13 @@ import { beforeEach, describe, expect, test } from "bun:test"; import { writeMailboxMessage, + writeMailboxMessages, deliverInboxItems, mailboxKey, MAX_MAILBOX_REFS, MAX_MAILBOX_FRAME_BYTES, } from "./write.js"; +import { decodeMailFrame } from "./frame.js"; import { getMailboxMessage, listUserMailbox } from "./read.js"; import { createInMemoryMailboxEventBus } from "./bus.js"; import { withTestDb, seedScope, TEST_VOCABULARY } from "./test-helpers.js"; @@ -827,3 +829,350 @@ describe("frame size hard cap", () => { }); }); +describe("caller-supplied messageId, direction, and messageKey", () => { + async function rawFrame(id: string): Promise { + const rows = await db.execute<{ raw: Uint8Array }>( + sql`SELECT raw FROM "mailbox"."principal_mail" WHERE id = ${id}`, + ); + return rows[0]!.raw; + } + + async function rowColumns(id: string): Promise<{ + direction: string; + message_id: string | null; + message_key: string | null; + }> { + const rows = await db.execute<{ + direction: string; + message_id: string | null; + message_key: string | null; + }>( + sql`SELECT direction, message_id, message_key FROM "mailbox"."principal_mail" WHERE id = ${id}`, + ); + return rows[0]!; + } + + test("a caller-supplied Message-ID round-trips to the stored frame header and the cached column", async () => { + const messageId = ""; + const written = await writeMailboxMessage(db, { + tenantId: "t1", + principalId: "p1", + address: "p1@t1.example", + fromAddress: "agent@t1.example", + subject: "Hello", + body: "World", + messageId, + }); + expect(written).not.toBeNull(); + + const raw = await rawFrame(written!.id); + const decoded = decodeMailFrame(raw); + expect(decoded?.messageId).toBe(messageId); + + const columns = await rowColumns(written!.id); + expect(columns.message_id).toBe(messageId); + }); + + test("an invalid caller-supplied messageId is refused with RangeError", async () => { + await expect( + writeMailboxMessage(db, { + tenantId: "t1", + principalId: "p1", + address: "p1@t1.example", + fromAddress: "agent@t1.example", + subject: "Hello", + body: "World", + messageId: "not-a-msg-id", + }), + ).rejects.toThrow(RangeError); + }); + + test("omitting messageId still mints one, as before", async () => { + const written = await writeMailboxMessage(db, { + tenantId: "t1", + principalId: "p1", + address: "p1@t1.example", + fromAddress: "agent@t1.example", + subject: "Hello", + body: "World", + }); + const columns = await rowColumns(written!.id); + expect(columns.message_id).toMatch(/^<.+@.+>$/); + }); + + test("direction defaults to inbound and can be set to outbound", async () => { + const inbound = await writeMailboxMessage(db, { + tenantId: "t1", + principalId: "p1", + address: "p1@t1.example", + fromAddress: "agent@t1.example", + subject: "Hello", + body: "World", + }); + expect((await rowColumns(inbound!.id)).direction).toBe("inbound"); + + const outbound = await writeMailboxMessage(db, { + tenantId: "t1", + principalId: "p1", + address: "p1@t1.example", + fromAddress: "agent@t1.example", + subject: "Sent", + body: "World", + direction: "outbound", + }); + expect((await rowColumns(outbound!.id)).direction).toBe("outbound"); + }); + + test("an explicit messageKey is honored over the default transport key", async () => { + const written = await writeMailboxMessage(db, { + tenantId: "t1", + principalId: "p1", + address: "p1@t1.example", + fromAddress: "agent@t1.example", + subject: "Hello", + body: "World", + messageKey: "custom:my-key", + }); + expect((await rowColumns(written!.id)).message_key).toBe("custom:my-key"); + }); + + test("omitting messageKey defaults to the package's transport key, keyed off messageId", async () => { + const messageId = ""; + const written = await writeMailboxMessage(db, { + tenantId: "t1", + principalId: "p1", + address: "p1@t1.example", + fromAddress: "agent@t1.example", + subject: "Hello", + body: "World", + messageId, + }); + expect((await rowColumns(written!.id)).message_key).toBe( + mailboxKey.transport(messageId, "p1"), + ); + + // A retry with the SAME caller-supplied messageId therefore dedupes. + const retry = await writeMailboxMessage(db, { + tenantId: "t1", + principalId: "p1", + address: "p1@t1.example", + fromAddress: "agent@t1.example", + subject: "Hello", + body: "World", + messageId, + }); + expect(retry).toBeNull(); + }); +}); + +describe("writeMailboxMessages (one-transaction batch write)", () => { + async function mailRows(): Promise< + Array<{ id: string; principal_id: string; direction: string }> + > { + return db.execute<{ id: string; principal_id: string; direction: string }>( + sql`SELECT id, principal_id, direction FROM "mailbox"."principal_mail"`, + ); + } + + test("a batch of three — one outbound for the sender, two inbound for recipients — commits atomically", async () => { + const bus = createInMemoryMailboxEventBus(); + const receivedP1: Array<{ id: string; op?: string }> = []; + const receivedP2: Array<{ id: string; op?: string }> = []; + bus.subscribe({ tenantId: "t1", principalId: "p1" }, (e) => + receivedP1.push(e), + ); + bus.subscribe({ tenantId: "t1", principalId: "p2" }, (e) => + receivedP2.push(e), + ); + + const ids = await writeMailboxMessages( + db, + [ + { + scope: { tenantId: "t1", principalId: "p1" }, + args: { + tenantId: "t1", + principalId: "p1", + address: "p1@t1.example", + fromAddress: "p1@t1.example", + subject: "Sent", + body: "Hi p2", + direction: "outbound", + }, + }, + { + scope: { tenantId: "t1", principalId: "p1" }, + args: { + tenantId: "t1", + principalId: "p1", + address: "p1@t1.example", + fromAddress: "p1@t1.example", + subject: "Recv", + body: "Hi p1", + direction: "inbound", + }, + }, + { + scope: { tenantId: "t1", principalId: "p2" }, + args: { + tenantId: "t1", + principalId: "p2", + address: "p2@t1.example", + fromAddress: "p1@t1.example", + subject: "Recv", + body: "Hi p2", + direction: "inbound", + }, + }, + ], + { bus }, + ); + + expect(ids.length).toBe(3); + const rows = await mailRows(); + expect(rows.length).toBe(3); + // p1 receives two events (its outbound sent-copy and its inbound copy), + // p2 receives one (its inbound copy) — one bus event per written row. + expect(receivedP1.length).toBe(2); + expect(receivedP2.length).toBe(1); + expect(receivedP1.every((e) => e.op === "create")).toBe(true); + expect(receivedP2[0]?.op).toBe("create"); + }); + + test("a failing third item rolls back the whole batch, leaving zero rows", async () => { + await expect( + writeMailboxMessages(db, [ + { + scope: { tenantId: "t1", principalId: "p1" }, + args: { + tenantId: "t1", + principalId: "p1", + address: "p1@t1.example", + fromAddress: "p1@t1.example", + subject: "Sent", + body: "Hi p2", + direction: "outbound", + }, + }, + { + scope: { tenantId: "t1", principalId: "p2" }, + args: { + tenantId: "t1", + principalId: "p2", + address: "p2@t1.example", + fromAddress: "p1@t1.example", + subject: "Recv", + body: "Hi p2", + direction: "inbound", + }, + }, + { + scope: { tenantId: "t1", principalId: "nobody-seeded-this" }, + args: { + tenantId: "t1", + principalId: "nobody-seeded-this", + address: "ghost@t1.example", + fromAddress: "p1@t1.example", + subject: "Bad", + body: "Bad", + }, + }, + ]), + ).rejects.toThrow(); + + expect((await mailRows()).length).toBe(0); + }); + + test("retrying an already-committed batch (same messageIds, no messageKey) writes nothing and returns empty", async () => { + const messageId1 = ""; + const messageId2 = ""; + const items = [ + { + scope: { tenantId: "t1", principalId: "p1" }, + args: { + tenantId: "t1", + principalId: "p1", + address: "p1@t1.example", + fromAddress: "p1@t1.example", + subject: "Sent", + body: "Hi p2", + direction: "outbound" as const, + messageId: messageId1, + }, + }, + { + scope: { tenantId: "t1", principalId: "p2" }, + args: { + tenantId: "t1", + principalId: "p2", + address: "p2@t1.example", + fromAddress: "p1@t1.example", + subject: "Recv", + body: "Hi p2", + direction: "inbound" as const, + messageId: messageId2, + }, + }, + ]; + + const first = await writeMailboxMessages(db, items); + expect(first.length).toBe(2); + + const retry = await writeMailboxMessages(db, items); + expect(retry).toEqual([]); + expect((await mailRows()).length).toBe(2); + }); + + test("events fire only after commit, and only for newly-written rows", async () => { + const bus = createInMemoryMailboxEventBus(); + const received: Array<{ id: string }> = []; + bus.subscribe({ tenantId: "t1", principalId: "p1" }, (e) => + received.push(e), + ); + + const messageId = ""; + const item = { + scope: { tenantId: "t1", principalId: "p1" }, + args: { + tenantId: "t1", + principalId: "p1", + address: "p1@t1.example", + fromAddress: "p1@t1.example", + subject: "Hello", + body: "World", + messageId, + }, + }; + + const first = await writeMailboxMessages(db, [item], { bus }); + expect(first.length).toBe(1); + expect(received.map((e) => e.id)).toEqual(first); + + // Dedupe: the retry writes nothing and must not publish a second event. + const retry = await writeMailboxMessages(db, [item], { bus }); + expect(retry).toEqual([]); + expect(received.length).toBe(1); + }); + + test("a messageKey override is honored inside a batch", async () => { + const ids = await writeMailboxMessages(db, [ + { + scope: { tenantId: "t1", principalId: "p1" }, + args: { + tenantId: "t1", + principalId: "p1", + address: "p1@t1.example", + fromAddress: "p1@t1.example", + subject: "Hello", + body: "World", + messageKey: "custom:batch-key", + }, + }, + ]); + const rows = await db.execute<{ message_key: string }>( + sql`SELECT message_key FROM "mailbox"."principal_mail" WHERE id = ${ids[0]}`, + ); + expect(rows[0]?.message_key).toBe("custom:batch-key"); + }); +}); + From 8801627baf3fe21249ec269294f1af705620b0d1 Mon Sep 17 00:00:00 2001 From: Sawyer Date: Fri, 4 Sep 2026 18:20:38 -0700 Subject: [PATCH 2/5] write: caller-supplied Message-ID, direction, and message key; batch write WriteMailboxMessageArgs gains messageId (validated as a bracketed msg-id, becomes the frame's Message-ID and the cached column) and direction (default inbound). Omitting messageKey no longer leaves a row unkeyed: it defaults to mailboxKey.transport(messageId, principalId), the same shape persist.ts's transport dual-write already uses, so a retry with the same caller-supplied messageId dedupes without a caller minting its own key. New writeMailboxMessages(db, items, opts?) writes an arbitrary batch of { scope, args } pairs in one transaction, with per-row onConflictDoNothing dedupe and bus events published only after commit for rows actually inserted. deliverInboxItems is unchanged. --- src/write.ts | 153 +++++++++++++++++++++++++++++++++++++++++++++++---- 1 file changed, 142 insertions(+), 11 deletions(-) diff --git a/src/write.ts b/src/write.ts index 3e092fd..9b139b0 100644 --- a/src/write.ts +++ b/src/write.ts @@ -5,6 +5,7 @@ import { mailbox, principalMail } from "./schema.js"; import type { MailboxDb } from "./db.js"; import { publishMailboxEvent, type MailboxEventBus } from "./bus.js"; import { + assertMsgId, buildMailFrame, generateMailboxMessageId, headerValue, @@ -116,8 +117,32 @@ export type WriteMailboxMessageArgs = { fromAddress: string; subject: string; body: string; - /** Idempotency key; a second write with the same key is a no-op (returns null). */ + /** + * Idempotency key; a second write with the same key is a no-op (returns + * null). Omitted, a write still gets a stable key of its own: the package's + * transport key (`mailboxKey.transport`), derived from the frame's + * `messageId` and the recipient `principalId` — so a caller that retries + * the exact same `messageId` collapses onto one row without having to mint + * its own key, while two independent writes with different (minted) + * `messageId`s never collide. + */ messageKey?: string; + /** + * The complete msg-id (angle brackets included) this write's frame carries + * as its `Message-ID:` header, and the value cached in + * `principal_mail.message_id`. `RangeError` (via `assertMsgId`) when it is + * not a bracketed msg-id. Omitted, one is minted the way it always was — + * `generateMailboxMessageId`. + */ + messageId?: string; + /** + * `"inbound"` (default) or `"outbound"`. The mailbox row's own copy of who + * sent it: an inbound row is delivered mail, an outbound row is the + * sender's durable copy of a message they sent. Purely a stored fact — + * this package's inbox views stay inbound-only regardless of what a caller + * writes here (see ARCHITECTURE.md's Known limits). + */ + direction?: "inbound" | "outbound"; inReplyTo?: string; /** * The thread's ancestry, oldest first; each entry a bracketed msg-id. Emitted @@ -161,9 +186,16 @@ type MailboxInsertTx = { * in agreement — not two independent trims that could drift apart. */ function normalizeThreadingArgs< - T extends { inReplyTo?: string; references?: string[] }, + T extends { + messageId?: string; + inReplyTo?: string; + references?: string[]; + }, >(args: T): T { const normalized: T = { ...args }; + if (args.messageId !== undefined) { + normalized.messageId = headerValue(args.messageId); + } if (args.inReplyTo !== undefined) { normalized.inReplyTo = headerValue(args.inReplyTo); } @@ -174,18 +206,21 @@ function normalizeThreadingArgs< } /** - * Encode args into a durable MIME frame. Mint a fresh Message-ID each call. + * Encode args into a durable MIME frame. * - * The minted id is returned alongside the bytes rather than re-parsed out of - * them: it is what the row's `message_id` cache stores, and re-decoding a frame - * this function just built to recover a value it already had is work with a - * failure mode attached. + * Uses the caller's `messageId` (already validated as a bracketed msg-id by + * `assertMsgId` below) when supplied, else mints a fresh one exactly as + * before. Either way the id is returned alongside the bytes rather than + * re-parsed out of them: it is what the row's `message_id` cache stores, and + * re-decoding a frame this function just built to recover a value it already + * had is work with a failure mode attached. */ function encodeMailboxFrame(args: WriteMailboxMessageArgs): { raw: Uint8Array; messageId: string; } { - const messageId = generateMailboxMessageId(args.fromAddress); + if (args.messageId !== undefined) assertMsgId(args.messageId, "messageId"); + const messageId = args.messageId ?? generateMailboxMessageId(args.fromAddress); const frameArgs: Parameters[0] = { from: args.fromAddress, to: args.address, @@ -237,7 +272,8 @@ async function insertMailboxMessage( messageId: string, ): Promise<{ id: string } | null> { assertMailboxFrameBytes(raw); - const refs = boundRefs(args.refs, args.messageKey ?? null); + const messageKey = args.messageKey ?? mailboxKey.transport(messageId, args.principalId); + const refs = boundRefs(args.refs, messageKey); // The management row is created EAGERLY with the message: every mutation and // the unread count are then plain operations on `mailbox`, and the unread @@ -251,11 +287,11 @@ async function insertMailboxMessage( tenantId: args.tenantId, principalId: args.principalId, address: args.address, - direction: "inbound", + direction: args.direction ?? "inbound", raw: Buffer.from(raw), subject: args.subject, fromAddress: args.fromAddress, - messageKey: args.messageKey ?? null, + messageKey, messageId, inReplyTo: args.inReplyTo ?? null, refs: refs ?? null, @@ -350,11 +386,21 @@ export async function writeMailboxMessage( * not dedupe against the new encoding and cannot false-collide with it — no * migration is performed; redelivery after upgrade may insert a second row. */ +// `transport` is the default `writeMailboxMessage` / `writeMailboxMessages` +// fall back to when a caller supplies no `messageKey` of its own: keyed on +// the frame's own `messageId` (caller-supplied or minted) plus the recipient +// `principalId`, matching `persist.ts`'s transport dual-write key shape +// (`transport:mid::`) without importing from it — +// that file owns a second fallback (content-hash) for frames with no +// Message-ID at all, which never happens on this package's own write path, +// where a `messageId` is always present by the time a row is inserted. export const mailboxKey = { inbox: (source: string, externalId: string) => `inbox2:${source.length}:${source}:${externalId}`, gate: (gateId: string) => `gate:${gateId}`, run: (runId: string) => `run:${runId}`, + transport: (messageId: string, principalId: string) => + `transport:mid:${messageId}:${principalId}`, } as const; export type InboxItem = { @@ -499,3 +545,88 @@ export async function deliverInboxItems( return results; } + +/** One item of a `writeMailboxMessages` batch: an address plus the scope it lands in. */ +export type WriteMailboxMessagesItem = { + scope: MailboxScopeIds; + args: WriteMailboxMessageArgs; +}; + +export type WriteMailboxMessagesOpts = { + /** Best-effort live signal per inserted row, published only after commit. */ + bus?: MailboxEventBus; +}; + +/** + * Write an entire conversation turn — a sender's own outbound copy alongside + * every recipient's inbound copy, or any other mixed-scope, mixed-direction + * batch — as ONE transaction. This is the conversation path; `deliverInboxItems` + * remains the notify-item path (ingress adapters fanning one external item out + * to durable rows) and is unchanged by this function's existence. + * + * Each item is scope-checked, field-checked, encoded, and frame-asserted + * BEFORE the transaction opens — same prevalidation discipline as + * `deliverInboxItems` — so oversize or malformed input never begins a + * multi-row insert. All rows then commit together inside one + * `db.transaction`: a throw from any single item (a caller bug, or a control- + * plane FK the item's scope does not satisfy) rolls back every row the batch + * would otherwise have written, including ones already inserted earlier in + * the same call. + * + * Dedupe is per row, on `(tenantId, principalId, messageKey)` via + * `onConflictDoNothing` on the existing partial unique index — the same + * mechanism `writeMailboxMessage` and `deliverInboxItems` use. A row whose + * `messageKey` collides with one already committed is a no-op inside the + * transaction, not a rollback trigger; retrying an entire successful batch + * therefore commits nothing the second time and returns no ids. + * + * Returns the ids of rows this call actually inserted, in item order, + * skipping any item deduped by its messageKey. Bus events publish only after + * commit, one per written row — never for a deduped item, and never before + * the transaction is durable. + */ +export async function writeMailboxMessages( + db: MailboxDb, + items: WriteMailboxMessagesItem[], + opts?: WriteMailboxMessagesOpts, +): Promise { + type Prepared = { + scope: MailboxScopeIds; + writeArgs: WriteMailboxMessageArgs; + raw: Uint8Array; + messageId: string; + }; + const prepared: Prepared[] = []; + for (const { scope, args } of items) { + assertMailboxScope(scope); + const writeArgs: WriteMailboxMessageArgs = { + ...normalizeThreadingArgs(args), + tenantId: scope.tenantId, + principalId: scope.principalId, + }; + assertMailboxStringFieldsFit(writeArgs); + const { raw, messageId } = encodeMailboxFrame(writeArgs); + assertMailboxFrameBytes(raw); + prepared.push({ scope, writeArgs, raw, messageId }); + } + + type Inserted = { id: string; scope: MailboxScopeIds }; + const inserted: Inserted[] = await db.transaction(async (tx) => { + const inserted: Inserted[] = []; + for (const { scope, writeArgs, raw, messageId } of prepared) { + const written = await insertMailboxMessage(tx, writeArgs, raw, messageId); + if (written === null) continue; + inserted.push({ id: written.id, scope }); + } + return inserted; + }); + + // Post-commit only: live signals for newly inserted ids, one per row. + if (opts?.bus) { + for (const { id, scope } of inserted) { + publishMailboxEvent(opts.bus, scope, id, logger, "create"); + } + } + + return inserted.map((row) => row.id); +} From d75049b09e1cac4e169f00744eb48210db905f6c Mon Sep 17 00:00:00 2001 From: Sawyer Date: Fri, 4 Sep 2026 18:20:43 -0700 Subject: [PATCH 3/5] Update docs: caller-supplied write fields and the new batch write path ARCHITECTURE.md documents writeMailboxMessages as the conversation path alongside deliverInboxItems's notify-item path, and the messageId/ direction/messageKey defaulting behavior. README points to both from a new Write paths section. CHANGELOG records the addition. --- ARCHITECTURE.md | 31 +++++++++++++++++++++++++++++++ CHANGELOG.md | 29 +++++++++++++++++++++++++++++ README.md | 19 +++++++++++++++++++ 3 files changed, 79 insertions(+) diff --git a/ARCHITECTURE.md b/ARCHITECTURE.md index 265f584..6f28f26 100644 --- a/ARCHITECTURE.md +++ b/ARCHITECTURE.md @@ -252,6 +252,37 @@ optional host `enqueue` hook run only after commit, and only for newly inserted ids. Both side effects are best-effort: a throw is logged with the message id and never rejects the delivery. +**Two write paths, two shapes of batch.** `deliverInboxItems` is the +**notify-item path**: one external item, fanned out to every addressed +principal, keyed by `mailboxKey.inbox(source, externalId)` — unchanged by the +addition below. `writeMailboxMessages(db, items, opts?)` is the +**conversation path**: an arbitrary batch of `{ scope, args }` pairs — a +sender's own outbound copy alongside every recipient's inbound copy of the +same turn, mixed tenants and principals allowed — committed in the same +single-transaction-or-none shape, with per-row `onConflictDoNothing` dedupe on +the same `messageKey` partial unique index and bus events published only +after commit, one per row this call actually inserted. A throw from any one +item (an invalid scope, an oversize frame, a control-plane FK the item's +scope does not satisfy) rolls back every row the batch would otherwise have +written, including ones already inserted earlier in the same call — same +atomicity guarantee as `deliverInboxItems`, over a caller-shaped item instead +of an ingress-shaped one. + +**A write's Message-ID, direction, and dedupe key are now the caller's to +set.** `WriteMailboxMessageArgs.messageId` lets a caller hand the write path +the exact msg-id its own frame must carry (validated as a bracketed msg-id; +`RangeError` otherwise) instead of always minting one — needed when a +message's id has to be predictable ahead of the write, e.g. so a later +`inReplyTo` can reference it. `direction` (default `"inbound"`) is a stored +fact only; it does not change which rows the inbox views serve (still +inbound-only, see Known limits). And `messageKey`, when the caller omits it, +now defaults to `mailboxKey.transport(messageId, principalId)` — the same +`transport:mid::` shape `persist.ts`'s transport +dual-write already uses — rather than leaving the row unkeyed: a retry that +reuses the same caller-supplied `messageId` therefore dedupes for free, while +two writes that each mint their own `messageId` never collide. A caller- +supplied `messageKey` still overrides the default, exactly as before. + **Bus publish isolates listeners.** `publishMailboxEvent` invokes each subscriber independently; one throwing listener does not stop the others. SSE connections serialize writes, bound the pending queue, and close on overflow or diff --git a/CHANGELOG.md b/CHANGELOG.md index 77a70fc..8ab8998 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -20,6 +20,35 @@ always called out under their own heading. already carries refs. A throwing `resolveRefs` follows the existing dual-write contract: logged, upstream unaffected, no mailbox row for that frame. +- **Caller-supplied Message-ID, direction, and message key on writes; a + one-transaction batch write.** `WriteMailboxMessageArgs` gains + `messageId?: string` — when supplied it must be a bracketed msg-id + (validated with `assertMsgId`) and becomes the built frame's own + `Message-ID:` header and the cached `principal_mail.message_id`; + omitted, one is still minted exactly as before. It also gains + `direction?: "inbound" | "outbound"` (default `"inbound"`). Omitting + `messageKey` no longer leaves the row unkeyed: it now defaults to + `mailboxKey.transport(messageId, principalId)`, the same + `transport:mid::` shape `persist.ts`'s + transport dual-write already uses — so retrying a write with the same + caller-supplied `messageId` dedupes without the caller minting its own + key, while two independent (differently-minted) writes never collide. + A caller-supplied `messageKey` still overrides the default. + New `writeMailboxMessages(db, items, opts?)` writes an arbitrary batch + of `{ scope, args }` pairs — e.g. a sender's outbound copy alongside + every recipient's inbound copy of the same conversation turn — in ONE + transaction: every row is scope-checked, field-checked, encoded, and + frame-asserted before the transaction opens, all new rows commit + together or none do, and a throw from any single item (an invalid + scope, an oversize frame, an unknown control-plane principal) rolls + back the whole batch. Per-row dedupe still runs through + `onConflictDoNothing` on the existing `messageKey` partial unique + index, so a retried batch dedupes row-by-row without failing; the + function returns only the ids of rows this call actually inserted. + Bus events publish only after commit, one per written row. This is the + **conversation path**; `deliverInboxItems` remains the **notify-item + path** for ingress adapters and is unchanged. + - **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 392222d..99dd684 100644 --- a/README.md +++ b/README.md @@ -51,6 +51,25 @@ bun add github:corbitsdev/corbits-mailbox | `src/` | The published package. Owns `principal_mail` (the message, immutable) and `mailbox` (the management layer, created eagerly with each message). | | `examples/reference-host` | Mounts it on a real `@intx/hub-api` app against a live Postgres and asserts the acceptance scenarios end to end. | +## Write paths + +`src/write.ts` exports two batch write functions, each for a different shape +of caller: + +- **`deliverInboxItems`** — the notify-item path. One external item (an + ingress adapter: a mail connector, a webhook), fanned out to every + addressed principal, deduped on `mailboxKey.inbox(source, externalId)`. +- **`writeMailboxMessages`** — the conversation path. An arbitrary batch of + `{ scope, args }` pairs — for example a sender's own outbound copy + alongside every recipient's inbound copy of the same turn — committed in + one transaction with per-row dedupe on the `messageKey` unique index. + +Both commit every new row in the call as a single transaction (or none), and +publish bus events only after commit, one per row actually written. See +[ARCHITECTURE.md](./ARCHITECTURE.md) for the full write-path writeup, +including `writeMailboxMessage`'s caller-supplied `messageId`, `direction`, +and default `messageKey`. + ## Working on it ```sh From 2162136d8dca2ed8c5e6b2e5564f29ba7c74052f Mon Sep 17 00:00:00 2001 From: Sawyer Date: Fri, 4 Sep 2026 18:45:31 -0700 Subject: [PATCH 4/5] write: direction-aware default key, outbound pre-read, index exports (CL-7452) Fold critique findings into the caller-message-id write path: - export writeMailboxMessage(s) and their arg/opts types from the package entry point, so a consumer never has to reach into src/write.js directly - an outbound row is now created already-read (mailbox.read_at pinned to its own created_at), so the unread count and unread view exclude a sender's own copy without a direction predicate; listUserMailbox and getMailboxMessage gain an optional direction filter (default "inbound") for a caller that does need a principal's sent copies - the default messageKey now folds direction into mailboxKey.transport(messageId, principalId, direction): the inbound default is unchanged (byte-for-byte what persist.ts's transport dual-write already writes), while outbound gets a distinct :outbound suffix so the same caller-supplied Message-ID in both directions no longer collapses onto one row - writeMailboxMessages now returns Array<{ messageKey, id }> in item order, matching deliverInboxItems's DeliveredInboxItem shape, instead of a filtered id list; WriteMailboxMessagesItem's args drops the duplicated tenantId/principalId (Omit<..., "tenantId" | "principalId">) so scope is the only source of either --- src/index.ts | 3 + src/read.ts | 16 +++- src/write.test.ts | 235 +++++++++++++++++++++++++++++++++++++++++----- src/write.ts | 89 +++++++++++++----- 4 files changed, 290 insertions(+), 53 deletions(-) diff --git a/src/index.ts b/src/index.ts index 1ac31fc..f9b4e86 100644 --- a/src/index.ts +++ b/src/index.ts @@ -69,6 +69,7 @@ export type { export { writeMailboxMessage, + writeMailboxMessages, deliverInboxItems, mailboxKey, MAX_MAILBOX_REFS, @@ -77,6 +78,8 @@ export { } from "./write.js"; export type { WriteMailboxMessageArgs, + WriteMailboxMessagesItem, + WriteMailboxMessagesOpts, InboxItem, DeliverInboxItemsOpts, DeliveredInboxItem, diff --git a/src/read.ts b/src/read.ts index 9a61154..417d303 100644 --- a/src/read.ts +++ b/src/read.ts @@ -518,6 +518,14 @@ export type MailboxScope = { priorities: readonly string[]; /** Host seam for turning sender addresses into human labels; see `SenderDisplayResolver`. */ resolveSenderDisplays?: SenderDisplayResolver; + /** + * Which direction of mail to serve. Defaults to `"inbound"` — the + * long-standing contract, since the inbox has only ever shown delivered + * mail. `"outbound"` reads a principal's own sent copies; `"all"` returns + * both, e.g. for a thread reader that needs a sender's copy alongside its + * recipients' copies. + */ + direction?: "inbound" | "outbound" | "all"; }; export type MailboxPage = { @@ -541,10 +549,11 @@ export async function listUserMailbox( const sort: MailboxSort = scope.sort ?? "date"; const filter: MailboxFilter = scope.filter ?? {}; const PRIORITY_RANK = priorityRank(scope.priorities); + const direction = scope.direction ?? "inbound"; const conditions = [ eq(principalMail.tenantId, scope.tenantId), eq(principalMail.principalId, scope.principalId), - eq(principalMail.direction, "inbound"), + ...(direction === "all" ? [] : [eq(principalMail.direction, direction)]), ...viewConditions(scope.view), ...filterConditions(filter), ]; @@ -651,8 +660,11 @@ export async function getMailboxMessage( principalId: string; id: string; resolveSenderDisplays?: SenderDisplayResolver; + /** Defaults to `"inbound"`; see `MailboxScope.direction`. */ + direction?: "inbound" | "outbound" | "all"; }, ): Promise { + const direction = args.direction ?? "inbound"; const [row] = await db .select({ ...getTableColumns(principalMail), ...STATE_COLUMNS }) .from(principalMail) @@ -662,7 +674,7 @@ export async function getMailboxMessage( eq(principalMail.id, args.id), eq(principalMail.tenantId, args.tenantId), eq(principalMail.principalId, args.principalId), - eq(principalMail.direction, "inbound"), + ...(direction === "all" ? [] : [eq(principalMail.direction, direction)]), ), ) .limit(1); diff --git a/src/write.test.ts b/src/write.test.ts index e2e14fa..565d1f4 100644 --- a/src/write.test.ts +++ b/src/write.test.ts @@ -9,6 +9,7 @@ import { } from "./write.js"; import { decodeMailFrame } from "./frame.js"; import { getMailboxMessage, listUserMailbox } from "./read.js"; +import { countUnreadActiveMailbox } from "./mutations.js"; import { createInMemoryMailboxEventBus } from "./bus.js"; import { withTestDb, seedScope, TEST_VOCABULARY } from "./test-helpers.js"; import type { MailboxDb } from "./db.js"; @@ -991,8 +992,6 @@ describe("writeMailboxMessages (one-transaction batch write)", () => { { scope: { tenantId: "t1", principalId: "p1" }, args: { - tenantId: "t1", - principalId: "p1", address: "p1@t1.example", fromAddress: "p1@t1.example", subject: "Sent", @@ -1003,8 +1002,6 @@ describe("writeMailboxMessages (one-transaction batch write)", () => { { scope: { tenantId: "t1", principalId: "p1" }, args: { - tenantId: "t1", - principalId: "p1", address: "p1@t1.example", fromAddress: "p1@t1.example", subject: "Recv", @@ -1015,8 +1012,6 @@ describe("writeMailboxMessages (one-transaction batch write)", () => { { scope: { tenantId: "t1", principalId: "p2" }, args: { - tenantId: "t1", - principalId: "p2", address: "p2@t1.example", fromAddress: "p1@t1.example", subject: "Recv", @@ -1045,8 +1040,6 @@ describe("writeMailboxMessages (one-transaction batch write)", () => { { scope: { tenantId: "t1", principalId: "p1" }, args: { - tenantId: "t1", - principalId: "p1", address: "p1@t1.example", fromAddress: "p1@t1.example", subject: "Sent", @@ -1057,8 +1050,6 @@ describe("writeMailboxMessages (one-transaction batch write)", () => { { scope: { tenantId: "t1", principalId: "p2" }, args: { - tenantId: "t1", - principalId: "p2", address: "p2@t1.example", fromAddress: "p1@t1.example", subject: "Recv", @@ -1069,8 +1060,6 @@ describe("writeMailboxMessages (one-transaction batch write)", () => { { scope: { tenantId: "t1", principalId: "nobody-seeded-this" }, args: { - tenantId: "t1", - principalId: "nobody-seeded-this", address: "ghost@t1.example", fromAddress: "p1@t1.example", subject: "Bad", @@ -1083,15 +1072,13 @@ describe("writeMailboxMessages (one-transaction batch write)", () => { expect((await mailRows()).length).toBe(0); }); - test("retrying an already-committed batch (same messageIds, no messageKey) writes nothing and returns empty", async () => { + test("retrying an already-committed batch (same messageIds, no messageKey) writes nothing and returns null ids", async () => { const messageId1 = ""; const messageId2 = ""; const items = [ { scope: { tenantId: "t1", principalId: "p1" }, args: { - tenantId: "t1", - principalId: "p1", address: "p1@t1.example", fromAddress: "p1@t1.example", subject: "Sent", @@ -1103,8 +1090,6 @@ describe("writeMailboxMessages (one-transaction batch write)", () => { { scope: { tenantId: "t1", principalId: "p2" }, args: { - tenantId: "t1", - principalId: "p2", address: "p2@t1.example", fromAddress: "p1@t1.example", subject: "Recv", @@ -1119,7 +1104,7 @@ describe("writeMailboxMessages (one-transaction batch write)", () => { expect(first.length).toBe(2); const retry = await writeMailboxMessages(db, items); - expect(retry).toEqual([]); + expect(retry.map((r) => r.id)).toEqual([null, null]); expect((await mailRows()).length).toBe(2); }); @@ -1134,8 +1119,6 @@ describe("writeMailboxMessages (one-transaction batch write)", () => { const item = { scope: { tenantId: "t1", principalId: "p1" }, args: { - tenantId: "t1", - principalId: "p1", address: "p1@t1.example", fromAddress: "p1@t1.example", subject: "Hello", @@ -1146,21 +1129,21 @@ describe("writeMailboxMessages (one-transaction batch write)", () => { const first = await writeMailboxMessages(db, [item], { bus }); expect(first.length).toBe(1); - expect(received.map((e) => e.id)).toEqual(first); + expect(received.map((e) => e.id)).toEqual( + first.map((row) => row.id).filter((id): id is string => id !== null), + ); // Dedupe: the retry writes nothing and must not publish a second event. const retry = await writeMailboxMessages(db, [item], { bus }); - expect(retry).toEqual([]); + expect(retry.map((r) => r.id)).toEqual([null]); expect(received.length).toBe(1); }); test("a messageKey override is honored inside a batch", async () => { - const ids = await writeMailboxMessages(db, [ + const results = await writeMailboxMessages(db, [ { scope: { tenantId: "t1", principalId: "p1" }, args: { - tenantId: "t1", - principalId: "p1", address: "p1@t1.example", fromAddress: "p1@t1.example", subject: "Hello", @@ -1169,10 +1152,210 @@ describe("writeMailboxMessages (one-transaction batch write)", () => { }, }, ]); + expect(results[0]?.messageKey).toBe("custom:batch-key"); const rows = await db.execute<{ message_key: string }>( - sql`SELECT message_key FROM "mailbox"."principal_mail" WHERE id = ${ids[0]}`, + sql`SELECT message_key FROM "mailbox"."principal_mail" WHERE id = ${results[0]?.id}`, ); expect(rows[0]?.message_key).toBe("custom:batch-key"); }); + + test("returns { messageKey, id } per item, in item order, including the default-keyed items", async () => { + const messageId = ""; + const results = await writeMailboxMessages(db, [ + { + scope: { tenantId: "t1", principalId: "p1" }, + args: { + address: "p1@t1.example", + fromAddress: "p1@t1.example", + subject: "Sent", + body: "Hi p2", + direction: "outbound", + messageId, + }, + }, + { + scope: { tenantId: "t1", principalId: "p2" }, + args: { + address: "p2@t1.example", + fromAddress: "p1@t1.example", + subject: "Recv", + body: "Hi p2", + direction: "inbound", + messageId, + }, + }, + ]); + expect(results).toEqual([ + { messageKey: `transport:mid:${messageId}:p1:outbound`, id: expect.any(String) }, + { messageKey: `transport:mid:${messageId}:p2`, id: expect.any(String) }, + ]); + }); + + test("invalid scope on a later item is refused before any row is written", async () => { + await expect( + writeMailboxMessages(db, [ + { + scope: { tenantId: "t1", principalId: "p1" }, + args: { + address: "p1@t1.example", + fromAddress: "p1@t1.example", + subject: "Hello", + body: "World", + }, + }, + { + scope: { tenantId: "t1", principalId: " " }, + args: { + address: "p1@t1.example", + fromAddress: "p1@t1.example", + subject: "Hello", + body: "World", + }, + }, + ]), + ).rejects.toThrow(RangeError); + expect((await mailRows()).length).toBe(0); + }); + + test("mailbox management row is created for outbound rows written through the batch path too", async () => { + const results = await writeMailboxMessages(db, [ + { + scope: { tenantId: "t1", principalId: "p1" }, + args: { + address: "p1@t1.example", + fromAddress: "p1@t1.example", + subject: "Hello", + body: "World", + direction: "outbound", + }, + }, + ]); + const rows = await db.execute<{ n: number }>( + sql`SELECT count(*)::int AS n FROM "mailbox"."mailbox" WHERE id = ${results[0]?.id}`, + ); + expect(rows[0]!.n).toBe(1); + }); +}); + +describe("outbound rows and the inbox read model", () => { + const base = { + tenantId: "t1", + principalId: "p1", + address: "p1@t1.example", + fromAddress: "p1@t1.example", + subject: "Hello", + body: "World", + }; + + test("an outbound row is created already-read: excluded from the unread view and count without a direction predicate", async () => { + const written = await writeMailboxMessage(db, { + ...base, + direction: "outbound", + }); + expect(written).not.toBeNull(); + + const scope = { tenantId: "t1", principalId: "p1" }; + const page = await listUserMailbox(db, { + ...scope, + priorities: TEST_VOCABULARY.priorities, + view: "unread", + limit: 50, + }); + expect(page.items.length).toBe(0); + expect( + await getMailboxMessage(db, { ...scope, id: written!.id }), + ).toBeNull(); + expect( + await getMailboxMessage(db, { ...scope, id: written!.id, direction: "all" }), + ).not.toBeNull(); + + expect(await countUnreadActiveMailbox(db, scope)).toBe(0); + }); + + test("listUserMailbox and getMailboxMessage accept an explicit direction filter", async () => { + const outbound = await writeMailboxMessage(db, { + ...base, + direction: "outbound", + messageId: "", + }); + const inbound = await writeMailboxMessage(db, { + ...base, + direction: "inbound", + messageId: "", + }); + const scope = { tenantId: "t1", principalId: "p1" }; + + const outboundPage = await listUserMailbox(db, { + ...scope, + priorities: TEST_VOCABULARY.priorities, + view: "all", + limit: 50, + direction: "outbound", + }); + expect(outboundPage.items.map((i) => i.id)).toEqual([outbound!.id]); + + const allPage = await listUserMailbox(db, { + ...scope, + priorities: TEST_VOCABULARY.priorities, + view: "all", + limit: 50, + direction: "all", + }); + expect(new Set(allPage.items.map((i) => i.id))).toEqual( + new Set([outbound!.id, inbound!.id]), + ); + + expect( + await getMailboxMessage(db, { + ...scope, + id: outbound!.id, + direction: "outbound", + }), + ).not.toBeNull(); + }); +}); + +describe("default messageKey collisions", () => { + const base = { + tenantId: "t1", + principalId: "p1", + address: "p1@t1.example", + fromAddress: "p1@t1.example", + subject: "Hello", + body: "World", + }; + + test("same caller Message-ID to the same principal in both directions writes two distinct rows", async () => { + const messageId = ""; + const results = await writeMailboxMessages(db, [ + { + scope: { tenantId: "t1", principalId: "p1" }, + args: { ...base, direction: "outbound", messageId, subject: "Sent" }, + }, + { + scope: { tenantId: "t1", principalId: "p1" }, + args: { ...base, direction: "inbound", messageId, subject: "Recv" }, + }, + ]); + expect(results.length).toBe(2); + expect(results.every((r) => r.id !== null)).toBe(true); + }); + + test("two distinct messages that reuse one caller Message-ID for the same principal collide", async () => { + const messageId = ""; + const a = await writeMailboxMessage(db, { ...base, messageId, subject: "A" }); + const b = await writeMailboxMessage(db, { ...base, messageId, subject: "B" }); + expect(a).not.toBeNull(); + expect(b).toBeNull(); + }); + + test("default write key equals persist.ts's transport key shape for the same Message-ID + principal", async () => { + const messageId = ""; + const written = await writeMailboxMessage(db, { ...base, messageId }); + const rows = await db.execute<{ message_key: string }>( + sql`SELECT message_key FROM "mailbox"."principal_mail" WHERE id = ${written!.id}`, + ); + expect(rows[0]!.message_key).toBe(`transport:mid:${messageId}:p1`); + }); }); diff --git a/src/write.ts b/src/write.ts index 9b139b0..9b12bd5 100644 --- a/src/write.ts +++ b/src/write.ts @@ -270,9 +270,11 @@ async function insertMailboxMessage( args: WriteMailboxMessageArgs, raw: Uint8Array, messageId: string, -): Promise<{ id: string } | null> { +): Promise<{ id: string; messageKey: string } | null> { assertMailboxFrameBytes(raw); - const messageKey = args.messageKey ?? mailboxKey.transport(messageId, args.principalId); + const direction = args.direction ?? "inbound"; + const messageKey = + args.messageKey ?? mailboxKey.transport(messageId, args.principalId, direction); const refs = boundRefs(args.refs, messageKey); // The management row is created EAGERLY with the message: every mutation and @@ -287,7 +289,7 @@ async function insertMailboxMessage( tenantId: args.tenantId, principalId: args.principalId, address: args.address, - direction: args.direction ?? "inbound", + direction, raw: Buffer.from(raw), subject: args.subject, fromAddress: args.fromAddress, @@ -304,20 +306,26 @@ async function insertMailboxMessage( ], where: sql`${principalMail.messageKey} IS NOT NULL`, }) - .returning({ id: principalMail.id }); + .returning({ id: principalMail.id, createdAt: principalMail.createdAt }); const inserted = rows[0]; if (!inserted) return null; + // An outbound row is the sender's own durable copy of a message they sent, + // not something to notify them about — it is created already-read (readAt + // pinned to the same createdAt Postgres just minted) so the unread count + // and the unread view exclude it without either needing a direction + // predicate of their own. await tx.insert(mailbox).values({ id: inserted.id, tenantId: args.tenantId, principalId: args.principalId, + readAt: direction === "outbound" ? inserted.createdAt : null, priority: args.priority ?? null, classification: args.classification ?? null, status: args.status ?? null, }); - return inserted; + return { id: inserted.id, messageKey }; } /** @@ -389,18 +397,31 @@ export async function writeMailboxMessage( // `transport` is the default `writeMailboxMessage` / `writeMailboxMessages` // fall back to when a caller supplies no `messageKey` of its own: keyed on // the frame's own `messageId` (caller-supplied or minted) plus the recipient -// `principalId`, matching `persist.ts`'s transport dual-write key shape -// (`transport:mid::`) without importing from it — -// that file owns a second fallback (content-hash) for frames with no -// Message-ID at all, which never happens on this package's own write path, -// where a `messageId` is always present by the time a row is inserted. +// `principalId`. For the (default) `"inbound"` direction this matches +// `persist.ts`'s transport dual-write key shape +// (`transport:mid::`) BYTE FOR BYTE and without +// importing from it — a frame persist already delivered and a direct inbound +// write for the same Message-ID + principal dedupe onto the same row, as +// they always have. `"outbound"` gets a `:outbound` suffix instead of +// silently sharing the inbound key: a sender's own copy of a turn and a +// recipient's (or their own) inbound copy of the identical caller-supplied +// Message-ID must NOT collapse onto one row. `persist.ts` owns a second +// fallback (content-hash) for frames with no Message-ID at all, which never +// happens on this package's own write path, where a `messageId` is always +// present by the time a row is inserted. export const mailboxKey = { inbox: (source: string, externalId: string) => `inbox2:${source.length}:${source}:${externalId}`, gate: (gateId: string) => `gate:${gateId}`, run: (runId: string) => `run:${runId}`, - transport: (messageId: string, principalId: string) => - `transport:mid:${messageId}:${principalId}`, + transport: ( + messageId: string, + principalId: string, + direction: "inbound" | "outbound" = "inbound", + ) => + direction === "outbound" + ? `transport:mid:${messageId}:${principalId}:outbound` + : `transport:mid:${messageId}:${principalId}`, } as const; export type InboxItem = { @@ -546,10 +567,14 @@ export async function deliverInboxItems( return results; } -/** One item of a `writeMailboxMessages` batch: an address plus the scope it lands in. */ +/** + * One item of a `writeMailboxMessages` batch: an address plus the scope it + * lands in. `args` omits `tenantId`/`principalId` — `scope` is the SOLE + * source of both, so there is no second copy that could disagree with it. + */ export type WriteMailboxMessagesItem = { scope: MailboxScopeIds; - args: WriteMailboxMessageArgs; + args: Omit; }; export type WriteMailboxMessagesOpts = { @@ -580,21 +605,23 @@ export type WriteMailboxMessagesOpts = { * transaction, not a rollback trigger; retrying an entire successful batch * therefore commits nothing the second time and returns no ids. * - * Returns the ids of rows this call actually inserted, in item order, - * skipping any item deduped by its messageKey. Bus events publish only after - * commit, one per written row — never for a deduped item, and never before - * the transaction is durable. + * Returns one result per item, IN ITEM ORDER — `{ messageKey, id }`, `id` + * null exactly when that item's messageKey deduped against an existing row + * (matching `deliverInboxItems`'s `DeliveredInboxItem` shape). Bus events + * publish only after commit, one per written row — never for a deduped item, + * and never before the transaction is durable. */ export async function writeMailboxMessages( db: MailboxDb, items: WriteMailboxMessagesItem[], opts?: WriteMailboxMessagesOpts, -): Promise { +): Promise { type Prepared = { scope: MailboxScopeIds; writeArgs: WriteMailboxMessageArgs; raw: Uint8Array; messageId: string; + messageKey: string; }; const prepared: Prepared[] = []; for (const { scope, args } of items) { @@ -607,18 +634,30 @@ export async function writeMailboxMessages( assertMailboxStringFieldsFit(writeArgs); const { raw, messageId } = encodeMailboxFrame(writeArgs); assertMailboxFrameBytes(raw); - prepared.push({ scope, writeArgs, raw, messageId }); + // Computed here, once, rather than left to `insertMailboxMessage`'s own + // fallback — this is the value returned to the caller for EVERY item, + // including one that dedupes and never reaches an insert. + const messageKey = + writeArgs.messageKey ?? + mailboxKey.transport(messageId, scope.principalId, writeArgs.direction ?? "inbound"); + writeArgs.messageKey = messageKey; + prepared.push({ scope, writeArgs, raw, messageId, messageKey }); } type Inserted = { id: string; scope: MailboxScopeIds }; - const inserted: Inserted[] = await db.transaction(async (tx) => { + const { results, inserted } = await db.transaction(async (tx) => { + const results: DeliveredInboxItem[] = []; const inserted: Inserted[] = []; - for (const { scope, writeArgs, raw, messageId } of prepared) { + for (const { scope, writeArgs, raw, messageId, messageKey } of prepared) { const written = await insertMailboxMessage(tx, writeArgs, raw, messageId); - if (written === null) continue; + if (written === null) { + results.push({ messageKey, id: null }); + continue; + } + results.push({ messageKey, id: written.id }); inserted.push({ id: written.id, scope }); } - return inserted; + return { results, inserted }; }); // Post-commit only: live signals for newly inserted ids, one per row. @@ -628,5 +667,5 @@ export async function writeMailboxMessages( } } - return inserted.map((row) => row.id); + return results; } From af0a1e1222dced0aa4caf21181edb3134c83a072 Mon Sep 17 00:00:00 2001 From: Sawyer Date: Fri, 4 Sep 2026 18:45:36 -0700 Subject: [PATCH 5/5] Update docs: direction-aware write key, outbound unread exclusion (CL-7452) Move the write-path default-key behavior under Changed (it changed shape mid-flight, not merely landed), and document the outbound pre-read ruling, the listUserMailbox/getMailboxMessage direction filter, the writeMailboxMessages return shape, and the Omit-based WriteMailboxMessagesItem in both CHANGELOG.md and ARCHITECTURE.md. --- ARCHITECTURE.md | 46 ++++++++++++++++++++++++++---------- CHANGELOG.md | 62 +++++++++++++++++++++++++++++++++++-------------- 2 files changed, 78 insertions(+), 30 deletions(-) diff --git a/ARCHITECTURE.md b/ARCHITECTURE.md index 6f28f26..65fce6d 100644 --- a/ARCHITECTURE.md +++ b/ARCHITECTURE.md @@ -266,7 +266,13 @@ item (an invalid scope, an oversize frame, a control-plane FK the item's scope does not satisfy) rolls back every row the batch would otherwise have written, including ones already inserted earlier in the same call — same atomicity guarantee as `deliverInboxItems`, over a caller-shaped item instead -of an ingress-shaped one. +of an ingress-shaped one. Each item's `args` is +`Omit` — `scope` is the +sole source of both, so there is no second copy of the scope an item could +disagree with. `writeMailboxMessages` returns one `{ messageKey, id }` entry +per item, in item order — matching `deliverInboxItems`'s `DeliveredInboxItem` +shape — with `id: null` exactly for an item whose messageKey deduped against +an existing row, rather than a filtered array of inserted ids. **A write's Message-ID, direction, and dedupe key are now the caller's to set.** `WriteMailboxMessageArgs.messageId` lets a caller hand the write path @@ -274,14 +280,26 @@ the exact msg-id its own frame must carry (validated as a bracketed msg-id; `RangeError` otherwise) instead of always minting one — needed when a message's id has to be predictable ahead of the write, e.g. so a later `inReplyTo` can reference it. `direction` (default `"inbound"`) is a stored -fact only; it does not change which rows the inbox views serve (still -inbound-only, see Known limits). And `messageKey`, when the caller omits it, -now defaults to `mailboxKey.transport(messageId, principalId)` — the same -`transport:mid::` shape `persist.ts`'s transport -dual-write already uses — rather than leaving the row unkeyed: a retry that -reuses the same caller-supplied `messageId` therefore dedupes for free, while -two writes that each mint their own `messageId` never collide. A caller- -supplied `messageKey` still overrides the default, exactly as before. +fact: an outbound row is the sender's own durable copy, and is created +already-read — its `mailbox.read_at` is pinned to its own `created_at` at +insert — so it is excluded from the unread count and the unread view without +either needing a direction predicate of its own. `listUserMailbox` and +`getMailboxMessage` accept an optional `direction?: "inbound" | "outbound" | +"all"` (default `"inbound"`, preserving today's contract) so a thread reader +can fetch a principal's own sent copies or both directions together — see +Known limits. And `messageKey`, when the caller omits it, now defaults to +`mailboxKey.transport(messageId, principalId, direction)`: for the default +`"inbound"` direction this is `transport:mid::` — +the same shape `persist.ts`'s transport dual-write already uses, byte for +byte, so a frame `persist.ts` already delivered and a direct inbound write +for the same Message-ID + principal still dedupe onto the same row — while +`"outbound"` gets a `:outbound` suffix, so a sender's own copy of a turn +never collapses onto an inbound copy that reuses the identical +caller-supplied `messageId` for the same principal. A retry that reuses the +same caller-supplied `messageId` (and direction) therefore dedupes for free +— the write returns `null` — while two writes that each mint their own +`messageId`, or that differ in direction, never collide. A caller-supplied +`messageKey` still overrides the default, exactly as before. **Bus publish isolates listeners.** `publishMailboxEvent` invokes each subscriber independently; one throwing listener does not stop the others. SSE @@ -434,9 +452,13 @@ actual mail transport — this package neither sends nor receives SMTP. documented in the package README. - **`sort=priority` pays a cross-table join** on top of a rank that was never index-servable; see the measurements above. -- **List routes read inbound rows.** The `direction` column admits outbound - rows and the write path can create them, but the inbox views are - inbound-only; there is no "sent" view and no send route. +- **List routes default to inbound rows.** The `direction` column admits + outbound rows and the write path can create them; `listUserMailbox` and + `getMailboxMessage` default to `"inbound"` (preserving the mounted route + table's existing behavior) but accept `direction: "outbound" | "all"` for + a caller — a thread reader, not yet a mounted route — that needs a + principal's own sent copies. There is still no "sent" view or send route + on the mounted API itself. - **No search.** Filtering is by view, priority, classification, status and assignee. There is no full-text index over subjects or bodies. - **Reordering the host's `priorities` invalidates in-flight priority diff --git a/CHANGELOG.md b/CHANGELOG.md index 8ab8998..467e1f0 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -20,21 +20,7 @@ always called out under their own heading. already carries refs. A throwing `resolveRefs` follows the existing dual-write contract: logged, upstream unaffected, no mailbox row for that frame. -- **Caller-supplied Message-ID, direction, and message key on writes; a - one-transaction batch write.** `WriteMailboxMessageArgs` gains - `messageId?: string` — when supplied it must be a bracketed msg-id - (validated with `assertMsgId`) and becomes the built frame's own - `Message-ID:` header and the cached `principal_mail.message_id`; - omitted, one is still minted exactly as before. It also gains - `direction?: "inbound" | "outbound"` (default `"inbound"`). Omitting - `messageKey` no longer leaves the row unkeyed: it now defaults to - `mailboxKey.transport(messageId, principalId)`, the same - `transport:mid::` shape `persist.ts`'s - transport dual-write already uses — so retrying a write with the same - caller-supplied `messageId` dedupes without the caller minting its own - key, while two independent (differently-minted) writes never collide. - A caller-supplied `messageKey` still overrides the default. - New `writeMailboxMessages(db, items, opts?)` writes an arbitrary batch +- New `writeMailboxMessages(db, items, opts?)` writes an arbitrary batch of `{ scope, args }` pairs — e.g. a sender's outbound copy alongside every recipient's inbound copy of the same conversation turn — in ONE transaction: every row is scope-checked, field-checked, encoded, and @@ -43,11 +29,18 @@ always called out under their own heading. scope, an oversize frame, an unknown control-plane principal) rolls back the whole batch. Per-row dedupe still runs through `onConflictDoNothing` on the existing `messageKey` partial unique - index, so a retried batch dedupes row-by-row without failing; the - function returns only the ids of rows this call actually inserted. - Bus events publish only after commit, one per written row. This is the + index, so a retried batch dedupes row-by-row without failing. Each + item's `args` omits `tenantId`/`principalId` (`Omit`) — `scope` is the sole source of both, so + there is no second copy of the scope that could disagree with it. Bus + events publish only after commit, one per written row. This is the **conversation path**; `deliverInboxItems` remains the **notify-item path** for ingress adapters and is unchanged. +- `listUserMailbox` and `getMailboxMessage` accept an optional + `direction?: "inbound" | "outbound" | "all"` (default `"inbound"`), so a + thread reader can fetch a principal's own sent copies (`"outbound"`) or + both directions together (`"all"`) alongside the existing inbox-only + default. - **Threading headers on the frame and in the list projection.** `buildMailFrame` accepts `references` — the thread's ancestry, oldest @@ -108,6 +101,39 @@ always called out under their own heading. ### Changed +- **Caller-supplied Message-ID, direction, and message key on writes.** + `WriteMailboxMessageArgs` gains `messageId?: string` — when supplied it + must be a bracketed msg-id (validated with `assertMsgId`) and becomes + the built frame's own `Message-ID:` header and the cached + `principal_mail.message_id`; omitted, one is still minted exactly as + before. It also gains `direction?: "inbound" | "outbound"` (default + `"inbound"`). Omitting `messageKey` no longer leaves the row unkeyed: it + now defaults to `mailboxKey.transport(messageId, principalId, direction)`. + For the default `"inbound"` direction this is + `transport:mid::` — the exact shape + `persist.ts`'s transport dual-write already uses, byte for byte — so a + frame `persist.ts` already delivered and a direct inbound write for the + same Message-ID + principal dedupe onto the same row, as before, and + retrying a write with the same caller-supplied `messageId` dedupes + without the caller minting its own key. `"outbound"` gets a + `:outbound` suffix instead, so a sender's own copy of a turn never + collides with an inbound copy that reuses the identical caller-supplied + `messageId` for the same principal — two independent (differently + keyed) writes still never collide either way. A caller-supplied + `messageKey` still overrides the default. A *colliding* caller + `messageId` (same effective key) is a no-op: the write returns `null` + rather than a second row. + An outbound row is also created already-read — its `mailbox.read_at` is + pinned to its own `created_at` at insert — so it never counts toward + `countUnreadActiveMailbox` or appears in the unread view without either + needing a direction predicate of its own; `listUserMailbox` and + `getMailboxMessage` still default to `"inbound"` only, unaffected by + this. + `writeMailboxMessages` returns `Array<{ messageKey: string; id: string | + null }>`, one entry per item, in item order — matching + `deliverInboxItems`'s `DeliveredInboxItem` shape — rather than a + filtered array of inserted ids; `id` is `null` exactly when that item's + messageKey deduped against an existing row. - **Inbox list no longer loads or decodes full MIME frames.** List selects every `principal_mail` column except `raw`, and projects `subject` / `from` from the denormalized caches only — no list `snippet`, and list `date` / `messageId` /