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..24dd77c 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -11,6 +11,37 @@ 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. 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`, `trash`, `archive`, `restore`, `enrich`, `assign`) alongside the existing diff --git a/src/frame.test.ts b/src/frame.test.ts index 77de3f8..a6e8cfb 100644 --- a/src/frame.test.ts +++ b/src/frame.test.ts @@ -66,6 +66,82 @@ 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("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"); diff --git a/src/frame.ts b/src/frame.ts index 90e21be..2742e24 100644 --- a/src/frame.ts +++ b/src/frame.ts @@ -35,6 +35,70 @@ 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 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]+>$/; + +/** + * 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 +114,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 +133,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 +162,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 +237,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.test.ts b/src/migrations.test.ts index 294ec40..2a3a1ec 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,204 @@ 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("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); @@ -274,6 +475,7 @@ describe("runMailboxMigrations", () => { ); expect(rows.map((r) => [r.id, r.count])).toEqual([ ["0001_principal_mailbox", "1"], + ["0002_mail_threading_headers", "1"], ]); }); }); @@ -477,7 +679,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/migrations.ts b/src/migrations.ts index 20d6407..656a13d 100644 --- a/src/migrations.ts +++ b/src/migrations.ts @@ -135,6 +135,89 @@ 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. + // + // 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 ( + 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", + replace( + convert_from(clean_bytes, 'LATIN1'), + chr(13) || chr(10), chr(10) + ) AS head + 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" + AND (h."message_id" IS NOT NULL OR h."in_reply_to" IS NOT NULL)`, + ], + }, ]; const DIALECT = new PgDialect(); diff --git a/src/persist.test.ts b/src/persist.test.ts index dfb3f3b..fdf5455 100644 --- a/src/persist.test.ts +++ b/src/persist.test.ts @@ -96,6 +96,89 @@ 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("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. diff --git a/src/persist.ts b/src/persist.ts index a3051ef..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, @@ -222,7 +222,15 @@ 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; + // 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 @@ -242,6 +250,8 @@ export function createMailboxPersist( raw: Buffer.from(raw), subject, fromAddress, + messageId, + inReplyTo, messageKey: transportMessageKey( messageId, raw, 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/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-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); }); }); 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.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. diff --git a/src/write.ts b/src/write.ts index 3f98e4f..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"]); @@ -112,6 +116,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 @@ -133,10 +143,45 @@ 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. + * + * 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 +191,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 +206,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 +231,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 +253,8 @@ async function insertMailboxMessage( subject: args.subject, fromAddress: args.fromAddress, messageKey: args.messageKey ?? null, + messageId, + inReplyTo: args.inReplyTo ?? null, refs: refs ?? null, }) .onConflictDoNothing({ @@ -249,19 +300,20 @@ 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 // 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 +363,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 +419,7 @@ export async function deliverInboxItems( messageKey: string; writeArgs: WriteMailboxMessageArgs; raw: Uint8Array; + messageId: string; }; const prepared: Prepared[] = []; for (const item of items) { @@ -378,23 +435,32 @@ 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 normalizedWriteArgs = normalizeThreadingArgs(writeArgs); + const { raw, messageId } = encodeMailboxFrame(normalizedWriteArgs); assertMailboxFrameBytes(raw); - prepared.push({ item, messageKey, writeArgs, raw }); + prepared.push({ + item, + messageKey, + writeArgs: normalizedWriteArgs, + 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;