From 4623126355224dd02343ecb23f4d062d62cac387 Mon Sep 17 00:00:00 2001
From: TJ Baker <1617679+zaridan@users.noreply.github.com>
Date: Mon, 13 Jul 2026 17:23:48 -0700
Subject: [PATCH 1/3] =?UTF-8?q?feat(db):=20inbound-ingestion=20persistence?=
 =?UTF-8?q?=20schema=20=E2=80=94=20migrations=20009-012=20(HT-36)?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Adds the durable state the inbound pipeline needs, all mailbox-namespaced: 009 mailboxes (isolation anchor; address UNIQUE, status CHECK); 010 mailbox_oauth_tokens (per-mailbox OAuth secrets); 011 gmail_watch_state (Gmail historyId cursor + watch expiry, kept out of the provider-agnostic mailboxes table); 012 inbound_deliveries (delivery ledger with a UNIQUE (mailbox_id, provider_message_id) claim key — the idempotency record, claim/lease, and retry queue, spec §4).

SECURITY: both the refresh token AND the short-lived access token are stored as ciphertext (bytea), not plaintext — a DB dump alone must not yield usable mailbox access; HT-38 owns the crypto. inbound_deliveries.conversation_id/thread_id are ON DELETE SET NULL so the ingestion fact survives conversation deletion (invariant #1).

Schema + migration tests only; no store methods or pipeline code. Implements specs/mail/inbound-ingestion.md §4 + gmail-push.md §4/§6. Gates green: typecheck, biome lint, 403 tests.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
---
 src/db/migrate.test.ts  | 313 ++++++++++++++++++++++++++++++++++++++++
 src/db/migrate.ts       | 261 +++++++++++++++++++++++++++++++++
 src/db/postgres.test.ts |   4 +
 3 files changed, 578 insertions(+)

diff --git a/src/db/migrate.test.ts b/src/db/migrate.test.ts
index d3b7cd7..d3508c1 100644
--- a/src/db/migrate.test.ts
+++ b/src/db/migrate.test.ts
@@ -48,6 +48,10 @@ describe('migrate', () => {
       { id: 6, name: 'tags_and_assignee' },
       { id: 7, name: 'note_thread_direction' },
       { id: 8, name: 'customer_viewed_at' },
+      { id: 9, name: 'mailboxes' },
+      { id: 10, name: 'mailbox_oauth_tokens' },
+      { id: 11, name: 'gmail_watch_state' },
+      { id: 12, name: 'inbound_deliveries' },
     ])
   })
 
@@ -66,6 +70,10 @@ describe('migrate', () => {
       { id: 6 },
       { id: 7 },
       { id: 8 },
+      { id: 9 },
+      { id: 10 },
+      { id: 11 },
+      { id: 12 },
     ])
   })
 
@@ -526,4 +534,309 @@ describe('migrate', () => {
       ).rejects.toThrow()
     }
   })
