From 26c9822bb79a748f0076d02caeb9ff60b49d0867 Mon Sep 17 00:00:00 2001 From: Sawyer Cutler Date: Sun, 2 Aug 2026 09:56:29 -0700 Subject: [PATCH 1/3] Name the operation on mailbox change events The live event carried only an id, so a listener could not tell a read from an archive from an assignment without re-fetching and diffing against remembered state. Add an optional MailboxEventOp (create/mark_read/mark_unread/trash/ archive/restore/enrich/assign) to MailboxEvent and thread it through every publishMailboxEvent call site: the two delivery paths and the transport dual-write publish "create"; mountMailbox's mutation routes publish their own action, reusing the same identifiers applyMailboxBulkAction already uses so single and bulk mutations agree. op is optional on MailboxEventSchema and on publishMailboxEvent, so an existing listener reading only id, or an existing caller omitting op, is unaffected. Document delivery semantics that were previously unstated: events can be missed, duplicated, or arrive out of order, in the bus module doc comment and the README's SSE client contract. Closes CL-5018 --- ARCHITECTURE.md | 19 ++- CHANGELOG.md | 17 ++ .../reference-host/test/acceptance.test.ts | 3 +- packages/mailbox/README.md | 28 +++- packages/mailbox/src/bus.test.ts | 64 ++++++++ packages/mailbox/src/bus.ts | 55 ++++++- packages/mailbox/src/index.ts | 7 +- packages/mailbox/src/mount-event-op.test.ts | 149 ++++++++++++++++++ packages/mailbox/src/mount.ts | 49 ++++-- packages/mailbox/src/persist.test.ts | 13 +- packages/mailbox/src/persist.ts | 1 + packages/mailbox/src/write.test.ts | 33 +++- packages/mailbox/src/write.ts | 2 + 13 files changed, 413 insertions(+), 27 deletions(-) create mode 100644 packages/mailbox/src/bus.test.ts create mode 100644 packages/mailbox/src/mount-event-op.test.ts diff --git a/ARCHITECTURE.md b/ARCHITECTURE.md index 29964ab..c969c26 100644 --- a/ARCHITECTURE.md +++ b/ARCHITECTURE.md @@ -230,6 +230,18 @@ subscriber independently; one throwing listener does not stop the others. SSE connections serialize writes, bound the pending queue, and close on overflow or write failure rather than buffering forever. +**The event names the operation that fired.** `publishMailboxEvent` takes an +optional `op` (`MailboxEventOp`: `create`, `mark_read`, `mark_unread`, `trash`, +`archive`, `restore`, `enrich`, `assign`) and, when given one, includes it on +the published event. Every call site in this package passes one — the two +delivery paths (`writeMailboxMessage`, `deliverInboxItems`) and the transport +dual-write (`createMailboxPersist`) publish `create`; `mountMailbox`'s route +table passes the mutation's own identifier, reusing `MailboxBulkAction`'s +vocabulary for the single-message verbs so "read one" and "read fifty" report +the same op. `op` is optional on `MailboxEventSchema` — additive, not a +reshape: a listener built against the original `{ type, id }` shape still +validates and still works. + **Triage enriches the message, not a task.** `priority`, `classification`, `status` and `assignee` are columns on the message's management row, not a spawned work item. Delegation is the `assignee` ref: the item stays in the @@ -343,8 +355,11 @@ actual mail transport — this package neither sends nor receives SMTP. - **SSE events are non-durable nudges.** Publication is best-effort after commit, each connection's queue is bounded at `MAX_PENDING_SSE_EVENTS` (100), and a consumer that stops reading is disconnected rather than buffered for. - The client contract — reconnect and refetch the list and unread count on any - disconnect — is documented in the package README. + Events can be missed (dropped publish, overflow disconnect), duplicated (an + undeduped redelivery, or a broker-backed bus redelivering), or arrive out of + order (no cross-replica ordering guarantee). The client contract — reconnect + and refetch the list and unread count on any disconnect, and never trust + event arrival order over a refetch — is documented in the package README. - **`sort=priority` pays a cross-table join** on top of a rank that was never index-servable; see the measurements above. - **List routes read inbound rows.** The `direction` column admits outbound diff --git a/CHANGELOG.md b/CHANGELOG.md index 0aeb85a..46b1eb2 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -9,6 +9,23 @@ always called out under their own heading. ## [Unreleased] +### Added + +- **Live events name the operation that fired.** `MailboxEvent` gains an + optional `op` (`MailboxEventOp`: `create`, `mark_read`, `mark_unread`, + `trash`, `archive`, `restore`, `enrich`, `assign`) alongside the existing + `id` — a listener can react to a specific kind of change without + re-fetching and diffing the whole message. `op` is additive: it is + optional on `MailboxEventSchema` and on `publishMailboxEvent`, so an + existing listener reading only `id`, or an existing caller of + `publishMailboxEvent` that omits `op`, is unaffected. `MAILBOX_EVENT_OPS` + and `MailboxEventOp` are now exported. +- **Delivery semantics are documented.** Events can be missed (best-effort + publish, bounded SSE queue with overflow disconnect), duplicated (an + undeduped redelivery, or a broker-backed bus a host supplies redelivering), + or arrive out of order (no cross-replica ordering guarantee) — see the + README's SSE client contract and `MailboxEventBus`'s doc comment. + ### Security - Require `drizzle-orm` `>= 0.45.2` (peer and dev pins, plus a root diff --git a/examples/reference-host/test/acceptance.test.ts b/examples/reference-host/test/acceptance.test.ts index f6ed867..a9ab222 100644 --- a/examples/reference-host/test/acceptance.test.ts +++ b/examples/reference-host/test/acceptance.test.ts @@ -355,8 +355,9 @@ describe("reference host", () => { expect(text).toContain("event: mailbox"); // The id in the frame is the row that was just written, not merely "some" // event — a stream echoing the wrong id would pass a substring check. + // `op` names the operation that produced it — a new message is a `create`. expect(text).toContain( - JSON.stringify({ type: "mailbox", id: written!.id }), + JSON.stringify({ type: "mailbox", id: written!.id, op: "create" }), ); }); diff --git a/packages/mailbox/README.md b/packages/mailbox/README.md index 2eaa133..f181a92 100644 --- a/packages/mailbox/README.md +++ b/packages/mailbox/README.md @@ -114,10 +114,18 @@ All routes carry `describeRoute`, so they appear in the host's OpenAPI document. ### The SSE client contract -**Events are non-durable nudges, not data.** Each event carries only an id to -refetch; Postgres holds the truth. The server queues at most -`MAX_PENDING_SSE_EVENTS` (100) events per connection — a consumer that stops -reading is **disconnected**, not buffered for. So the contract for any client: +**Events are non-durable nudges, not data.** Each event carries an id to +refetch and, when the publisher knows it, `op` — which operation fired +(`create`, `mark_read`, `mark_unread`, `trash`, `archive`, `restore`, +`enrich`, `assign`). `op` is **additive**: it is optional on the wire, so a +client that only reads `id` (the original shape) keeps working unchanged, +and a client that wants to react to a specific kind of change (e.g. badge a +new arrival differently from a read receipt) can switch on it instead of +re-fetching and diffing every message. Postgres remains the source of truth +either way — `op` narrows what changed, it does not replace a refetch when +you need the new state. The server queues at most `MAX_PENDING_SSE_EVENTS` +(100) events per connection — a consumer that stops reading is +**disconnected**, not buffered for. So the contract for any client: - On **any** disconnect — network drop, server restart, or an overflow close — reconnect and **refetch from the API**: the list and the unread count. Never @@ -125,6 +133,18 @@ reading is **disconnected**, not buffered for. So the contract for any client: - Do not treat the stream as a change log. It may drop events (publish is best-effort after commit) and the server may close a stream whose consumer stops draining it. +- **Events can be missed, duplicated, or arrive out of order** — this is a + best-effort nudge channel, not a durable log: + - *Missed*: publish failures are logged and swallowed, never retried; an + overflowing connection is disconnected, not buffered for. + - *Duplicated*: an inbox item redelivered without a stable dedupe key + inserts a second row and publishes a second `create`; a broker-backed bus + a host supplies for multi-replica fan-out may itself redeliver. Treat a + repeat of an already-applied `op` for the same `id` as a no-op. + - *Out of order*: the default in-memory bus preserves publish order within + one process for one mailbox, but a broker-backed bus, or multiple + replicas publishing concurrently, gives no such guarantee. Never infer + "later event = later state" from arrival order. Four more behaviors to know before wiring a UI: diff --git a/packages/mailbox/src/bus.test.ts b/packages/mailbox/src/bus.test.ts new file mode 100644 index 0000000..ed42067 --- /dev/null +++ b/packages/mailbox/src/bus.test.ts @@ -0,0 +1,64 @@ +import { describe, test, expect } from "bun:test"; +import { type } from "arktype"; +import { + MailboxEventSchema, + publishMailboxEvent, + type MailboxEvent, +} from "./bus.js"; + +const SCOPE = { tenantId: "t1", principalId: "p1" }; +const noopLogger = { error: () => {} }; + +describe("MailboxEventSchema", () => { + test("accepts the pre-existing shape: type and id, no op", () => { + // The additive contract: a listener (or a stored/replayed event) built + // before `op` existed must still validate. + const result = MailboxEventSchema({ type: "mailbox", id: "row-1" }); + expect(result instanceof type.errors).toBe(false); + }); + + test("accepts a known op", () => { + const result = MailboxEventSchema({ + type: "mailbox", + id: "row-1", + op: "mark_read", + }); + expect(result instanceof type.errors).toBe(false); + }); + + test("rejects an op outside the known vocabulary", () => { + const result = MailboxEventSchema({ + type: "mailbox", + id: "row-1", + op: "delete_everything", + }); + expect(result instanceof type.errors).toBe(true); + }); +}); + +describe("publishMailboxEvent", () => { + test("omitting op publishes the original two-field shape", () => { + const seen: MailboxEvent[] = []; + const bus = { + publish: (_scope: typeof SCOPE, event: MailboxEvent) => { + seen.push(event); + }, + subscribe: () => () => {}, + }; + publishMailboxEvent(bus, SCOPE, "row-1", noopLogger); + expect(seen).toEqual([{ type: "mailbox", id: "row-1" }]); + expect("op" in seen[0]!).toBe(false); + }); + + test("passing op includes it on the published event", () => { + const seen: MailboxEvent[] = []; + const bus = { + publish: (_scope: typeof SCOPE, event: MailboxEvent) => { + seen.push(event); + }, + subscribe: () => () => {}, + }; + publishMailboxEvent(bus, SCOPE, "row-1", noopLogger, "archive"); + expect(seen).toEqual([{ type: "mailbox", id: "row-1", op: "archive" }]); + }); +}); diff --git a/packages/mailbox/src/bus.ts b/packages/mailbox/src/bus.ts index 21f8fd0..0620d2f 100644 --- a/packages/mailbox/src/bus.ts +++ b/packages/mailbox/src/bus.ts @@ -1,8 +1,38 @@ import { type } from "arktype"; +/** + * The operation that produced an event, when the publisher knows it. Mirrors + * `MailboxBulkAction` in mutations.ts (`mark_read`, `mark_unread`, `trash`, + * `archive`, `restore`) plus the three operations mutations.ts does not own: + * `create` (a new message landed, from `writeMailboxMessage`, + * `deliverInboxItems`, or `createMailboxPersist`), `enrich` (triage stamp), + * `assign` (delegation). Duplicated here rather than imported from + * mutations.ts to avoid a bus.ts -> mutations.ts -> write.ts -> bus.ts import + * cycle; mount.ts's route table keeps the two lists in sync. + */ +export const MAILBOX_EVENT_OPS = [ + "create", + "mark_read", + "mark_unread", + "trash", + "archive", + "restore", + "enrich", + "assign", +] as const; +export type MailboxEventOp = (typeof MAILBOX_EVENT_OPS)[number]; + +/** + * `op` is optional and additive: a listener that only reads `id` (the + * original shape) keeps working unchanged. A listener that wants to react to + * a specific kind of change without re-fetching and diffing can switch on + * `op` when present, and still fall back to a refetch when it is absent + * (e.g. an older publisher, or a host bus that does not round-trip it). + */ export const MailboxEventSchema = type({ type: "'mailbox'", id: "string", + "op?": type.enumerated(...MAILBOX_EVENT_OPS), }); export type MailboxEvent = typeof MailboxEventSchema.infer; @@ -29,6 +59,25 @@ export type MailboxEventScope = { tenantId: string; principalId: string }; * subscribers for that scope from receiving the event. Events are * best-effort nudges, so a per-listener failure is swallowable; starving * healthy connections is not. + * + * Delivery semantics — the same for every bus, in-memory or host-supplied: + * + * - **Events can be missed.** Publish is best-effort after commit (a + * publish failure is logged and swallowed, never retried), and the SSE + * route disconnects a consumer whose queue exceeds `MAX_PENDING_SSE_EVENTS` + * rather than buffering for it. A listener must treat `id`/`op` as a hint + * to refetch, not as a complete change log. + * - **Events can be duplicated.** A redelivered inbox item without a stable + * dedupe key inserts a second row and publishes a second `create`; a + * broker-backed bus a host supplies may itself redeliver. Handling a + * repeat of an already-applied `op` for the same `id` must be a no-op, not + * an error. + * - **Events are not guaranteed in order across scopes' worth of concurrent + * writers.** The in-memory default preserves publish order within one + * process for a given (tenant, principal) scope; a broker-backed bus, or + * multiple replicas publishing concurrently, offers no such guarantee. + * A listener must not infer "later event = later state" — refetch the + * specific message (or the list) rather than trusting event arrival order. */ export interface MailboxEventBus { publish(scope: MailboxEventScope, event: MailboxEvent): void; @@ -40,15 +89,19 @@ export interface MailboxEventBus { * may be broker-backed and therefore may throw, and a publish failure must * never turn a committed write into a caller-visible error the client will * retry forever. The failure is logged and swallowed. + * + * `op` is optional — callers that do not know (or do not care to name) the + * operation may omit it, and the event still carries `id` alone as before. */ export function publishMailboxEvent( bus: MailboxEventBus, scope: MailboxEventScope, id: string, logger: { error: (message: string, data?: Record) => void }, + op?: MailboxEventOp, ): void { try { - bus.publish(scope, { type: "mailbox", id }); + bus.publish(scope, op ? { type: "mailbox", id, op } : { type: "mailbox", id }); } catch (err) { logger.error("mailbox event publish failed for {rowId}", { rowId: id, diff --git a/packages/mailbox/src/index.ts b/packages/mailbox/src/index.ts index 2f3aafb..1ac31fc 100644 --- a/packages/mailbox/src/index.ts +++ b/packages/mailbox/src/index.ts @@ -55,11 +55,16 @@ export type { MailboxScopeIds } from "./write.js"; export { purgeTenantMailbox, purgePrincipalMailbox } from "./purge.js"; -export { createInMemoryMailboxEventBus, MailboxEventSchema } from "./bus.js"; +export { + createInMemoryMailboxEventBus, + MailboxEventSchema, + MAILBOX_EVENT_OPS, +} from "./bus.js"; export type { MailboxEventBus, MailboxEvent, MailboxEventScope, + MailboxEventOp, } from "./bus.js"; export { diff --git a/packages/mailbox/src/mount-event-op.test.ts b/packages/mailbox/src/mount-event-op.test.ts new file mode 100644 index 0000000..279d069 --- /dev/null +++ b/packages/mailbox/src/mount-event-op.test.ts @@ -0,0 +1,149 @@ +// CL-5018: the live event must name the operation that fired so a listener +// can react without re-fetching and diffing against remembered state. +import { beforeEach, describe, expect, test } from "bun:test"; +import { Hono } from "hono"; +import { mountMailbox } from "./mount.js"; +import { createInMemoryMailboxEventBus, type MailboxEvent } from "./bus.js"; +import { writeMailboxMessage } from "./write.js"; +import { withTestDb, seedScope, TEST_VOCABULARY } from "./test-helpers.js"; +import type { MailboxDb } from "./db.js"; + +let db: MailboxDb; +const P = { tenantId: "t1", principalId: "p1" }; + +beforeEach(async () => { + db = await withTestDb(); + await seedScope(db, P.tenantId, P.principalId); +}); + +function buildApp(bus: ReturnType) { + const app = new Hono(); + mountMailbox(app, { + db, + bus, + resolvePrincipal: () => P, + vocabulary: TEST_VOCABULARY, + }); + return app; +} + +async function seedMessage(messageKey: string): Promise { + const written = await writeMailboxMessage(db, { + ...P, + address: "p1@t1.example", + fromAddress: "a@t1.example", + subject: "Hi", + body: "Body", + messageKey, + }); + return written!.id; +} + +describe("single-message mutations publish their op", () => { + const cases: Array<{ verb: string; op: string; seedKey: string }> = [ + { verb: "read", op: "mark_read", seedKey: "op-read" }, + { verb: "unread", op: "mark_unread", seedKey: "op-unread" }, + { verb: "trash", op: "trash", seedKey: "op-trash" }, + { verb: "archive", op: "archive", seedKey: "op-archive" }, + { verb: "restore", op: "restore", seedKey: "op-restore" }, + ]; + + for (const { verb, op, seedKey } of cases) { + test(`POST .../${verb} publishes op "${op}"`, async () => { + const id = await seedMessage(seedKey); + const bus = createInMemoryMailboxEventBus(); + const received: MailboxEvent[] = []; + bus.subscribe(P, (e) => received.push(e)); + const app = buildApp(bus); + + const res = await app.request(`/me/inbox/${id}/${verb}`, { + method: "POST", + }); + + expect(res.status).toBe(200); + expect(received).toEqual([{ type: "mailbox", id, op }]); + }); + } +}); + +describe("enrich and assign publish their own op", () => { + test("enrich publishes op 'enrich'", async () => { + const id = await seedMessage("op-enrich"); + const bus = createInMemoryMailboxEventBus(); + const received: MailboxEvent[] = []; + bus.subscribe(P, (e) => received.push(e)); + const app = buildApp(bus); + + const res = await app.request(`/me/inbox/${id}/enrich`, { + method: "POST", + headers: { "content-type": "application/json" }, + body: JSON.stringify({ priority: "urgent" }), + }); + + expect(res.status).toBe(200); + expect(received).toEqual([{ type: "mailbox", id, op: "enrich" }]); + }); + + test("assign publishes op 'assign'", async () => { + const id = await seedMessage("op-assign"); + const bus = createInMemoryMailboxEventBus(); + const received: MailboxEvent[] = []; + bus.subscribe(P, (e) => received.push(e)); + const app = buildApp(bus); + + const res = await app.request(`/me/inbox/${id}/assign`, { + method: "POST", + headers: { "content-type": "application/json" }, + body: JSON.stringify({ assignee: "teammate-1" }), + }); + + expect(res.status).toBe(200); + expect(received).toEqual([{ type: "mailbox", id, op: "assign" }]); + }); +}); + +describe("bulk mutation publishes the requested action as op", () => { + test("bulk trash publishes op 'trash' for every updated id", async () => { + const idA = await seedMessage("op-bulk-a"); + const idB = await seedMessage("op-bulk-b"); + const bus = createInMemoryMailboxEventBus(); + const received: MailboxEvent[] = []; + bus.subscribe(P, (e) => received.push(e)); + const app = buildApp(bus); + + const res = await app.request("/me/inbox/bulk", { + method: "POST", + headers: { "content-type": "application/json" }, + body: JSON.stringify({ action: "trash", ids: [idA, idB] }), + }); + + expect(res.status).toBe(200); + expect(received).toHaveLength(2); + expect(received.every((e) => e.op === "trash")).toBe(true); + expect(new Set(received.map((e) => e.id))).toEqual(new Set([idA, idB])); + }); +}); + +describe("delivery publishes op 'create'", () => { + test("writeMailboxMessage against the same bus a mount would use", async () => { + const bus = createInMemoryMailboxEventBus(); + const received: MailboxEvent[] = []; + bus.subscribe(P, (e) => received.push(e)); + + await writeMailboxMessage( + db, + { + ...P, + address: "p1@t1.example", + fromAddress: "a@t1.example", + subject: "New", + body: "Body", + messageKey: "op-create-write", + }, + bus, + ); + + expect(received).toHaveLength(1); + expect(received[0]?.op).toBe("create"); + }); +}); diff --git a/packages/mailbox/src/mount.ts b/packages/mailbox/src/mount.ts index fdfe5b7..60a364a 100644 --- a/packages/mailbox/src/mount.ts +++ b/packages/mailbox/src/mount.ts @@ -8,6 +8,7 @@ import { publishMailboxEvent, type MailboxEvent, type MailboxEventBus, + type MailboxEventOp, } from "./bus.js"; import { countUnreadActiveMailbox, @@ -210,24 +211,40 @@ const ID_PARAM = { }; // The five single-message mutations differ only in verb and handler, so they -// are registered from one table instead of five near-identical blocks. +// are registered from one table instead of five near-identical blocks. `op` +// is the event op published on success — same identifiers `applyMailboxBulkAction` +// takes as its `action`, so a listener sees the same op for "read one" and +// "read fifty". const SINGLE_MUTATIONS = [ - { verb: "read", summary: "Mark a message read", run: markMailboxMessageRead }, + { + verb: "read", + summary: "Mark a message read", + run: markMailboxMessageRead, + op: "mark_read", + }, { verb: "unread", summary: "Mark an active message unread", run: markMailboxMessageUnread, + op: "mark_unread", + }, + { + verb: "trash", + summary: "Trash a message", + run: trashMailboxMessage, + op: "trash", }, - { verb: "trash", summary: "Trash a message", run: trashMailboxMessage }, { verb: "archive", summary: "Archive a message (refused once trashed)", run: archiveMailboxMessage, + op: "archive", }, { verb: "restore", summary: "Restore a message out of archive or trash", run: restoreMailboxMessage, + op: "restore", }, ] as const; @@ -262,8 +279,12 @@ export function mountMailbox( ); } - function publish(scope: ResolvedPrincipal, id: string): void { - publishMailboxEvent(bus, scope, id, logger); + function publish( + scope: ResolvedPrincipal, + id: string, + op: MailboxEventOp, + ): void { + publishMailboxEvent(bus, scope, id, logger, op); } app.get( @@ -432,7 +453,9 @@ export function mountMailbox( tags: TAGS, summary: "Server-sent stream of mailbox events for the caller", description: - "Emits a `mailbox` event per affected message id, plus a heartbeat comment every 25s.", + "Emits a `mailbox` event per affected message id, plus a heartbeat comment every 25s. " + + "Each event also carries `op` (create/mark_read/mark_unread/trash/archive/restore/enrich/assign) " + + "when the publisher knows it — optional and additive, so a listener reading only `id` still works.", responses: { 200: { description: "text/event-stream" }, 403: { description: "No resolvable principalId" }, @@ -554,6 +577,7 @@ export function mountMailbox( principalId: string; id: string; }) => Promise, + op: MailboxEventOp, ) { if (!isUuid(id)) { return c.json({ error: "Message id must be a UUID" }, 400); @@ -564,11 +588,11 @@ export function mountMailbox( } const ok = await run({ ...resolved, id }); if (!ok) return c.json({ error: "Message not found" }, 404); - publish(resolved, id); + publish(resolved, id, op); return c.json({ id, ok: true as const }); } - for (const { verb, summary, run } of SINGLE_MUTATIONS) { + for (const { verb, summary, run, op } of SINGLE_MUTATIONS) { app.post( `/me/inbox/:id/${verb}`, describeRoute({ @@ -582,7 +606,8 @@ export function mountMailbox( 404: { description: "No message in scope for this action" }, }, }), - (c) => singleMutation(c, c.req.param("id"), (scope) => run(db, scope)), + (c) => + singleMutation(c, c.req.param("id"), (scope) => run(db, scope), op), ); } @@ -638,7 +663,7 @@ export function mountMailbox( throw err; } if (!ok) return c.json({ error: "Message not found" }, 404); - publish(resolved, id); + publish(resolved, id, "enrich"); return c.json({ id, ok: true as const }); }, ); @@ -683,7 +708,7 @@ export function mountMailbox( body.assignee, ); if (!ok) return c.json({ error: "Message not found" }, 404); - publish(resolved, id); + publish(resolved, id, "assign"); return c.json({ id, ok: true as const }); }, ); @@ -734,7 +759,7 @@ export function mountMailbox( throw err; } for (const r of results) { - if (r.ok) publish(resolved, r.id); + if (r.ok) publish(resolved, r.id, body.action); } return c.json({ updated: results.filter((r) => r.ok).length, diff --git a/packages/mailbox/src/persist.test.ts b/packages/mailbox/src/persist.test.ts index 0e4c91e..dfb3f3b 100644 --- a/packages/mailbox/src/persist.test.ts +++ b/packages/mailbox/src/persist.test.ts @@ -247,13 +247,15 @@ describe("dual-write independence", () => { }); describe("announcements", () => { - test("each inserted row is published to its own principalId", async () => { + test("each inserted row is published to its own principalId, as a `create`", async () => { const bus = createInMemoryMailboxEventBus(); const forUser1: string[] = []; const forUser2: string[] = []; - bus.subscribe({ tenantId: "acme", principalId: "user-1" }, (e) => - forUser1.push(e.id), - ); + const opsUser1: Array = []; + bus.subscribe({ tenantId: "acme", principalId: "user-1" }, (e) => { + forUser1.push(e.id); + opsUser1.push(e.op); + }); bus.subscribe({ tenantId: "acme", principalId: "user-2" }, (e) => forUser2.push(e.id), ); @@ -273,6 +275,9 @@ describe("announcements", () => { expect(forUser1).toHaveLength(1); expect(forUser2).toHaveLength(1); expect(forUser1[0]).not.toBe(forUser2[0]); + // Dual-write inserts are always new mail, never a mutation of an + // existing row, so the op is unconditionally `create`. + expect(opsUser1).toEqual(["create"]); }); test("onRow reports the row id, principalId and sender", async () => { diff --git a/packages/mailbox/src/persist.ts b/packages/mailbox/src/persist.ts index eab4144..a3051ef 100644 --- a/packages/mailbox/src/persist.ts +++ b/packages/mailbox/src/persist.ts @@ -126,6 +126,7 @@ export function createMailboxPersist( { tenantId: row.tenantId, principalId: row.principalId }, row.id, logger, + "create", ); } if (!opts.onRow) return; diff --git a/packages/mailbox/src/write.test.ts b/packages/mailbox/src/write.test.ts index a4291d7..83e8abf 100644 --- a/packages/mailbox/src/write.test.ts +++ b/packages/mailbox/src/write.test.ts @@ -190,9 +190,9 @@ describe("writeMailboxMessage", () => { test("successful write publishes to the bus", async () => { const bus = createInMemoryMailboxEventBus(); - const received: string[] = []; + const received: Array<{ id: string; op?: string }> = []; bus.subscribe({ tenantId: "t1", principalId: "p1" }, (event) => - received.push(event.id), + received.push(event), ); await writeMailboxMessage( db, @@ -208,6 +208,9 @@ describe("writeMailboxMessage", () => { bus, ); expect(received.length).toBe(1); + // A new message is a `create` — a listener can tell delivery apart from + // a mutation without re-fetching and diffing. + expect(received[0]?.op).toBe("create"); }); }); @@ -495,6 +498,32 @@ describe("deliverInboxItems", () => { expect(enqueued).toEqual([]); }); + test("a newly delivered item publishes a `create` event", async () => { + const bus = createInMemoryMailboxEventBus(); + const received: Array<{ id: string; op?: string }> = []; + bus.subscribe({ tenantId: "t1", principalId: "p1" }, (event) => + received.push(event), + ); + await deliverInboxItems( + db, + [ + { + tenantId: "t1", + principalId: "p1", + address: "p1@t1.example", + fromAddress: "sender@ext.example", + subject: "Delivered", + body: "Body", + source: "gmail", + externalId: "op-create", + }, + ], + { bus }, + ); + expect(received).toHaveLength(1); + expect(received[0]?.op).toBe("create"); + }); + test("bus publish and enqueue run only after a successful batch commit", async () => { const bus = createInMemoryMailboxEventBus(); const received: string[] = []; diff --git a/packages/mailbox/src/write.ts b/packages/mailbox/src/write.ts index 63bd248..3f98e4f 100644 --- a/packages/mailbox/src/write.ts +++ b/packages/mailbox/src/write.ts @@ -271,6 +271,7 @@ export async function writeMailboxMessage( { tenantId: args.tenantId, principalId: args.principalId }, row.id, logger, + "create", ); } return { id: row.id }; @@ -413,6 +414,7 @@ export async function deliverInboxItems( { tenantId: item.tenantId, principalId: item.principalId }, id, logger, + "create", ); } if (!opts?.enqueue) continue; From 4102acba92a42ee79d72cfb00d1e6d0356bf4be6 Mon Sep 17 00:00:00 2001 From: Sawyer Cutler Date: Sun, 2 Aug 2026 10:22:28 -0700 Subject: [PATCH 2/3] Require op at publishMailboxEvent call sites, reconcile op vocabularies - publishMailboxEvent's op parameter is now required; every call site already named one, so this makes the guarantee compiler-enforced instead of relying on convention. MailboxEventSchema's op field stays optional on the wire for callers outside this package and for historical events replayed from before the field existed. - Add a test asserting every MailboxBulkAction value has a matching MailboxEventOp, so a future bulk verb without a corresponding op fails a test instead of silently drifting. - Reword the duplication bullets in bus.ts and the README so the default (deduped, via a stable dedupe key) reads as the default and the undeduped-redelivery case reads as the exception it is. - Align the ordering-guarantee wording between bus.ts and the README, and back-port the README's parallel missed/duplicated/out-of-order structure into bus.ts's doc comment. - Document why MailboxEventOp keeps its own name instead of aliasing MailboxBulkAction: it's a strict superset (create/enrich/assign aren't bulk actions), so collapsing the names would claim an equivalence the two vocabularies don't have. --- ARCHITECTURE.md | 46 +++++++++++++++-------- CHANGELOG.md | 19 ++++++---- packages/mailbox/README.md | 10 +++-- packages/mailbox/src/bus.test.ts | 28 +++++++------- packages/mailbox/src/bus.ts | 63 ++++++++++++++++++++++---------- 5 files changed, 106 insertions(+), 60 deletions(-) diff --git a/ARCHITECTURE.md b/ARCHITECTURE.md index c969c26..073efac 100644 --- a/ARCHITECTURE.md +++ b/ARCHITECTURE.md @@ -230,17 +230,29 @@ subscriber independently; one throwing listener does not stop the others. SSE connections serialize writes, bound the pending queue, and close on overflow or write failure rather than buffering forever. -**The event names the operation that fired.** `publishMailboxEvent` takes an -optional `op` (`MailboxEventOp`: `create`, `mark_read`, `mark_unread`, `trash`, -`archive`, `restore`, `enrich`, `assign`) and, when given one, includes it on -the published event. Every call site in this package passes one — the two -delivery paths (`writeMailboxMessage`, `deliverInboxItems`) and the transport -dual-write (`createMailboxPersist`) publish `create`; `mountMailbox`'s route -table passes the mutation's own identifier, reusing `MailboxBulkAction`'s -vocabulary for the single-message verbs so "read one" and "read fifty" report -the same op. `op` is optional on `MailboxEventSchema` — additive, not a -reshape: a listener built against the original `{ type, id }` shape still -validates and still works. +**The event names the operation that fired.** `publishMailboxEvent` takes a +required `op` (`MailboxEventOp`: `create`, `mark_read`, `mark_unread`, `trash`, +`archive`, `restore`, `enrich`, `assign`) and includes it on the published +event. Every call site in this package passes one — the two delivery paths +(`writeMailboxMessage`, `deliverInboxItems`) and the transport dual-write +(`createMailboxPersist`) publish `create`; `mountMailbox`'s route table passes +the mutation's own identifier, reusing `MailboxBulkAction`'s vocabulary for +the single-message verbs so "read one" and "read fifty" report the same op. +`op` stays *optional on `MailboxEventSchema`* even though it is required to +publish — additive, not a reshape: a listener built against the original +`{ type, id }` shape still validates, and a historical event replayed from +before this field existed still passes. Requiring it on `publishMailboxEvent` +is what keeps every call site *in this package* honest going forward; it +cannot reach a caller outside the package, which is the other reason the +schema field has to stay optional. + +`MailboxEventOp` deliberately keeps its own name and vocabulary rather than +reusing `MailboxBulkAction`. It is a superset — `create`, `enrich`, and +`assign` are not bulk actions, and never will be — so aliasing the two would +claim an equivalence that does not hold. `mount.ts`'s route table is the one +place that has to know both: it maps HTTP verbs to `MailboxBulkAction` values +that also happen to be valid `MailboxEventOp` values, and a test in +`bus.test.ts` keeps that overlap from drifting silently. **Triage enriches the message, not a task.** `priority`, `classification`, `status` and `assignee` are columns on the message's management row, not a @@ -355,11 +367,13 @@ actual mail transport — this package neither sends nor receives SMTP. - **SSE events are non-durable nudges.** Publication is best-effort after commit, each connection's queue is bounded at `MAX_PENDING_SSE_EVENTS` (100), and a consumer that stops reading is disconnected rather than buffered for. - Events can be missed (dropped publish, overflow disconnect), duplicated (an - undeduped redelivery, or a broker-backed bus redelivering), or arrive out of - order (no cross-replica ordering guarantee). The client contract — reconnect - and refetch the list and unread count on any disconnect, and never trust - event arrival order over a refetch — is documented in the package README. + Events can be missed (dropped publish, overflow disconnect); duplicated, + but only when there is no stable dedupe key to prevent it — an inbox item + redelivered without one, or a broker-backed bus itself redelivering; or + arrive out of order (no cross-replica ordering guarantee). The client + contract — reconnect and refetch the list and unread count on any + disconnect, and never trust event arrival order over a refetch — is + documented in the package README. - **`sort=priority` pays a cross-table join** on top of a rank that was never index-servable; see the measurements above. - **List routes read inbound rows.** The `direction` column admits outbound diff --git a/CHANGELOG.md b/CHANGELOG.md index 46b1eb2..fd1ff3a 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -15,15 +15,18 @@ always called out under their own heading. optional `op` (`MailboxEventOp`: `create`, `mark_read`, `mark_unread`, `trash`, `archive`, `restore`, `enrich`, `assign`) alongside the existing `id` — a listener can react to a specific kind of change without - re-fetching and diffing the whole message. `op` is additive: it is - optional on `MailboxEventSchema` and on `publishMailboxEvent`, so an - existing listener reading only `id`, or an existing caller of - `publishMailboxEvent` that omits `op`, is unaffected. `MAILBOX_EVENT_OPS` - and `MailboxEventOp` are now exported. + re-fetching and diffing the whole message. `op` is additive on the wire: + it is optional on `MailboxEventSchema`, so an existing listener reading + only `id` is unaffected, and a historical event replayed from before this + field existed still validates. `publishMailboxEvent` requires `op` — every + call site in this package always knew the operation, and the parameter now + enforces that a future call site can't silently regress to an op-less + event. `MAILBOX_EVENT_OPS` and `MailboxEventOp` are now exported. - **Delivery semantics are documented.** Events can be missed (best-effort - publish, bounded SSE queue with overflow disconnect), duplicated (an - undeduped redelivery, or a broker-backed bus a host supplies redelivering), - or arrive out of order (no cross-replica ordering guarantee) — see the + publish, bounded SSE queue with overflow disconnect); duplicated, but only + when there is no stable dedupe key to prevent it — an undeduped inbox + redelivery, or a broker-backed bus a host supplies redelivering itself; or + arrive out of order (no cross-replica ordering guarantee) — see the README's SSE client contract and `MailboxEventBus`'s doc comment. ### Security diff --git a/packages/mailbox/README.md b/packages/mailbox/README.md index f181a92..7543abf 100644 --- a/packages/mailbox/README.md +++ b/packages/mailbox/README.md @@ -137,10 +137,12 @@ you need the new state. The server queues at most `MAX_PENDING_SSE_EVENTS` best-effort nudge channel, not a durable log: - *Missed*: publish failures are logged and swallowed, never retried; an overflowing connection is disconnected, not buffered for. - - *Duplicated*: an inbox item redelivered without a stable dedupe key - inserts a second row and publishes a second `create`; a broker-backed bus - a host supplies for multi-replica fan-out may itself redeliver. Treat a - repeat of an already-applied `op` for the same `id` as a no-op. + - *Duplicated*: redelivery is deduped by default — an inbox item with a + stable external identifier inserts once, on conflict-do-nothing. Only a + redelivered item with no such identifier inserts a second row and + publishes a second `create`; a broker-backed bus a host supplies for + multi-replica fan-out may also redeliver on its own. Treat a repeat of + an already-applied `op` for the same `id` as a no-op. - *Out of order*: the default in-memory bus preserves publish order within one process for one mailbox, but a broker-backed bus, or multiple replicas publishing concurrently, gives no such guarantee. Never infer diff --git a/packages/mailbox/src/bus.test.ts b/packages/mailbox/src/bus.test.ts index ed42067..1d8885b 100644 --- a/packages/mailbox/src/bus.test.ts +++ b/packages/mailbox/src/bus.test.ts @@ -1,10 +1,12 @@ import { describe, test, expect } from "bun:test"; import { type } from "arktype"; import { + MAILBOX_EVENT_OPS, MailboxEventSchema, publishMailboxEvent, type MailboxEvent, } from "./bus.js"; +import { MAILBOX_BULK_ACTIONS } from "./mutations.js"; const SCOPE = { tenantId: "t1", principalId: "p1" }; const noopLogger = { error: () => {} }; @@ -36,21 +38,21 @@ describe("MailboxEventSchema", () => { }); }); -describe("publishMailboxEvent", () => { - test("omitting op publishes the original two-field shape", () => { - const seen: MailboxEvent[] = []; - const bus = { - publish: (_scope: typeof SCOPE, event: MailboxEvent) => { - seen.push(event); - }, - subscribe: () => () => {}, - }; - publishMailboxEvent(bus, SCOPE, "row-1", noopLogger); - expect(seen).toEqual([{ type: "mailbox", id: "row-1" }]); - expect("op" in seen[0]!).toBe(false); +describe("MAILBOX_EVENT_OPS vs MAILBOX_BULK_ACTIONS", () => { + // MAILBOX_EVENT_OPS is duplicated from MAILBOX_BULK_ACTIONS rather than + // importing it (to avoid a bus.ts -> mutations.ts -> write.ts -> bus.ts + // cycle), and nothing at runtime enforces that the copy stays in sync. + // This is that enforcement: a bulk action added to mutations.ts without a + // matching entry here fails this test instead of silently losing its op. + test("every bulk action has a matching event op", () => { + for (const action of MAILBOX_BULK_ACTIONS) { + expect(MAILBOX_EVENT_OPS).toContain(action); + } }); +}); - test("passing op includes it on the published event", () => { +describe("publishMailboxEvent", () => { + test("publishes op on the event", () => { const seen: MailboxEvent[] = []; const bus = { publish: (_scope: typeof SCOPE, event: MailboxEvent) => { diff --git a/packages/mailbox/src/bus.ts b/packages/mailbox/src/bus.ts index 0620d2f..019cc72 100644 --- a/packages/mailbox/src/bus.ts +++ b/packages/mailbox/src/bus.ts @@ -8,7 +8,19 @@ import { type } from "arktype"; * `deliverInboxItems`, or `createMailboxPersist`), `enrich` (triage stamp), * `assign` (delegation). Duplicated here rather than imported from * mutations.ts to avoid a bus.ts -> mutations.ts -> write.ts -> bus.ts import - * cycle; mount.ts's route table keeps the two lists in sync. + * cycle; mount.ts's route table keeps the two lists in sync, and + * `bus.test.ts` asserts every `MailboxBulkAction` value is a member of this + * list. + * + * This is a deliberately different name from its two siblings, not an + * accident: mount.ts's HTTP route table calls the same five shared values a + * "verb" (the path segment), mutations.ts calls them an "action", and this + * is an "op". The three names share five values because a single-message + * mutation and its bulk equivalent report the same op, but `MailboxEventOp` + * is a strict superset — `create`, `enrich`, `assign` are not bulk actions + * and never will be — so this stays its own vocabulary rather than + * importing/aliasing `MailboxBulkAction`, which would claim an equivalence + * the two sets don't have. */ export const MAILBOX_EVENT_OPS = [ "create", @@ -23,11 +35,19 @@ export const MAILBOX_EVENT_OPS = [ export type MailboxEventOp = (typeof MAILBOX_EVENT_OPS)[number]; /** - * `op` is optional and additive: a listener that only reads `id` (the - * original shape) keeps working unchanged. A listener that wants to react to - * a specific kind of change without re-fetching and diffing can switch on - * `op` when present, and still fall back to a refetch when it is absent - * (e.g. an older publisher, or a host bus that does not round-trip it). + * `op` is optional and additive on the wire: a listener that only reads `id` + * (the original shape) keeps working unchanged. A listener that wants to + * react to a specific kind of change without re-fetching and diffing can + * switch on `op` when present, and still fall back to a refetch when it is + * absent. + * + * It stays optional here — not because a caller *inside this package* might + * reasonably omit it (every call site names one) — but because two things + * outside this package's control cannot: a caller outside this package that + * predates `op` and has no reason to know about it, and a historical event + * replayed from before this field existed. `publishMailboxEvent` below makes + * `op` a required parameter precisely so no future call site here can + * silently produce one of the events this schema still has to tolerate. */ export const MailboxEventSchema = type({ type: "'mailbox'", @@ -67,15 +87,17 @@ export type MailboxEventScope = { tenantId: string; principalId: string }; * route disconnects a consumer whose queue exceeds `MAX_PENDING_SSE_EVENTS` * rather than buffering for it. A listener must treat `id`/`op` as a hint * to refetch, not as a complete change log. - * - **Events can be duplicated.** A redelivered inbox item without a stable - * dedupe key inserts a second row and publishes a second `create`; a - * broker-backed bus a host supplies may itself redeliver. Handling a - * repeat of an already-applied `op` for the same `id` must be a no-op, not - * an error. - * - **Events are not guaranteed in order across scopes' worth of concurrent - * writers.** The in-memory default preserves publish order within one - * process for a given (tenant, principal) scope; a broker-backed bus, or - * multiple replicas publishing concurrently, offers no such guarantee. + * - **Events can be duplicated, but only when there is no stable dedupe key + * to prevent it.** Redelivery is deduped by default — an inbox item with a + * stable external identifier inserts once, on conflict-do-nothing. Only a + * redelivered item that carries no such identifier inserts a second row + * and publishes a second `create`; a broker-backed bus a host supplies may + * also redeliver on its own. Handling a repeat of an already-applied `op` + * for the same `id` must be a no-op, not an error. + * - **Events can arrive out of order.** The in-memory default preserves + * publish order within one process for one mailbox (a (tenant, principal) + * pair — same scope as the fan-out guarantee above); a broker-backed bus, + * or multiple replicas publishing concurrently, offers no such guarantee. * A listener must not infer "later event = later state" — refetch the * specific message (or the list) rather than trusting event arrival order. */ @@ -90,18 +112,21 @@ export interface MailboxEventBus { * never turn a committed write into a caller-visible error the client will * retry forever. The failure is logged and swallowed. * - * `op` is optional — callers that do not know (or do not care to name) the - * operation may omit it, and the event still carries `id` alone as before. + * `op` is required here even though it is optional on `MailboxEventSchema`: + * every call site in this package knows what it just did, and requiring it + * is what keeps a future call site from silently regressing to an + * op-less event. The schema field stays optional for the callers this + * function's signature cannot reach — see the comment above it. */ export function publishMailboxEvent( bus: MailboxEventBus, scope: MailboxEventScope, id: string, logger: { error: (message: string, data?: Record) => void }, - op?: MailboxEventOp, + op: MailboxEventOp, ): void { try { - bus.publish(scope, op ? { type: "mailbox", id, op } : { type: "mailbox", id }); + bus.publish(scope, { type: "mailbox", id, op }); } catch (err) { logger.error("mailbox event publish failed for {rowId}", { rowId: id, From a6de1bd6ef643144d2fa1dcd01dae0121cd636d8 Mon Sep 17 00:00:00 2001 From: Sawyer Cutler Date: Sun, 2 Aug 2026 10:34:58 -0700 Subject: [PATCH 3/3] Keep the op test table's literal types narrow Array<{ op: string }> widened each case's op through inference, losing the literal type publishMailboxEvent's now-required op parameter needs. Annotate the table with MailboxEventOp instead of string so the compiler still enforces the union at the one place a test constructs it. --- packages/mailbox/src/mount-event-op.test.ts | 8 ++++++-- 1 file changed, 6 insertions(+), 2 deletions(-) diff --git a/packages/mailbox/src/mount-event-op.test.ts b/packages/mailbox/src/mount-event-op.test.ts index 279d069..ddb7733 100644 --- a/packages/mailbox/src/mount-event-op.test.ts +++ b/packages/mailbox/src/mount-event-op.test.ts @@ -3,7 +3,11 @@ import { beforeEach, describe, expect, test } from "bun:test"; import { Hono } from "hono"; import { mountMailbox } from "./mount.js"; -import { createInMemoryMailboxEventBus, type MailboxEvent } from "./bus.js"; +import { + createInMemoryMailboxEventBus, + type MailboxEvent, + type MailboxEventOp, +} from "./bus.js"; import { writeMailboxMessage } from "./write.js"; import { withTestDb, seedScope, TEST_VOCABULARY } from "./test-helpers.js"; import type { MailboxDb } from "./db.js"; @@ -40,7 +44,7 @@ async function seedMessage(messageKey: string): Promise { } describe("single-message mutations publish their op", () => { - const cases: Array<{ verb: string; op: string; seedKey: string }> = [ + const cases: Array<{ verb: string; op: MailboxEventOp; seedKey: string }> = [ { verb: "read", op: "mark_read", seedKey: "op-read" }, { verb: "unread", op: "mark_unread", seedKey: "op-unread" }, { verb: "trash", op: "trash", seedKey: "op-trash" },