From 0a30280f94ae1fbc8675e2a5389610178e1700fa Mon Sep 17 00:00:00 2001 From: Sawyer Date: Fri, 4 Sep 2026 18:08:45 -0700 Subject: [PATCH 1/6] Add tests for thread reads and lookup by Message-ID (CL-7447) Pins the two properties a thread read has to get right: ancestry is never fabricated, and it never changes when the reader turns the page. Also covers the cached `references` column, the 0003 backfill's unfolding of folded `References:` continuation lines, and the msg-id lookup's scope. --- src/migrations.test.ts | 84 ++++++- src/schema-check.test.ts | 17 +- src/schema-ddl-parity.test.ts | 27 ++- src/thread.test.ts | 407 ++++++++++++++++++++++++++++++++++ 4 files changed, 513 insertions(+), 22 deletions(-) create mode 100644 src/thread.test.ts diff --git a/src/migrations.test.ts b/src/migrations.test.ts index 2a3a1ec..7c36660 100644 --- a/src/migrations.test.ts +++ b/src/migrations.test.ts @@ -84,6 +84,7 @@ describe("runMailboxMigrations", () => { "message_key", "principal_id", "raw", + "references", "refs", "subject", "tenant_id", @@ -112,12 +113,15 @@ describe("runMailboxMigrations", () => { WHERE schemaname = 'mailbox' AND tablename = 'principal_mail' ORDER BY indexname`, ); - // The mail plane keeps exactly two access paths: the dedupe constraint - // and the keyset the default page seeks on. - // `schema-ddl-parity.test.ts` holds schema.ts to this same list. + // The mail plane keeps exactly four access paths: the dedupe constraint, + // the keyset the default page seeks on, and the two the thread read adds + // — the msg-id lookup and the GIN index serving the `refs` containment + // filter. `schema-ddl-parity.test.ts` holds schema.ts to this same list. expect(mailIndexes.map((i) => i.indexname)).toEqual([ "principal_mail_pkey", + "principal_mail_refs_idx", "principal_mail_tenant_id_principal_id_created_at_id_idx", + "principal_mail_tenant_id_principal_id_message_id_idx", "principal_mail_tenant_id_principal_id_message_key_idx", ]); @@ -385,6 +389,7 @@ describe("runMailboxMigrations", () => { expect(ledger.map((r) => r.id)).toEqual([ "0001_principal_mailbox", "0002_mail_threading_headers", + "0003_mail_references", ]); const rows = await db.execute<{ @@ -401,6 +406,77 @@ describe("runMailboxMigrations", () => { }); }); + test("0003 backfills the References chain, unfolding continuation lines", async () => { + // `References` is the header that FOLDS: RFC 2822 caps a line at 78 + // characters, so a real chain of more than a couple of ids arrives split + // across continuation lines. A backfill anchored to one line would cache + // only the first fragment, and every older message would then link to the + // wrong ancestor — worse than linking to none. + await fromEmpty(async ({ db }) => { + await runMailboxMigrations(db); + await db.execute( + sql`ALTER TABLE "mailbox"."principal_mail" DROP COLUMN "references"`, + ); + await db.execute( + sql`DELETE FROM "mailbox"."corbits_mailbox_migrations" + WHERE "id" = '0003_mail_references'`, + ); + await seedScope(db, "acme", "user-1"); + + const enc = new TextEncoder(); + const folded = enc.encode( + "From: bot@acme.example\r\n" + + "Message-ID: \r\n" + + "References: \r\n" + + "\t\r\n" + + " \r\n" + + "\r\nBody\r\n", + ); + const none = enc.encode( + "From: bot@acme.example\r\nSubject: no chain\r\n\r\nBody\r\n", + ); + // The body says `References:` at the start of a line; the header slice + // must not reach it, and a NUL after it must not abort the UPDATE. + const decoy = Uint8Array.from([ + ...enc.encode( + "From: bot@acme.example\r\nMessage-ID: \r\n" + + "\r\nReferences: \r\n", + ), + 0x00, + 0x41, + ]); + for (const [key, raw] of [ + ["refs-folded", folded], + ["refs-none", none], + ["refs-decoy", decoy], + ] 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; + references: string[] | null; + }>(sql`SELECT "message_key", "references" + FROM "mailbox"."principal_mail" ORDER BY "message_key"`); + expect(rows.map((r) => [r.message_key, r.references])).toEqual([ + ["refs-decoy", null], + ["refs-folded", [ + "", + "", + "", + ]], + ["refs-none", null], + ]); + }); + }); + 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 @@ -476,6 +552,7 @@ describe("runMailboxMigrations", () => { expect(rows.map((r) => [r.id, r.count])).toEqual([ ["0001_principal_mailbox", "1"], ["0002_mail_threading_headers", "1"], + ["0003_mail_references", "1"], ]); }); }); @@ -682,6 +759,7 @@ describe("runMailboxMigrations under concurrent cold start", () => { expect(ledger.map((r) => r.id)).toEqual([ "0001_principal_mailbox", "0002_mail_threading_headers", + "0003_mail_references", ]); }); diff --git a/src/schema-check.test.ts b/src/schema-check.test.ts index 4d7f797..2a12c29 100644 --- a/src/schema-check.test.ts +++ b/src/schema-check.test.ts @@ -161,14 +161,15 @@ describe("boot against a host table this package did not create", () => { test("rejects a pre-existing table with a column MISSING outright", async () => { const schema = "mailbox"; await inFreshSchema(schema, async ({ db }) => { - // A host `principal_mail` carrying the scope and the frame but none of - // the cached header columns. Nothing errors on such a schema: `subject` - // and `refs` are read through the codec, so every message would just - // quietly lose its "Related" row and fall back to the frame for its - // subject. + // A host `principal_mail` carrying the scope and the frame but not + // `subject`. Nothing errors on such a schema: `subject` is read through + // the codec, so every message would just quietly fall back to the frame + // for it. // - // Both missing columns are ones NO index covers — see the case below for - // why that distinction matters. + // The missing column is one NO index covers — see the case below for why + // that distinction matters. `refs` is present here for exactly that + // reason: it is GIN-indexed as of `0003_mail_references`, so its absence + // is now rejected by the DDL rather than by this check. await admin.unsafe(` CREATE TABLE "${schema}"."principal_mail" ( "id" text PRIMARY KEY, @@ -179,13 +180,13 @@ describe("boot against a host table this package did not create", () => { "raw" bytea NOT NULL, "from_address" text, "message_key" text, + "refs" jsonb, "created_at" timestamp NOT NULL DEFAULT now() )`); const failure = await bootFailure(runMailboxMigrations(db)); expect(failure).toBeInstanceOf(SchemaTypeMismatchError); expect((failure as SchemaTypeMismatchError).mismatches).toEqual([ "principal_mail.subject is missing (expected text)", - "principal_mail.refs is missing (expected jsonb)", ]); }); expect(await ledgerRows(schema)).toBe(0); diff --git a/src/schema-ddl-parity.test.ts b/src/schema-ddl-parity.test.ts index 1b5a488..f2fddca 100644 --- a/src/schema-ddl-parity.test.ts +++ b/src/schema-ddl-parity.test.ts @@ -29,7 +29,7 @@ afterAll(async () => { await client.end(); }); -/** `name(col asc, col desc)` plus `unique`/`partial` markers. */ +/** `name USING method(col asc, col desc)` plus `unique`/`partial` markers. */ type IndexDescriptor = string; // eslint-disable-next-line @typescript-eslint/no-explicit-any -- one canonicalizer @@ -61,20 +61,25 @@ function declaredIndexes(table: any): IndexDescriptor[] { config.where !== undefined ? "partial" : null, ].filter((flag) => flag !== null); const suffix = flags.length > 0 ? ` [${flags.join(" ")}]` : ""; - return `${config.name}(${columns})${suffix}`; + // The access method is part of the descriptor, not decoration: a GIN + // index and a btree index over the same column serve different queries, + // and `refs @> …` is only servable by the former. + const method = (config as { method?: string }).method ?? "btree"; + return `${config.name} USING ${method}(${columns})${suffix}`; }) .sort(); } -// `pg_get_indexdef` renders `CREATE [UNIQUE] INDEX ON USING btree -// ()[ WHERE ()]`, with DESC spelled out and ASC left implicit. +// `pg_get_indexdef` renders `CREATE [UNIQUE] INDEX ON USING +// ()[ WHERE ()]`, with DESC spelled out and ASC left +// implicit. function canonicalizeIndexDef(def: string): IndexDescriptor { const match = - /^CREATE (UNIQUE )?INDEX (\S+) ON \S+ USING btree \((.*?)\)( WHERE .*)?$/.exec( + /^CREATE (UNIQUE )?INDEX (\S+) ON \S+ USING (\S+) \((.*?)\)( WHERE .*)?$/.exec( def, ); if (match === null) throw new Error(`unparsed index definition: ${def}`); - const [, unique, name, columnList, where] = match; + const [, unique, name, method, columnList, where] = match; const columns = columnList! .split(", ") .map((column) => { @@ -88,7 +93,7 @@ function canonicalizeIndexDef(def: string): IndexDescriptor { where !== undefined ? "partial" : null, ].filter((flag) => flag !== null); const suffix = flags.length > 0 ? ` [${flags.join(" ")}]` : ""; - return `${name}(${columns})${suffix}`; + return `${name} USING ${method}(${columns})${suffix}`; } async function liveIndexes(table: string): Promise { @@ -121,14 +126,14 @@ describe("schema.ts vs. the DDL runMailboxMigrations actually creates", () => { const live = await liveIndexes("mailbox"); for (const column of ["priority", "classification", "status", "assignee"]) { expect(live).toContain( - `mailbox_tenant_id_principal_id_${column}_idx(tenant_id asc, principal_id asc, ${column} asc)`, + `mailbox_tenant_id_principal_id_${column}_idx USING btree(tenant_id asc, principal_id asc, ${column} asc)`, ); } // Bare single-column forms must stay absent: an index on a low-cardinality // column is not an access path the planner would choose. expect( live.filter((descriptor) => - /^mailbox_(priority|classification|status|assignee)_idx\(/.test( + /^mailbox_(priority|classification|status|assignee)_idx USING /.test( descriptor, ), ), @@ -137,7 +142,7 @@ describe("schema.ts vs. the DDL runMailboxMigrations actually creates", () => { it("keeps the keyset access path on the mail plane, where the split left it", async () => { expect(await liveIndexes("principal_mail")).toContain( - "principal_mail_tenant_id_principal_id_created_at_id_idx(tenant_id asc, principal_id asc, created_at desc, id desc)", + "principal_mail_tenant_id_principal_id_created_at_id_idx USING btree(tenant_id asc, principal_id asc, created_at desc, id desc)", ); }); @@ -147,7 +152,7 @@ describe("schema.ts vs. the DDL runMailboxMigrations actually creates", () => { const live = await liveIndexes("mailbox"); for (const name of ["archived_at", "trashed_at", "unread"]) { expect(live).toContain( - `mailbox_tenant_id_principal_id_${name}_idx(tenant_id asc, principal_id asc) [partial]`, + `mailbox_tenant_id_principal_id_${name}_idx USING btree(tenant_id asc, principal_id asc) [partial]`, ); } }); diff --git a/src/thread.test.ts b/src/thread.test.ts new file mode 100644 index 0000000..301aa75 --- /dev/null +++ b/src/thread.test.ts @@ -0,0 +1,407 @@ +// Thread reads are the one path that has to be right about *ancestry*, not +// just about scope: a fabricated parent silently reshapes a conversation, and +// a parent that changes when the reader turns the page is worse than none at +// all. Every test here pins one of those two properties. +import { beforeEach, describe, expect, test } from "bun:test"; +import { and, eq } from "drizzle-orm"; +import { writeMailboxMessage } from "./write.js"; +import { principalMail } from "./schema.js"; +import { + readMailboxThread, + readMailboxMessageByMessageId, + decodeMailboxThreadCursor, +} from "./thread.js"; +import { withTestDb, seedScope } from "./test-helpers.js"; +import type { MailboxDb } from "./db.js"; + +let db: MailboxDb; + +const WORKBENCH = { kind: "workbench", id: "wb-1" } as const; +const OTHER_WORKBENCH = { kind: "workbench", id: "wb-2" } as const; + +beforeEach(async () => { + db = await withTestDb(); + await seedScope(db, "t1", "p1", "p2"); +}); + +/** + * `writeMailboxMessage` mints the frame's Message-ID itself, so a test that + * wants to reply to a message has to read the minted id back — exactly as a + * caller threading a real conversation would. + */ +async function send(args: { + principalId?: string; + subject: string; + inReplyTo?: string; + references?: string[]; + refs?: { kind: string; id: string }[]; + messageKey: string; +}): Promise<{ id: string; messageId: string }> { + const principalId = args.principalId ?? "p1"; + const written = await writeMailboxMessage(db, { + tenantId: "t1", + principalId, + address: `${principalId}@t1.example`, + fromAddress: "sender@t1.example", + subject: args.subject, + body: "Body", + messageKey: args.messageKey, + ...(args.inReplyTo !== undefined ? { inReplyTo: args.inReplyTo } : {}), + ...(args.references !== undefined ? { references: args.references } : {}), + ...(args.refs !== undefined ? { refs: args.refs } : {}), + }); + const [row] = await db + .select({ messageId: principalMail.messageId }) + .from(principalMail) + .where(eq(principalMail.id, written!.id)); + return { id: written!.id, messageId: row!.messageId! }; +} + +describe("readMailboxThread", () => { + test("two replies with different In-Reply-To resolve to different parents", async () => { + const root = await send({ + subject: "Root", + refs: [WORKBENCH], + messageKey: "root", + }); + const first = await send({ + subject: "Re: Root", + inReplyTo: root.messageId, + references: [root.messageId], + refs: [WORKBENCH], + messageKey: "first", + }); + const second = await send({ + subject: "Re: Re: Root", + inReplyTo: first.messageId, + references: [root.messageId, first.messageId], + refs: [WORKBENCH], + messageKey: "second", + }); + + const page = await readMailboxThread( + db, + { tenantId: "t1", principalId: "p1" }, + { ref: WORKBENCH }, + ); + const byId = new Map(page.items.map((item) => [item.id, item])); + expect(page.items.map((item) => item.id)).toEqual([ + root.id, + first.id, + second.id, + ]); + expect(byId.get(root.id)?.parentId).toBeNull(); + expect(byId.get(first.id)?.parentId).toBe(root.id); + expect(byId.get(second.id)?.parentId).toBe(first.id); + }); + + test("falls back to References, newest ancestor first, when In-Reply-To names nothing present", async () => { + const root = await send({ + subject: "Root", + refs: [WORKBENCH], + messageKey: "root", + }); + const middle = await send({ + subject: "Middle", + inReplyTo: root.messageId, + references: [root.messageId], + refs: [WORKBENCH], + messageKey: "middle", + }); + // In-Reply-To names a message nobody in this mailbox has; References + // carries the whole chain, and the NEWEST present ancestor wins. + const leaf = await send({ + subject: "Leaf", + inReplyTo: "", + references: [root.messageId, middle.messageId], + refs: [WORKBENCH], + messageKey: "leaf", + }); + + const page = await readMailboxThread( + db, + { tenantId: "t1", principalId: "p1" }, + { ref: WORKBENCH }, + ); + expect(page.items.find((item) => item.id === leaf.id)?.parentId).toBe( + middle.id, + ); + }); + + test("a parent outside the principal's mailbox yields null, never a fabricated node", async () => { + const foreign = await send({ + principalId: "p2", + subject: "Not yours", + refs: [WORKBENCH], + messageKey: "foreign", + }); + const reply = await send({ + subject: "Re: Not yours", + inReplyTo: foreign.messageId, + references: [foreign.messageId], + refs: [WORKBENCH], + messageKey: "reply", + }); + + const page = await readMailboxThread( + db, + { tenantId: "t1", principalId: "p1" }, + { ref: WORKBENCH }, + ); + expect(page.items.map((item) => item.id)).toEqual([reply.id]); + expect(page.items[0]?.parentId).toBeNull(); + }); + + test("a parent outside the ref yields null", async () => { + const elsewhere = await send({ + subject: "Other workbench", + refs: [OTHER_WORKBENCH], + messageKey: "elsewhere", + }); + const reply = await send({ + subject: "Re: Other workbench", + inReplyTo: elsewhere.messageId, + refs: [WORKBENCH], + messageKey: "reply", + }); + + const page = await readMailboxThread( + db, + { tenantId: "t1", principalId: "p1" }, + { ref: WORKBENCH }, + ); + expect(page.items.map((item) => item.id)).toEqual([reply.id]); + expect(page.items[0]?.parentId).toBeNull(); + }); + + test("the refs filter excludes other workbenches", async () => { + await send({ + subject: "Other", + refs: [OTHER_WORKBENCH], + messageKey: "other", + }); + const mine = await send({ + subject: "Mine", + refs: [WORKBENCH], + messageKey: "mine", + }); + await send({ subject: "Unreffed", messageKey: "unreffed" }); + + const page = await readMailboxThread( + db, + { tenantId: "t1", principalId: "p1" }, + { ref: WORKBENCH }, + ); + expect(page.items.map((item) => item.subject)).toEqual(["Mine"]); + expect(page.items[0]?.id).toBe(mine.id); + }); + + test("a chain spanning a page boundary keeps parentId stable across pages", async () => { + // The parent of the first row on page two lives on page one, so a + // resolver that only looked at the current page would answer null for it. + const chain: { id: string; messageId: string }[] = []; + let previous: { id: string; messageId: string } | undefined; + for (let index = 0; index < 5; index += 1) { + const message = await send({ + subject: `Message ${index}`, + refs: [WORKBENCH], + messageKey: `chain-${index}`, + ...(previous !== undefined + ? { + inReplyTo: previous.messageId, + references: chain.map((entry) => entry.messageId), + } + : {}), + }); + chain.push(message); + previous = message; + } + + const whole = await readMailboxThread( + db, + { tenantId: "t1", principalId: "p1" }, + { ref: WORKBENCH }, + ); + const expected = new Map( + whole.items.map((item) => [item.id, item.parentId]), + ); + expect(whole.nextCursor).toBeUndefined(); + expect([...expected.values()]).toEqual([ + null, + chain[0]!.id, + chain[1]!.id, + chain[2]!.id, + chain[3]!.id, + ]); + + const paged: { id: string; parentId: string | null }[] = []; + let cursor: string | undefined; + do { + const page = await readMailboxThread( + db, + { tenantId: "t1", principalId: "p1" }, + { ref: WORKBENCH, limit: 2, ...(cursor !== undefined ? { cursor } : {}) }, + ); + for (const item of page.items) { + paged.push({ id: item.id, parentId: item.parentId }); + } + cursor = page.nextCursor; + } while (cursor !== undefined); + + expect(paged.map((item) => item.id)).toEqual(chain.map((one) => one.id)); + for (const item of paged) { + expect(item.parentId).toBe(expected.get(item.id)!); + } + }); + + test("projects the threading headers and state without loading raw", async () => { + const root = await send({ + subject: "Root", + refs: [WORKBENCH], + messageKey: "root", + }); + const reply = await send({ + subject: "Re: Root", + inReplyTo: root.messageId, + references: [root.messageId], + refs: [WORKBENCH], + messageKey: "reply", + }); + + const page = await readMailboxThread( + db, + { tenantId: "t1", principalId: "p1" }, + { ref: WORKBENCH }, + ); + const item = page.items.find((one) => one.id === reply.id)!; + expect(item.messageId).toBe(reply.messageId); + expect(item.inReplyTo).toBe(root.messageId); + expect(item.references).toEqual([root.messageId]); + expect(item.fromAddress).toBe("sender@t1.example"); + expect(item.subject).toBe("Re: Root"); + expect(item.read).toBe(false); + expect(item.archived).toBe(false); + expect(item.createdAt).toMatch( + /^\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}\.\d{6}Z$/, + ); + expect(page.items[0]?.references).toEqual([]); + }); + + test("refuses a cursor minted for a different ref", async () => { + await send({ subject: "Mine", refs: [WORKBENCH], messageKey: "mine" }); + await send({ subject: "Mine 2", refs: [WORKBENCH], messageKey: "mine2" }); + const first = await readMailboxThread( + db, + { tenantId: "t1", principalId: "p1" }, + { ref: WORKBENCH, limit: 1 }, + ); + expect(first.nextCursor).toBeDefined(); + expect( + readMailboxThread( + db, + { tenantId: "t1", principalId: "p1" }, + { ref: OTHER_WORKBENCH, cursor: first.nextCursor! }, + ), + ).rejects.toThrow(RangeError); + }); + + test("refuses a malformed cursor and an out-of-range limit", async () => { + expect(decodeMailboxThreadCursor("not-a-cursor")).toBeNull(); + expect( + readMailboxThread( + db, + { tenantId: "t1", principalId: "p1" }, + { ref: WORKBENCH, cursor: "not-a-cursor" }, + ), + ).rejects.toThrow(RangeError); + expect( + readMailboxThread( + db, + { tenantId: "t1", principalId: "p1" }, + { ref: WORKBENCH, limit: 0 }, + ), + ).rejects.toThrow(RangeError); + }); +}); + +describe("readMailboxMessageByMessageId", () => { + test("returns the principal's message", async () => { + const message = await send({ + subject: "Findable", + refs: [WORKBENCH], + messageKey: "findable", + }); + const found = await readMailboxMessageByMessageId( + db, + { tenantId: "t1", principalId: "p1" }, + message.messageId, + ); + expect(found?.id).toBe(message.id); + expect(found?.subject).toBe("Findable"); + expect(found?.refs).toEqual([WORKBENCH]); + }); + + test("another principal's message is null", async () => { + const foreign = await send({ + principalId: "p2", + subject: "Not yours", + messageKey: "foreign", + }); + expect( + await readMailboxMessageByMessageId( + db, + { tenantId: "t1", principalId: "p1" }, + foreign.messageId, + ), + ).toBeNull(); + // …and is readable by the principal it belongs to, so the null above is + // the scope filter rather than a lookup that never worked. + expect( + ( + await readMailboxMessageByMessageId( + db, + { tenantId: "t1", principalId: "p2" }, + foreign.messageId, + ) + )?.id, + ).toBe(foreign.id); + }); + + test("an unknown msg-id is null", async () => { + expect( + await readMailboxMessageByMessageId( + db, + { tenantId: "t1", principalId: "p1" }, + "", + ), + ).toBeNull(); + }); +}); + +describe("the cached references column", () => { + test("caches what the frame carries, so the thread read never touches raw", async () => { + const root = await send({ + subject: "Root", + refs: [WORKBENCH], + messageKey: "root", + }); + await send({ + subject: "Re: Root", + inReplyTo: root.messageId, + references: [root.messageId], + refs: [WORKBENCH], + messageKey: "reply", + }); + const rows = await db + .select({ references: principalMail.references }) + .from(principalMail) + .where( + and( + eq(principalMail.tenantId, "t1"), + eq(principalMail.principalId, "p1"), + eq(principalMail.subject, "Re: Root"), + ), + ); + expect(rows[0]?.references).toEqual([root.messageId]); + }); +}); From 24c55bff38cd8eec055b08c35c5bb616ca393b5e Mon Sep 17 00:00:00 2001 From: Sawyer Date: Fri, 4 Sep 2026 18:08:53 -0700 Subject: [PATCH 2/6] Thread reads over References, scoped by a ref (CL-7447) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `readMailboxThread` returns the conversation under one entity ref, oldest first and keyset-paged, with `parentId` resolved by RFC 5256 References linking across the whole ref-scoped set — never by subject, and never fabricated when the ancestor is absent. `readMailboxMessageByMessageId` looks one message up by its sender-minted id, scoped to the mailbox. Both run on the list projection and never load `raw`, which is what the cached threading columns exist for. Migration 0003 adds the third of them, `references`, backfilling it from each legacy row's frame on the same NUL-safe terms 0002 uses plus the unfolding a folded header needs, and creates the two access paths the reads need: (tenant_id, principal_id, message_id) and a GIN index serving the refs containment filter. --- src/index.ts | 21 +++ src/migrations.ts | 87 ++++++++++ src/persist.ts | 7 + src/read.ts | 10 +- src/schema.ts | 27 +++ src/thread.ts | 414 ++++++++++++++++++++++++++++++++++++++++++++++ src/write.ts | 6 + 7 files changed, 567 insertions(+), 5 deletions(-) create mode 100644 src/thread.ts diff --git a/src/index.ts b/src/index.ts index f9b4e86..c7b37ff 100644 --- a/src/index.ts +++ b/src/index.ts @@ -123,6 +123,27 @@ export type { MailboxPage, } from "./read.js"; +// Thread reads: the conversation under one entity ref, parents resolved by +// RFC 5256 References linking (never by subject), and the msg-id lookup. +export { + readMailboxThread, + readMailboxMessageByMessageId, + canonicalMailboxThreadRef, + encodeMailboxThreadCursor, + decodeMailboxThreadCursor, + MailboxThreadMessageSchema, + MailboxThreadResponseSchema, + DEFAULT_MAILBOX_THREAD_LIMIT, + MAX_MAILBOX_THREAD_LIMIT, +} from "./thread.js"; +export type { + MailboxThreadScope, + MailboxThreadArgs, + MailboxThreadMessage, + MailboxThreadPage, + MailboxThreadCursor, +} from "./thread.js"; + export { markMailboxMessageRead, markMailboxMessageUnread, diff --git a/src/migrations.ts b/src/migrations.ts index 656a13d..642f266 100644 --- a/src/migrations.ts +++ b/src/migrations.ts @@ -218,6 +218,93 @@ export const MIGRATIONS: Migration[] = [ AND (h."message_id" IS NOT NULL OR h."in_reply_to" IS NOT NULL)`, ], }, + { + // The third threading header, plus the two access paths a thread read + // needs. `references` completes what 0002 started: RFC 5256 linking walks + // `In-Reply-To` first and then the `References` chain newest-first, and + // `readMailboxThread` runs on the list path, which never loads `raw`. + id: "0003_mail_references", + statements: [ + sql`ALTER TABLE "mailbox"."principal_mail" + ADD COLUMN IF NOT EXISTS "references" jsonb`, + // Msg-id lookup: `readMailboxMessageByMessageId`, and the one ancestor + // map query `readMailboxThread` issues per read. + sql`CREATE INDEX IF NOT EXISTS "principal_mail_tenant_id_principal_id_message_id_idx" + ON "mailbox"."principal_mail" ("tenant_id", "principal_id", "message_id")`, + // The ref filter is jsonb containment; only GIN serves it. Default + // `jsonb_ops`, matching what `schema.ts` declares. + sql`CREATE INDEX IF NOT EXISTS "principal_mail_refs_idx" + ON "mailbox"."principal_mail" USING gin ("refs")`, + // Backfill from the frozen frame, on exactly the terms 0002 backfills + // its two columns: the header section is sliced out of `raw` at the + // BYTEA level so a body line beginning `References:` cannot be mistaken + // for a header, and every NUL byte is stripped from that slice before + // `convert_from` runs, because Postgres `text` cannot hold 0x00 in any + // encoding and one legacy frame carrying one would otherwise abort the + // whole UPDATE — and with it every boot, forever. + // + // One step 0002 did not need: `References` is the header that FOLDS. + // RFC 2822 §2.2.3 caps a line at 78 characters, so a chain of more than + // a couple of ids is written across continuation lines, and a regex + // anchored to one line would see only the first fragment. The header + // section is unfolded first — every newline followed by whitespace + // collapses to a single space — after which each header is one line and + // the value is everything to the end of it. + // + // Msg-ids are then extracted with the same `<[^<>]+>` shape + // `parseMsgIdList` uses at runtime, in order, so a row backfilled here + // and a row written after the upgrade project the same chain. A frame + // with no References (or none that parse) keeps the column NULL rather + // than storing an empty array: absent and empty are the same thing here, + // and NULL is the cheaper of the two. + sql`UPDATE "mailbox"."principal_mail" AS pm + SET "references" = h."references" + FROM ( + SELECT "id", + ( + SELECT jsonb_agg(m[1] ORDER BY ord) + FROM regexp_matches( + COALESCE(substring(head from '(?ni)^References:[ \t]*(.*)$'), ''), + '<[^<>]+>', 'g' + ) WITH ORDINALITY AS matched(m, ord) + ) AS "references" + FROM ( + SELECT "id", + regexp_replace( + replace( + convert_from(clean_bytes, 'LATIN1'), + chr(13) || chr(10), chr(10) + ), + chr(10) || '[ \t]+', ' ', 'g' + ) 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 "references" IS NULL + ) sliced + ) cleaned + ) heads + ) h + WHERE pm."id" = h."id" + AND h."references" IS NOT NULL`, + ], + }, ]; const DIALECT = new PgDialect(); diff --git a/src/persist.ts b/src/persist.ts index 12f172c..98facf7 100644 --- a/src/persist.ts +++ b/src/persist.ts @@ -292,6 +292,12 @@ export function createMailboxPersist( // 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; + // The whole chain, oldest first, on the same terms: bracketed msg-ids + // only, and NULL rather than `[]` for a frame that carries none — what + // migration 0003's backfill derives from the same header text. + const references = decoded === null || decoded.references.length === 0 + ? null + : decoded.references; // Resolved ONCE per frame, before the transaction — every recipient row // gets the same refs, and a resolver that hits an upstream entity does one @@ -343,6 +349,7 @@ export function createMailboxPersist( messageId, inReplyTo, refs: refs ?? null, + references, messageKey: transportMessageKey( messageId, raw, diff --git a/src/read.ts b/src/read.ts index 417d303..a77f6f5 100644 --- a/src/read.ts +++ b/src/read.ts @@ -345,14 +345,14 @@ function toISODate(dateHeader: string | undefined, createdAt: Date): string { // One bad backfill would otherwise emit a warn line per bad row per page per // request — steady-state log spam that buries the signal. Bad rows are // collected per read and reported once, with a bounded sample of ids. -type DroppedRefs = { rowIds: string[]; summary: string | null }; +export type DroppedRefs = { rowIds: string[]; summary: string | null }; const DROPPED_REFS_SAMPLE = 5; -function newDroppedRefs(): DroppedRefs { +export function newDroppedRefs(): DroppedRefs { return { rowIds: [], summary: null }; } -function reportDroppedRefs(dropped: DroppedRefs): void { +export function reportDroppedRefs(dropped: DroppedRefs): void { if (dropped.rowIds.length === 0) return; logger.warn("mailbox refs column failed schema; dropped for {rows} row(s)", { rows: dropped.rowIds.length, @@ -388,7 +388,7 @@ function readRowRefs( // `raw` is intentionally absent from the row type: list selects every // principal_mail column except it, and toMailboxMessage never needs it // (the caller decodes outside and threads the result through `decoded`). -function toMailboxMessage( +export function toMailboxMessage( row: Omit, decoded: DecodedFrame | null, dropped: DroppedRefs, @@ -480,7 +480,7 @@ function filterConditions(filter: MailboxFilter): SQL[] { } /** The management columns, projected through the LEFT JOIN. */ -const STATE_COLUMNS = { +export const STATE_COLUMNS = { readAt: mailbox.readAt, archivedAt: mailbox.archivedAt, trashedAt: mailbox.trashedAt, diff --git a/src/schema.ts b/src/schema.ts index 7603fb6..0ae74a9 100644 --- a/src/schema.ts +++ b/src/schema.ts @@ -99,6 +99,14 @@ export const principalMail = mailboxPgSchema.table( // parser rejects, or one carrying no such header, still persists. messageId: text("message_id"), inReplyTo: text("in_reply_to"), + // The `References:` chain, OLDEST FIRST, as a jsonb array of bracketed + // msg-ids — cached for the same reason `in_reply_to` is, and needed by the + // same reader: `readMailboxThread` resolves a parent by RFC 5256 linking + // (In-Reply-To, then References newest-first), and it runs on the list + // path, which never loads `raw`. Plain `jsonb` with no `$type()` + // for the reason spelled out on `refs`: nothing in Postgres constrains the + // blob's shape, so every reader validates it instead. + references: jsonb("references"), // 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 @@ -143,6 +151,25 @@ export const principalMail = mailboxPgSchema.table( uniqueIndex("principal_mail_tenant_id_principal_id_message_key_idx") .on(t.tenantId, t.principalId, t.messageKey) .where(sql`${t.messageKey} IS NOT NULL`), + // Msg-id lookup, scoped the way every other access path here is. It serves + // both `readMailboxMessageByMessageId` and the ancestor map + // `readMailboxThread` builds — the latter probes it once per read with an + // `IN` list of every msg-id the page references, so this is the difference + // between one index scan and a sequential scan of the mailbox's history. + // Not unique: a msg-id is the SENDER's identifier, and nothing stops two + // externally-delivered frames from carrying the same one. + // Created by migration `0003_mail_references`. + index("principal_mail_tenant_id_principal_id_message_id_idx").on( + t.tenantId, + t.principalId, + t.messageId, + ), + // The ref filter is `refs @> [{"kind":…,"id":…}]`, which only a GIN index + // can serve. Default `jsonb_ops` rather than `jsonb_path_ops`: the + // containment query is what both support, and staying on the default + // opclass keeps this index renderable by `drizzle-kit` from the table + // object above exactly as the migration creates it. + index("principal_mail_refs_idx").using("gin", t.refs), ], ); diff --git a/src/thread.ts b/src/thread.ts new file mode 100644 index 0000000..baa82de --- /dev/null +++ b/src/thread.ts @@ -0,0 +1,414 @@ +// Thread reads: the conversation under one entity ref, and the msg-id lookup +// that makes an externally-delivered reply findable. +// +// The whole module runs on the LIST path — it never selects `principal_mail.raw` +// and never decodes a MIME frame. That is not an optimization: a thread is read +// on every conversation open, and a projection that had to decode one frame per +// row would make the cached threading columns (`message_id`, `in_reply_to`, +// `references`) pointless. They exist for exactly this reader. + +import { and, asc, eq, inArray, sql, type SQL } from "drizzle-orm"; +import { type } from "arktype"; +import { getLogger } from "@intx/log"; +import { base64urlDecode, base64urlEncode } from "@intx/types"; +import { mailbox, principalMail } from "./schema.js"; +import type { MailboxDb } from "./db.js"; +import { + MailboxRefSchema, + PRINCIPAL_MAIL_LIST_COLUMNS, + STATE_COLUMNS, + newDroppedRefs, + reportDroppedRefs, + toMailboxMessage, + type MailboxMessage, + type MailboxRef, +} from "./read.js"; + +const logger = getLogger(["corbits-mailbox", "thread"]); + +export const DEFAULT_MAILBOX_THREAD_LIMIT = 50; +/** Same ceiling the HTTP list surface enforces; a thread page is not cheaper. */ +export const MAX_MAILBOX_THREAD_LIMIT = 200; + +/** The (tenant, principal) mailbox a thread read is answered from. */ +export type MailboxThreadScope = { tenantId: string; principalId: string }; + +/** + * One message as the thread read projects it. + * + * `references` is ALWAYS present, `[]` for a message with no ancestry — a chain + * of no ancestors is an empty chain, not an absent one, and a client walking it + * should never have to branch. `parentId` is likewise always present and is + * `null`, never omitted and never invented, when the nearest ancestor is not in + * this mailbox under this ref. + * + * `createdAt` is Postgres's own microsecond rendering, the same string the + * cursor is minted from — deliberately not a JS `Date`, which holds only + * milliseconds. + */ +export const MailboxThreadMessageSchema = type({ + id: "string", + messageId: "string", + "inReplyTo?": "string", + references: "string[]", + fromAddress: "string", + "subject?": "string", + createdAt: "string", + read: "boolean", + archived: "boolean", + parentId: "string | null", +}); +export type MailboxThreadMessage = typeof MailboxThreadMessageSchema.infer; + +export const MailboxThreadResponseSchema = type({ + messages: MailboxThreadMessageSchema.array(), + "nextCursor?": "string", +}); + +export type MailboxThreadPage = { + items: MailboxThreadMessage[]; + nextCursor?: string; +}; + +export type MailboxThreadArgs = { + /** The entity the thread hangs off; matched on `kind` and `id` alone. */ + ref: MailboxRef; + cursor?: string; + /** 1..`MAX_MAILBOX_THREAD_LIMIT`; defaults to `DEFAULT_MAILBOX_THREAD_LIMIT`. */ + limit?: number; +}; + +// The same microsecond rendering `to_char` produces below, and the same shape +// the list cursor pins — a cursor is interpolated into a `::timestamp` cast, so +// it must be exactly what this package MINTS rather than merely something +// `new Date()` tolerates. +const CURSOR_CREATED_AT = /^\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}\.\d{6}Z$/; + +const MailboxThreadCursorSchema = type({ + createdAt: "string", + id: "string", + /** Canonical rendering of the ref the page was minted under. */ + ref: "string", +}); +export type MailboxThreadCursor = typeof MailboxThreadCursorSchema.infer; + +/** + * A stable string identifying which ref a thread page was produced under. + * + * JSON-encoded as a pair rather than joined with a separator: a `kind` or `id` + * containing the separator would otherwise let two different refs render the + * same string, and a cursor is only meaningful against the exact result set it + * was minted from. + */ +export function canonicalMailboxThreadRef(ref: MailboxRef): string { + return JSON.stringify([ref.kind, ref.id]); +} + +export function encodeMailboxThreadCursor( + row: { createdAt: string; id: string }, + ref: MailboxRef, +): string { + const payload: MailboxThreadCursor = { + createdAt: row.createdAt, + id: row.id, + ref: canonicalMailboxThreadRef(ref), + }; + return base64urlEncode(new TextEncoder().encode(JSON.stringify(payload))); +} + +/** + * Decode an opaque thread cursor, or null when it is malformed — bad base64, + * non-JSON, wrong shape, or a `createdAt` that is not the exact rendering this + * package mints. Null so a route can answer 400 rather than hand a crafted + * value to Postgres. + */ +export function decodeMailboxThreadCursor( + raw: string, +): MailboxThreadCursor | null { + let json: string; + try { + // `base64urlDecode` is `atob`-backed and DOES throw on a non-base64 + // character, unlike `Buffer.from(raw, "base64url")`. + json = new TextDecoder().decode(base64urlDecode(raw)); + } catch { + return null; + } + let parsed: unknown; + try { + parsed = JSON.parse(json); + } catch { + return null; + } + const result = MailboxThreadCursorSchema(parsed); + if (result instanceof type.errors) return null; + if (!CURSOR_CREATED_AT.test(result.createdAt)) return null; + if (Number.isNaN(new Date(result.createdAt).getTime())) return null; + return result; +} + +// The stored `references` blob is validated ON READ for the same reason `refs` +// is: nothing in Postgres constrains its shape, and a row written by an older +// version (or by the host directly) still reaches this projection. A blob that +// fails degrades to no ancestry — logged — rather than failing the read. +const MsgIdListSchema = type("string[]"); + +function readRowReferences(stored: unknown, rowId: string): string[] { + if (stored === null || stored === undefined) return []; + const parsed = MsgIdListSchema(stored); + if (parsed instanceof type.errors) { + logger.warn("mailbox references column failed schema; dropped for {rowId}", { + rowId, + summary: parsed.summary, + }); + return []; + } + return parsed; +} + +/** + * The ref predicate: jsonb containment, so a stored ref carrying an extra + * `label` still matches a `{ kind, id }` query. Served by the GIN index + * `principal_mail_refs_idx`. + */ +function refCondition(ref: MailboxRef) { + return sql`${principalMail.refs} @> ${JSON.stringify([{ kind: ref.kind, id: ref.id }])}::jsonb`; +} + +function assertThreadArgs(args: MailboxThreadArgs): number { + const ref = MailboxRefSchema(args.ref); + if (ref instanceof type.errors) { + throw new RangeError(`invalid mailbox thread ref: ${ref.summary}`); + } + const limit = args.limit ?? DEFAULT_MAILBOX_THREAD_LIMIT; + if ( + !Number.isSafeInteger(limit) || + limit < 1 || + limit > MAX_MAILBOX_THREAD_LIMIT + ) { + throw new RangeError( + `mailbox thread limit must be an integer in 1..${MAX_MAILBOX_THREAD_LIMIT}`, + ); + } + return limit; +} + +/** + * Resolve the cursor, refusing one minted for a different ref. + * + * A keyset cursor is only meaningful against the result set that produced it. + * Paging a cursor from one ref into another ref's thread would silently skip + * every message older than the cursor, so this is a `RangeError` — the same + * posture the list path takes when a cursor's view, sort or filter disagrees. + */ +function resolveThreadCursor( + args: MailboxThreadArgs, +): MailboxThreadCursor | undefined { + if (args.cursor === undefined) return undefined; + const cursor = decodeMailboxThreadCursor(args.cursor); + if (cursor === null) throw new RangeError("malformed mailbox thread cursor"); + if (cursor.ref !== canonicalMailboxThreadRef(args.ref)) { + throw new RangeError("mailbox thread cursor was minted for a different ref"); + } + return cursor; +} + +/** + * Read the conversation under one entity ref, oldest first, keyset-paged on + * `(created_at, id)` and scoped to `(tenantId, principalId)`. + * + * **Parents are resolved by RFC 5256 References linking, over the whole + * ref-scoped set — never by subject.** For each message the candidate ancestors + * are its `In-Reply-To` followed by its `References` chain walked + * newest-to-oldest, and the first candidate that is present in THIS mailbox + * under THIS ref wins. An ancestor that is not present yields `parentId: null`: + * a message whose parent lives in someone else's mailbox, or under a different + * ref, is a root of what this reader can see, and inventing a node for it would + * be a lie about the conversation. + * + * The ancestor lookup deliberately spans the whole ref-scoped set rather than + * the current page: a chain crossing a page boundary must not report a parent + * on one page and `null` on another, which is exactly what a page-local resolve + * would do. It costs ONE extra query per read — a msg-id map over the ids the + * page actually references, served by + * `principal_mail_tenant_id_principal_id_message_id_idx`. + * + * Throws `RangeError` on a malformed ref, an out-of-range limit, and a cursor + * that is malformed or was minted for a different ref. + */ +export async function readMailboxThread( + db: MailboxDb, + scope: MailboxThreadScope, + args: MailboxThreadArgs, +): Promise { + const limit = assertThreadArgs(args); + const cursor = resolveThreadCursor(args); + + const scopeConditions = [ + eq(principalMail.tenantId, scope.tenantId), + eq(principalMail.principalId, scope.principalId), + eq(principalMail.direction, "inbound"), + refCondition(args.ref), + ]; + const conditions = [...scopeConditions]; + if (cursor) { + // Row-value comparison matching the ORDER BY exactly, with the cast on the + // CURSOR and never on the column — `timestamp → timestamptz` is STABLE, so + // a cast on the column side cannot serve an index condition, and a + // `timestamptz` literal would resolve through the session's TimeZone and + // seek to a different row on a non-UTC host. Same rule as `listUserMailbox`. + conditions.push( + sql`(${principalMail.createdAt}, ${principalMail.id}) > (${cursor.createdAt}::timestamp, ${cursor.id})`, + ); + } + + const rows = await db + .select({ + id: principalMail.id, + messageId: principalMail.messageId, + inReplyTo: principalMail.inReplyTo, + references: principalMail.references, + fromAddress: principalMail.fromAddress, + subject: principalMail.subject, + // Postgres renders the timestamp; a JS Date would drop the microseconds + // the cursor is minted from. No `AT TIME ZONE`: the column is + // `timestamp without time zone` already holding UTC. + createdAtText: sql`to_char(${principalMail.createdAt}, 'YYYY-MM-DD"T"HH24:MI:SS.US"Z"')`, + readAt: mailbox.readAt, + archivedAt: mailbox.archivedAt, + }) + .from(principalMail) + .leftJoin(mailbox, eq(mailbox.id, principalMail.id)) + .where(and(...conditions)) + .orderBy(asc(principalMail.createdAt), asc(principalMail.id)) + .limit(limit + 1); + + const hasMore = rows.length > limit; + const pageRows = hasMore ? rows.slice(0, limit) : rows; + + const projected = pageRows.map((row) => ({ + row, + references: readRowReferences(row.references, row.id), + })); + + // Every msg-id the page could possibly link to, resolved in one query + // against the whole ref-scoped set. + const referenced = new Set(); + for (const { row, references } of projected) { + if (row.inReplyTo !== null) referenced.add(row.inReplyTo); + for (const reference of references) referenced.add(reference); + } + const ancestors = await ancestorIdsByMessageId( + db, + scopeConditions, + [...referenced], + ); + + const items = projected.map(({ row, references }) => { + // RFC 5256: the immediate parent first, then the chain newest-to-oldest. + const candidates = [ + ...(row.inReplyTo !== null ? [row.inReplyTo] : []), + ...[...references].reverse(), + ]; + let parentId: string | null = null; + for (const candidate of candidates) { + const found = ancestors.get(candidate); + // A frame naming its own msg-id is not its own parent. + if (found !== undefined && found !== row.id) { + parentId = found; + break; + } + } + const item: MailboxThreadMessage = { + id: row.id, + // The row id is the last resort, not the cache: a frame with no + // Message-ID still needs a stable handle. Same rule as the list path. + messageId: row.messageId ?? row.id, + references, + fromAddress: row.fromAddress ?? "", + createdAt: row.createdAtText, + read: row.readAt !== null, + archived: row.archivedAt !== null, + parentId, + }; + if (row.inReplyTo !== null) item.inReplyTo = row.inReplyTo; + if (row.subject !== null) item.subject = row.subject; + return item; + }); + + const page: MailboxThreadPage = { items }; + if (hasMore) { + // `hasMore` means rows.length > limit >= 1, so the page is non-empty. + const last = pageRows[pageRows.length - 1]!; + page.nextCursor = encodeMailboxThreadCursor( + { createdAt: last.createdAtText, id: last.id }, + args.ref, + ); + } + return page; +} + +/** + * Map every supplied msg-id to the id of the message carrying it, within the + * same scope and ref the thread page was read under. Ties (nothing makes a + * msg-id unique — it is the sender's identifier) resolve to the OLDEST + * carrier, so a parent does not change when a duplicate arrives later. + */ +async function ancestorIdsByMessageId( + db: MailboxDb, + scopeConditions: SQL[], + messageIds: string[], +): Promise> { + if (messageIds.length === 0) return new Map(); + const rows = await db + .select({ id: principalMail.id, messageId: principalMail.messageId }) + .from(principalMail) + .where( + and(...scopeConditions, inArray(principalMail.messageId, messageIds)), + ) + .orderBy(asc(principalMail.createdAt), asc(principalMail.id)); + const byMessageId = new Map(); + for (const row of rows) { + if (row.messageId === null) continue; + if (!byMessageId.has(row.messageId)) byMessageId.set(row.messageId, row.id); + } + return byMessageId; +} + +/** + * Look one message up by its `Message-ID`, scoped to (tenantId, principalId). + * Returns null when this mailbox holds no such message — including when + * another principal's does, which is the whole point of the scope. + * + * Nothing makes a msg-id unique (it is the sender's identifier, and two + * externally-delivered frames may carry the same one), so the OLDEST match + * wins — a stable answer rather than whichever row the planner reached first. + * + * Served from the cached `message_id` column, on the list projection: this is + * a lookup, not a detail read, and it never loads `raw`. + */ +export async function readMailboxMessageByMessageId( + db: MailboxDb, + scope: MailboxThreadScope, + messageId: string, +): Promise { + const [row] = await db + .select({ ...PRINCIPAL_MAIL_LIST_COLUMNS, ...STATE_COLUMNS }) + .from(principalMail) + .leftJoin(mailbox, eq(mailbox.id, principalMail.id)) + .where( + and( + eq(principalMail.tenantId, scope.tenantId), + eq(principalMail.principalId, scope.principalId), + eq(principalMail.direction, "inbound"), + eq(principalMail.messageId, messageId), + ), + ) + .orderBy(asc(principalMail.createdAt), asc(principalMail.id)) + .limit(1); + if (!row) return null; + + const dropped = newDroppedRefs(); + const message = toMailboxMessage(row, null, dropped); + reportDroppedRefs(dropped); + return message; +} diff --git a/src/write.ts b/src/write.ts index 9b12bd5..1d90211 100644 --- a/src/write.ts +++ b/src/write.ts @@ -296,6 +296,12 @@ async function insertMailboxMessage( messageKey, messageId, inReplyTo: args.inReplyTo ?? null, + // Absent and empty are the same chain, and NULL is the cheaper of the + // two — the same rule migration 0003's backfill applies. + references: + args.references === undefined || args.references.length === 0 + ? null + : args.references, refs: refs ?? null, }) .onConflictDoNothing({ From 9f33fd3b7ba9d5a7f7018704f2bd1d5e1c450a59 Mon Sep 17 00:00:00 2001 From: Sawyer Date: Fri, 4 Sep 2026 18:08:57 -0700 Subject: [PATCH 3/6] Update docs: thread reads and Message-ID lookup (CL-7447) --- ARCHITECTURE.md | 35 ++++++++++++++++++++++++++++++++++- CHANGELOG.md | 31 +++++++++++++++++++++++++++++++ README.md | 33 +++++++++++++++++++++++++++++++++ 3 files changed, 98 insertions(+), 1 deletion(-) diff --git a/ARCHITECTURE.md b/ARCHITECTURE.md index 65fce6d..ae25bf1 100644 --- a/ARCHITECTURE.md +++ b/ARCHITECTURE.md @@ -110,6 +110,7 @@ mounting more than one passes the same function to each. | --- | --- | | `mount.ts` | HTTP surface. Parsing, validation, status codes, SSE. No SQL. | | `read.ts` | List (cached columns, no `raw`) and detail (frame-decoded) projection, keyset paging, snippets on detail. | +| `thread.ts` | The conversation under one entity ref: keyset-paged oldest-first, parents resolved by RFC 5256 References linking, plus the msg-id lookup. | | `mutations.ts` | Read/unread, archive, trash, restore, bulk, enrich, assign. | | `write.ts` | `writeMailboxMessage` / `deliverInboxItems` — the host-facing write API. | | `persist.ts` | The transport dual-write wrapper and the `authorizeSender` seam. | @@ -354,7 +355,39 @@ stays on `principal_mail`, matching the list's `ORDER BY` and its row-value cursor seek exactly, so the default (highest-traffic) page remains a single-table index scan that stops at `limit + 1` rows. The triage indexes and the three partial view indexes (`unread`, `archived_at`, `trashed_at`) live on -`mailbox`. +`mailbox`. The thread read adds two more on `principal_mail`: +`(tenant_id, principal_id, message_id)` — not unique, since a msg-id is the +*sender's* identifier and nothing stops two delivered frames carrying the same +one — and a GIN index on `refs`, the only kind that can serve the `refs @> …` +containment filter the ref scope is expressed as. + +### Thread reads + +`readMailboxThread(db, scope, { ref, cursor?, limit? })` answers the +conversation under one entity ref: oldest first, keyset-paged on +`(created_at, id)`, scoped to `(tenant_id, principal_id)` and filtered by +jsonb containment on `refs`. + +**Parents are resolved by RFC 5256 References linking, never by subject.** For +each message the candidate ancestors are its `In-Reply-To` followed by its +`References` chain walked newest-to-oldest, and the first candidate present in +*this* mailbox under *this* ref wins. An ancestor that is not present yields +`parentId: null` — a message whose parent lives in another principal's mailbox, +or under a different ref, is a root of what this reader can see, and inventing +a node for it would be a lie about the conversation. + +The ancestor lookup spans the whole ref-scoped set rather than the current +page, so a chain crossing a page boundary cannot report a parent on one page +and `null` on another. It costs one extra query per read — a msg-id map over +the ids the page actually references, served by +`principal_mail_tenant_id_principal_id_message_id_idx`. + +The whole module runs on the list path and never selects `raw`. That is what +the cached `message_id`, `in_reply_to` and `references` columns exist for: a +thread is read on every conversation open, and decoding one MIME frame per row +would make the cache pointless. `readMailboxMessageByMessageId(db, scope, +messageId)` is the same posture — a scoped lookup on the list projection, +oldest match winning, `null` when this mailbox holds no such message. ### What the split costs, measured diff --git a/CHANGELOG.md b/CHANGELOG.md index 467e1f0..d1c0603 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -41,6 +41,37 @@ always called out under their own heading. thread reader can fetch a principal's own sent copies (`"outbound"`) or both directions together (`"all"`) alongside the existing inbox-only default. +- **Thread reads over `References`, and lookup by `Message-ID`.** + `readMailboxThread(db, scope, { ref, cursor?, limit? })` returns the + principal's messages carrying a `refs` entry equal to `ref`, oldest + first, keyset-paged on `(created_at, id)`. Each message projects `id`, + `messageId`, `inReplyTo`, `references`, `fromAddress`, `subject`, + `createdAt`, its read/archived state, and `parentId` — resolved by + **RFC 5256 References linking** (`In-Reply-To` first, then the + `References` chain newest-to-oldest) across the whole ref-scoped set, + never by subject grouping. A parent that is not in this mailbox under + this ref yields `parentId: null` rather than a fabricated node, and the + ancestor lookup spans the whole ref-scoped set rather than the current + page, so a chain crossing a page boundary keeps a stable `parentId`. + Cursors are bound to the ref that minted them; paging one into another + ref is a `RangeError`, as is a malformed cursor or a limit outside + `1..200`. `readMailboxMessageByMessageId(db, scope, messageId)` looks + one message up by its `Message-ID`, scoped to `(tenantId, principalId)`, + oldest match winning since nothing makes a msg-id unique. + `principal_mail` gains a cached `references` column, populated on the + write and transport-persist paths; migration `0003_mail_references` + adds it, backfills it from each existing row's `raw` — unfolding the + `References:` continuation lines RFC 2822 line limits force, with the + same `bytea`-level header slicing and NUL stripping `0002` uses — and + creates the two indexes the reads need: + `(tenant_id, principal_id, message_id)` and a GIN index on `refs`. + Both surfaces run on the list projection and never load `raw`. + Exported alongside them: `MailboxThreadMessageSchema`, + `MailboxThreadResponseSchema`, `canonicalMailboxThreadRef`, + `encodeMailboxThreadCursor`, `decodeMailboxThreadCursor`, + `DEFAULT_MAILBOX_THREAD_LIMIT`, `MAX_MAILBOX_THREAD_LIMIT`, and the + `MailboxThreadScope` / `MailboxThreadArgs` / `MailboxThreadMessage` / + `MailboxThreadPage` / `MailboxThreadCursor` types. - **Threading headers on the frame and in the list projection.** `buildMailFrame` accepts `references` — the thread's ancestry, oldest diff --git a/README.md b/README.md index 99dd684..4c4cff1 100644 --- a/README.md +++ b/README.md @@ -70,6 +70,39 @@ publish bus events only after commit, one per row actually written. See including `writeMailboxMessage`'s caller-supplied `messageId`, `direction`, and default `messageKey`. +## Thread reads + +```ts +import { + readMailboxThread, + readMailboxMessageByMessageId, +} from "@corbits/mailbox"; + +// The conversation under one entity ref, oldest first, keyset-paged. +const page = await readMailboxThread( + db, + { tenantId, principalId }, + { ref: { kind: "workbench", id: "wb-1" }, limit: 50 }, +); +// page.items: { id, messageId, inReplyTo?, references, fromAddress, subject?, +// createdAt, read, archived, parentId } +// page.nextCursor: pass back as `cursor` for the next page. + +// One message by its Message-ID, scoped to this mailbox. +const message = await readMailboxMessageByMessageId( + db, + { tenantId, principalId }, + "", +); +``` + +`parentId` is resolved by RFC 5256 References linking — `In-Reply-To` first, +then the `References` chain newest-to-oldest — across the whole ref-scoped set, +not just the current page. It is `null`, never fabricated, when the nearest +ancestor is not in this mailbox under this ref. Subjects are never used to +group. A cursor is bound to the ref that minted it; paging it into a different +ref is a `RangeError`, as is a malformed cursor or an out-of-range limit. + ## Working on it ```sh From 3f382d0a4bf9ce3075a3677364f74e7defeffbc0 Mon Sep 17 00:00:00 2001 From: Sawyer Date: Fri, 4 Sep 2026 18:37:09 -0700 Subject: [PATCH 4/6] Add tests for thread cycle-breaking and the thread keyset index (CL-7447) Folds critique coverage (self-reference, mutual cycle, out-of-order ancestor, duplicate Message-ID tie-break, cross-tenant scoping, exact ref match, malformed references) into src/thread.test.ts, and adds an EXPLAIN test pinning that a large, time-clustered ref still pages via an Index Scan with a Limit rather than a full sort. Updates the migration-ledger index-name assertions for the new keyset index ahead of the implementation that adds it. --- src/migrations.test.ts | 10 +- src/thread.test.ts | 253 ++++++++++++++++++++++++++++++++++++++++- 2 files changed, 258 insertions(+), 5 deletions(-) diff --git a/src/migrations.test.ts b/src/migrations.test.ts index 7c36660..5307c33 100644 --- a/src/migrations.test.ts +++ b/src/migrations.test.ts @@ -113,13 +113,15 @@ describe("runMailboxMigrations", () => { WHERE schemaname = 'mailbox' AND tablename = 'principal_mail' ORDER BY indexname`, ); - // The mail plane keeps exactly four access paths: the dedupe constraint, - // the keyset the default page seeks on, and the two the thread read adds - // — the msg-id lookup and the GIN index serving the `refs` containment - // filter. `schema-ddl-parity.test.ts` holds schema.ts to this same list. + // The mail plane keeps exactly five access paths: the dedupe constraint, + // the keyset the default page seeks on, and the three the thread read + // adds — the msg-id lookup, the GIN index serving the `refs` containment + // filter, and the thread's own oldest-first keyset. + // `schema-ddl-parity.test.ts` holds schema.ts to this same list. expect(mailIndexes.map((i) => i.indexname)).toEqual([ "principal_mail_pkey", "principal_mail_refs_idx", + "principal_mail_tenant_id_principal_id_created_at_id_asc_idx", "principal_mail_tenant_id_principal_id_created_at_id_idx", "principal_mail_tenant_id_principal_id_message_id_idx", "principal_mail_tenant_id_principal_id_message_key_idx", diff --git a/src/thread.test.ts b/src/thread.test.ts index 301aa75..2f7a5fa 100644 --- a/src/thread.test.ts +++ b/src/thread.test.ts @@ -3,7 +3,7 @@ // a parent that changes when the reader turns the page is worse than none at // all. Every test here pins one of those two properties. import { beforeEach, describe, expect, test } from "bun:test"; -import { and, eq } from "drizzle-orm"; +import { and, eq, sql } from "drizzle-orm"; import { writeMailboxMessage } from "./write.js"; import { principalMail } from "./schema.js"; import { @@ -378,6 +378,257 @@ describe("readMailboxMessageByMessageId", () => { }); }); +describe("thread edge cases: cycles, tie-breaks, scope", () => { + /** + * These tests write rows directly (bypassing `writeMailboxMessage`, which + * mints its own Message-ID) so a scenario can pin exact msg-ids, exact + * `createdAt` ordering, and — for the cycle tests — headers a real MIME + * frame would never carry on its own but that RFC 5256 step 1.B still + * requires a reader to survive. + */ + async function insertRaw(args: { + id: string; + tenantId?: string; + messageId: string | null; + inReplyTo?: string | null; + references?: string[] | null; + createdAt: string; + refs?: unknown; + }): Promise { + await db.execute(sql` + INSERT INTO "mailbox"."principal_mail" + ("id","tenant_id","principal_id","address","direction","raw","message_id","in_reply_to","references","refs","created_at") + VALUES (${args.id}, ${args.tenantId ?? "t1"}, 'p1', 'p1@t1.example', 'inbound', ${Buffer.from("x")}, + ${args.messageId}, ${args.inReplyTo ?? null}, + ${ + args.references === undefined || args.references === null + ? null + : JSON.stringify(args.references) + }::jsonb, + ${JSON.stringify(args.refs ?? [WORKBENCH])}::jsonb, ${args.createdAt}::timestamp) + `); + } + + test("a message whose References names its own Message-ID is not its own parent", async () => { + await insertRaw({ + id: "a", + messageId: "", + inReplyTo: "", + references: [""], + createdAt: "2026-01-01T00:00:00.000001Z", + }); + const page = await readMailboxThread( + db, + { tenantId: "t1", principalId: "p1" }, + { ref: WORKBENCH }, + ); + expect(page.items[0]?.parentId).toBeNull(); + }); + + test("a mutual References cycle is broken: the LATER-created message becomes the root", async () => { + await insertRaw({ + id: "a", + messageId: "", + inReplyTo: "", + createdAt: "2026-01-01T00:00:00.000001Z", + }); + await insertRaw({ + id: "b", + messageId: "", + inReplyTo: "", + createdAt: "2026-01-01T00:00:00.000002Z", + }); + const page = await readMailboxThread( + db, + { tenantId: "t1", principalId: "p1" }, + { ref: WORKBENCH }, + ); + const byId = new Map(page.items.map((i) => [i.id, i.parentId])); + // b is later-created, so cutting b's edge breaks the cycle: a -> b -> null. + expect(byId.get("a")).toBe("b"); + expect(byId.get("b")).toBeNull(); + }); + + test("a child that is OLDER than its parent still resolves (ancestor map is not ordering-bound)", async () => { + await insertRaw({ + id: "child", + messageId: "", + inReplyTo: "", + createdAt: "2026-01-01T00:00:00.000001Z", + }); + await insertRaw({ + id: "parent", + messageId: "", + createdAt: "2026-01-02T00:00:00.000001Z", + }); + const page = await readMailboxThread( + db, + { tenantId: "t1", principalId: "p1" }, + { ref: WORKBENCH, limit: 1 }, + ); + expect(page.items[0]?.id).toBe("child"); + expect(page.items[0]?.parentId).toBe("parent"); + }); + + test("duplicate Message-ID: oldest carrier wins, and self is skipped even when a duplicate exists", async () => { + await insertRaw({ + id: "dup-old", + messageId: "", + createdAt: "2026-01-01T00:00:00.000001Z", + }); + await insertRaw({ + id: "dup-new", + messageId: "", + inReplyTo: "", + createdAt: "2026-01-01T00:00:00.000002Z", + }); + await insertRaw({ + id: "reply", + messageId: "", + inReplyTo: "", + createdAt: "2026-01-01T00:00:00.000003Z", + }); + const page = await readMailboxThread( + db, + { tenantId: "t1", principalId: "p1" }, + { ref: WORKBENCH }, + ); + const byId = new Map(page.items.map((i) => [i.id, i.parentId])); + expect(byId.get("reply")).toBe("dup-old"); + // dup-new names its own msg-id; the oldest carrier is dup-old, which is + // NOT itself, so it links there. + expect(byId.get("dup-new")).toBe("dup-old"); + const found = await readMailboxMessageByMessageId( + db, + { tenantId: "t1", principalId: "p1" }, + "", + ); + expect(found?.id).toBe("dup-old"); + }); + + test("same principalId under another tenant is invisible to lookup and thread", async () => { + await seedScope(db, "t2", "p1"); + await insertRaw({ + id: "other-tenant", + tenantId: "t2", + messageId: "", + createdAt: "2026-01-01T00:00:00.000001Z", + }); + expect( + await readMailboxMessageByMessageId( + db, + { tenantId: "t1", principalId: "p1" }, + "", + ), + ).toBeNull(); + const page = await readMailboxThread( + db, + { tenantId: "t1", principalId: "p1" }, + { ref: WORKBENCH }, + ); + expect(page.items).toEqual([]); + }); + + test("ref match is exact on kind and id; a ref carrying an extra label still matches", async () => { + await insertRaw({ + id: "labelled", + messageId: "", + createdAt: "2026-01-01T00:00:00.000001Z", + refs: [{ kind: "workbench", id: "wb-1", label: "L" }], + }); + await insertRaw({ + id: "prefix", + messageId: "", + createdAt: "2026-01-01T00:00:00.000002Z", + refs: [{ kind: "workbench", id: "wb-10" }], + }); + await insertRaw({ + id: "kind", + messageId: "", + createdAt: "2026-01-01T00:00:00.000003Z", + refs: [{ kind: "thread", id: "wb-1" }], + }); + const page = await readMailboxThread( + db, + { tenantId: "t1", principalId: "p1" }, + { ref: WORKBENCH }, + ); + expect(page.items.map((i) => i.id)).toEqual(["labelled"]); + }); + + test("a malformed references blob degrades to [] rather than failing the read", async () => { + await insertRaw({ + id: "bad", + messageId: "", + createdAt: "2026-01-01T00:00:00.000001Z", + }); + await db.execute( + sql`UPDATE "mailbox"."principal_mail" SET "references" = '{"not":"a list"}'::jsonb WHERE id = 'bad'`, + ); + const page = await readMailboxThread( + db, + { tenantId: "t1", principalId: "p1" }, + { ref: WORKBENCH }, + ); + expect(page.items[0]?.references).toEqual([]); + }); + + test("EXPLAIN: ref-filtered thread page uses the GIN index once the table is large", async () => { + await db.execute(sql` + INSERT INTO "mailbox"."principal_mail" ("tenant_id","principal_id","address","direction","raw","message_id","refs","created_at") + SELECT 't1','p1','p1@t1.example','inbound', ${Buffer.from("x")}, '', + jsonb_build_array(jsonb_build_object('kind','workbench','id','wb-' || (g % 500))), + now() - (g || ' seconds')::interval + FROM generate_series(1, 50000) g`); + await db.execute(sql`ANALYZE "mailbox"."principal_mail"`); + const plan = await db.execute<{ "QUERY PLAN": string }>(sql` + EXPLAIN SELECT pm.id FROM "mailbox"."principal_mail" pm + LEFT JOIN "mailbox"."mailbox" m ON m.id = pm.id + WHERE pm.tenant_id = 't1' AND pm.principal_id = 'p1' AND pm.direction = 'inbound' + AND pm.refs @> '[{"kind":"workbench","id":"wb-1"}]'::jsonb + ORDER BY pm.created_at ASC, pm.id ASC LIMIT 51`); + const text = plan.map((r) => r["QUERY PLAN"]).join("\n"); + expect(text).toContain("principal_mail_refs_idx"); + const lookup = await db.execute<{ "QUERY PLAN": string }>(sql` + EXPLAIN SELECT pm.id FROM "mailbox"."principal_mail" pm + WHERE pm.tenant_id = 't1' AND pm.principal_id = 'p1' AND pm.direction = 'inbound' + AND pm.refs @> '[{"kind":"workbench","id":"wb-1"}]'::jsonb + AND pm.message_id IN ('','')`); + const ltext = lookup.map((r) => r["QUERY PLAN"]).join("\n"); + expect(ltext).toContain("principal_mail_tenant_id_principal_id_message_id_idx"); + // Bulk inserts of this size (needed for a realistic planner decision) run + // past bun's default per-test timeout. + }, 30000); + + test("EXPLAIN: a large, time-clustered ref pages via an Index Scan with a Limit, not a full sort", async () => { + // 300k rows in the mailbox; the 50k newest of them all carry the SAME + // ref, so the ref is both large (in absolute row count) and clustered at + // one end of the created_at range — the shape that makes a page have to + // choose between scanning the ordered btree with a Filter, or bitmapping + // the GIN index and sorting every one of the ref's rows. + await db.execute(sql` + INSERT INTO "mailbox"."principal_mail" ("tenant_id","principal_id","address","direction","raw","message_id","refs","created_at") + SELECT 't1','p1','p1@t1.example','inbound', ${Buffer.from("x")}, '', + jsonb_build_array(jsonb_build_object( + 'kind','workbench', + 'id', CASE WHEN g <= 50000 THEN 'wb-1' ELSE 'wb-' || (g % 6000) END + )), + now() - (g || ' seconds')::interval + FROM generate_series(1, 300000) g`); + await db.execute(sql`ANALYZE "mailbox"."principal_mail"`); + const plan = await db.execute<{ "QUERY PLAN": string }>(sql` + EXPLAIN (ANALYZE, BUFFERS) SELECT pm.id FROM "mailbox"."principal_mail" pm + LEFT JOIN "mailbox"."mailbox" m ON m.id = pm.id + WHERE pm.tenant_id = 't1' AND pm.principal_id = 'p1' AND pm.direction = 'inbound' + AND pm.refs @> '[{"kind":"workbench","id":"wb-1"}]'::jsonb + ORDER BY pm.created_at ASC, pm.id ASC LIMIT 51`); + const text = plan.map((r) => r["QUERY PLAN"]).join("\n"); + expect(text).toContain("Limit"); + expect(text).toContain("Index Scan"); + expect(text).not.toContain("Sort Key"); + }, 30000); +}); + describe("the cached references column", () => { test("caches what the frame carries, so the thread read never touches raw", async () => { const root = await send({ From 2c1c36f04fafc9c03e5729c328465c3844ae41ee Mon Sep 17 00:00:00 2001 From: Sawyer Date: Fri, 4 Sep 2026 18:37:20 -0700 Subject: [PATCH 5/6] Break reference cycles; add thread keyset index; diagnose refs earlier (CL-7447) Three fixes surfaced by critique: - readMailboxThread resolved parentId per row from an ancestor map, but never checked whether the resolved chain looped back on itself (RFC 5256 step 1.B). A self-reference was already excluded, but a mutual or longer cycle among delivered frames' In-Reply-To/References would resolve every member to a non-null parent forever. Ancestors are now fetched breadth-first (beyond the page itself, when a chain reaches further) into a small graph, and every edge that would close a loop is cut before a page is projected: the later-created message in the cycle (ties broken by id) becomes a root. A defensive node cap bounds the walk against a pathological reference graph. - Added `principal_mail_tenant_id_principal_id_created_at_id_asc_idx` in the (still unshipped) 0003_mail_references migration, matching readMailboxThread's own oldest-first ORDER BY verbatim rather than depending on the planner reversing the list path's DESC index. schema.ts mirrors it for schema-ddl-parity. - Migration.assertColumnsBeforeStatement lets a migration call assertExpectedColumnTypes mid-run; 0003 uses it ahead of its GIN index so a host whose principal_mail predates this package (missing `refs`) fails with the named SchemaTypeMismatchError instead of a raw Postgres "column \"refs\" does not exist". Also replaces thread.ts's per-row references-column warn with the same once-per-read dropped/report pattern read.ts uses for `refs`. --- src/migrations.ts | 41 ++++++- src/schema.ts | 13 +++ src/thread.ts | 270 ++++++++++++++++++++++++++++++++++++++-------- 3 files changed, 279 insertions(+), 45 deletions(-) diff --git a/src/migrations.ts b/src/migrations.ts index 642f266..b13bdc8 100644 --- a/src/migrations.ts +++ b/src/migrations.ts @@ -20,7 +20,21 @@ const LOCK_KEY = 0x0a27_2c01; // statement changes its checksum, and the runner then refuses to boot rather // than silently leaving old environments on the old schema while fresh ones get // the new one. -export type Migration = { id: string; statements: SQL[] }; +export type Migration = { + id: string; + statements: SQL[]; + /** + * When set, the runner calls `assertExpectedColumnTypes` immediately before + * executing `statements[assertColumnsBeforeStatement]`, inside this + * migration's own step transaction. Purely a runner-ordering knob — it does + * not change `statements` and so cannot change `migrationChecksum`. See + * `0003_mail_references`, where it exists so a host whose "principal_mail" + * predates this package (and so is missing "refs") fails here with the named + * `SchemaTypeMismatchError` instead of at the CREATE INDEX statement below, + * as a raw, unfriendly Postgres "column \"refs\" does not exist". + */ + assertColumnsBeforeStatement?: number; +}; export const MIGRATIONS: Migration[] = [ { @@ -219,11 +233,17 @@ export const MIGRATIONS: Migration[] = [ ], }, { - // The third threading header, plus the two access paths a thread read + // The third threading header, plus the three access paths a thread read // needs. `references` completes what 0002 started: RFC 5256 linking walks // `In-Reply-To` first and then the `References` chain newest-first, and // `readMailboxThread` runs on the list path, which never loads `raw`. id: "0003_mail_references", + // See `assertColumnsBeforeStatement` on `Migration`: this runs the column + // check right before the GIN index below, which is the first statement in + // this migration that would otherwise fail with a raw Postgres error + // ("column \"refs\" does not exist") on a host whose "principal_mail" + // predates this package, rather than the named diagnostic. + assertColumnsBeforeStatement: 3, statements: [ sql`ALTER TABLE "mailbox"."principal_mail" ADD COLUMN IF NOT EXISTS "references" jsonb`, @@ -231,6 +251,18 @@ export const MIGRATIONS: Migration[] = [ // map query `readMailboxThread` issues per read. sql`CREATE INDEX IF NOT EXISTS "principal_mail_tenant_id_principal_id_message_id_idx" ON "mailbox"."principal_mail" ("tenant_id", "principal_id", "message_id")`, + // The keyset access path `readMailboxThread` actually orders by: + // oldest-first, `(created_at, id)`, scoped to `(tenant_id, principal_id)`. + // `0001`'s index already covers a query shaped like this via a BACKWARD + // scan (it is declared `created_at DESC, id DESC`, for the list path's + // newest-first order) — but a plan is easier to reason about, and to + // pin in a test, when the thread path has its own index matching its + // own `ORDER BY` verbatim rather than depending on Postgres choosing to + // scan another index in reverse. Same three leading columns + // `principal_mail_tenant_id_principal_id_created_at_id_idx` carries, in + // the direction `readMailboxThread` actually asks for. + sql`CREATE INDEX IF NOT EXISTS "principal_mail_tenant_id_principal_id_created_at_id_asc_idx" + ON "mailbox"."principal_mail" ("tenant_id", "principal_id", "created_at" ASC, "id" ASC)`, // The ref filter is jsonb containment; only GIN serves it. Default // `jsonb_ops`, matching what `schema.ts` declares. sql`CREATE INDEX IF NOT EXISTS "principal_mail_refs_idx" @@ -411,7 +443,10 @@ export async function runMailboxMigrations(db: MailboxDb): Promise { continue; } await tx.transaction(async (step) => { - for (const statement of migration.statements) { + for (const [index, statement] of migration.statements.entries()) { + if (migration.assertColumnsBeforeStatement === index) { + await assertExpectedColumnTypes(step); + } await step.execute(statement); } await step.execute( diff --git a/src/schema.ts b/src/schema.ts index 0ae74a9..99d6d5b 100644 --- a/src/schema.ts +++ b/src/schema.ts @@ -164,6 +164,19 @@ export const principalMail = mailboxPgSchema.table( t.principalId, t.messageId, ), + // `readMailboxThread`'s own keyset order: oldest first, `(created_at, id)`, + // scoped the same way every other access path here is. The list-path index + // above covers the same three columns in the opposite direction and a + // backward scan of it would serve this query too, but this index lets the + // thread path's plan match its `ORDER BY` directly rather than depending + // on the planner choosing to scan the other index in reverse. + // Created by migration `0003_mail_references`. + index("principal_mail_tenant_id_principal_id_created_at_id_asc_idx").on( + t.tenantId, + t.principalId, + t.createdAt.asc(), + t.id.asc(), + ), // The ref filter is `refs @> [{"kind":…,"id":…}]`, which only a GIN index // can serve. Default `jsonb_ops` rather than `jsonb_path_ops`: the // containment query is what both support, and staying on the default diff --git a/src/thread.ts b/src/thread.ts index baa82de..12c9ca9 100644 --- a/src/thread.ts +++ b/src/thread.ts @@ -152,14 +152,39 @@ export function decodeMailboxThreadCursor( // fails degrades to no ancestry — logged — rather than failing the read. const MsgIdListSchema = type("string[]"); -function readRowReferences(stored: unknown, rowId: string): string[] { +// One bad backfill would otherwise emit a warn line per bad row per page per +// request — the same steady-state log spam `read.ts`'s `DroppedRefs` exists to +// avoid. Collected per read and reported once, with a bounded sample of ids. +const DROPPED_REFERENCES_SAMPLE = 5; + +type DroppedReferences = { rowIds: string[]; summary: string | null }; + +function newDroppedReferences(): DroppedReferences { + return { rowIds: [], summary: null }; +} + +function reportDroppedReferences(dropped: DroppedReferences): void { + if (dropped.rowIds.length === 0) return; + logger.warn( + "mailbox references column failed schema; dropped for {rows} row(s)", + { + rows: dropped.rowIds.length, + sampleRowIds: dropped.rowIds.slice(0, DROPPED_REFERENCES_SAMPLE), + summary: dropped.summary, + }, + ); +} + +function readRowReferences( + stored: unknown, + rowId: string, + dropped: DroppedReferences, +): string[] { if (stored === null || stored === undefined) return []; const parsed = MsgIdListSchema(stored); if (parsed instanceof type.errors) { - logger.warn("mailbox references column failed schema; dropped for {rowId}", { - rowId, - summary: parsed.summary, - }); + dropped.rowIds.push(rowId); + dropped.summary ??= parsed.summary; return []; } return parsed; @@ -212,6 +237,29 @@ function resolveThreadCursor( return cursor; } +/** + * One node of the ref-scoped ancestry graph — a message and just enough of it + * to resolve (and, when necessary, cut) its candidate parent edge. `createdAt` + * is the same sortable microsecond text the cursor is minted from, so nodes + * from different queries (the page, and any ancestor batches fetched to walk + * a chain) compare with a plain string `<`. + */ +type ThreadNode = { + id: string; + messageId: string | null; + inReplyTo: string | null; + references: string[]; + createdAt: string; +}; + +// Defensive cap on how many nodes a single read will walk while resolving +// ancestry and breaking cycles. A real conversation's chain is nowhere near +// this deep; the cap exists so a pathological or adversarial reference graph +// degrades (bailing out of further expansion, which can only ever turn a +// resolved parent into `null`, never fabricate one) rather than reading an +// unbounded number of rows. +const MAX_THREAD_ANCESTRY_NODES = 2000; + /** * Read the conversation under one entity ref, oldest first, keyset-paged on * `(created_at, id)` and scoped to `(tenantId, principalId)`. @@ -225,11 +273,25 @@ function resolveThreadCursor( * ref, is a root of what this reader can see, and inventing a node for it would * be a lie about the conversation. * + * **`parentId` chains are acyclic.** Nothing stops a delivered frame's + * `In-Reply-To`/`References` from naming a msg-id that (directly, or through + * further ancestors) points back at the frame itself — RFC 5256 step 1.B calls + * this out explicitly. Left unresolved a cycle would either loop a client's + * ancestry walk forever or silently make a message a descendant of one of its + * own descendants, so before a page is projected, every resolved parent edge + * that would close a loop is cut: the LATER-created message in the cycle (ties + * broken by id) becomes a root (`parentId: null`) instead, and every other + * message in the cycle keeps its resolved parent. Which edge is cut is + * deterministic and depends only on the cycle's members, never on where the + * cursor happens to land, so a cycle's shape does not change from one page to + * the next. + * * The ancestor lookup deliberately spans the whole ref-scoped set rather than * the current page: a chain crossing a page boundary must not report a parent * on one page and `null` on another, which is exactly what a page-local resolve - * would do. It costs ONE extra query per read — a msg-id map over the ids the - * page actually references, served by + * would do. Ancestors are fetched breadth-first, one batch per hop, so a chain + * (or a cycle) reaching beyond the messages the page directly names is still + * resolved correctly; each batch is served by * `principal_mail_tenant_id_principal_id_message_id_idx`. * * Throws `RangeError` on a malformed ref, an out-of-range limit, and a cursor @@ -285,39 +347,102 @@ export async function readMailboxThread( const hasMore = rows.length > limit; const pageRows = hasMore ? rows.slice(0, limit) : rows; + const dropped = newDroppedReferences(); const projected = pageRows.map((row) => ({ row, - references: readRowReferences(row.references, row.id), + references: readRowReferences(row.references, row.id, dropped), })); + reportDroppedReferences(dropped); + + // The ancestry graph: every node discovered so far, by id, plus the + // oldest-carrier msg-id -> id map candidates resolve through. Seeded with + // the page itself, then expanded breadth-first to whatever the page's rows + // (and, in turn, THEIR ancestors) name — the graph a cycle could hide in. + const nodes = new Map(); + const byMessageId = new Map(); + + function addNode(node: ThreadNode): void { + if (!nodes.has(node.id)) nodes.set(node.id, node); + if (node.messageId === null) return; + const existingId = byMessageId.get(node.messageId); + if (existingId === undefined) { + byMessageId.set(node.messageId, node.id); + return; + } + // Oldest carrier wins — nothing makes a msg-id unique (it is the sender's + // identifier), so ties resolve to whichever row sorts first, deterministic + // and stable regardless of fetch order. + const existing = nodes.get(existingId)!; + if ( + node.createdAt < existing.createdAt || + (node.createdAt === existing.createdAt && node.id < existingId) + ) { + byMessageId.set(node.messageId, node.id); + } + } - // Every msg-id the page could possibly link to, resolved in one query - // against the whole ref-scoped set. - const referenced = new Set(); for (const { row, references } of projected) { - if (row.inReplyTo !== null) referenced.add(row.inReplyTo); - for (const reference of references) referenced.add(reference); + addNode({ + id: row.id, + messageId: row.messageId, + inReplyTo: row.inReplyTo, + references, + createdAt: row.createdAtText, + }); } - const ancestors = await ancestorIdsByMessageId( - db, - scopeConditions, - [...referenced], - ); - const items = projected.map(({ row, references }) => { - // RFC 5256: the immediate parent first, then the chain newest-to-oldest. + // Breadth-first expansion: each hop resolves one more round of msg-ids that + // the nodes discovered so far point at, until nothing new turns up or the + // safety cap is hit. Bounded and cheap in the overwhelmingly common case + // (no cycle, a chain a few hops deep) and the only way to prove a cycle + // absent rather than merely absent from the current page. + let frontier = new Set(); + for (const node of nodes.values()) { + if (node.inReplyTo !== null) frontier.add(node.inReplyTo); + for (const reference of node.references) frontier.add(reference); + } + const queried = new Set(); + while (frontier.size > 0 && nodes.size < MAX_THREAD_ANCESTRY_NODES) { + const toFetch = [...frontier].filter((messageId) => !queried.has(messageId)); + for (const messageId of toFetch) queried.add(messageId); + frontier = new Set(); + if (toFetch.length === 0) break; + const fetched = await fetchThreadNodesByMessageId(db, scopeConditions, toFetch); + for (const node of fetched) { + addNode(node); + if (node.inReplyTo !== null && !queried.has(node.inReplyTo)) { + frontier.add(node.inReplyTo); + } + for (const reference of node.references) { + if (!queried.has(reference)) frontier.add(reference); + } + } + } + + // Candidate parent, per node, before cycle-breaking: RFC 5256's + // In-Reply-To-first, then References newest-to-oldest, first candidate + // present under this scope and ref — excluding the node itself, since a + // frame naming its own msg-id is not its own parent. + const rawParent = new Map(); + for (const node of nodes.values()) { const candidates = [ - ...(row.inReplyTo !== null ? [row.inReplyTo] : []), - ...[...references].reverse(), + ...(node.inReplyTo !== null ? [node.inReplyTo] : []), + ...[...node.references].reverse(), ]; - let parentId: string | null = null; + let parent: string | null = null; for (const candidate of candidates) { - const found = ancestors.get(candidate); - // A frame naming its own msg-id is not its own parent. - if (found !== undefined && found !== row.id) { - parentId = found; + const found = byMessageId.get(candidate); + if (found !== undefined && found !== node.id) { + parent = found; break; } } + rawParent.set(node.id, parent); + } + + const finalParent = resolveAcyclicParents(nodes, rawParent); + + const items = projected.map(({ row, references }) => { const item: MailboxThreadMessage = { id: row.id, // The row id is the last resort, not the cache: a frame with no @@ -328,7 +453,7 @@ export async function readMailboxThread( createdAt: row.createdAtText, read: row.readAt !== null, archived: row.archivedAt !== null, - parentId, + parentId: finalParent.get(row.id) ?? null, }; if (row.inReplyTo !== null) item.inReplyTo = row.inReplyTo; if (row.subject !== null) item.subject = row.subject; @@ -348,30 +473,91 @@ export async function readMailboxThread( } /** - * Map every supplied msg-id to the id of the message carrying it, within the - * same scope and ref the thread page was read under. Ties (nothing makes a - * msg-id unique — it is the sender's identifier) resolve to the OLDEST - * carrier, so a parent does not change when a duplicate arrives later. + * Break every reference cycle in the candidate-parent graph, per RFC 5256 + * step 1.B: walk each node's raw-parent chain with a per-walk visited set, and + * when a walk revisits a node already on its own path, the path from that node + * to the end IS the cycle. Cut it by nulling out the parent of the + * LATER-created member (ties broken by the larger id) — that member becomes a + * root; every other member of the cycle keeps its raw parent. A node's outcome + * never depends on which node the walk started from, only on the cycle's own + * membership, so the result is the same regardless of `nodes` iteration order. + */ +function resolveAcyclicParents( + nodes: Map, + rawParent: Map, +): Map { + const finalParent = new Map(); + const done = new Set(); + + function isLater(a: string, b: string): boolean { + const nodeA = nodes.get(a)!; + const nodeB = nodes.get(b)!; + if (nodeA.createdAt !== nodeB.createdAt) { + return nodeA.createdAt > nodeB.createdAt; + } + return a > b; + } + + for (const start of nodes.keys()) { + if (done.has(start)) continue; + const path: string[] = []; + let current: string | null = start; + while (current !== null && !done.has(current)) { + const cycleStart = path.indexOf(current); + if (cycleStart !== -1) { + const cycle = path.slice(cycleStart); + const cut = cycle.reduce((worst, candidate) => + isLater(candidate, worst) ? candidate : worst, + ); + finalParent.set(cut, null); + break; + } + path.push(current); + current = rawParent.get(current) ?? null; + } + for (const id of path) { + if (!finalParent.has(id)) finalParent.set(id, rawParent.get(id) ?? null); + done.add(id); + } + } + return finalParent; +} + +/** + * Fetch full ancestry nodes (not just ids) for a batch of msg-ids, within the + * same scope and ref the thread page was read under — the per-hop query the + * breadth-first ancestor walk issues, served by + * `principal_mail_tenant_id_principal_id_message_id_idx`. */ -async function ancestorIdsByMessageId( +async function fetchThreadNodesByMessageId( db: MailboxDb, scopeConditions: SQL[], messageIds: string[], -): Promise> { - if (messageIds.length === 0) return new Map(); +): Promise { + if (messageIds.length === 0) return []; const rows = await db - .select({ id: principalMail.id, messageId: principalMail.messageId }) + .select({ + id: principalMail.id, + messageId: principalMail.messageId, + inReplyTo: principalMail.inReplyTo, + references: principalMail.references, + createdAtText: sql`to_char(${principalMail.createdAt}, 'YYYY-MM-DD"T"HH24:MI:SS.US"Z"')`, + }) .from(principalMail) .where( and(...scopeConditions, inArray(principalMail.messageId, messageIds)), ) .orderBy(asc(principalMail.createdAt), asc(principalMail.id)); - const byMessageId = new Map(); - for (const row of rows) { - if (row.messageId === null) continue; - if (!byMessageId.has(row.messageId)) byMessageId.set(row.messageId, row.id); - } - return byMessageId; + const dropped = newDroppedReferences(); + const nodes = rows.map((row) => ({ + id: row.id, + messageId: row.messageId, + inReplyTo: row.inReplyTo, + references: readRowReferences(row.references, row.id, dropped), + createdAt: row.createdAtText, + })); + reportDroppedReferences(dropped); + return nodes; } /** From 0e1bbaa50391d204060647577926910d7132d20a Mon Sep 17 00:00:00 2001 From: Sawyer Date: Fri, 4 Sep 2026 18:37:24 -0700 Subject: [PATCH 6/6] Update docs: acyclic parentId, thread keyset index, earlier refs diagnostic (CL-7447) --- ARCHITECTURE.md | 44 ++++++++++++++++++++++++++++++++++++++------ CHANGELOG.md | 40 ++++++++++++++++++++++++++++------------ 2 files changed, 66 insertions(+), 18 deletions(-) diff --git a/ARCHITECTURE.md b/ARCHITECTURE.md index ae25bf1..5a223ac 100644 --- a/ARCHITECTURE.md +++ b/ARCHITECTURE.md @@ -355,11 +355,22 @@ stays on `principal_mail`, matching the list's `ORDER BY` and its row-value cursor seek exactly, so the default (highest-traffic) page remains a single-table index scan that stops at `limit + 1` rows. The triage indexes and the three partial view indexes (`unread`, `archived_at`, `trashed_at`) live on -`mailbox`. The thread read adds two more on `principal_mail`: +`mailbox`. The thread read adds three more on `principal_mail`: `(tenant_id, principal_id, message_id)` — not unique, since a msg-id is the *sender's* identifier and nothing stops two delivered frames carrying the same -one — and a GIN index on `refs`, the only kind that can serve the `refs @> …` -containment filter the ref scope is expressed as. +one — a GIN index on `refs`, the only kind that can serve the `refs @> …` +containment filter the ref scope is expressed as, and +`(tenant_id, principal_id, created_at, id)` matching `readMailboxThread`'s own +oldest-first `ORDER BY` verbatim. That last one covers the same three leading +columns the list path's own keyset index does, in the opposite direction; a +backward scan of the list's index already serves the thread query, but a +dedicated index removes the dependence on the planner choosing to scan the +other one in reverse. Whichever index a page's plan uses, a ref whose messages +cluster at one end of the principal's own `created_at` history while a page +seeks from the other end still costs a `Filter` proportional to how much +*unrelated* history sits between them — no index shape fixes that; only +clustering by ref would, and this package deliberately holds no opinion on +physical row order. See "What the split costs, measured" below. ### Thread reads @@ -376,11 +387,22 @@ each message the candidate ancestors are its `In-Reply-To` followed by its or under a different ref, is a root of what this reader can see, and inventing a node for it would be a lie about the conversation. +**`parentId` chains are acyclic.** RFC 5256 step 1.B calls out that nothing +stops a delivered frame's `In-Reply-To`/`References` from naming a msg-id that, +directly or through further ancestors, points back at the frame itself. Before +a page is projected, the candidate-parent graph is walked (breadth-first, +beyond the page itself when a chain reaches further) and every edge that would +close a loop is cut: the LATER-created message in the cycle (ties broken by +id) becomes a root instead, deterministically — the cut depends only on the +cycle's own membership, never on which page or cursor triggered the read. + The ancestor lookup spans the whole ref-scoped set rather than the current page, so a chain crossing a page boundary cannot report a parent on one page -and `null` on another. It costs one extra query per read — a msg-id map over -the ids the page actually references, served by -`principal_mail_tenant_id_principal_id_message_id_idx`. +and `null` on another. It costs one query per hop of the ancestry graph — a +msg-id map over the ids referenced so far, served by +`principal_mail_tenant_id_principal_id_message_id_idx` — capped defensively at +`MAX_THREAD_ANCESTRY_NODES` so a pathological reference graph degrades a +resolved parent to `null` rather than reading an unbounded number of rows. The whole module runs on the list path and never selects `raw`. That is what the cached `message_id`, `in_reply_to` and `references` columns exist for: a @@ -445,6 +467,16 @@ every boot of every replica. host's columns through our codec. The expectation is derived from the drizzle table objects, so it cannot drift; a rejected boot rolls the ledger row back with it. +- **A migration can also run that same check early**, via + `Migration.assertColumnsBeforeStatement`. `0003_mail_references` sets it: a + host whose `principal_mail` predates this package leaves `refs` missing (it + is only ever declared inline in `0001`'s `CREATE TABLE`, which no-ops against + a pre-existing table), and without the early check the first statement to + notice would be `0003`'s `CREATE INDEX ... USING gin ("refs")` — a raw + Postgres "column \"refs\" does not exist" instead of the named + `SchemaTypeMismatchError` diagnostic. The knob only changes when the runner + calls the check, never `Migration.statements`, so it cannot change + `migrationChecksum`. **Everything lands in the `mailbox` schema, fully qualified.** Nothing resolves through `search_path`, so the host's own setting cannot redirect or shadow diff --git a/CHANGELOG.md b/CHANGELOG.md index d1c0603..872d90b 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -49,23 +49,39 @@ always called out under their own heading. `createdAt`, its read/archived state, and `parentId` — resolved by **RFC 5256 References linking** (`In-Reply-To` first, then the `References` chain newest-to-oldest) across the whole ref-scoped set, - never by subject grouping. A parent that is not in this mailbox under - this ref yields `parentId: null` rather than a fabricated node, and the - ancestor lookup spans the whole ref-scoped set rather than the current - page, so a chain crossing a page boundary keeps a stable `parentId`. - Cursors are bound to the ref that minted them; paging one into another - ref is a `RangeError`, as is a malformed cursor or a limit outside - `1..200`. `readMailboxMessageByMessageId(db, scope, messageId)` looks - one message up by its `Message-ID`, scoped to `(tenantId, principalId)`, - oldest match winning since nothing makes a msg-id unique. + never by subject grouping. **`parentId` chains are acyclic** — RFC 5256 + step 1.B calls out that a delivered frame's `In-Reply-To`/`References` + can name a msg-id that, directly or through further ancestors, points + back at the frame itself; a cycle among the resolved candidate parents + is detected and cut (the LATER-created message in the cycle, ties broken + by id, becomes a root) before a page is projected, so a client's + ancestry walk always terminates. A parent that is not in this mailbox + under this ref yields `parentId: null` rather than a fabricated node, + and the ancestor lookup spans the whole ref-scoped set rather than the + current page, so a chain crossing a page boundary keeps a stable + `parentId`. Cursors are bound to the ref that minted them; paging one + into another ref is a `RangeError`, as is a malformed cursor or a limit + outside `1..200`. `readMailboxMessageByMessageId(db, scope, messageId)` + looks one message up by its `Message-ID`, scoped to + `(tenantId, principalId)`, oldest match winning since nothing makes a + msg-id unique. `principal_mail` gains a cached `references` column, populated on the write and transport-persist paths; migration `0003_mail_references` adds it, backfills it from each existing row's `raw` — unfolding the `References:` continuation lines RFC 2822 line limits force, with the same `bytea`-level header slicing and NUL stripping `0002` uses — and - creates the two indexes the reads need: - `(tenant_id, principal_id, message_id)` and a GIN index on `refs`. - Both surfaces run on the list projection and never load `raw`. + creates the three indexes the reads need: + `(tenant_id, principal_id, message_id)`, a GIN index on `refs`, and + `(tenant_id, principal_id, created_at, id)` matching `readMailboxThread`'s + own oldest-first `ORDER BY` (the list path's index over the same three + columns, in the opposite direction, already served this query via a + backward scan; this one lets the thread path's plan match its `ORDER BY` + directly). Both surfaces run on the list projection and never load `raw`. + A host whose own `principal_mail` predates this package (and so is + missing `refs` — see `ARCHITECTURE.md`'s "Migrations" section) now fails + `0003` with the named `SchemaTypeMismatchError` diagnostic rather than a + raw Postgres "column \"refs\" does not exist" at the `CREATE INDEX` + statement that needs it. Exported alongside them: `MailboxThreadMessageSchema`, `MailboxThreadResponseSchema`, `canonicalMailboxThreadRef`, `encodeMailboxThreadCursor`, `decodeMailboxThreadCursor`,