+
+  it('migration 009 creates mailboxes with a default status, enforces the address UNIQUE constraint and the status CHECK', async () => {
+    db = await createPgliteDb()
+    await migrate(db)
+
+    const [mailbox] = await db.query<{
+      id: string
+      status: string
+      created_at: string
+      updated_at: string
+    }>(
+      `INSERT INTO mailboxes (address, provider) VALUES ($1, $2)
+       RETURNING id, status, created_at, updated_at`,
+      ['support@example.test', 'gmail'],
+    )
+    expect(mailbox.id).toMatch(/^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$/)
+    expect(mailbox.status).toBe('active')
+    expect(mailbox.created_at).toBeDefined()
+    expect(mailbox.updated_at).toBeDefined()
+
+    // A second mailbox at the SAME address collides — gmail-push.md §3 needs
+    // emailAddress to resolve to exactly one mailbox.
+    await expect(
+      db.query('INSERT INTO mailboxes (address, provider) VALUES ($1, $2)', [
+        'support@example.test',
+        'gmail',
+      ]),
+    ).rejects.toThrow()
+
+    // A different address with an explicit, legal non-default status is fine.
+    await expect(
+      db.query('INSERT INTO mailboxes (address, provider, status) VALUES ($1, $2, $3)', [
+        'ops@example.test',
+        'gmail',
+        'needs_reconnect',
+      ]),
+    ).resolves.toBeDefined()
+
+    // An out-of-domain status is rejected.
+    await expect(
+      db.query('INSERT INTO mailboxes (address, provider, status) VALUES ($1, $2, $3)', [
+        'billing@example.test',
+        'gmail',
+        'bogus',
+      ]),
+    ).rejects.toThrow()
+  })
+
+  it('migration 010 stores OAuth ciphertext bytes keyed one-to-one by mailbox, enforces NOT NULL and the FK, and cascades on mailbox delete', async () => {
+    db = await createPgliteDb()
+    await migrate(db)
+
+    const [mailbox] = await db.query<{ id: string }>(
+      'INSERT INTO mailboxes (address, provider) VALUES ($1, $2) RETURNING id',
+      ['support@example.test', 'gmail'],
+    )
+
+    const ciphertext = new Uint8Array([1, 2, 3, 253, 254, 255])
+    const accessCiphertext = new Uint8Array([10, 20, 30, 250, 251, 252])
+    const [token] = await db.query<{
+      mailbox_id: string
+      refresh_token_ciphertext: Uint8Array
+      access_token_ciphertext: Uint8Array | null
+      scopes: string | null
+    }>(
+      `INSERT INTO mailbox_oauth_tokens (mailbox_id, refresh_token_ciphertext, access_token_ciphertext, scopes)
+       VALUES ($1, $2, $3, $4)
+       RETURNING mailbox_id, refresh_token_ciphertext, access_token_ciphertext, scopes`,
+      [mailbox.id, ciphertext, accessCiphertext, 'https://www.googleapis.com/auth/gmail.readonly'],
+    )
+    expect(token.mailbox_id).toBe(mailbox.id)
+    // Both secrets round-trip as genuine bytes, not a re-encoded string —
+    // same proof shape as src/db/postgres.test.ts's bytea round-trip test.
+    expect(Buffer.from(token.refresh_token_ciphertext)).toEqual(Buffer.from(ciphertext))
+    expect(Buffer.from(token.access_token_ciphertext as Uint8Array)).toEqual(
+      Buffer.from(accessCiphertext),
+    )
+    expect(token.scopes).toBe('https://www.googleapis.com/auth/gmail.readonly')
+
+    // A second row for the SAME mailbox collides — mailbox_id is the PK
+    // (one OAuth grant per connected mailbox).
+    await expect(
+      db.query(
+        'INSERT INTO mailbox_oauth_tokens (mailbox_id, refresh_token_ciphertext) VALUES ($1, $2)',
+        [mailbox.id, ciphertext],
+      ),
+    ).rejects.toThrow()
+
+    // A nonexistent mailbox_id violates the FK.
+    await expect(
+      db.query(
+        'INSERT INTO mailbox_oauth_tokens (mailbox_id, refresh_token_ciphertext) VALUES ($1, $2)',
+        ['00000000-0000-0000-0000-000000000000', ciphertext],
+      ),
+    ).rejects.toThrow()
+
+    // A row with no ciphertext at all violates NOT NULL — there is no legal
+    // "connected but tokenless" row.
+    const [bareMailbox] = await db.query<{ id: string }>(
+      'INSERT INTO mailboxes (address, provider) VALUES ($1, $2) RETURNING id',
+      ['bare@example.test', 'gmail'],
+    )
+    await expect(
+      db.query('INSERT INTO mailbox_oauth_tokens (mailbox_id) VALUES ($1)', [bareMailbox.id]),
+    ).rejects.toThrow()
+
+    // Deleting the mailbox cascades to its token row.
+    await db.query('DELETE FROM mailboxes WHERE id = $1', [mailbox.id])
+    const remaining = await db.query(
+      'SELECT mailbox_id FROM mailbox_oauth_tokens WHERE mailbox_id = $1',
+      [mailbox.id],
+    )
+    expect(remaining).toEqual([])
+  })
+
+  it('migration 011 stores a nullable Gmail cursor keyed one-to-one by mailbox, and cascades on mailbox delete', async () => {
+    db = await createPgliteDb()
+    await migrate(db)
+
+    const [mailbox] = await db.query<{ id: string }>(
+      'INSERT INTO mailboxes (address, provider) VALUES ($1, $2) RETURNING id',
+      ['support@example.test', 'gmail'],
+    )
+
+    // No cursor yet (between connect and the first successful watch()) is legal.
+    const [bare] = await db.query<{
+      history_id: string | null
+      watch_expiration: string | null
+    }>(
+      'INSERT INTO gmail_watch_state (mailbox_id) VALUES ($1) RETURNING history_id, watch_expiration',
+      [mailbox.id],
+    )
+    expect(bare.history_id).toBeNull()
+    expect(bare.watch_expiration).toBeNull()
+
+    // A second row for the SAME mailbox collides — mailbox_id is the PK.
+    await expect(
+      db.query('INSERT INTO gmail_watch_state (mailbox_id) VALUES ($1)', [mailbox.id]),
+    ).rejects.toThrow()
+
+    // Once watch() succeeds, both columns are populated — history_id stays a
+    // string (Gmail's own wire type), never coerced to a number.
+    await db.query(
+      'UPDATE gmail_watch_state SET history_id = $1, watch_expiration = $2 WHERE mailbox_id = $3',
+      ['123456789', '2026-07-20T00:00:00.000Z', mailbox.id],
+    )
+    const [updated] = await db.query<{ history_id: string | null }>(
+      'SELECT history_id FROM gmail_watch_state WHERE mailbox_id = $1',
+      [mailbox.id],
+    )
+    expect(updated.history_id).toBe('123456789')
+
+    // Deleting the mailbox cascades to its watch-state row.
+    await db.query('DELETE FROM mailboxes WHERE id = $1', [mailbox.id])
+    const remaining = await db.query(
+      'SELECT mailbox_id FROM gmail_watch_state WHERE mailbox_id = $1',
+      [mailbox.id],
+    )
+    expect(remaining).toEqual([])
+  })
+
+  it('migration 012 enforces the (mailbox_id, provider_message_id) claim key and its defaults', async () => {
+    db = await createPgliteDb()
+    await migrate(db)
+
+    const [mailbox] = await db.query<{ id: string }>(
+      'INSERT INTO mailboxes (address, provider) VALUES ($1, $2) RETURNING id',
+      ['support@example.test', 'gmail'],
+    )
+
+    const [delivery] = await db.query<{
+      id: string
+      status: string
+      attempts: number
+      last_error: string | null
+      conversation_id: string | null
+      thread_id: string | null
+    }>(
+      `INSERT INTO inbound_deliveries (mailbox_id, provider_message_id) VALUES ($1, $2)
+       RETURNING id, status, attempts, last_error, conversation_id, thread_id`,
+      [mailbox.id, 'gmail-msg-1'],
+    )
+    expect(delivery.id).toMatch(/^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$/)
+    expect(delivery.status).toBe('received')
+    expect(delivery.attempts).toBe(0)
+    expect(delivery.last_error).toBeNull()
+    expect(delivery.conversation_id).toBeNull()
+    expect(delivery.thread_id).toBeNull()
+
+    // A plain second INSERT of the SAME (mailbox_id, provider_message_id)
+    // violates the unique claim key outright.
+    await expect(
+      db.query('INSERT INTO inbound_deliveries (mailbox_id, provider_message_id) VALUES ($1, $2)', [
+        mailbox.id,
+        'gmail-msg-1',
+      ]),
+    ).rejects.toThrow()
+
+    // The EXACT claim pattern the pipeline uses (spec §3 step 1): a conflict
+    // is absorbed, not thrown — 0 rows back, so the caller re-reads the
+    // winner's row instead of double-processing.
+    const claimed = await db.query(
+      `INSERT INTO inbound_deliveries (mailbox_id, provider_message_id)
+       VALUES ($1, $2)
+       ON CONFLICT (mailbox_id, provider_message_id) DO NOTHING
+       RETURNING id`,
+      [mailbox.id, 'gmail-msg-1'],
+    )
+    expect(claimed).toEqual([])
+
+    // The SAME provider_message_id at a DIFFERENT mailbox is not a
+    // collision — the claim key is the pair, not provider_message_id alone.
+    const [otherMailbox] = await db.query<{ id: string }>(
+      'INSERT INTO mailboxes (address, provider) VALUES ($1, $2) RETURNING id',
+      ['ops@example.test', 'gmail'],
+    )
+    await expect(
+      db.query('INSERT INTO inbound_deliveries (mailbox_id, provider_message_id) VALUES ($1, $2)', [
+        otherMailbox.id,
+        'gmail-msg-1',
+      ]),
+    ).resolves.toBeDefined()
+  })
+
+  it('migration 012 CHECKs status against the closed set, spelled dead-letter with a hyphen', async () => {
+    db = await createPgliteDb()
+    await migrate(db)
+
+    const [mailbox] = await db.query<{ id: string }>(
+      'INSERT INTO mailboxes (address, provider) VALUES ($1, $2) RETURNING id',
+      ['support@example.test', 'gmail'],
+    )
+
+    for (const status of ['received', 'stored', 'suppressed', 'failed', 'dead-letter']) {
+      await expect(
+        db.query(
+          'INSERT INTO inbound_deliveries (mailbox_id, provider_message_id, status) VALUES ($1, $2, $3)',
+          [mailbox.id, `msg-${status}`, status],
+        ),
+      ).resolves.toBeDefined()
+    }
+
+    // The ticket text's underscore spelling is NOT the spec's — rejected.
+    await expect(
+      db.query(
+        'INSERT INTO inbound_deliveries (mailbox_id, provider_message_id, status) VALUES ($1, $2, $3)',
+        [mailbox.id, 'msg-bad-spelling', 'dead_letter'],
+      ),
+    ).rejects.toThrow()
+
+    await expect(
+      db.query(
+        'INSERT INTO inbound_deliveries (mailbox_id, provider_message_id, status) VALUES ($1, $2, $3)',
+        [mailbox.id, 'msg-bogus', 'bogus'],
+      ),
+    ).rejects.toThrow()
+  })
+
+  it('migration 012 ties conversation_id/thread_id to real rows via FK, and clears them (SET NULL) rather than deleting the ledger row when the conversation is removed', async () => {
+    db = await createPgliteDb()
+    await migrate(db)
+
+    const [mailbox] = await db.query<{ id: string }>(
+      'INSERT INTO mailboxes (address, provider) VALUES ($1, $2) RETURNING id',
+      ['support@example.test', 'gmail'],
+    )
+    const [conversation] = await db.query<{ id: string }>(
+      'INSERT INTO conversations (customer_email) VALUES ($1) RETURNING id',
+      ['customer@example.test'],
+    )
+    const [thread] = await db.query<{ id: string }>(
+      `INSERT INTO threads (conversation_id, direction, from_address)
+       VALUES ($1, 'inbound', $2) RETURNING id`,
+      [conversation.id, 'customer@example.test'],
+    )
+
+    // A nonexistent conversation_id violates the FK.
+    await expect(
+      db.query(
+        `INSERT INTO inbound_deliveries (mailbox_id, provider_message_id, status, conversation_id, thread_id)
+         VALUES ($1, $2, 'stored', $3, $4)`,
+        [mailbox.id, 'gmail-msg-1', '00000000-0000-0000-0000-000000000000', thread.id],
+      ),
+    ).rejects.toThrow()
+
+    const [delivery] = await db.query<{ id: string }>(
+      `INSERT INTO inbound_deliveries (mailbox_id, provider_message_id, status, conversation_id, thread_id)
+       VALUES ($1, $2, 'stored', $3, $4) RETURNING id`,
+      [mailbox.id, 'gmail-msg-2', conversation.id, thread.id],
+    )
+
+    // Deleting the conversation (which cascades to its thread, migration
+    // 001) must NOT delete the ledger row — the ingestion fact survives;
+    // only the now-unresolvable pointers clear.
+    await db.query('DELETE FROM conversations WHERE id = $1', [conversation.id])
+
+    const [afterDelete] = await db.query<{
+      id: string
+      conversation_id: string | null
+      thread_id: string | null
+    }>('SELECT id, conversation_id, thread_id FROM inbound_deliveries WHERE id = $1', [delivery.id])
+    expect(afterDelete.id).toBe(delivery.id)
+    expect(afterDelete.conversation_id).toBeNull()
+    expect(afterDelete.thread_id).toBeNull()
+  })
 })
