corbits-mailbox: createMailboxPersist accepts refs at insert time

ARCHITECTURE.md

@@ -58,15 +58,16 @@ tables foreign-key to both. Nothing changed in Interchange to make that work —
 the coupling lives entirely on this side.
 
 One further seam lives outside `mountMailbox`, on the write side:
-`createMailboxPersist(db, { upstream, authorizeSender, bus?, onRow? })` wraps a
-host's own mail-persist function so every addressed principal also gets a
-durable row. `authorizeSender(address) => { tenantId, domain } | null` is the
-host's decision — whether a sender address belongs to a *live* agent instance
-is not a schema fact. Returning `null` skips the mailbox write entirely while
-the frame still goes upstream. On the recipient side the package does consult
-the control plane: an address whose local part matches no known principal in
-the authorized tenant is skipped with a warning rather than minting a phantom
-mailbox row, and never costs the frame's real recipients their durable copy.
+`createMailboxPersist(db, { upstream, authorizeSender, bus?, onRow?, resolveRefs? })`
+wraps a host's own mail-persist function so every addressed principal also
+gets a durable row. `authorizeSender(address) => { tenantId, domain } | null`
+is the host's decision — whether a sender address belongs to a *live* agent
+instance is not a schema fact. Returning `null` skips the mailbox write
+entirely while the frame still goes upstream. On the recipient side the
+package does consult the control plane: an address whose local part matches
+no known principal in the authorized tenant is skipped with a warning rather
+than minting a phantom mailbox row, and never costs the frame's real
+recipients their durable copy.
 
 The wrapper's contract is **dual-write independence in both directions**: an
 `upstream` throw still attempts the mailbox write and then re-throws the
@@ -75,6 +76,31 @@ persist that upstream already completed. Under retry, the mailbox side is
 idempotent: package-owned transport `messageKey`s plus `onConflictDoNothing`
 collapse duplicate frames without failing the call or re-announcing.
 
+`resolveRefs(args)` — `args` is `MailboxPersistArgs` plus the resolved
+`senderAuthorization` and the `decoded` frame (or `null` if the parser
+rejected it) — is called ONCE per frame, before the transaction opens, not
+once per recipient: a host pointing every row at the same upstream entity
+does one lookup, not N. It runs AFTER `upstream` resolves and serially with
+it, so its latency adds to the call rather than overlapping. Its result is
+validated with `MailboxRefArraySchema` and capped at `MAX_MAILBOX_REFS` the
+same way `writeMailboxMessage`'s `refs` argument is (excess entries
+truncated from the end of the list, logged with `messageId` and
+`senderAddress`, never a throw) — so a resolver must return a small set with
+the load-bearing ref FIRST, since anything past the cap is silently dropped.
+Refs are then stored on every recipient row of that frame INSIDE the same
+transaction — so the post-commit bus `create` event and any SSE subscriber
+already see refs on the row once it's readable.
+
+Refs are frozen at the FIRST successful insert for a given frame: a retry
+(same idempotency key) still calls `resolveRefs` — it is not skipped — but
+because `onConflictDoNothing` writes no row on a retry, a different result
+from that second call is simply discarded; only the first call's refs ever
+land. A throwing `resolveRefs` is handled exactly like any other
+pre-transaction failure: it falls under the same dual-write contract as the
+rest of the wrapper — logged (naming `resolveRefs` as the failing stage),
+upstream unaffected (already ran, or still will, independently of this), and
+no mailbox row for that frame.
+
 `resolvePrincipal`'s signature is identical across the Corbits cores, so a host
 mounting more than one passes the same function to each.
 

CHANGELOG.md

@@ -11,6 +11,15 @@ always called out under their own heading.
 
 ### Added
 
+- **`createMailboxPersist` accepts `resolveRefs` at insert time.** The new
+  `resolveRefs(args)` option — `args` is the persist args plus the resolved
+  `senderAuthorization` and the `decoded` frame (or `null`) — is called once
+  per frame, before the transaction opens, and its result (validated with
+  `MailboxRefArraySchema`, capped at `MAX_MAILBOX_REFS`) is stored on every
+  recipient row inside that same transaction, so the post-commit bus event
+  already carries refs. A throwing `resolveRefs` follows the existing
+  dual-write contract: logged, upstream unaffected, no mailbox row for that
+  frame.
 - **Threading headers on the frame and in the list projection.**
   `buildMailFrame` accepts `references` — the thread's ancestry, oldest
   first — and emits it as a folded `References:` header; `In-Reply-To`

README.md

@@ -10,6 +10,30 @@ Requires `@intx` 0.2.2 or newer.
 
 See [ARCHITECTURE.md](./ARCHITECTURE.md) for the data model.
 
