fix(store): allocate uids in Postgres and enforce uid and Message-ID uniqueness

README.md

@@ -116,6 +116,8 @@ Takes the same arguments as `@intx/db`'s `runMigrations`. `schema` holds the hos
 | `assertMailboxScope`, `assertMailboxTenantId`, `assertMailboxFrameBytes`, `MailboxScopeIdsSchema` | Scope and frame-size checks the write paths run.                                             |
 | `MAX_MAILBOX_PAGE_LIMIT`, `MAX_MAILBOX_RECIPIENTS`, `MAX_MAILBOX_FRAME_BYTES`                     | Page size (200), recipients per frame, and frame size (1 MiB) limits.                        |
 
+The interface's synchronous `append` must return a uid before it reaches Postgres, so it cannot allocate one atomically and two stores appending at once can lose a write. Write with `appendMessage`, `writeMailboxMessage` or `createMailboxPersist` instead.
+
 ## Using with Interchange
 
 The host's tenant middleware sets `tenant` and `principal` on the context; the routes read the caller from there, the same principal `requireGrant` authorizes. Grant each person `mailbox:*` with the `read`, `create` and `manage` actions they need.

e2e/migrations.test.ts

@@ -85,6 +85,7 @@ describe("runMailboxMigrations", () => {
         "raw",
         "references",
         "refs",
+        "sender_address",
         "subject",
         "tenant_id",
         "to_addresses",
@@ -109,10 +110,12 @@ describe("runMailboxMigrations", () => {
       // filter, and the thread's own oldest-first keyset.
       // The schema.ts parity suite below holds schema.ts to this same list.
       expect(mailIndexes.map((i) => i.indexname)).toEqual([
+        "principal_mail_folder_message_id_sender_address_idx",
         "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_folder_uid_idx",
         "principal_mail_tenant_id_principal_id_message_id_idx",
         "principal_mail_tenant_id_principal_id_message_key_idx",
       ]);
@@ -154,6 +157,80 @@ describe("runMailboxMigrations", () => {
     });
   });
 
+  test("0008 renumbers duplicate uids and clears repeated Message-IDs without dropping a row", async () => {
+    await fromEmpty(async ({ db }) => {
+      await applyMailboxMigrations(db, "public");
+      await admin.unsafe(`
+        DROP INDEX "mailbox"."principal_mail_tenant_id_principal_id_folder_uid_idx";
+        DROP INDEX "mailbox"."principal_mail_folder_message_id_sender_address_idx";
+        INSERT INTO "tenant" VALUES ('t8') ON CONFLICT DO NOTHING;
+        INSERT INTO "principal" VALUES ('p8', 't8', 'p8') ON CONFLICT DO NOTHING;
+        INSERT INTO "mailbox"."mailbox_state" VALUES ('t8', 'p8', 'INBOX', 1, 3, 3);
+        INSERT INTO "mailbox"."principal_mail"
+          ("id", "tenant_id", "principal_id", "address", "direction", "raw",
+           "message_id", "sender_address", "created_at", "uid", "modseq")
+        VALUES
+          ('a', 't8', 'p8', 'x', 'inbound', '\\x00', '<m@x>', 'f@x', '2026-01-01', 1, 1),
+          ('b', 't8', 'p8', 'x', 'inbound', '\\x00', '<m@x>', 'f@x', '2026-01-02', 2, 2),
+          ('c', 't8', 'p8', 'x', 'inbound', '\\x00', '<n@x>', 'f@x', '2026-01-03', 2, 3),
+          ('d', 't8', 'p8', 'x', 'inbound', '\\x00', NULL, 'f@x', '2026-01-04', 2, 3),
+          ('e', 't8', 'p8', 'x', 'inbound', '\\x00', '<m@x>', 'g@x', '2026-01-05', 5, 4);
+      `);
+
+      await applyMailboxMigrations(db, "public");
+
+      const rows = await admin<
+        { id: string; uid: string; message_id: string | null }[]
+      >`SELECT "id", "uid", "message_id" FROM "mailbox"."principal_mail"
+        WHERE "tenant_id" = 't8' ORDER BY "id"`;
+      expect(rows.map((r) => [r.id, Number(r.uid), r.message_id])).toEqual([
+        ["a", 1, "<m@x>"],
+        ["b", 2, null],
+        ["c", 6, "<n@x>"],
+        ["d", 7, null],
+        ["e", 5, "<m@x>"],
+      ]);
+      const [state] = await admin<{ uid_next: string }[]>`
+        SELECT "uid_next" FROM "mailbox"."mailbox_state" WHERE "tenant_id" = 't8'`;
+      expect(Number(state!.uid_next)).toBe(8);
+    });
+  });
+
+  test("0008 renumbers a mailbox with no mailbox_state row and creates one past its uids", async () => {
+    await fromEmpty(async ({ db }) => {
+      await applyMailboxMigrations(db, "public");
+      await admin.unsafe(`
+        DROP INDEX "mailbox"."principal_mail_tenant_id_principal_id_folder_uid_idx";
+        INSERT INTO "tenant" VALUES ('t8') ON CONFLICT DO NOTHING;
+        INSERT INTO "principal" VALUES ('p8', 't8', 'p8') ON CONFLICT DO NOTHING;
+        INSERT INTO "mailbox"."principal_mail"
+          ("id", "tenant_id", "principal_id", "address", "direction", "raw",
+           "sender_address", "created_at", "folder", "uid", "modseq")
+        VALUES
+          ('a', 't8', 'p8', 'x', 'inbound', '\\x00', 'f@x', '2026-01-01', 'Archive', 1, 1),
+          ('b', 't8', 'p8', 'x', 'inbound', '\\x00', 'f@x', '2026-01-02', 'Archive', 1, 2),
+          ('c', 't8', 'p8', 'x', 'inbound', '\\x00', 'f@x', '2026-01-03', 'Archive', 2, 3);
+      `);
+
+      await applyMailboxMigrations(db, "public");
+
+      const rows = await admin<{ id: string; uid: string }[]>`
+        SELECT "id", "uid" FROM "mailbox"."principal_mail"
+        WHERE "tenant_id" = 't8' ORDER BY "id"`;
+      expect(rows.map((r) => [r.id, Number(r.uid)])).toEqual([
+        ["a", 1],
+        ["b", 3],
+        ["c", 2],
+      ]);
+      const state = await admin<{ uid_next: string; highest_modseq: string }[]>`
+        SELECT "uid_next", "highest_modseq" FROM "mailbox"."mailbox_state"
+        WHERE "tenant_id" = 't8' AND "folder" = 'Archive'`;
+      expect(
+        state.map((r) => [Number(r.uid_next), Number(r.highest_modseq)]),
+      ).toEqual([[4, 3]]);
+    });
+  });
+
   test("0005 leaves uid and modseq NOT NULL: every write path is the native store now", async () => {
     await fromEmpty(async ({ db }) => {
       await applyMailboxMigrations(db, "public");

e2e/native-store.test.ts

@@ -129,7 +129,11 @@ describe("native MailboxStore over the principal mailbox tables", () => {
 
     // Moving again gets the next uid in Archive, not a collision.
     const inbox2 = await store.open("INBOX");
-    const uid2 = inbox2.append(new Uint8Array([10]), envelope(), []);
+    const uid2 = inbox2.append(
+      new Uint8Array([10]),
+      envelope({ messageId: "<m2@example.com>" }),
+      [],
+    );
     await inbox2.settled;
     const secondMove = await store.move("INBOX", uid2, "Archive");
     expect(secondMove).toBe(2);

e2e/persist.test.ts

@@ -599,3 +599,33 @@ describe("frame size and recipient hard caps", () => {
     expect(calls).toHaveLength(2);
   });
 });
+
+describe("Message-ID dedupe", () => {
+  test("keys on the authorized envelope sender, not the From header", async () => {
+    const persist = createMailboxPersist(db, {
+      upstream: async () => undefined,
+      authorizeSender: () => ACTIVE,
+    });
+    // A forged From reusing SENDER's Message-ID from another envelope sender
+    // lands first, and SENDER's own mail still delivers.
+    await persist(args({ senderAddress: "intruder@acme.example" }));
+    await persist(args());
+    // A retry from SENDER with a reformatted From is still a duplicate.
+    await persist(
+      args({
+        raw: buildMailFrame({
+          from: `Heartbeat <${SENDER}>`,
+          to: "usr_user-1@acme.example",
+          subject: "Run finished",
+          body: "Body",
+          messageId: "<fixed@acme.example>",
+        }),
+      }),
+    );
+    const rows = await rowsFor("acme", "user-1");
+    expect(rows.map((r) => r.senderAddress).sort()).toEqual([
+      SENDER,
+      "intruder@acme.example",
+    ]);
+  });
+});

e2e/upgrade-from-0.1.0.test.ts

@@ -4,6 +4,7 @@ import {
   buildMailFrame,
   createMailboxPersist,
   moveNativeMailboxMessage,
+  openNativeMailboxStore as openPublishedStore,
   runMailboxMigrations as runPublishedMigrations,
 } from "@corbits/mailbox-0.1.0";
 import { runMailboxMigrations, type MailboxDb } from "../src/index.js";
@@ -23,15 +24,17 @@ afterAll(async () => {
 type Snapshot = {
   mail: { folder: string; flags: string[] }[];
   state: unknown[];
+  columns: string[];
+  indexes: string[];
 };
 
 async function snapshot(db: MailboxDb): Promise<Snapshot> {
   const [row] = await db.execute<{ snapshot: Snapshot }>(sql`
     SELECT jsonb_build_object(
-      'mail', (SELECT jsonb_agg(to_jsonb(m) ORDER BY m."id") FROM "mailbox"."principal_mail" m),
+      'mail', (SELECT jsonb_agg(to_jsonb(m) - 'sender_address' ORDER BY m."id") FROM "mailbox"."principal_mail" m),
       'state', (SELECT jsonb_agg(to_jsonb(s) ORDER BY s."tenant_id", s."principal_id", s."folder")
                   FROM "mailbox"."mailbox_state" s),
-      'columns', (SELECT jsonb_agg(jsonb_build_array(table_name, column_name, data_type, is_nullable, column_default)
+      'columns', (SELECT jsonb_agg(concat_ws(' ', table_name, column_name, data_type, is_nullable, column_default)
                            ORDER BY table_name, column_name)
                     FROM information_schema.columns
                     WHERE table_schema = 'mailbox' AND table_name <> 'corbits_mailbox_migrations'),
@@ -102,8 +105,73 @@ test("a database migrated by the published 0.1.0 runner upgrades with every row
 
   await runMailboxMigrations(config, { schema: "public" });
   expect(await ledgerExists(db)).toBe(false);
-  expect(await snapshot(db)).toEqual(before);
+  const after = await snapshot(db);
+  // 0008 adds sender_address, backfilled from the From, and the uid and
+  // Message-ID unique indexes, and nothing else.
+  expect(after.mail).toEqual(before.mail);
+  expect(after.state).toEqual(before.state);
+  expect(after.columns.filter((c) => !before.columns.includes(c))).toEqual([
+    "principal_mail sender_address text NO ''::text",
+  ]);
+  expect(before.columns.filter((c) => !after.columns.includes(c))).toEqual([]);
+  const senders = await db.execute<{ same: boolean }>(
+    sql`SELECT "sender_address" = "from_address" AS "same" FROM "mailbox"."principal_mail"`,
+  );
+  expect(senders.map((r) => r.same)).toEqual([true, true]);
+  expect(after.indexes.filter((def) => !before.indexes.includes(def))).toEqual([
+    "CREATE UNIQUE INDEX principal_mail_folder_message_id_sender_address_idx ON mailbox.principal_mail USING btree (tenant_id, principal_id, folder, message_id, sender_address) WHERE (message_id IS NOT NULL)",
+    "CREATE UNIQUE INDEX principal_mail_tenant_id_principal_id_folder_uid_idx ON mailbox.principal_mail USING btree (tenant_id, principal_id, folder, uid)",
+  ]);
 
   await runMailboxMigrations(config, { schema: "public" });
-  expect(await snapshot(db)).toEqual(before);
+  expect(await snapshot(db)).toEqual(after);
+});
+
+test("a 0.1.0 database whose racing appends left duplicate uids and uid_next behind upgrades", async () => {
+  const { db, config, close } = await createEmptyTestDb();
+  try {
+    await runPublishedMigrations(db);
+    await seedScope(db, "t2", "carol");
+    const scope = { tenantId: "t2", principalId: "carol", folder: "INBOX" };
+    const envelope = (messageId: string) => ({
+      messageId,
+      from: "f@t2.example",
+      to: ["carol@t2.example"],
+      subject: messageId,
+      date: new Date(),
+      inReplyTo: undefined,
+      references: [],
+      interchangeType: undefined,
+      interchangeCorrelationId: undefined,
+    });
+    // Two 0.1.0 stores open at uid_next 1: one appends uids 1-3, then the other
+    // appends uid 1 and writes uid_next back to 2.
+    const first = await openPublishedStore(db, scope);
+    const second = await openPublishedStore(db, scope);
+    for (const id of ["<a1@x>", "<a2@x>", "<a3@x>"]) {
+      first.append(new Uint8Array([1]), envelope(id), []);
+    }
+    await first.settled;
+    second.append(new Uint8Array([1]), envelope("<b1@x>"), []);
+    await second.settled;
+
+    await runMailboxMigrations(config, { schema: "public" });
+
+    const rows = await db.execute<{ message_id: string; uid: string }>(
+      sql`SELECT "message_id", "uid" FROM "mailbox"."principal_mail"
+        WHERE "tenant_id" = 't2' ORDER BY "uid"`,
+    );
+    expect(rows.map((r) => [r.message_id, Number(r.uid)])).toEqual([
+      ["<a1@x>", 1],
+      ["<a2@x>", 2],
+      ["<a3@x>", 3],
+      ["<b1@x>", 4],
+    ]);
+    const [state] = await db.execute<{ uid_next: string }>(
+      sql`SELECT "uid_next" FROM "mailbox"."mailbox_state" WHERE "tenant_id" = 't2'`,
+    );
+    expect(Number(state!.uid_next)).toBe(5);
+  } finally {
+    await close();
+  }
 });

e2e/write-failure.test.ts

@@ -102,7 +102,7 @@ describe("a refused append", () => {
       interchangeCorrelationId: undefined,
     };
     await expect(
-      store.appendMessage(new Uint8Array([1]), envelope, []),
+      store.appendMessage(new Uint8Array([1]), envelope, [], envelope.from),
     ).rejects.toThrow();
     expect(store.uidNext).toBe(1);
     expect(store.highestModSeq).toBe(0);

e2e/write.test.ts

@@ -193,3 +193,36 @@ describe("deliverInboxItems", () => {
     expect(results[0]!.id).not.toBeNull();
   });
 });
+
+describe("concurrent appends to one mailbox", () => {
+  test("five writes get five distinct uids", async () => {
+    const written = await Promise.all(
+      [0, 1, 2, 3, 4].map((i) =>
+        writeMailboxMessage(db, args({ subject: `m${i}` })),
+      ),
+    );
+    expect(written.map((w) => w?.uid).sort()).toEqual([1, 2, 3, 4, 5]);
+    const store = await openNativeMailboxStore(db, {
+      tenantId: "t1",
+      principalId: "p1",
+      folder: "INBOX",
+    });
+    expect(store.messages.map((m) => m.uid)).toEqual([1, 2, 3, 4, 5]);
+    expect(store.uidNext).toBe(6);
+  });
+
+  test("three writes with one Message-ID land one row", async () => {
+    const written = await Promise.all(
+      [0, 1, 2].map(() =>
+        writeMailboxMessage(db, args({ messageId: "<same@t1.example>" })),
+      ),
+    );
+    expect(written.filter((w) => w !== null)).toHaveLength(1);
+    const store = await openNativeMailboxStore(db, {
+      tenantId: "t1",
+      principalId: "p1",
+      folder: "INBOX",
+    });
+    expect(store.messages).toHaveLength(1);
+  });
+});