diff --git a/src/db/migrate.ts b/src/db/migrate.ts
index 0a0831d..5fed64b 100644
--- a/src/db/migrate.ts
+++ b/src/db/migrate.ts
@@ -318,6 +318,247 @@ ALTER TABLE threads ADD CONSTRAINT threads_customer_viewed_at_outbound_only CHEC
 );
 `
 
+/**
+ * Migration 009 — `mailboxes`, the inbound-ingestion namespace anchor
+ * (HT-36; specs/mail/inbound-ingestion.md §2, §7).
+ *
+ * One row per connected mailbox. `id` is `mailboxId` everywhere else in the
+ * mail-ingestion specs (inbound-ingestion.md §2, gmail-push.md §3) — the
+ * value every other table this migration group adds is namespaced by, and
+ * the anchor for storage, blob keys, and dedup today, and tenancy later
+ * (inbound-ingestion.md §7: "the schema carries mailboxId from day one...
+ * but behavior is single-tenant for the dogfood").
+ *
+ * - `address` is UNIQUE: gmail-push.md §3 resolves a push notification's
+ *   `emailAddress` to "a known, active connected mailbox" and rejects
+ *   anything that doesn't map to exactly one — a duplicate address would
+ *   make that resolution ambiguous, so uniqueness is enforced here rather
+ *   than trusted to application code.
+ * - `provider` is plain `text`, deliberately NOT CHECK-constrained (unlike
+ *   `status` below). Constraining it to a fixed list would couple a
+ *   provider-agnostic pipeline (inbound-ingestion.md's own framing) to a
+ *   schema migration every time a new transport ships an adapter
+ *   (`src/providers/inbound-email.ts` already anticipates "Postmark inbound,
+ *   SES inbound, etc." arriving as adapter code, not schema changes);
+ *   `'gmail'` is simply the only value written today.
+ * - `status` IS CHECK-constrained — a mailbox's own lifecycle is a small,
+ *   engine-owned set, matching this file's standing convention of
+ *   CHECK-constraining every closed-set lifecycle column
+ *   (`conversations.status`, `threads.direction`, `threads.delivery_status`).
+ *   `'needs_reconnect'` is the state gmail-push.md §5 (an expired/404 history
+ *   cursor) and §6 (a failed `watch()` renewal) put a mailbox into —
+ *   operator-visible and resolvable, never a silent failure. `'paused'` is
+ *   the deliberate dogfood response to §5's expired-cursor case ("pause the
+ *   mailbox and flag it for manual rebaseline"). Default `'active'`: a
+ *   mailbox starts usable the moment it is connected (HT-40).
+ *
+ * No `updated_at` trigger: exactly like `conversations`/`threads`, this
+ * schema has no auto-bump mechanism anywhere — `updated_at` is maintained by
+ * whichever application code writes the row (a later ticket for this table;
+ * HT-36 is schema only).
+ */
+const MIGRATION_009_MAILBOXES = `
+CREATE TABLE mailboxes (
+  id uuid PRIMARY KEY DEFAULT gen_random_uuid(),
+  address text NOT NULL UNIQUE,
+  provider text NOT NULL,
+  status text NOT NULL DEFAULT 'active' CHECK (status IN ('active','paused','needs_reconnect')),
+  created_at timestamptz NOT NULL DEFAULT now(),
+  updated_at timestamptz NOT NULL DEFAULT now()
+);
+`
+
+/**
+ * Migration 010 — `mailbox_oauth_tokens`, per-mailbox OAuth credential
+ * storage (HT-36, schema only; gmail-push.md §7: "OAuth token
+ * acquisition/refresh → HT-38; the connect/consent flow → HT-40").
+ *
+ * `mailbox_id` is the PRIMARY KEY, not a separate surrogate `id` — this is a
+ * per-mailbox singleton (one OAuth grant per connected mailbox today), the
+ * same 1:1-sidecar shape `gmail_watch_state` below uses, deliberately kept
+ * consistent between the two.
+ *
+ * ## This migration stores ciphertext. It does not encrypt anything.
+ *
+ * `refresh_token_ciphertext` is `bytea` — opaque encrypted bytes — and
+ * `NOT NULL` because a row only exists once an OAuth grant actually produced
+ * a refresh token (HT-40); there is no legal "connected but tokenless" row.
+ * **No encryption or decryption logic exists anywhere in this codebase
+ * yet.** HT-38 ("OAuth token acquisition/refresh") is the ticket that
+ * implements the actual encrypt/decrypt and is the only code ever meant to
+ * hold plaintext; this migration only reserves the column shape a
+ * ciphertext value will live in. `bytea` (not `text`) because encrypted
+ * output is arbitrary binary, not necessarily valid text — and because
+ * `SqlValue` (`src/db/client.ts`) already treats `Uint8Array` as a
+ * first-class bindable value precisely for columns like this one (see the
+ * `pg`/PGlite round-trip proof in `src/db/postgres.test.ts`).
+ *
+ * `access_token_ciphertext`/`access_token_expires_at` are the short-lived
+ * (~1h, for Gmail) OAuth access-token cache — nullable (absent until the
+ * first token exchange). The access token is ALSO stored as ciphertext
+ * (`bytea`), not plaintext: it is itself a bearer credential that grants
+ * mailbox access for its whole lifetime, so a database dump alone must not
+ * yield usable mailbox access even for that ~1h window. Encrypting BOTH
+ * secrets means an attacker needs the encryption key (held only by HT-38's
+ * code, never the DB) to use either — a plaintext access-token column would
+ * hand a DB thief ~1h of live mailbox access for free, defeating the point
+ * of encrypting the refresh token beside it. As with the refresh token,
+ * HT-38 owns the encrypt/decrypt; this migration only reserves the column.
+ *
+ * `scopes` is raw nullable `text` — the OAuth token endpoint's own
+ * space-delimited `scope` string (RFC 6749 §5.1), stored verbatim and
+ * unparsed, not a `jsonb` array like `conversations.tags`. This is provider
+ * metadata for audit/debugging, not a queried or filtered feature, so no
+ * structure is imposed on it until something actually needs one — the
+ * `jsonb` alternative is noted as an open option in the implementation
+ * report.
+ *
+ * `ON DELETE CASCADE` mirrors this schema's one existing FK precedent
+ * (`threads.conversation_id`, migration 001): a token row has no purpose
+ * once its owning mailbox is gone.
+ */
+const MIGRATION_010_MAILBOX_OAUTH_TOKENS = `
+CREATE TABLE mailbox_oauth_tokens (
+  mailbox_id uuid PRIMARY KEY REFERENCES mailboxes(id) ON DELETE CASCADE,
+  refresh_token_ciphertext bytea NOT NULL,
+  access_token_ciphertext bytea,
+  access_token_expires_at timestamptz,
+  scopes text,
+  updated_at timestamptz NOT NULL DEFAULT now()
+);
+`
+
+/**
+ * Migration 011 — `gmail_watch_state`, per-mailbox Gmail push cursor state
+ * (HT-36; gmail-push.md §4 "the cursor", §6 "watch() renewal").
+ *
+ * Kept as its own table, OUT of the generic `mailboxes` schema, on purpose:
+ * inbound-ingestion.md's pipeline is provider-agnostic and never reads this
+ * table — only the Gmail transport (gmail-push.md) does — so a future
+ * non-Gmail provider (the forwarding-address transport, or any other) adds
+ * nothing here and this table needs no change for it to ship. Same
+ * 1:1-sidecar shape as `mailbox_oauth_tokens`: `mailbox_id` is the PRIMARY
+ * KEY (one watch state per mailbox), not a separate surrogate `id`.
+ *
+ * - `history_id` is `text`, not an integer type, even though Gmail's
+ *   `historyId` is numeric-looking. Gmail's own API represents it as a
+ *   string, the engine only ever treats it as an opaque watermark —
+ *   compared and passed back to `history.list?startHistoryId=`, never
+ *   arithmetic'd (gmail-push.md §1: "historyId is a watermark, not a
+ *   message id") — and `text` sidesteps any bigint range/precision question
+ *   entirely rather than assuming Gmail's values always fit one. Nullable:
+ *   a mailbox between connection and its first successful `watch()` call
+ *   has no cursor yet.
+ * - `watch_expiration` is nullable `timestamptz`: `watch()`'s returned
+ *   expiration (~7 days out, gmail-push.md §6), null until the first
+ *   successful `watch()`.
+ *
+ * No `created_at` (unlike `mailboxes`/`inbound_deliveries`): this is a 1:1
+ * mutable operational state whose "created" moment adds nothing beyond its
+ * owning mailbox's own `created_at` — only `updated_at` is meaningful here,
+ * tracking cursor freshness.
+ */
+const MIGRATION_011_GMAIL_WATCH_STATE = `
+CREATE TABLE gmail_watch_state (
+  mailbox_id uuid PRIMARY KEY REFERENCES mailboxes(id) ON DELETE CASCADE,
+  history_id text,
+  watch_expiration timestamptz,
+  updated_at timestamptz NOT NULL DEFAULT now()
+);
+`
+
+/**
+ * Migration 012 — `inbound_deliveries`, the delivery ledger (HT-36;
+ * specs/mail/inbound-ingestion.md §4).
+ *
+ * One row per `(mailbox_id, provider_message_id)` — simultaneously the
+ * **idempotency record**, the **claim/lease**, and the **retry queue** (spec
+ * §4's own three-way framing). `provider_message_id`, not the RFC
+ * `Message-ID`, is the dedup authority: the inbound `Message-ID` is optional
+ * and entirely sender-controlled (`NewThread.messageId` permits `null`,
+ * `src/store/conversations.ts`), while the transport's own message id is
+ * stable and provider-issued (spec §4).
+ *
+ * `id` is a conventional surrogate `uuid` PRIMARY KEY (matching every other
+ * table in this schema), separate from the UNIQUE claim key below — the
+ * same "surrogate PK + a separate business-key unique index" shape
+ * migration 003 already uses for `threads`' own idempotency key
+ * (`threads_conversation_idempotency_key_idx`).
+ *
+ * ## The claim key
+ *
+ * `inbound_deliveries_mailbox_id_provider_message_id_key` is what the
+ * ingest pipeline's step 1 targets (spec §3 step 1): `INSERT ... ON
+ * CONFLICT (mailbox_id, provider_message_id) DO NOTHING RETURNING *`. A
+ * fresh insert means the caller owns processing this delivery; a conflict
+ * means a concurrent or prior delivery already claimed or completed it, and
+ * the caller must return THAT row's outcome rather than double-process
+ * (spec §3 step 1, §8's "two concurrent deliveries... exactly one
+ * conversation" acceptance case). Ordinary `UNIQUE`, not partial: unlike
+ * `threads.idempotency_key` (optional, migration 003),
+ * `provider_message_id` is always present (spec §2: the transport rejects a
+ * delivery it cannot resolve to a `providerMessageId`), so every row
+ * participates in the constraint.
+ *
+ * `status` defaults to `'received'` — the state a row is inserted in at the
+ * step-1 claim, before parse/thread/store (steps 2-5) even run. The CHECK
+ * list is spelled `'dead-letter'` (hyphen) to match
+ * specs/mail/inbound-ingestion.md §4's own spelling, used consistently
+ * throughout that spec (and matching the industry-standard "dead-letter
+ * queue" term); HT-36's ticket text listed the same value with an
+ * underscore (`dead_letter`) in one place, which reads as a transcription
+ * slip against the spec's consistent hyphenated usage — flagged for
+ * explicit confirmation in the implementation report rather than resolved
+ * silently.
+ *
+ * `attempts`/`last_error` are the retry-queue bookkeeping the spec's "retry
+ * queue" framing implies (§4) — no schema-level opinion on the attempts
+ * ceiling or backoff; that policy belongs to the worker that consumes this
+ * table (a later ticket).
+ *
+ * `conversation_id`/`thread_id` are the ledger's recorded OUTCOME (spec §3
+ * step 5, §4: "recording the resulting conversationId/threadId"), nullable
+ * because most statuses (`received`, `suppressed`, `failed`, `dead-letter`)
+ * never resolve to one. Declared as real FKs, matching this schema's
+ * unbroken convention that every id-shaped reference column is one, but
+ * `ON DELETE SET NULL` rather than `CASCADE` (migration 001's `threads`
+ * choice): unlike a thread, which has no meaning without its conversation,
+ * a ledger row's audit/idempotency value ("we received message X for
+ * mailbox Y, and here is what happened") does not depend on the
+ * conversation it produced still existing — invariant #1's
+ * never-silently-lost applies to the fact of ingestion, not just the
+ * resulting conversation, so the ledger row survives and only the
+ * now-unresolvable pointer clears.
+ *
+ * No cross-column CHECK tying `status` to `conversation_id`/`thread_id`
+ * nullability (e.g. "non-null iff `stored`") — deliberately deferred: the
+ * exact invariant depends on retry/dead-letter edge cases the consuming
+ * store methods (a later ticket) haven't been written yet to pin down, and
+ * this ticket is schema-only. Worth adding once that implementation settles
+ * the question for real.
+ *
+ * No index beyond the UNIQUE claim key: the ticket's own framing ("the
+ * unique index IS the claim key") reads as the one index this migration
+ * needs; a `status`-scoped index for a future retry-sweep/dead-letter-review
+ * query is deferred to whichever ticket implements that query, so as not to
+ * carry write-time index cost for a read pattern that doesn't exist yet.
+ */
+const MIGRATION_012_INBOUND_DELIVERIES = `
+CREATE TABLE inbound_deliveries (
+  id uuid PRIMARY KEY DEFAULT gen_random_uuid(),
+  mailbox_id uuid NOT NULL REFERENCES mailboxes(id) ON DELETE CASCADE,
+  provider_message_id text NOT NULL,
+  status text NOT NULL DEFAULT 'received' CHECK (status IN ('received','stored','suppressed','failed','dead-letter')),
+  attempts integer NOT NULL DEFAULT 0,
+  last_error text,
+  conversation_id uuid REFERENCES conversations(id) ON DELETE SET NULL,
+  thread_id uuid REFERENCES threads(id) ON DELETE SET NULL,
+  created_at timestamptz NOT NULL DEFAULT now(),
+  updated_at timestamptz NOT NULL DEFAULT now()
+);
+CREATE UNIQUE INDEX inbound_deliveries_mailbox_id_provider_message_id_key ON inbound_deliveries (mailbox_id, provider_message_id);
+`
+
 /**
  * Every migration, in the order they must apply. `id` is the sole ordering
  * key (ascending) — array position is not relied upon, so re-sorting this
@@ -360,6 +601,26 @@ const MIGRATIONS: Migration[] = [
     name: 'customer_viewed_at',
     sql: MIGRATION_008_CUSTOMER_VIEWED_AT,
   },
+  {
+    id: 9,
+    name: 'mailboxes',
+    sql: MIGRATION_009_MAILBOXES,
+  },
+  {
+    id: 10,
+    name: 'mailbox_oauth_tokens',
+    sql: MIGRATION_010_MAILBOX_OAUTH_TOKENS,
+  },
+  {
+    id: 11,
+    name: 'gmail_watch_state',
+    sql: MIGRATION_011_GMAIL_WATCH_STATE,
+  },
+  {
+    id: 12,
+    name: 'inbound_deliveries',
+    sql: MIGRATION_012_INBOUND_DELIVERIES,
+  },
 ]
 
 /**
diff --git a/src/db/postgres.test.ts b/src/db/postgres.test.ts
index 8a44d52..854c028 100644
--- a/src/db/postgres.test.ts
+++ b/src/db/postgres.test.ts
@@ -297,6 +297,10 @@ describe('createPostgresDb with a schema option', () => {
     expect(placed.rows.map((r) => r.table_name)).toEqual([
       '_migrations',
       'conversations',
+      'gmail_watch_state',
+      'inbound_deliveries',
+      'mailbox_oauth_tokens',
+      'mailboxes',
       'threads',
     ])
 

From 0a2605806b2a8a0222fff085145aa6aaf825a12c Mon Sep 17 00:00:00 2001
From: TJ Baker <1617679+zaridan@users.noreply.github.com>
Date: Mon, 13 Jul 2026 18:31:30 -0700
Subject: [PATCH 2/3] fix(db): drop redundant
 inbound_deliveries.conversation_id (CodeRabbit, HT-36)
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Two independent FKs (conversation_id, thread_id) let a ledger row pair a conversation with a thread from a DIFFERENT conversation — a corrupt outcome. A thread already belongs to exactly one conversation (threads.conversation_id NOT NULL), so thread_id alone records the outcome and the conversation is derivable; removing the redundant column makes the mismatch impossible by construction rather than merely checked. spec §4 wording and the migration test updated to match.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
---
 specs/mail/inbound-ingestion.md |  6 +++--
 src/db/migrate.test.ts          | 39 ++++++++++++++++-----------------
 src/db/migrate.ts               | 31 ++++++++++++++------------
 3 files changed, 40 insertions(+), 36 deletions(-)

diff --git a/specs/mail/inbound-ingestion.md b/specs/mail/inbound-ingestion.md
index 8a087e9..b58e1d5 100644
--- a/specs/mail/inbound-ingestion.md
+++ b/specs/mail/inbound-ingestion.md
@@ -102,14 +102,16 @@ as data and as a *secondary* duplicate signal, never as the dedup key.
 **The delivery ledger** (a table, HT-36) is one row per `(mailboxId, providerMessageId)`
 with a **unique constraint** on that pair, carrying `status` (`received` | `stored` |
 `suppressed` | `failed` | `dead-letter`), `attempts`, `last_error`, and the resulting
-`conversationId`/`threadId`. It is simultaneously the **idempotency record** (§3 step 1),
+`threadId` (the produced/appended thread; its conversation follows from
+`threads.conversationId` and is not stored as a redundant second column). It is
+simultaneously the **idempotency record** (§3 step 1),
 the **claim/lease**, and the **retry queue**.
 
 **The claim, the store write, and the outcome are one atomic unit.** The step-5 store write
 (`createConversation`/`appendThread`) and the ledger's `received → stored` transition —
 recording the resulting ids — commit in a **single transaction**, so the ledger row *is*
 the idempotency record: a retry re-hits the §3-step-1 claim, finds a `stored` row, and
-returns its recorded `conversationId` without re-writing. A crash *before* that commit
+returns its recorded outcome (its `threadId`) without re-writing. A crash *before* that commit
 leaves the row at `received` and no conversation, and the retry redoes the whole unit
 cleanly. This is what closes the "successful conversation write, then failed ledger update,
 then duplicate conversation on retry" window — the write and its record are never
diff --git a/src/db/migrate.test.ts b/src/db/migrate.test.ts
index d3508c1..d6790d6 100644
--- a/src/db/migrate.test.ts
+++ b/src/db/migrate.test.ts
@@ -709,18 +709,16 @@ describe('migrate', () => {
       status: string
       attempts: number
       last_error: string | null
-      conversation_id: string | null
       thread_id: string | null
     }>(
       `INSERT INTO inbound_deliveries (mailbox_id, provider_message_id) VALUES ($1, $2)