+## Dual-write persist
+
+```ts
+import { createMailboxPersist } from "@corbits/mailbox";
+
+const persist = createMailboxPersist(db, {
+  upstream: hostMailTransport.persist,
+  authorizeSender: (address) => resolveActiveInstance(address),
+  // Called once per frame, before the transaction — every recipient row
+  // gets the same refs, so a bus subscriber sees them on the `create` event.
+  resolveRefs: ({ decoded }) =>
+    decoded ? [{ kind: "workbench", id: decoded.messageId ?? "" }] : undefined,
+});
+```
+
+`resolveRefs` runs after `upstream` resolves and serially with it, and its
+refs are frozen at the frame's first successful insert — a retry still runs
+the resolver but a different result on that later call is discarded. Return
+a small set with the load-bearing ref first: the list is capped at
+`MAX_MAILBOX_REFS` by truncating from the end.
+
+See ARCHITECTURE.md's persist section for the full contract, including
+`resolveRefs`'s dual-write-failure semantics.
+
 ## Install
 
 ```sh

src/persist.test.ts

@@ -9,6 +9,7 @@ import {
   type MailboxPersistArgs,
   type SenderAuthorization,
   type PersistedMailboxRow,
+  type ResolveMailboxRefs,
 } from "./persist.js";
 import { createInMemoryMailboxEventBus } from "./bus.js";
 import { buildMailFrame } from "./frame.js";
@@ -17,8 +18,10 @@ import { withTestDb, seedScope } from "./test-helpers.js";
 import type { MailboxDb } from "./db.js";
 import {
   MAX_MAILBOX_FRAME_BYTES,
+  MAX_MAILBOX_REFS,
   assertMailboxFrameBytes,
 } from "./write.js";
+import { getMailboxMessage, type MailboxRef } from "./read.js";
 
 let db: MailboxDb;
 
@@ -520,6 +523,223 @@ describe("transport insert idempotency", () => {
   });
 });
 
+describe("resolveRefs", () => {
+  test("refs are visible to a bus subscriber on the create event", async () => {
+    const bus = createInMemoryMailboxEventBus();
+    const seenIds: string[] = [];
+    bus.subscribe({ tenantId: "acme", principalId: "user-1" }, (e) =>
+      seenIds.push(e.id),
+    );
+    const refs: MailboxRef[] = [{ kind: "workbench", id: "thread-1" }];
+    const { upstream } = recordingUpstream();
+    const persist = createMailboxPersist(db, {
+      upstream,
+      authorizeSender: () => ACTIVE,
+      bus,
+      resolveRefs: () => refs,
+    });
+
+    await persist(args());
+
+    // By the time the bus fires, the insert has already committed — a
+    // subscriber reading the row by the announced id sees refs already
+    // stamped, not a later-arriving update.
+    expect(seenIds).toHaveLength(1);
+    const message = await getMailboxMessage(db, {
+      tenantId: "acme",
+      principalId: "user-1",
+      id: seenIds[0]!,
+    });
+    expect(message?.refs).toEqual(refs);
+  });
+
+  test("resolveRefs is called once for three recipients, and every row gets its refs", async () => {
+    await seedScope(db, "acme", "user-3");
+    const refs: MailboxRef[] = [{ kind: "workbench", id: "thread-1" }];
+    const calls: unknown[] = [];
+    const resolveRefs: ResolveMailboxRefs = (a) => {
+      calls.push(a);
+      return refs;
+    };
+    const { upstream } = recordingUpstream();
+    const persist = createMailboxPersist(db, {
+      upstream,
+      authorizeSender: () => ACTIVE,
+      resolveRefs,
+    });
+
+    await persist(
+      args({
+        recipients: [
+          "usr_user-1@acme.example",
+          "usr_user-2@acme.example",
+          "usr_user-3@acme.example",
+        ],
+      }),
+    );
+
+    expect(calls).toHaveLength(1);
+    for (const principalId of ["user-1", "user-2", "user-3"]) {
+      const [row] = await rowsFor("acme", principalId);
+      const message = await getMailboxMessage(db, {
+        tenantId: "acme",
+        principalId,
+        id: row!.id,
+      });
+      expect(message?.refs).toEqual(refs);
+    }
+  });
+
+  test("a throwing resolveRefs writes zero rows; upstream still completes", async () => {
+    const { upstream, result } = recordingUpstream();
+    const persist = createMailboxPersist(db, {
+      upstream,
+      authorizeSender: () => ACTIVE,
+      resolveRefs: () => {
+        throw new Error("resolver exploded");
+      },
+    });
+
+    expect(await persist(args())).toBe(result);
+    expect(await rowsFor("acme", "user-1")).toHaveLength(0);
+  });
+
+  test("over-cap refs from resolveRefs are truncated to MAX_MAILBOX_REFS", async () => {
+    const refs: MailboxRef[] = Array.from({ length: MAX_MAILBOX_REFS + 5 }, (_, i) => ({
+      kind: "workbench",
+      id: `thread-${i}`,
+    }));
+    const { upstream } = recordingUpstream();
+    const persist = createMailboxPersist(db, {
+      upstream,
+      authorizeSender: () => ACTIVE,
+      resolveRefs: () => refs,
+    });
+
+    await persist(args());
+
+    const [row] = await rowsFor("acme", "user-1");
+    const message = await getMailboxMessage(db, {
+      tenantId: "acme",
+      principalId: "user-1",
+      id: row!.id,
+    });
+    expect(message?.refs?.length).toBe(MAX_MAILBOX_REFS);
+  });
+
+  test("retried frame with a different resolver result keeps the FIRST refs", async () => {
+    let call = 0;
+    const { upstream } = recordingUpstream();
+    const persist = createMailboxPersist(db, {
+      upstream,
+      authorizeSender: () => ACTIVE,
+      resolveRefs: () => [{ kind: "workbench", id: `attempt-${++call}` }],
+    });
+
+    await persist(args());
+    await persist(args());
+
+    const all = await rowsFor("acme", "user-1");
+    expect(all).toHaveLength(1);
+    expect(call).toBe(2);
+    const message = await getMailboxMessage(db, {
+      tenantId: "acme",
+      principalId: "user-1",
+      id: all[0]!.id,
+    });
+    expect(message?.refs).toEqual([{ kind: "workbench", id: "attempt-1" }]);
+  });
+
+  test("resolver returning undefined stores SQL NULL, not []", async () => {
+    const { upstream } = recordingUpstream();
+    const persist = createMailboxPersist(db, {
+      upstream,
+      authorizeSender: () => ACTIVE,
+      resolveRefs: () => undefined,
+    });
+
+    await persist(args());
+
+    const [row] = await rowsFor("acme", "user-1");
+    expect(row!.refs).toBeNull();
+  });
+
+  test("resolver returning [] stores SQL NULL, not []", async () => {
+    const { upstream } = recordingUpstream();
+    const persist = createMailboxPersist(db, {
+      upstream,
+      authorizeSender: () => ACTIVE,
+      resolveRefs: () => [],
+    });
+
+    await persist(args());
+
+    const [row] = await rowsFor("acme", "user-1");
+    expect(row!.refs).toBeNull();
+  });
+
+  test("schema-invalid resolver output writes zero rows; upstream result still returned", async () => {
+    const { upstream, result } = recordingUpstream();
+    const persist = createMailboxPersist(db, {
+      upstream,
+      authorizeSender: () => ACTIVE,
+      resolveRefs: () => [{ kind: 42, id: "x" }] as unknown as MailboxRef[],
+    });
+
+    expect(await persist(args())).toBe(result);
+    expect(await rowsFor("acme", "user-1")).toHaveLength(0);
+  });
+
+  test("resolver runs after upstream resolves (serial), adding its latency to the call", async () => {
+    const order: string[] = [];
+    const persist = createMailboxPersist(db, {
+      upstream: async () => {
+        await Bun.sleep(150);
+        order.push("upstream");
+        return [{ delivered: true }];
+      },
+      authorizeSender: () => ACTIVE,
+      resolveRefs: async () => {
+        order.push("resolver-start");
+        await Bun.sleep(150);
+        order.push("resolver-end");
+        return undefined;
+      },
+    });
+
+    const t0 = performance.now();
+    await persist(args());
+    const elapsed = performance.now() - t0;
+
+    expect(order).toEqual(["upstream", "resolver-start", "resolver-end"]);
+    expect(elapsed).toBeGreaterThanOrEqual(290);
+  });
+
+  test("the 21st ref (a workbench ref appended last) is silently dropped", async () => {
+    const refs: MailboxRef[] = Array.from({ length: MAX_MAILBOX_REFS }, (_, i) => ({
+      kind: "thread",
+      id: `t-${i}`,
+    }));
+    refs.push({ kind: "workbench", id: "must-be-present" });
+    const { upstream } = recordingUpstream();
+    const persist = createMailboxPersist(db, {
+      upstream,
+      authorizeSender: () => ACTIVE,
+      resolveRefs: () => refs,
+    });
+
+    await persist(args());
+
+    const [row] = await rowsFor("acme", "user-1");
+    const message = await getMailboxMessage(db, {
+      tenantId: "acme",
+      principalId: "user-1",
+      id: row!.id,
+    });
+    expect(message?.refs?.some((r) => r.kind === "workbench")).toBe(false);
+  });
+});
+
 describe("frame size and recipient hard caps", () => {
   test("assertMailboxFrameBytes accepts at-cap and throws RangeError one byte over", () => {
     // Dual-write swallows the RangeError inside attemptMailboxWrite; the pure