diff --git a/ARCHITECTURE.md b/ARCHITECTURE.md index 265f584..65fce6d 100644 --- a/ARCHITECTURE.md +++ b/ARCHITECTURE.md @@ -252,6 +252,55 @@ 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. 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 +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: 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 connections serialize writes, bound the pending queue, and close on overflow or @@ -403,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 77a70fc..467e1f0 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -20,6 +20,28 @@ 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. +- 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. 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 first — and emits it as a folded `References:` header; `In-Reply-To` @@ -79,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` / 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 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 6ddb89a..565d1f4 100644 --- a/src/write.test.ts +++ b/src/write.test.ts @@ -1,12 +1,15 @@ 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 { countUnreadActiveMailbox } from "./mutations.js"; import { createInMemoryMailboxEventBus } from "./bus.js"; import { withTestDb, seedScope, TEST_VOCABULARY } from "./test-helpers.js"; import type { MailboxDb } from "./db.js"; @@ -827,3 +830,532 @@ 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: { + address: "p1@t1.example", + fromAddress: "p1@t1.example", + subject: "Sent", + body: "Hi p2", + direction: "outbound", + }, + }, + { + scope: { tenantId: "t1", principalId: "p1" }, + args: { + address: "p1@t1.example", + fromAddress: "p1@t1.example", + subject: "Recv", + body: "Hi p1", + direction: "inbound", + }, + }, + { + scope: { tenantId: "t1", principalId: "p2" }, + args: { + 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: { + address: "p1@t1.example", + fromAddress: "p1@t1.example", + subject: "Sent", + body: "Hi p2", + direction: "outbound", + }, + }, + { + scope: { tenantId: "t1", principalId: "p2" }, + args: { + address: "p2@t1.example", + fromAddress: "p1@t1.example", + subject: "Recv", + body: "Hi p2", + direction: "inbound", + }, + }, + { + scope: { tenantId: "t1", principalId: "nobody-seeded-this" }, + args: { + 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 null ids", async () => { + const messageId1 = ""; + const messageId2 = ""; + const items = [ + { + scope: { tenantId: "t1", principalId: "p1" }, + args: { + 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: { + 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.map((r) => r.id)).toEqual([null, null]); + 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: { + 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.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.map((r) => r.id)).toEqual([null]); + expect(received.length).toBe(1); + }); + + test("a messageKey override is honored inside a batch", async () => { + const results = await writeMailboxMessages(db, [ + { + scope: { tenantId: "t1", principalId: "p1" }, + args: { + address: "p1@t1.example", + fromAddress: "p1@t1.example", + subject: "Hello", + body: "World", + messageKey: "custom:batch-key", + }, + }, + ]); + 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 = ${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 3e092fd..9b12bd5 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, @@ -235,9 +270,12 @@ async function insertMailboxMessage( args: WriteMailboxMessageArgs, raw: Uint8Array, messageId: string, -): Promise<{ id: string } | null> { +): Promise<{ id: string; messageKey: string } | null> { assertMailboxFrameBytes(raw); - const refs = boundRefs(args.refs, args.messageKey ?? null); + 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 // the unread count are then plain operations on `mailbox`, and the unread @@ -251,11 +289,11 @@ async function insertMailboxMessage( tenantId: args.tenantId, principalId: args.principalId, address: args.address, - direction: "inbound", + direction, raw: Buffer.from(raw), subject: args.subject, fromAddress: args.fromAddress, - messageKey: args.messageKey ?? null, + messageKey, messageId, inReplyTo: args.inReplyTo ?? null, refs: refs ?? null, @@ -268,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 }; } /** @@ -350,11 +394,34 @@ 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`. 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, + direction: "inbound" | "outbound" = "inbound", + ) => + direction === "outbound" + ? `transport:mid:${messageId}:${principalId}:outbound` + : `transport:mid:${messageId}:${principalId}`, } as const; export type InboxItem = { @@ -499,3 +566,106 @@ export async function deliverInboxItems( return results; } + +/** + * 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: Omit; +}; + +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 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 { + type Prepared = { + scope: MailboxScopeIds; + writeArgs: WriteMailboxMessageArgs; + raw: Uint8Array; + messageId: string; + messageKey: 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); + // 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 { results, inserted } = await db.transaction(async (tx) => { + const results: DeliveredInboxItem[] = []; + const inserted: Inserted[] = []; + for (const { scope, writeArgs, raw, messageId, messageKey } of prepared) { + const written = await insertMailboxMessage(tx, writeArgs, raw, messageId); + if (written === null) { + results.push({ messageKey, id: null }); + continue; + } + results.push({ messageKey, id: written.id }); + inserted.push({ id: written.id, scope }); + } + return { results, 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 results; +}