-       RETURNING id, status, attempts, last_error, conversation_id, thread_id`,
+       RETURNING id, status, attempts, last_error, thread_id`,
       [mailbox.id, 'gmail-msg-1'],
     )
     expect(delivery.id).toMatch(/^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$/)
     expect(delivery.status).toBe('received')
     expect(delivery.attempts).toBe(0)
     expect(delivery.last_error).toBeNull()
-    expect(delivery.conversation_id).toBeNull()
     expect(delivery.thread_id).toBeNull()
 
     // A plain second INSERT of the SAME (mailbox_id, provider_message_id)
@@ -792,7 +790,7 @@ describe('migrate', () => {
     ).rejects.toThrow()
   })
 
-  it('migration 012 ties conversation_id/thread_id to real rows via FK, and clears them (SET NULL) rather than deleting the ledger row when the conversation is removed', async () => {
+  it('migration 012 ties thread_id to a real thread via FK, and clears it (SET NULL) rather than deleting the ledger row when the thread is removed', async () => {
     db = await createPgliteDb()
     await migrate(db)
 
@@ -810,33 +808,34 @@ describe('migrate', () => {
       [conversation.id, 'customer@example.test'],
     )
 
-    // A nonexistent conversation_id violates the FK.
+    // A nonexistent thread_id violates the FK.
     await expect(
       db.query(
-        `INSERT INTO inbound_deliveries (mailbox_id, provider_message_id, status, conversation_id, thread_id)
-         VALUES ($1, $2, 'stored', $3, $4)`,
-        [mailbox.id, 'gmail-msg-1', '00000000-0000-0000-0000-000000000000', thread.id],
+        `INSERT INTO inbound_deliveries (mailbox_id, provider_message_id, status, thread_id)
+         VALUES ($1, $2, 'stored', $3)`,
+        [mailbox.id, 'gmail-msg-1', '00000000-0000-0000-0000-000000000000'],
       ),
     ).rejects.toThrow()
 
+    // The recorded outcome is the thread; its conversation is derivable via
+    // threads.conversation_id, so there is no separate conversation_id column
+    // that could be paired with a thread from a different conversation.
     const [delivery] = await db.query<{ id: string }>(
-      `INSERT INTO inbound_deliveries (mailbox_id, provider_message_id, status, conversation_id, thread_id)
-       VALUES ($1, $2, 'stored', $3, $4) RETURNING id`,
-      [mailbox.id, 'gmail-msg-2', conversation.id, thread.id],
+      `INSERT INTO inbound_deliveries (mailbox_id, provider_message_id, status, thread_id)
+       VALUES ($1, $2, 'stored', $3) RETURNING id`,
+      [mailbox.id, 'gmail-msg-2', thread.id],
     )
 
-    // Deleting the conversation (which cascades to its thread, migration
-    // 001) must NOT delete the ledger row — the ingestion fact survives;
-    // only the now-unresolvable pointers clear.
+    // Deleting the conversation cascades to its thread (migration 001), which
+    // must NOT delete the ledger row — the ingestion fact survives; only the
+    // now-unresolvable thread pointer clears.
     await db.query('DELETE FROM conversations WHERE id = $1', [conversation.id])
 
-    const [afterDelete] = await db.query<{
-      id: string
-      conversation_id: string | null
-      thread_id: string | null
-    }>('SELECT id, conversation_id, thread_id FROM inbound_deliveries WHERE id = $1', [delivery.id])
+    const [afterDelete] = await db.query<{ id: string; thread_id: string | null }>(
+      'SELECT id, thread_id FROM inbound_deliveries WHERE id = $1',
+      [delivery.id],
+    )
     expect(afterDelete.id).toBe(delivery.id)
-    expect(afterDelete.conversation_id).toBeNull()
     expect(afterDelete.thread_id).toBeNull()
   })
 })
diff --git a/src/db/migrate.ts b/src/db/migrate.ts
index 5fed64b..3f9119e 100644
--- a/src/db/migrate.ts
+++ b/src/db/migrate.ts
@@ -516,19 +516,23 @@ CREATE TABLE gmail_watch_state (
  * ceiling or backoff; that policy belongs to the worker that consumes this
  * table (a later ticket).
  *
- * `conversation_id`/`thread_id` are the ledger's recorded OUTCOME (spec §3
- * step 5, §4: "recording the resulting conversationId/threadId"), nullable
- * because most statuses (`received`, `suppressed`, `failed`, `dead-letter`)
- * never resolve to one. Declared as real FKs, matching this schema's
- * unbroken convention that every id-shaped reference column is one, but
- * `ON DELETE SET NULL` rather than `CASCADE` (migration 001's `threads`
- * choice): unlike a thread, which has no meaning without its conversation,
- * a ledger row's audit/idempotency value ("we received message X for
- * mailbox Y, and here is what happened") does not depend on the
- * conversation it produced still existing — invariant #1's
- * never-silently-lost applies to the fact of ingestion, not just the
- * resulting conversation, so the ledger row survives and only the
- * now-unresolvable pointer clears.
+ * `thread_id` is the ledger's recorded OUTCOME (spec §3 step 5, §4),
+ * nullable because most statuses (`received`, `suppressed`, `failed`,
+ * `dead-letter`) never resolve to one. The resulting CONVERSATION is
+ * deliberately NOT a second column: a thread belongs to exactly one
+ * conversation (`threads.conversation_id`, NOT NULL, migration 001), so
+ * `thread_id` already determines it — a separate `conversation_id` would be
+ * derivable-but-denormalized, and two independent FKs would let a row pair a
+ * `conversation_id` with a `thread_id` from a DIFFERENT conversation, a
+ * corrupt outcome the schema simply should not be able to represent. Derive
+ * the conversation with a join to `threads` when an audit query needs it.
+ * Declared a real FK (this schema's convention for id-shaped columns) but
+ * `ON DELETE SET NULL` rather than `CASCADE` (unlike migration 001's
+ * `threads`): a ledger row's audit/idempotency value ("we received message X
+ * for mailbox Y, and here is what happened") does not depend on the thread it
+ * produced still existing — invariant #1's never-silently-lost applies to the
+ * fact of ingestion, so the ledger row survives and only the now-unresolvable
+ * pointer clears.
  *
  * No cross-column CHECK tying `status` to `conversation_id`/`thread_id`
  * nullability (e.g. "non-null iff `stored`") — deliberately deferred: the
@@ -551,7 +555,6 @@ CREATE TABLE inbound_deliveries (
   status text NOT NULL DEFAULT 'received' CHECK (status IN ('received','stored','suppressed','failed','dead-letter')),
   attempts integer NOT NULL DEFAULT 0,
   last_error text,
-  conversation_id uuid REFERENCES conversations(id) ON DELETE SET NULL,
   thread_id uuid REFERENCES threads(id) ON DELETE SET NULL,
   created_at timestamptz NOT NULL DEFAULT now(),
   updated_at timestamptz NOT NULL DEFAULT now()

From fce12bffc1c2bd65a4fdca628109f5ebdc072626 Mon Sep 17 00:00:00 2001
From: TJ Baker <1617679+zaridan@users.noreply.github.com>
Date: Mon, 13 Jul 2026 18:37:36 -0700
Subject: [PATCH 3/3] docs(spec): finish threadId consistency in the ledger
 atomicity contract (CodeRabbit, HT-36)
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Two remaining phrases in §3 step 5 and §4 ('resulting conversationId/threadId', 'resulting ids') still implied the removed conversation_id column. Both now say 'resulting threadId', consistent with migration 012.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
---
 specs/mail/inbound-ingestion.md | 4 ++--
 1 file changed, 2 insertions(+), 2 deletions(-)

diff --git a/specs/mail/inbound-ingestion.md b/specs/mail/inbound-ingestion.md
index b58e1d5..8254f28 100644
--- a/specs/mail/inbound-ingestion.md
+++ b/specs/mail/inbound-ingestion.md
@@ -83,7 +83,7 @@ Ordered, applied to each received message. Idempotent by step 1, so a whole re-r
      **`not-found`**, likewise fall back to a fresh conversation (the token verified but no
      such row exists — pathological, but the mail is still ingested, never lost).
    - The store write **and** the ledger row's `received → stored` transition (recording the
-     resulting `conversationId`/`threadId`) commit in **one transaction** — see §4.
+     resulting `threadId`) commit in **one transaction** — see §4.
 
 **Attachments belong to the pipeline, not the transport** (§2). After the parse (step 2),
 attachment bytes are written to the `BlobStore` under a **mailbox-namespaced** key
@@ -109,7 +109,7 @@ the **claim/lease**, and the **retry queue**.
 
 **The claim, the store write, and the outcome are one atomic unit.** The step-5 store write
 (`createConversation`/`appendThread`) and the ledger's `received → stored` transition —
-recording the resulting ids — commit in a **single transaction**, so the ledger row *is*
+recording the resulting `threadId` — commit in a **single transaction**, so the ledger row *is*
 the idempotency record: a retry re-hits the §3-step-1 claim, finds a `stored` row, and
 returns its recorded outcome (its `threadId`) without re-writing. A crash *before* that commit
 leaves the row at `received` and no conversation, and the retry redoes the whole unit