From 25358ab0cce67fba3a1656cceb8a07feac62d654 Mon Sep 17 00:00:00 2001 From: Sawyer Date: Fri, 4 Sep 2026 16:33:00 -0700 Subject: [PATCH 01/12] Add tests for References and the cached threading headers A frame is frozen at rest, so the properties worth pinning are the ones nobody can fix later: a three-deep References chain round-trips oldest first and folds inside the 78-character line limit, a value that is not a bracketed msg-id is refused, the list projection serves both headers without loading raw, the persist path caches them off a frame it did not build, and 0002 backfills legacy rows whose columns are null - including one whose raw is not valid UTF-8, which must not abort the statement. --- src/frame.test.ts | 58 ++++++++++++++++++++++++++++++ src/migrations.test.ts | 77 +++++++++++++++++++++++++++++++++++++++- src/persist.test.ts | 42 ++++++++++++++++++++++ src/read.test.ts | 63 ++++++++++++++++++++++++++++++-- src/schema-check.test.ts | 4 +-- 5 files changed, 238 insertions(+), 6 deletions(-) diff --git a/src/frame.test.ts b/src/frame.test.ts index 77de3f8..2f83c2a 100644 --- a/src/frame.test.ts +++ b/src/frame.test.ts @@ -66,6 +66,64 @@ describe("buildMailFrame headers", () => { expect(h.has("bcc")).toBe(false); }); + test("References is emitted oldest first, folded, and round-trips", () => { + // A three-deep chain is the shortest one where order and folding both + // matter: a client walks References oldest-first to place a reply under its + // root, and RFC 2822 caps a header line at 78 characters — which a chain + // passes within a few ids. + const chain = [ + "", + "", + "", + ]; + const raw = frame({ references: chain, inReplyTo: chain[2] }); + const text = new TextDecoder().decode(raw); + // Folded: continuation lines begin with the single space RFC 2822 requires. + expect(text).toContain("\r\n <"); + for (const line of text.split("\r\n")) { + expect(line.length).toBeLessThanOrEqual(78); + } + const decoded = decodeMailFrame(raw); + expect(decoded?.references).toEqual(chain); + // In-Reply-To stays a SINGLE msg-id — the immediate parent, not the chain. + expect(decoded?.inReplyTo).toBe(chain[2]!); + expect(decoded?.messageId).toBe(""); + }); + + test("no References header when the chain is absent or empty", () => { + expect(headers(frame()).has("references")).toBe(false); + expect(headers(frame({ references: [] })).has("references")).toBe(false); + expect(decodeMailFrame(frame())?.references).toEqual([]); + expect(decodeMailFrame(frame())?.inReplyTo).toBeNull(); + }); + + test("a frame with no Message-ID decodes to a null messageId", () => { + const raw = new TextEncoder().encode( + "From: a@b.c\r\nTo: d@e.f\r\nSubject: s\r\n\r\nbody\r\n", + ); + const decoded = decodeMailFrame(raw); + expect(decoded?.messageId).toBeNull(); + expect(decoded?.inReplyTo).toBeNull(); + expect(decoded?.references).toEqual([]); + }); + + test("a threading value that is not a bracketed msg-id is refused", () => { + // A frame is frozen at rest: an unthreadable header written today is + // unthreadable forever, so this is refused at the builder rather than + // stored and discovered by an MTA. + expect(() => frame({ messageId: "fixed-id@example.com" })).toThrow( + RangeError, + ); + expect(() => frame({ messageId: "<>" })).toThrow( + RangeError, + ); + expect(() => frame({ inReplyTo: "parent@example.com" })).toThrow(RangeError); + expect(() => + frame({ references: ["", "not-an-id"] }), + ).toThrow(RangeError); + expect(() => frame({ references: [""] })).toThrow(RangeError); + }); + test("the body round-trips through decodeMailFrame", () => { const decoded = decodeMailFrame(frame({ body: "line one\nline two" })); expect(decoded?.body).toBe("line one\nline two"); diff --git a/src/migrations.test.ts b/src/migrations.test.ts index 294ec40..36c05a3 100644 --- a/src/migrations.test.ts +++ b/src/migrations.test.ts @@ -16,6 +16,7 @@ import { MigrationChecksumError, runMailboxMigrations, } from "./migrations.js"; +import { buildMailFrame } from "./frame.js"; import { createHostControlPlane, seedScope, @@ -78,6 +79,8 @@ describe("runMailboxMigrations", () => { "direction", "from_address", "id", + "in_reply_to", + "message_id", "message_key", "principal_id", "raw", @@ -263,6 +266,74 @@ describe("runMailboxMigrations", () => { }); }); + test("0002 backfills the threading headers from legacy rows' raw", async () => { + // The state every already-deployed host is in at upgrade: rows written + // before the cached columns existed, so `raw` carries the headers and the + // columns are NULL. Without the backfill, threading would silently begin at + // the upgrade and every older message would project no parent. + await fromEmpty(async ({ db }) => { + // Build the pre-0002 schema, then seed through it. + await runMailboxMigrations(db); + await db.execute( + sql`ALTER TABLE "mailbox"."principal_mail" + DROP COLUMN "message_id", DROP COLUMN "in_reply_to"`, + ); + await db.execute( + sql`DELETE FROM "mailbox"."corbits_mailbox_migrations" + WHERE "id" = '0002_mail_threading_headers'`, + ); + await seedScope(db, "acme", "user-1"); + + const threaded = buildMailFrame({ + from: "bot@acme.example", + to: "user-1@acme.example", + subject: "Re: legacy", + body: "Body", + messageId: "", + inReplyTo: "", + references: ["", ""], + }); + // A frame with neither header, and one whose bytes are not valid UTF-8: + // both must survive the backfill statement rather than abort it. + const headerless = new TextEncoder().encode( + "From: bot@acme.example\r\nSubject: no ids\r\n\r\nBody\r\n", + ); + const invalidUtf8 = Uint8Array.from([ + ...new TextEncoder().encode("From: bot@acme.example\r\nMessage-ID: \r\n\r\n"), + 0xff, + 0xfe, + ]); + for (const [key, raw] of [ + ["legacy-threaded", threaded], + ["legacy-headerless", headerless], + ["legacy-invalid-utf8", invalidUtf8], + ] as const) { + await db.execute(sql` + INSERT INTO "mailbox"."principal_mail" + ("tenant_id","principal_id","address","direction","raw","message_key") + VALUES ('acme','user-1','user-1@acme.example','inbound', + ${Buffer.from(raw)}, ${key}) + `); + } + + await runMailboxMigrations(db); + + const rows = await db.execute<{ + message_key: string; + message_id: string | null; + in_reply_to: string | null; + }>(sql`SELECT "message_key", "message_id", "in_reply_to" + FROM "mailbox"."principal_mail" ORDER BY "message_key"`); + expect( + rows.map((r) => [r.message_key, r.message_id, r.in_reply_to]), + ).toEqual([ + ["legacy-headerless", null, null], + ["legacy-invalid-utf8", "", null], + ["legacy-threaded", "", ""], + ]); + }); + }); + test("is idempotent: running twice does not error and applies once", async () => { await fromEmpty(async ({ db }) => { await runMailboxMigrations(db); @@ -274,6 +345,7 @@ describe("runMailboxMigrations", () => { ); expect(rows.map((r) => [r.id, r.count])).toEqual([ ["0001_principal_mailbox", "1"], + ["0002_mail_threading_headers", "1"], ]); }); }); @@ -477,7 +549,10 @@ describe("runMailboxMigrations under concurrent cold start", () => { const ledger = await admin.unsafe( `SELECT id FROM mailbox.corbits_mailbox_migrations ORDER BY id`, ); - expect(ledger.map((r) => r.id)).toEqual(["0001_principal_mailbox"]); + expect(ledger.map((r) => r.id)).toEqual([ + "0001_principal_mailbox", + "0002_mail_threading_headers", + ]); }); test("a second wave against an already-migrated schema is a no-op for all", async () => { diff --git a/src/persist.test.ts b/src/persist.test.ts index dfb3f3b..f1ee30c 100644 --- a/src/persist.test.ts +++ b/src/persist.test.ts @@ -96,6 +96,48 @@ describe("sender auth", () => { expect(await rowsFor("acme", "user-2")).toHaveLength(1); }); + test("the inbound frame's threading headers are cached on the row", async () => { + // The persist path receives a frame it did not build, so the caches come + // off the decode — the list projection reads them without loading `raw`. + const { upstream } = recordingUpstream(); + const persist = createMailboxPersist(db, { + upstream, + authorizeSender: () => ACTIVE, + }); + + await persist( + args({ + raw: buildMailFrame({ + from: SENDER, + to: "usr_user-1@acme.example", + subject: "Re: run", + body: "Body", + messageId: "", + inReplyTo: "", + references: ["", ""], + }), + }), + ); + + const [row] = await rowsFor("acme", "user-1"); + expect(row?.messageId).toBe(""); + expect(row?.inReplyTo).toBe(""); + }); + + test("a frame with no threading headers leaves both caches NULL", async () => { + const { upstream } = recordingUpstream(); + const persist = createMailboxPersist(db, { + upstream, + authorizeSender: () => ACTIVE, + }); + + await persist(args()); + + const [row] = await rowsFor("acme", "user-1"); + expect(row?.messageId).toBe(""); + expect(row?.inReplyTo).toBeNull(); + }); + test("an unauthorized sender writes NO mailbox row but is still delegated upstream", async () => { // The reference predicate is "active instance only": a sender the host // cannot resolve to a live instance is refused here. diff --git a/src/read.test.ts b/src/read.test.ts index c906452..c10ce14 100644 --- a/src/read.test.ts +++ b/src/read.test.ts @@ -51,6 +51,63 @@ describe("listUserMailbox", () => { expect(page.items[0]?.subject).toBe("For p1"); }); + test("carries the threading headers from the cached columns", async () => { + // The list path never selects `raw`, so a client can only thread a page if + // these come off the cached columns — which is why they are cached at all. + const written = await writeMailboxMessage(db, { + tenantId: "t1", + principalId: "p1", + address: "p1@t1.example", + fromAddress: "a@t1.example", + subject: "Re: thread", + body: "Body", + messageKey: "threaded", + inReplyTo: "", + references: ["", ""], + }); + const page = await listUserMailbox(db, { + priorities: TEST_VOCABULARY.priorities, + tenantId: "t1", + principalId: "p1", + limit: 50, + view: "all", + }); + const item = page.items[0]; + expect(item?.inReplyTo).toBe(""); + // A real minted Message-ID, not the row id fallback. + expect(item?.messageId).toMatch(/^<[^<>]+@t1\.example>$/); + expect(item?.messageId).not.toBe(written!.id); + + // Detail re-derives both from the frame and agrees with the list. + const detail = await getMailboxMessage(db, { + tenantId: "t1", + principalId: "p1", + id: written!.id, + }); + expect(detail?.inReplyTo).toBe(""); + expect(detail?.messageId).toBe(item?.messageId ?? ""); + }); + + test("omits inReplyTo for a message that is not a reply", async () => { + await writeMailboxMessage(db, { + tenantId: "t1", + principalId: "p1", + address: "p1@t1.example", + fromAddress: "a@t1.example", + subject: "Root", + body: "Body", + messageKey: "root", + }); + const page = await listUserMailbox(db, { + priorities: TEST_VOCABULARY.priorities, + tenantId: "t1", + principalId: "p1", + limit: 50, + view: "all", + }); + expect(page.items[0]?.inReplyTo).toBeUndefined(); + }); + test("keyset pagination: limit+1 detects hasMore and mints nextCursor", async () => { for (let i = 0; i < 3; i++) { await writeMailboxMessage(db, { @@ -245,12 +302,12 @@ describe("getMailboxMessage", () => { view: "all", }); const item = page.items.find((m) => m.id === written!.id); - // List never decodes the frame, so no snippet and no body. - // messageId falls back to the row id (no Message-ID header without raw). + // List never decodes the frame, so no snippet and no body. messageId comes + // off the cached column — the minted id, not the row id fallback. expect(item?.snippet).toBeUndefined(); expect(item?.subject).toBe("Long"); expect(item?.from).toBe("a@t1.example"); - expect(item?.messageId).toBe(written!.id); + expect(item?.messageId).toMatch(/^<[^<>]+@t1\.example>$/); expect(item?.to).toEqual(["p1@t1.example"]); const detail = await getMailboxMessage(db, { diff --git a/src/schema-check.test.ts b/src/schema-check.test.ts index abe8d15..4d7f797 100644 --- a/src/schema-check.test.ts +++ b/src/schema-check.test.ts @@ -12,7 +12,7 @@ import { afterAll, beforeAll, describe, expect, test } from "bun:test"; import postgres from "postgres"; import { drizzle } from "drizzle-orm/postgres-js"; import { sql } from "drizzle-orm"; -import { runMailboxMigrations } from "./migrations.js"; +import { MIGRATIONS, runMailboxMigrations } from "./migrations.js"; import { expectedColumnTypes, SchemaTypeMismatchError, @@ -120,7 +120,7 @@ describe("boot against a host table this package did not create", () => { test("a fresh, correct database boots and records the migration", async () => { await inFreshSchema("mbx_check_ok", async ({ db }) => { await runMailboxMigrations(db); - expect(await ledgerRows("mailbox")).toBe(1); + expect(await ledgerRows("mailbox")).toBe(MIGRATIONS.length); }); }); From 6613465bd6a7836039a2058fdc93e4023ba60927 Mon Sep 17 00:00:00 2001 From: Sawyer Date: Fri, 4 Sep 2026 16:33:09 -0700 Subject: [PATCH 02/12] References on frames and threading headers in the list projection An inbox page never loads raw, so a client could not thread it: the only place Message-ID and In-Reply-To lived was inside the frame. They join subject and from_address as cached columns, written on all three durable paths and backfilled from raw by 0002 so threading does not begin at the upgrade. References is emitted folded and oldest first, the order RFC 2822 defines and every threading client walks; In-Reply-To stays the single immediate parent. Every threading value is validated as a bracketed msg-id at the builder, which is the last point a caller can still be blamed - past it the bytes are frozen and an unthreadable header is unthreadable forever. --- src/frame.ts | 85 +++++++++++++++++++++++++++++++++++++++++++++-- src/migrations.ts | 41 +++++++++++++++++++++++ src/persist.ts | 5 ++- src/read.ts | 12 ++++++- src/schema.ts | 8 +++++ src/write.ts | 43 +++++++++++++++++++----- 6 files changed, 182 insertions(+), 12 deletions(-) diff --git a/src/frame.ts b/src/frame.ts index 90e21be..cab6891 100644 --- a/src/frame.ts +++ b/src/frame.ts @@ -35,6 +35,55 @@ export function generateMailboxMessageId(fromAddress: string): string { return `<${crypto.randomUUID()}@${domain === "" ? MESSAGE_ID_FALLBACK_DOMAIN : domain}>`; } +/** + * A msg-id as every threading header carries it: exactly one pair of angle + * brackets, an `@`, and no whitespace. The same shape `@intx/mime`'s + * `generateMessageId` and `generateMailboxMessageId` return. + */ +const MSG_ID = /^<[^<>\s]+@[^<>\s]+>$/; + +/** + * Refuse a threading header value that is not a bracketed msg-id. + * + * `RangeError` for the same reason the frame-byte cap throws it: a caller that + * hands this builder `uuid@host` (or `<>`) has a bug, and a frame is + * frozen at rest — an unthreadable `References:` written today is unthreadable + * forever. Rejecting at the builder is the last point where the caller can + * still be blamed precisely. + */ +export function assertMsgId(value: string, field: string): void { + if (!MSG_ID.test(value)) { + throw new RangeError( + `mailbox frame ${field} must be a bracketed msg-id (), got ${JSON.stringify(value)}`, + ); + } +} + +// RFC 2822 §2.1.1 caps a line at 78 characters; a References chain on a deep +// thread passes that within a handful of ids. Folded before each id that would +// overflow, with the single leading space RFC 2822 §2.2.3 requires — an +// unfolding parser joins the pieces back into one space-separated value. +const HEADER_LINE_MAX = 78; + +function foldReferences(references: readonly string[]): string { + const lines: string[] = ["References:"]; + for (const reference of references) { + const current = lines[lines.length - 1]!; + if (current.length + 1 + reference.length > HEADER_LINE_MAX) { + lines.push(` ${reference}`); + continue; + } + lines[lines.length - 1] = `${current} ${reference}`; + } + return lines.join("\r\n"); +} + +/** Split an unfolded `References:` value into its msg-ids, oldest first. */ +export function parseMsgIdList(value: string | undefined): string[] { + if (value === undefined) return []; + return value.match(/<[^<>]+>/g) ?? []; +} + export type MailFrameArgs = { from: string; to: string; @@ -50,6 +99,14 @@ export type MailFrameArgs = { messageId: string; /** Also a complete msg-id, brackets included. */ inReplyTo?: string; + /** + * The thread's ancestry, OLDEST FIRST — the order RFC 2822 §3.6.4 defines + * and every threading client walks. Each entry is a complete msg-id, + * brackets included. `In-Reply-To` stays a single msg-id (the immediate + * parent); this is the whole chain, and the two are independent — a caller + * supplying one is not obliged to supply the other. + */ + references?: string[]; }; /** @@ -61,15 +118,27 @@ export type MailFrameArgs = { */ export function buildMailFrame(args: MailFrameArgs): Uint8Array { const from = headerValue(args.from); + const messageId = headerValue(args.messageId); + assertMsgId(messageId, "messageId"); const headers = [ `From: ${from}`, `To: ${headerValue(args.to)}`, `Subject: ${headerValue(args.subject)}`, `Date: ${formatRFC2822Date(new Date())}`, - `Message-ID: ${headerValue(args.messageId)}`, + `Message-ID: ${messageId}`, ]; if (args.inReplyTo !== undefined) { - headers.push(`In-Reply-To: ${headerValue(args.inReplyTo)}`); + const inReplyTo = headerValue(args.inReplyTo); + assertMsgId(inReplyTo, "inReplyTo"); + headers.push(`In-Reply-To: ${inReplyTo}`); + } + if (args.references !== undefined && args.references.length > 0) { + const references = args.references.map((reference) => { + const value = headerValue(reference); + assertMsgId(value, "references entry"); + return value; + }); + headers.push(foldReferences(references)); } const body = args.body.replace(/\r?\n/g, "\r\n"); return new TextEncoder().encode(`${headers.join("\r\n")}\r\n\r\n${body}\r\n`); @@ -78,6 +147,15 @@ export function buildMailFrame(args: MailFrameArgs): Uint8Array { export type DecodedFrame = { headers: Map; body: string; + /** + * The threading headers, parsed once here rather than re-derived by every + * reader. `messageId` and `inReplyTo` are null when the frame carries no such + * header; `references` is `[]`, oldest first, for the same case — a chain of + * no ancestors is an empty chain, not an absent one. + */ + messageId: string | null; + inReplyTo: string | null; + references: string[]; }; function normalizeMailText(bytes: Uint8Array): string { @@ -144,5 +222,8 @@ export function decodeMailFrame(raw: Uint8Array): DecodedFrame | null { return { headers: parsed.headers, body: extractFrameBody(raw, parsed.headers, parsed.bodyOffset), + messageId: parsed.headers.get("message-id") ?? null, + inReplyTo: parsed.headers.get("in-reply-to") ?? null, + references: parseMsgIdList(parsed.headers.get("references")), }; } diff --git a/src/migrations.ts b/src/migrations.ts index 20d6407..8b0f592 100644 --- a/src/migrations.ts +++ b/src/migrations.ts @@ -135,6 +135,47 @@ export const MIGRATIONS: Migration[] = [ WHERE "trashed_at" IS NOT NULL`, ], }, + { + // The threading headers, cached alongside `subject` and `from_address` so + // the list projection can render a thread without loading `raw`. Additive + // and nullable: nothing about an existing row becomes invalid, and a frame + // carrying neither header keeps both columns NULL. + id: "0002_mail_threading_headers", + statements: [ + sql`ALTER TABLE "mailbox"."principal_mail" + ADD COLUMN IF NOT EXISTS "message_id" text`, + sql`ALTER TABLE "mailbox"."principal_mail" + ADD COLUMN IF NOT EXISTS "in_reply_to" text`, + // Backfill from the frozen frame, so rows written before these columns + // existed project the same headers a row written after them does — + // otherwise threading would silently start at the upgrade. + // + // `raw` is decoded as LATIN1 rather than UTF8: LATIN1 accepts every byte + // sequence, and a single non-UTF8 byte anywhere in one frame would + // otherwise abort the whole statement. Header names and msg-ids are + // ASCII, so nothing is lost. Only the header section is searched (the + // text before the first blank line), so a body line that happens to + // begin `Message-ID:` cannot be mistaken for a header. + sql`UPDATE "mailbox"."principal_mail" AS pm + SET "message_id" = h."message_id", "in_reply_to" = h."in_reply_to" + FROM ( + SELECT "id", + substring(head from '(?ni)^Message-ID:[[:space:]]*(<[^<>]+>)') AS "message_id", + substring(head from '(?ni)^In-Reply-To:[[:space:]]*(<[^<>]+>)') AS "in_reply_to" + FROM ( + SELECT "id", + split_part( + replace(convert_from("raw", 'LATIN1'), chr(13) || chr(10), chr(10)), + chr(10) || chr(10), 1 + ) AS head + FROM "mailbox"."principal_mail" + WHERE "message_id" IS NULL AND "in_reply_to" IS NULL + ) heads + ) h + WHERE pm."id" = h."id" + AND (h."message_id" IS NOT NULL OR h."in_reply_to" IS NOT NULL)`, + ], + }, ]; const DIALECT = new PgDialect(); diff --git a/src/persist.ts b/src/persist.ts index a3051ef..384abf5 100644 --- a/src/persist.ts +++ b/src/persist.ts @@ -222,7 +222,8 @@ export function createMailboxPersist( const decoded = decodeMailFrame(raw); const subject = decoded?.headers.get("subject") ?? null; const fromAddress = decoded?.headers.get("from") ?? null; - const messageId = decoded?.headers.get("message-id") ?? null; + const messageId = decoded?.messageId ?? null; + const inReplyTo = decoded?.inReplyTo ?? null; // Mail rows and their management rows commit together: the management row // is created eagerly with the message (see `writeMailboxMessage`), and a @@ -242,6 +243,8 @@ export function createMailboxPersist( raw: Buffer.from(raw), subject, fromAddress, + messageId, + inReplyTo, messageKey: transportMessageKey( messageId, raw, diff --git a/src/read.ts b/src/read.ts index dd05e30..9a61154 100644 --- a/src/read.ts +++ b/src/read.ts @@ -279,6 +279,12 @@ export const MailboxMessageSchema = type({ "subject?": "string", date: "string", messageId: "string", + /** + * The immediate parent's msg-id, when the message has one. Served from the + * cached column on the list path and from the frame on detail, so a client + * can thread a page without fetching every message's `raw`. + */ + "inReplyTo?": "string", read: "boolean", "snippet?": "string", "refs?": MailboxRefArraySchema, @@ -402,11 +408,15 @@ function toMailboxMessage( from: headers?.get("from") ?? row.fromAddress ?? "", to, date: toISODate(headers?.get("date"), row.createdAt), - messageId: headers?.get("message-id") ?? row.id, + // header -> cached column -> the row id. The row id is the last resort, not + // the cache: a frame with no Message-ID still needs a stable handle. + messageId: headers?.get("message-id") ?? row.messageId ?? row.id, read: row.readAt !== null, }; const subject = headers?.get("subject") ?? row.subject ?? undefined; if (subject !== undefined) message.subject = subject; + const inReplyTo = headers?.get("in-reply-to") ?? row.inReplyTo ?? undefined; + if (inReplyTo !== undefined) message.inReplyTo = inReplyTo; if (decoded !== null && decoded.body.length > 0) { message.snippet = decoded.body.slice(0, SNIPPET_MAX_CHARS); } diff --git a/src/schema.ts b/src/schema.ts index 8f6c58f..7603fb6 100644 --- a/src/schema.ts +++ b/src/schema.ts @@ -91,6 +91,14 @@ export const principalMail = mailboxPgSchema.table( subject: text("subject"), fromAddress: text("from_address"), messageKey: text("message_key"), + // The threading headers, cached off `raw` at write for the SAME reason + // `subject` and `from_address` are: list never loads the MIME frame, so a + // client renders a thread from the projection alone instead of fetching + // every message's `raw` to read two headers. `raw` stays authoritative — + // detail re-derives both from the frame. Nullable because a frame the MIME + // parser rejects, or one carrying no such header, still persists. + messageId: text("message_id"), + inReplyTo: text("in_reply_to"), // Plain `jsonb` with NO `$type()`. A `$type` here is a claim // the column cannot keep: nothing in Postgres constrains this blob's shape, // and a row written by an older version (or by the host directly) will diff --git a/src/write.ts b/src/write.ts index 3f98e4f..bdf20c3 100644 --- a/src/write.ts +++ b/src/write.ts @@ -112,6 +112,12 @@ export type WriteMailboxMessageArgs = { /** Idempotency key; a second write with the same key is a no-op (returns null). */ messageKey?: string; inReplyTo?: string; + /** + * The thread's ancestry, oldest first; each entry a bracketed msg-id. Emitted + * as a folded `References:` header on the frame this write builds — see + * `buildMailFrame`. `RangeError` on an entry that is not a bracketed msg-id. + */ + references?: string[]; refs?: MailboxRef[]; /** * Triage known at write time. Values are the HOST's vocabulary — this @@ -135,8 +141,16 @@ type MailboxInsertTx = { /** * Encode args into a durable MIME frame. Mint a fresh Message-ID each call. + * + * 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. */ -function encodeMailboxFrame(args: WriteMailboxMessageArgs): Uint8Array { +function encodeMailboxFrame(args: WriteMailboxMessageArgs): { + raw: Uint8Array; + messageId: string; +} { const messageId = generateMailboxMessageId(args.fromAddress); const frameArgs: Parameters[0] = { from: args.fromAddress, @@ -146,7 +160,8 @@ function encodeMailboxFrame(args: WriteMailboxMessageArgs): Uint8Array { messageId, }; if (args.inReplyTo !== undefined) frameArgs.inReplyTo = args.inReplyTo; - return buildMailFrame(frameArgs); + if (args.references !== undefined) frameArgs.references = args.references; + return { raw: buildMailFrame(frameArgs), messageId }; } /** @@ -160,9 +175,11 @@ function assertMailboxStringFieldsFit(args: { fromAddress: string; address: string; inReplyTo?: string; + references?: string[]; }): void { const fields = [args.body, args.subject, args.fromAddress, args.address]; if (args.inReplyTo !== undefined) fields.push(args.inReplyTo); + if (args.references !== undefined) fields.push(...args.references); for (const field of fields) { if (Buffer.byteLength(field) >= MAX_MAILBOX_FRAME_BYTES) { throw new RangeError( @@ -183,6 +200,7 @@ async function insertMailboxMessage( tx: MailboxInsertTx, args: WriteMailboxMessageArgs, raw: Uint8Array, + messageId: string, ): Promise<{ id: string } | null> { assertMailboxFrameBytes(raw); const refs = boundRefs(args.refs, args.messageKey ?? null); @@ -204,6 +222,8 @@ async function insertMailboxMessage( subject: args.subject, fromAddress: args.fromAddress, messageKey: args.messageKey ?? null, + messageId, + inReplyTo: args.inReplyTo ?? null, refs: refs ?? null, }) .onConflictDoNothing({ @@ -257,11 +277,11 @@ export async function writeMailboxMessage( assertMailboxStringFieldsFit(args); // Encode and size-check the built frame before opening a transaction so // oversize input never pays for a begin/rollback. - const raw = encodeMailboxFrame(args); + const { raw, messageId } = encodeMailboxFrame(args); assertMailboxFrameBytes(raw); // One transaction for the mail row and its management row. const row = await db.transaction(async (tx) => - insertMailboxMessage(tx, args, raw), + insertMailboxMessage(tx, args, raw, messageId), ); if (!row) return null; @@ -311,6 +331,10 @@ export type InboxItem = { body: string; source: string; externalId: string; + /** The immediate parent's msg-id, brackets included. */ + inReplyTo?: string; + /** The thread's ancestry, oldest first; see `WriteMailboxMessageArgs`. */ + references?: string[]; refs?: MailboxRef[]; // An adapter that already knows an item's triage verdict stamps it // at delivery rather than writing the row and immediately updating it. @@ -363,6 +387,7 @@ export async function deliverInboxItems( messageKey: string; writeArgs: WriteMailboxMessageArgs; raw: Uint8Array; + messageId: string; }; const prepared: Prepared[] = []; for (const item of items) { @@ -378,23 +403,25 @@ export async function deliverInboxItems( body: item.body, messageKey, }; + if (item.inReplyTo !== undefined) writeArgs.inReplyTo = item.inReplyTo; + if (item.references !== undefined) writeArgs.references = item.references; if (item.refs !== undefined) writeArgs.refs = item.refs; if (item.priority !== undefined) writeArgs.priority = item.priority; if (item.classification !== undefined) { writeArgs.classification = item.classification; } if (item.status !== undefined) writeArgs.status = item.status; - const raw = encodeMailboxFrame(writeArgs); + const { raw, messageId } = encodeMailboxFrame(writeArgs); assertMailboxFrameBytes(raw); - prepared.push({ item, messageKey, writeArgs, raw }); + prepared.push({ item, messageKey, writeArgs, raw, messageId }); } type Inserted = { id: string; item: InboxItem }; const { results, inserted } = await db.transaction(async (tx) => { const results: DeliveredInboxItem[] = []; const inserted: Inserted[] = []; - for (const { item, messageKey, writeArgs, raw } of prepared) { - const written = await insertMailboxMessage(tx, writeArgs, raw); + for (const { item, messageKey, writeArgs, raw, messageId } of prepared) { + const written = await insertMailboxMessage(tx, writeArgs, raw, messageId); if (written === null) { results.push({ messageKey, id: null }); continue; From 34eafa6c51963987e9205c093d3faefde7ae2a0d Mon Sep 17 00:00:00 2001 From: Sawyer Date: Fri, 4 Sep 2026 16:33:09 -0700 Subject: [PATCH 03/12] Update docs: threading headers on principal_mail The data model block listed every principal_mail column, so it now lists the two cached threading columns as well. --- ARCHITECTURE.md | 3 ++- CHANGELOG.md | 19 +++++++++++++++++++ 2 files changed, 21 insertions(+), 1 deletion(-) diff --git a/ARCHITECTURE.md b/ARCHITECTURE.md index 7219477..2b41d04 100644 --- a/ARCHITECTURE.md +++ b/ARCHITECTURE.md @@ -106,7 +106,8 @@ never in a database of their own. Every row in either belongs to exactly one ``` principal_mail the message as delivered. IMMUTABLE. id, tenant_id, principal_id, address, direction, raw, - subject, from_address, message_key, refs, created_at + subject, from_address, message_id, in_reply_to, + message_key, refs, created_at mailbox the management layer, keyed by mail id. Mutable. read_at, archived_at, trashed_at (universal) diff --git a/CHANGELOG.md b/CHANGELOG.md index b728206..36a1f76 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -11,6 +11,25 @@ always called out under their own heading. ### Added +- **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` + stays the single immediate parent. Every threading value must be a + bracketed msg-id (``); anything else is a `RangeError` at + the builder, because a frame is frozen at rest and an unthreadable + header written today is unthreadable forever. `decodeMailFrame` now + returns `messageId`, `inReplyTo` (both `string | null`) and + `references` (`string[]`, oldest first) parsed alongside the header map. + `principal_mail` gains `message_id` and `in_reply_to` as cached + columns, populated on the `writeMailboxMessage`, `deliverInboxItems` + and `createMailboxPersist` paths, and `MailboxMessage` gains an + optional `inReplyTo` — so a client threads an inbox page from the list + projection alone, which never loads `raw`. `WriteMailboxMessageArgs` + and `InboxItem` accept `inReplyTo` and `references`. Migration + `0002_mail_threading_headers` adds both columns and backfills them from + each existing row's `raw`, so threading does not silently begin at the + upgrade. + - **Live events name the operation that fired.** `MailboxEvent` gains an optional `op` (`MailboxEventOp`: `create`, `mark_read`, `mark_unread`, `trash`, `archive`, `restore`, `enrich`, `assign`) alongside the existing From 13c9b8651f1cfd77e3e8bc53b50a80f6cdcac497 Mon Sep 17 00:00:00 2001 From: Sawyer Date: Fri, 4 Sep 2026 16:51:28 -0700 Subject: [PATCH 04/12] Add tests for NUL-safe 0002 threading-header backfill --- src/migrations.test.ts | 130 +++++++++++++++++++++++++++++++++++++++++ 1 file changed, 130 insertions(+) diff --git a/src/migrations.test.ts b/src/migrations.test.ts index 36c05a3..2a3a1ec 100644 --- a/src/migrations.test.ts +++ b/src/migrations.test.ts @@ -334,6 +334,136 @@ describe("runMailboxMigrations", () => { }); }); + test("0002 survives a legacy frame with a NUL byte in its body", async () => { + // Postgres `text` cannot hold 0x00 in any encoding — a single legacy + // frame with a NUL anywhere in `raw` used to abort the whole UPDATE (and + // with it the ledger insert), which meant every subsequent boot failed + // forever. This is RED against the pre-fix backfill (LATIN1-decoding the + // entire `raw`, NUL included) and GREEN once only the NUL-stripped header + // slice reaches `convert_from`. + await fromEmpty(async ({ db }) => { + await runMailboxMigrations(db); + await db.execute( + sql`ALTER TABLE "mailbox"."principal_mail" + DROP COLUMN "message_id", DROP COLUMN "in_reply_to"`, + ); + await db.execute( + sql`DELETE FROM "mailbox"."corbits_mailbox_migrations" + WHERE "id" = '0002_mail_threading_headers'`, + ); + await seedScope(db, "acme", "user-1"); + + const enc = new TextEncoder(); + const ok = enc.encode( + "From: a@b.c\r\nMessage-ID: \r\n\r\nBody\r\n", + ); + const nulBody = Uint8Array.from([ + ...enc.encode( + "From: a@b.c\r\nMessage-ID: \r\n" + + "Content-Type: application/octet-stream\r\n\r\n", + ), + 0x00, + 0x41, + ]); + for (const [key, raw] of [ + ["nul-ok", ok], + ["nul-body", nulBody], + ] as const) { + await db.execute(sql` + INSERT INTO "mailbox"."principal_mail" + ("tenant_id","principal_id","address","direction","raw","message_key") + VALUES ('acme','user-1','user-1@acme.example','inbound', + ${Buffer.from(raw)}, ${key}) + `); + } + + await runMailboxMigrations(db); + + const ledger = await db.execute<{ id: string }>( + sql`SELECT "id" FROM "mailbox"."corbits_mailbox_migrations" ORDER BY "id"`, + ); + expect(ledger.map((r) => r.id)).toEqual([ + "0001_principal_mailbox", + "0002_mail_threading_headers", + ]); + + const rows = await db.execute<{ + message_key: string; + message_id: string | null; + }>( + sql`SELECT "message_key", "message_id" FROM "mailbox"."principal_mail" + ORDER BY "message_key"`, + ); + expect(rows.map((r) => [r.message_key, r.message_id])).toEqual([ + ["nul-body", ""], + ["nul-ok", ""], + ]); + }); + }); + + test("0002 backfill agrees with the runtime decoder on non-bracketed and multi-id In-Reply-To", async () => { + // Characterization of the shared rule (see persist.ts): the FIRST + // bracketed msg-id if present, else NULL. `parseMsgIdList` is what the + // runtime path now uses too, so a frame decoded before or after the + // upgrade projects the same cached `in_reply_to`. + await fromEmpty(async ({ db }) => { + await runMailboxMigrations(db); + await db.execute( + sql`ALTER TABLE "mailbox"."principal_mail" + DROP COLUMN "message_id", DROP COLUMN "in_reply_to"`, + ); + await db.execute( + sql`DELETE FROM "mailbox"."corbits_mailbox_migrations" + WHERE "id" = '0002_mail_threading_headers'`, + ); + await seedScope(db, "acme", "user-1"); + const enc = new TextEncoder(); + const cases = [ + [ + "bare", + "From: a@b.c\r\nMessage-ID: \r\nIn-Reply-To: foo@bar\r\n\r\nBody\r\n", + ], + [ + "multi", + "From: a@b.c\r\nMessage-ID: \r\nIn-Reply-To: \r\n\r\nBody\r\n", + ], + [ + "folded", + "From: a@b.c\r\nMessage-ID:\r\n \r\nIn-Reply-To:\r\n\t\r\n\r\nBody\r\n", + ], + [ + "lf", + "From: a@b.c\nMessage-ID: \nIn-Reply-To: \n\nMessage-ID: \nBody\n", + ], + ] as const; + for (const [key, text] of cases) { + await db.execute(sql` + INSERT INTO "mailbox"."principal_mail" + ("tenant_id","principal_id","address","direction","raw","message_key") + VALUES ('acme','user-1','user-1@acme.example','inbound', + ${Buffer.from(enc.encode(text))}, ${key}) + `); + } + await runMailboxMigrations(db); + const rows = await db.execute<{ + message_key: string; + message_id: string | null; + in_reply_to: string | null; + }>( + sql`SELECT "message_key","message_id","in_reply_to" + FROM "mailbox"."principal_mail" ORDER BY "message_key"`, + ); + expect( + rows.map((r) => [r.message_key, r.message_id, r.in_reply_to]), + ).toEqual([ + ["bare", "", null], + ["folded", "", ""], + ["lf", "", ""], + ["multi", "", ""], + ]); + }); + }); + test("is idempotent: running twice does not error and applies once", async () => { await fromEmpty(async ({ db }) => { await runMailboxMigrations(db); From 0b709f453eaf6a8c900cde5c66fd8d0cd276d423 Mon Sep 17 00:00:00 2001 From: Sawyer Date: Fri, 4 Sep 2026 16:51:28 -0700 Subject: [PATCH 05/12] migrations: make the 0002 backfill survive a NUL byte in raw --- src/migrations.ts | 64 +++++++++++++++++++++++++++++++++++++++-------- 1 file changed, 53 insertions(+), 11 deletions(-) diff --git a/src/migrations.ts b/src/migrations.ts index 8b0f592..656a13d 100644 --- a/src/migrations.ts +++ b/src/migrations.ts @@ -150,12 +150,34 @@ export const MIGRATIONS: Migration[] = [ // existed project the same headers a row written after them does — // otherwise threading would silently start at the upgrade. // - // `raw` is decoded as LATIN1 rather than UTF8: LATIN1 accepts every byte - // sequence, and a single non-UTF8 byte anywhere in one frame would - // otherwise abort the whole statement. Header names and msg-ids are - // ASCII, so nothing is lost. Only the header section is searched (the - // text before the first blank line), so a body line that happens to - // begin `Message-ID:` cannot be mistaken for a header. + // The header section is sliced out of `raw` at the BYTEA level, before + // any text conversion runs, by searching for the first blank-line + // separator (`\r\n\r\n`, falling back to a bare `\n\n` for frames that + // never had their line endings normalized) — a body line that happens + // to begin `Message-ID:` still cannot be mistaken for a header, exactly + // as the original text-level split guaranteed. No separator at all + // means no reliable header/body boundary, so the whole frame is + // searched, matching the pre-existing degrade for a headerless frame. + // + // Postgres `text` cannot hold a NUL byte (0x00) in ANY encoding — that is + // a server-side invariant, not an encoding limitation, so choosing UTF8 + // over LATIN1 would not have helped, and Postgres's own `bytea` has no + // `replace`/`regexp_replace` overload to lean on either (PG16). A single + // legacy frame with a NUL anywhere in `raw` (a binary attachment is the + // common case) used to abort the entire UPDATE — and with it, every + // boot, forever, because the migration ledger row for 0002 would never + // commit. + // + // `clean_bytes` rebuilds the (already header-only) bytea slice one byte + // at a time via `get_byte`/`set_byte`, keeping every byte except 0x00, + // so NUL cannot reach `convert_from` at all — the conversion can no + // longer fail on this row. LATIN1 still accepts every remaining byte + // value 0-255, so header names and msg-ids (both ASCII) survive + // unchanged. A NUL is not expected in a real header, so stripping it + // here costs nothing for well-formed mail and only spares the migration + // from a legacy body's bytes it should never have been reading. The + // per-byte scan runs only over the (small, already-sliced) header + // section of rows still missing both cached columns, once, at upgrade. sql`UPDATE "mailbox"."principal_mail" AS pm SET "message_id" = h."message_id", "in_reply_to" = h."in_reply_to" FROM ( @@ -164,12 +186,32 @@ export const MIGRATIONS: Migration[] = [ substring(head from '(?ni)^In-Reply-To:[[:space:]]*(<[^<>]+>)') AS "in_reply_to" FROM ( SELECT "id", - split_part( - replace(convert_from("raw", 'LATIN1'), chr(13) || chr(10), chr(10)), - chr(10) || chr(10), 1 + replace( + convert_from(clean_bytes, 'LATIN1'), + chr(13) || chr(10), chr(10) ) AS head - FROM "mailbox"."principal_mail" - WHERE "message_id" IS NULL AND "in_reply_to" IS NULL + FROM ( + SELECT sliced."id", + COALESCE( + (SELECT string_agg(set_byte(decode('00', 'hex'), 0, b), ''::bytea ORDER BY i) + FROM generate_series(0, octet_length(sliced.head_bytes) - 1) AS i, + LATERAL (SELECT get_byte(sliced.head_bytes, i) AS b) AS byte + WHERE b <> 0), + ''::bytea + ) AS clean_bytes + FROM ( + SELECT "id", + CASE + WHEN position(decode('0d0a0d0a', 'hex') IN "raw") > 0 + THEN substring("raw" FOR position(decode('0d0a0d0a', 'hex') IN "raw") - 1) + WHEN position(decode('0a0a', 'hex') IN "raw") > 0 + THEN substring("raw" FOR position(decode('0a0a', 'hex') IN "raw") - 1) + ELSE "raw" + END AS head_bytes + FROM "mailbox"."principal_mail" + WHERE "message_id" IS NULL AND "in_reply_to" IS NULL + ) sliced + ) cleaned ) heads ) h WHERE pm."id" = h."id" From f5e689adb675628011f243437f2a95e93a6c4dfb Mon Sep 17 00:00:00 2001 From: Sawyer Date: Fri, 4 Sep 2026 16:51:28 -0700 Subject: [PATCH 06/12] Add test for cached inReplyTo agreeing with the frame header --- src/write.test.ts | 40 ++++++++++++++++++++++++++++++++++++++-- 1 file changed, 38 insertions(+), 2 deletions(-) diff --git a/src/write.test.ts b/src/write.test.ts index 83e8abf..6ddb89a 100644 --- a/src/write.test.ts +++ b/src/write.test.ts @@ -6,9 +6,9 @@ import { MAX_MAILBOX_REFS, MAX_MAILBOX_FRAME_BYTES, } from "./write.js"; -import { getMailboxMessage } from "./read.js"; +import { getMailboxMessage, listUserMailbox } from "./read.js"; import { createInMemoryMailboxEventBus } from "./bus.js"; -import { withTestDb, seedScope } from "./test-helpers.js"; +import { withTestDb, seedScope, TEST_VOCABULARY } from "./test-helpers.js"; import type { MailboxDb } from "./db.js"; import { sql } from "drizzle-orm"; @@ -50,6 +50,42 @@ describe("writeMailboxMessage", () => { expect(second).toBeNull(); }); + test("cached inReplyTo agrees between the list and detail projections", async () => { + // The list projection serves `principal_mail.in_reply_to` (the cached + // column); detail serves the header out of `raw`. Before normalizing + // `inReplyTo` once on the way in, an untrimmed caller value was cached + // verbatim while `buildMailFrame` trimmed the same value into the header + // — so the same message projected two different inReplyTo strings + // depending only on which route read it. + const written = await writeMailboxMessage(db, { + tenantId: "t1", + principalId: "p1", + address: "p1@t1.example", + fromAddress: "agent@t1.example", + subject: "ws", + body: "b", + inReplyTo: " ", + }); + expect(written).not.toBeNull(); + + const page = await listUserMailbox(db, { + tenantId: "t1", + principalId: "p1", + limit: 50, + view: "all", + priorities: TEST_VOCABULARY.priorities, + }); + const listed = page.items.find((m) => m.id === written!.id); + const detail = await getMailboxMessage(db, { + tenantId: "t1", + principalId: "p1", + id: written!.id, + }); + + expect(detail?.inReplyTo).toBe(""); + expect(listed?.inReplyTo).toBe(""); + }); + test("a write to a scope the control plane does not know is an FK rejection", async () => { // The FKs are the enforcement: a mailbox that cannot exist is a caller // bug, not a deliverable outcome. From 32164b1efeb95dbc2e44a3c5ef793ee1153359e1 Mon Sep 17 00:00:00 2001 From: Sawyer Date: Fri, 4 Sep 2026 16:51:28 -0700 Subject: [PATCH 07/12] write: normalize inReplyTo/references once so the cache matches the frame --- src/write.ts | 49 ++++++++++++++++++++++++++++++++++++++++++++----- 1 file changed, 44 insertions(+), 5 deletions(-) diff --git a/src/write.ts b/src/write.ts index bdf20c3..f6c2a18 100644 --- a/src/write.ts +++ b/src/write.ts @@ -4,7 +4,11 @@ import { getLogger } from "@intx/log"; import { mailbox, principalMail } from "./schema.js"; import type { MailboxDb } from "./db.js"; import { publishMailboxEvent, type MailboxEventBus } from "./bus.js"; -import { buildMailFrame, generateMailboxMessageId } from "./frame.js"; +import { + buildMailFrame, + generateMailboxMessageId, + headerValue, +} from "./frame.js"; import type { MailboxRef } from "./read.js"; const logger = getLogger(["corbits-mailbox", "write"]); @@ -139,6 +143,33 @@ type MailboxInsertTx = { insert: MailboxDb["insert"]; }; +/** + * Normalize the threading fields once, on the way in, so the value cached in + * `principal_mail.in_reply_to` and the value that ends up in the frame's + * `In-Reply-To:` header are the SAME string. + * + * `buildMailFrame` already runs every threading value through `headerValue` + * before writing it into `raw` — trimmed and newline-flattened. Without this, + * `insertMailboxMessage` cached `args.inReplyTo` untrimmed, so a caller + * passing `" "` produced a row whose list projection (served from + * the cached column) differed from its detail projection (served from the + * frame) for the exact same message. Applying the same normalization here, + * once, before either the cache write or the frame encode, is what keeps them + * in agreement — not two independent trims that could drift apart. + */ +function normalizeThreadingArgs< + T extends { inReplyTo?: string; references?: string[] }, +>(args: T): T { + const normalized: T = { ...args }; + if (args.inReplyTo !== undefined) { + normalized.inReplyTo = headerValue(args.inReplyTo); + } + if (args.references !== undefined) { + normalized.references = args.references.map(headerValue); + } + return normalized; +} + /** * Encode args into a durable MIME frame. Mint a fresh Message-ID each call. * @@ -269,10 +300,11 @@ async function insertMailboxMessage( */ export async function writeMailboxMessage( db: MailboxDb, - args: WriteMailboxMessageArgs, + rawArgs: WriteMailboxMessageArgs, bus?: MailboxEventBus, ): Promise<{ id: string } | null> { - assertMailboxScope(args); + assertMailboxScope(rawArgs); + const args = normalizeThreadingArgs(rawArgs); // Refuse obviously oversize string fields before allocating the full encode. assertMailboxStringFieldsFit(args); // Encode and size-check the built frame before opening a transaction so @@ -411,9 +443,16 @@ export async function deliverInboxItems( writeArgs.classification = item.classification; } if (item.status !== undefined) writeArgs.status = item.status; - const { raw, messageId } = encodeMailboxFrame(writeArgs); + const normalizedWriteArgs = normalizeThreadingArgs(writeArgs); + const { raw, messageId } = encodeMailboxFrame(normalizedWriteArgs); assertMailboxFrameBytes(raw); - prepared.push({ item, messageKey, writeArgs, raw, messageId }); + prepared.push({ + item, + messageKey, + writeArgs: normalizedWriteArgs, + raw, + messageId, + }); } type Inserted = { id: string; item: InboxItem }; From 7943fb014c1480fa76850ebe62ad63cf25d4e927 Mon Sep 17 00:00:00 2001 From: Sawyer Date: Fri, 4 Sep 2026 16:51:28 -0700 Subject: [PATCH 08/12] Add tests for cached inReplyTo on non-bracketed/multi-id headers --- src/persist.test.ts | 41 +++++++++++++++++++++++++++++++++++++++++ 1 file changed, 41 insertions(+) diff --git a/src/persist.test.ts b/src/persist.test.ts index f1ee30c..fdf5455 100644 --- a/src/persist.test.ts +++ b/src/persist.test.ts @@ -138,6 +138,47 @@ describe("sender auth", () => { expect(row?.inReplyTo).toBeNull(); }); + test("caches only the first bracketed msg-id from a non-bracketed or multi-id In-Reply-To", async () => { + // The persist path decodes a frame it never built, so `In-Reply-To` is + // NOT re-validated against `assertMsgId` (an external MTA's headers are + // not this package's frame to reject) — it can be a bare id, several + // ids, or otherwise malformed. The cached column must still agree with + // what migration `0002_mail_threading_headers`'s backfill produces for + // the same header text: the FIRST bracketed msg-id if present, else + // NULL — never the raw header value. + const { upstream } = recordingUpstream(); + const persist = createMailboxPersist(db, { + upstream, + authorizeSender: () => ACTIVE, + }); + const enc = new TextEncoder(); + + await persist( + args({ + raw: enc.encode( + `From: ${SENDER}\r\nTo: usr_user-1@acme.example\r\n` + + "Message-ID: \r\nIn-Reply-To: foo@bar\r\n\r\nBody\r\n", + ), + }), + ); + const [bare] = await rowsFor("acme", "user-1"); + expect(bare?.messageId).toBe(""); + expect(bare?.inReplyTo).toBeNull(); + + await db.execute(sql`DELETE FROM "mailbox"."principal_mail"`); + await persist( + args({ + raw: enc.encode( + `From: ${SENDER}\r\nTo: usr_user-1@acme.example\r\n` + + "Message-ID: \r\nIn-Reply-To: \r\n\r\nBody\r\n", + ), + }), + ); + const [multi] = await rowsFor("acme", "user-1"); + expect(multi?.messageId).toBe(""); + expect(multi?.inReplyTo).toBe(""); + }); + test("an unauthorized sender writes NO mailbox row but is still delegated upstream", async () => { // The reference predicate is "active instance only": a sender the host // cannot resolve to a live instance is refused here. From dd7696f83430ca455adde856f9b3e21d65146a6c Mon Sep 17 00:00:00 2001 From: Sawyer Date: Fri, 4 Sep 2026 16:51:29 -0700 Subject: [PATCH 09/12] persist: cache the first bracketed msg-id from In-Reply-To, like the backfill --- src/persist.ts | 11 +++++++++-- 1 file changed, 9 insertions(+), 2 deletions(-) diff --git a/src/persist.ts b/src/persist.ts index 384abf5..7a85395 100644 --- a/src/persist.ts +++ b/src/persist.ts @@ -21,7 +21,7 @@ import { getLogger } from "@intx/log"; import { hostPrincipal, mailbox, principalMail } from "./schema.js"; import type { MailboxDb } from "./db.js"; import { publishMailboxEvent, type MailboxEventBus } from "./bus.js"; -import { decodeMailFrame } from "./frame.js"; +import { decodeMailFrame, parseMsgIdList } from "./frame.js"; import { resolveMailboxRecipients } from "./recipients.js"; import { assertMailboxScope, @@ -223,7 +223,14 @@ export function createMailboxPersist( const subject = decoded?.headers.get("subject") ?? null; const fromAddress = decoded?.headers.get("from") ?? null; const messageId = decoded?.messageId ?? null; - const inReplyTo = decoded?.inReplyTo ?? null; + // Cache the same shape migration `0002_mail_threading_headers` backfills + // from legacy frames: the first BRACKETED msg-id in `In-Reply-To`, or + // `null` — never the raw header value. An externally delivered frame's + // `In-Reply-To` is not validated on this path (see `assertMsgId`'s + // JSDoc), so it can be a bare id, several ids, or otherwise malformed; + // caching that raw junk would make the cached column disagree with what + // an upgrade's backfill would have produced for the same frame. + const inReplyTo = parseMsgIdList(decoded?.headers.get("in-reply-to"))[0] ?? null; // Mail rows and their management rows commit together: the management row // is created eagerly with the message (see `writeMailboxMessage`), and a From 2f49742e7356d3b228a2c5655c3210b4489e6a59 Mon Sep 17 00:00:00 2001 From: Sawyer Date: Fri, 4 Sep 2026 16:51:29 -0700 Subject: [PATCH 10/12] Add test for a quoted local part in assertMsgId --- src/frame.test.ts | 18 ++++++++++++++++++ 1 file changed, 18 insertions(+) diff --git a/src/frame.test.ts b/src/frame.test.ts index 2f83c2a..a6e8cfb 100644 --- a/src/frame.test.ts +++ b/src/frame.test.ts @@ -124,6 +124,24 @@ describe("buildMailFrame headers", () => { expect(() => frame({ references: [""] })).toThrow(RangeError); }); + test("a quoted local part (RFC 5322 obs-id-left) is accepted", () => { + // `"john doe"@example.com` is a legal (if obsolete-syntax) local part; + // widened rather than documented as a limitation because a quoted local + // part is real, if rare, on host-supplied threading headers. + const h = headers( + frame({ messageId: '<"john doe"@example.com>' }), + ); + expect(h.get("message-id")).toBe('<"john doe"@example.com>'); + expect( + () => frame({ inReplyTo: '<"a b"@example.com>' }), + ).not.toThrow(); + // A quoted part still cannot contain a bare `<` or `>`, and an unescaped + // trailing quote must close the string before the `@`. + expect(() => frame({ messageId: '<"open@example.com>' })).toThrow( + RangeError, + ); + }); + test("the body round-trips through decodeMailFrame", () => { const decoded = decodeMailFrame(frame({ body: "line one\nline two" })); expect(decoded?.body).toBe("line one\nline two"); From 3e591f52b28a1dbee5fba45f5266c19de5485c50 Mon Sep 17 00:00:00 2001 From: Sawyer Date: Fri, 4 Sep 2026 16:51:29 -0700 Subject: [PATCH 11/12] frame: widen assertMsgId to accept a quoted local part (obs-id-left) --- src/frame.ts | 21 ++++++++++++++++++--- 1 file changed, 18 insertions(+), 3 deletions(-) diff --git a/src/frame.ts b/src/frame.ts index cab6891..2742e24 100644 --- a/src/frame.ts +++ b/src/frame.ts @@ -37,10 +37,25 @@ export function generateMailboxMessageId(fromAddress: string): string { /** * A msg-id as every threading header carries it: exactly one pair of angle - * brackets, an `@`, and no whitespace. The same shape `@intx/mime`'s - * `generateMessageId` and `generateMailboxMessageId` return. + * brackets, an `@`, and no whitespace outside a quoted local part. The same + * shape `@intx/mime`'s `generateMessageId` and `generateMailboxMessageId` + * return. + * + * The local part (`id-left`) accepts either a dot-atom token (no `<`, `>`, + * whitespace, or `"`) or an RFC 5322 quoted-string (`"..."`, backslash-escaped + * quotes and backslashes allowed inside) — `<"john doe"@example.com>` is a + * legal msg-id under `obs-id-left` and every mainstream MTA emits and accepts + * it. The domain part (`id-right`) stays a plain dot-atom token: this package + * mints only dot-atom domains, and widening it further is not needed to + * accept ids this package did not author. + * + * This validates the header SHAPE only, never who may claim it: a frame + * arriving on the persist path (`persist.ts`) is never re-validated against + * this regex — its `In-Reply-To`/`References` are cached (or dropped) exactly + * as decoded, because an external MTA's headers are not this package's frame + * to reject. */ -const MSG_ID = /^<[^<>\s]+@[^<>\s]+>$/; +const MSG_ID = /^<(?:[^<>\s"]+|"(?:[^"\\]|\\.)*")@[^<>\s]+>$/; /** * Refuse a threading header value that is not a bracketed msg-id. From b5c620c78a55e4b2cc956b6b5093ded6fa7ffca5 Mon Sep 17 00:00:00 2001 From: Sawyer Date: Fri, 4 Sep 2026 16:51:29 -0700 Subject: [PATCH 12/12] Update docs: NUL-safe backfill, quoted local part, threading cache fixes --- CHANGELOG.md | 14 +++++++++++++- 1 file changed, 13 insertions(+), 1 deletion(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 36a1f76..24dd77c 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -28,7 +28,19 @@ always called out under their own heading. and `InboxItem` accept `inReplyTo` and `references`. Migration `0002_mail_threading_headers` adds both columns and backfills them from each existing row's `raw`, so threading does not silently begin at the - upgrade. + upgrade. The backfill slices the header section out of `raw` at the + `bytea` level and strips any NUL byte from that slice before decoding + it, so a legacy frame with a NUL anywhere in its bytes (a binary + attachment, most commonly) cannot abort the migration. `assertMsgId` + accepts a quoted-string local part (`<"john doe"@example.com>`, + RFC 5322 `obs-id-left`) in addition to a dot-atom one. The cached + `in_reply_to` is normalized once on the way in — trimmed and + newline-flattened the same way `buildMailFrame` normalizes the header — + so the list and detail projections of the same message agree; on the + transport dual-write path (`createMailboxPersist`), `in_reply_to` caches + the first bracketed msg-id found in a decoded `In-Reply-To:` header, or + `null`, matching what the 0002 backfill derives from the same header + text rather than caching an unbracketed or multi-id header verbatim. - **Live events name the operation that fired.** `MailboxEvent` gains an optional `op` (`MailboxEventOp`: `create`, `mark_read`, `mark_unread`,