From dcecc812eb55a87fd2e560d0f401b7e5b7028edb Mon Sep 17 00:00:00 2001 From: TJ Baker <1617679+zaridan@users.noreply.github.com> Date: Tue, 14 Jul 2026 16:00:30 -0700 Subject: [PATCH 1/4] docs(deploy): Gmail inbound deployment + provisioning runbook (HT-43) The one-time operator steps to take the merged inbound engine live: GCP Internal OAuth app + Gmail/Pub-Sub provisioning, Supabase Postgres + Storage, Vercel env + cron, an env-var reference, and a post-deploy smoke checklist. Defines the endpoint/env contract the composition root builds to. Real credentials + the consent round-trip remain the operator's action (HT-44). Co-Authored-By: Claude Opus 4.8 --- specs/deploy/gmail-inbound-runbook.md | 200 ++++++++++++++++++++++++++ 1 file changed, 200 insertions(+) create mode 100644 specs/deploy/gmail-inbound-runbook.md diff --git a/specs/deploy/gmail-inbound-runbook.md b/specs/deploy/gmail-inbound-runbook.md new file mode 100644 index 0000000..70cb29a --- /dev/null +++ b/specs/deploy/gmail-inbound-runbook.md @@ -0,0 +1,200 @@ +# Gmail inbound — deployment & provisioning runbook (HT-43) + +Status: draft (HT-43). The one-time operator steps to take the merged engine +code (HT-34…HT-42) live: a deployed Vercel environment where **RIQ's own +inbound Gmail flows end-to-end** into a Helpthread conversation. This is the +"deployed, end to end" acceptance HT-43 owns; the actual **real Google +consent** that connects the mailbox is the last step and is tracked as +**HT-44**. + +Nothing here is engine code — it is accounts, credentials, and console +clicks. **Every real credential and every consent screen is the operator's +action, never the assistant's.** The engine reads all secrets from +environment variables (never hardcoded); this runbook is how those env values +come to exist. + +> Read alongside [gmail-push.md](../mail/gmail-push.md) §2/§7 (the webhook +> auth + the provisioning checklist this expands) and +> [gmail-connect.md](../mail/gmail-connect.md) §3 (the OAuth app + scopes). + +## 0. Architecture being deployed + +``` +Gmail mailbox ──watch()──▶ Cloud Pub/Sub topic ──push sub (OIDC JWT)──▶ + POST /api/v1/inbound/gmail (webhook: verify JWT → enqueue reconcile job → 2xx) + │ enqueue (durable INSERT into the PG job queue — commits BEFORE the 2xx) + ▼ + Vercel Cron ──GET /api/v1/internal/queue/drain (every minute)──▶ drain N jobs: + reconcile (history.list → messages.get raw) → idempotent ingest → conversation + Vercel Cron ──GET /api/v1/internal/cron/watch-maintenance (daily)──▶ re-arm watch() + sweep + +Operator connect: POST /api/v1/inbound/gmail/connect (Bearer) → consentUrl + → browser → Google consent → GET /callback → mailbox connected +Storage: Supabase Postgres (conversations, threads, mailboxes, tokens, job queue) + + Supabase Storage (attachment + oversized-raw blobs) +``` + +The queue is a **Postgres-backed, cron-drained durable queue** (not Vercel +Queues — which is still beta): the webhook's enqueue is a durable `INSERT` +that commits before the endpoint acks Pub/Sub, and a once-a-minute Vercel Cron +leases and processes a bounded batch. ~1-minute worst-case delivery latency is +well within a support desk's needs; the durability (never ack Pub/Sub before +the row commits) is what protects invariant #1. + +## Prerequisites + +- A **Google Workspace** account for the mailbox to connect (e.g. + `support@resonantiq.app`) — Workspace, because the OAuth app is **Internal** + (no CASA verification, no external-user consent screen). +- A **Google Cloud project** with billing (Pub/Sub needs a billing account; + volume here is negligible/free-tier). +- A **Supabase** project (Postgres + Storage). +- A **Vercel** project connected to this repo. +- The `gcloud` CLI (optional but the steps below give both console + CLI). + +--- + +## Part A — Google Cloud: OAuth app + Gmail + Pub/Sub + +Do this in the Google Cloud project that will own the push topic. + +### A1. Enable the APIs +Console → *APIs & Services → Enable APIs* → enable **Gmail API** and **Cloud +Pub/Sub API**. (CLI: `gcloud services enable gmail.googleapis.com pubsub.googleapis.com`.) + +### A2. The Internal OAuth app + client credentials +1. *APIs & Services → OAuth consent screen* → **Internal** user type. Fill + app name / support email. No scopes need adding on the screen for an + Internal app, but the app must be in **Published**/In-use state for your org. +2. *APIs & Services → Credentials → Create credentials → OAuth client ID* → + **Web application**. +3. Under **Authorized redirect URIs** add **exactly**: + `https:///api/v1/inbound/gmail/callback` + (must byte-match `PUBLIC_BASE_URL` + `/api/v1/inbound/gmail/callback`; see + gmail-connect.md §3.) +4. Save the **Client ID** → `GMAIL_OAUTH_CLIENT_ID` and **Client secret** → + `GMAIL_OAUTH_CLIENT_SECRET`. **These are the operator's to hold — never + commit them, never paste them to the assistant.** + +Scopes the connect flow requests (no console action; requested at consent +time): `https://www.googleapis.com/auth/gmail.readonly` + +`https://www.googleapis.com/auth/gmail.send` (gmail-connect.md §3, least +privilege). + +### A3. The Pub/Sub topic + push subscription +1. *Pub/Sub → Topics → Create topic*, e.g. `gmail-push`. Full name + `projects//topics/gmail-push` → `GMAIL_PUBSUB_TOPIC`. +2. **Grant Gmail permission to publish** to the topic: add principal + **`gmail-api-push@system.gserviceaccount.com`** with role **Pub/Sub + Publisher** on that topic. (Without this, `watch()` returns an error — this + is the single most common setup miss.) +3. Create a **service account** the push subscription will present as its OIDC + identity, e.g. `gmail-push-invoker@.iam.gserviceaccount.com` → + `GMAIL_PUSH_SERVICE_ACCOUNT`. +4. *Pub/Sub → Subscriptions → Create subscription* on that topic: + - Delivery type **Push**. + - Endpoint URL: `https:///api/v1/inbound/gmail`. + - **Enable authentication** → the service account from A3.3; audience = + the **exact** endpoint URL above (the webhook checks `aud` equals its own + URL — gmail-push.md §2). + - Full subscription name `projects//subscriptions/` → + `GMAIL_PUBSUB_SUBSCRIPTION` (the webhook rejects a push whose + `subscription` field isn't this exact value — gmail-push.md §2). + +> The initial `users.watch()` (which points the mailbox at the topic) is armed +> automatically by the **connect flow** (Part E) — you do not call it by hand. + +--- + +## Part B — Supabase: Postgres + Storage + +1. Create the Supabase project. From *Project Settings → Database → Connection + string*, take the **transaction-mode pooler** URI (**port 6543**, host + `...pooler.supabase.com`) → `DATABASE_URL`. (Port 6543, not 5432 — the + serverless-correct pooled connection; see `src/db/postgres.ts`.) +2. **Run migrations** against that database once (from a machine with the URL): + the engine's `migrate()` applies every migration including the new job-queue + table. (A `scripts/migrate.ts` one-shot is provided with the composition + root; or run against the direct 5432 URL for the one-time DDL.) +3. *Storage → Create bucket*, e.g. `helpthread-blobs` (**private**) → + `HELPTHREAD_BLOB_BUCKET`. +4. *Project Settings → API* → `SUPABASE_URL` and the **service_role** key → + `SUPABASE_SERVICE_ROLE_KEY` (server-side only; grants full storage access — + treat like a password, never expose to a browser). + +--- + +## Part C — Vercel: env vars, deploy, cron + +1. Set every variable from the [env reference](#env-reference) in *Project + Settings → Environment Variables* (Production). Generate the two secrets you + mint yourself: + - `HELPTHREAD_TOKEN_ENC_KEY` — a 32-byte key, base64 (`openssl rand -base64 32`). + Encrypts refresh tokens at rest; **losing/rotating it orphans every stored + token** (mailboxes must reconnect). + - `HELPTHREAD_API_TOKEN` — the Agent-inbox Bearer token (`openssl rand -base64 24`; ≥16 chars). + - `CRON_SECRET` — guards the internal cron/drain endpoints (`openssl rand -base64 24`). +2. `PUBLIC_BASE_URL` = your production URL (e.g. `https://desk.resonantiq.app`), + matching the OAuth redirect URI (A2.3) and the Pub/Sub push endpoint (A3.4). +3. Deploy. `vercel.json` (in the repo) declares the two Vercel Cron jobs: + - `*/1 * * * *` → `GET /api/v1/internal/queue/drain` (drain the job queue). + - `0 6 * * *` → `GET /api/v1/internal/cron/watch-maintenance` (daily renewal + sweep; UTC). + Vercel Cron invokes these as HTTP GETs; the handlers require the + `CRON_SECRET` (Vercel sends it as a bearer via the `Authorization` header on + cron requests) and are idempotent + lease-bounded. + + +## Env reference + +| Var | Source | Notes | +|---|---|---| +| `DATABASE_URL` | Supabase B1 | 6543 pooler URI | +| `SUPABASE_URL` | Supabase B4 | project URL | +| `SUPABASE_SERVICE_ROLE_KEY` | Supabase B4 | server-only secret | +| `HELPTHREAD_BLOB_BUCKET` | Supabase B3 | private bucket name | +| `GMAIL_OAUTH_CLIENT_ID` | Google A2 | | +| `GMAIL_OAUTH_CLIENT_SECRET` | Google A2 | secret | +| `GMAIL_PUBSUB_TOPIC` | Google A3.1 | `projects/…/topics/…` | +| `GMAIL_PUBSUB_SUBSCRIPTION` | Google A3.4 | `projects/…/subscriptions/…` | +| `GMAIL_PUSH_SERVICE_ACCOUNT` | Google A3.3 | the push SA email (JWT `email` claim) | +| `HELPTHREAD_TOKEN_ENC_KEY` | you mint (C1) | 32-byte base64; encrypts tokens at rest | +| `HELPTHREAD_API_TOKEN` | you mint (C1) | Agent-inbox Bearer, ≥16 chars | +| `CRON_SECRET` | you mint (C1) | guards internal cron endpoints | +| `PUBLIC_BASE_URL` | Vercel C2 | your prod origin, no trailing slash | +| `HELPTHREAD_MAIL_DOMAIN` | you choose | domain minted into outbound Message-IDs | +| `HELPTHREAD_SUPPORT_ADDRESS` | the mailbox | e.g. `support@resonantiq.app` | +| `HELPTHREAD_SIGNING_SECRET` | you mint | ≥32 chars; HMAC keyring for reply/state/view tokens | + +## Part E — Connect the mailbox (HT-44, operator action) + +With the deploy live and env set: +1. `POST https:///api/v1/inbound/gmail/connect` with + `Authorization: Bearer $HELPTHREAD_API_TOKEN` → returns `{ consentUrl }`. +2. Open `consentUrl` in a browser **signed into the mailbox's Google account**, + grant consent. Google redirects to `/callback`, which exchanges the code, + stores the encrypted refresh token, arms `watch()`, and seeds the cursor. + You should see a "Mailbox connected" page. +3. **This consent is the operator's action** — the assistant never completes it. + +## Part F — Post-deploy smoke checklist + +- [ ] `GET /api/v1/conversations` with the Bearer token → `200` (API + DB reachable). +- [ ] A wrong/no Bearer → `401`. +- [ ] `POST /connect` → a `consentUrl` whose `redirect_uri` matches A2.3 exactly. +- [ ] After connect: a `mailboxes` row (`status=active`), a `mailbox_oauth_tokens` + row (ciphertext, not plaintext), a `gmail_watch_state` row with a `history_id`. +- [ ] Send a test email **to** the connected mailbox → within ~1 min (the drain + tick) a new conversation appears (`GET /api/v1/conversations`). +- [ ] Pub/Sub subscription **oldest-unacked-message age** stays low (no backlog). +- [ ] The job-queue table: no rows stuck `dead_lettered_at IS NOT NULL`; + oldest `ready` job age stays under a minute or two. +- [ ] Reply from the Agent inbox → the reply arrives at the customer, and a + reply back **threads** into the same conversation (the sacred outbound-token + check — HT-44's live proof). + +## What this runbook does not cover + +- The Agent Inbox **UI** (HT-23) — this deploys the engine/API; the UI is separate. +- Multi-mailbox / GA onboarding — dogfood is one Workspace mailbox, Internal app. +- Vercel Queues — a future low-latency/managed `QueueProvider` adapter; the PG + queue is the dogfood implementation of the same interface. From e4f8b8279aac01151cce031d9ce0923f98884206 Mon Sep 17 00:00:00 2001 From: TJ Baker <1617679+zaridan@users.noreply.github.com> Date: Tue, 14 Jul 2026 16:00:30 -0700 Subject: [PATCH 2/4] feat(providers): Postgres-backed durable queue adapter + migration 013 (HT-43) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The production QueueProvider for the RIQ dogfood — a cron-drained durable queue on Supabase Postgres, chosen over Vercel Queues (beta) since it reuses the DB already required. Not a toy table: - migration 013 `queue_jobs`: run_after + locked_until (eligible + leased), attempts, dead_lettered_at (retained, never dropped — invariant #1), a partial unique index for (topic, dedupe_key) dedupe, a ready-jobs index. - enqueue: one durable INSERT (commits before the webhook acks Pub/Sub) with ON CONFLICT DO NOTHING dedupe. - drainOnce: atomic FOR UPDATE SKIP LOCKED claim (concurrent drains never double-process), attempts bumped at claim, ack deletes, retry reschedules with capped exponential backoff, dead-letter on ceiling/explicit — retained. - getStats: ready count / oldest-ready age / dead-letter count for the smoke checklist + alerting. 12 PGlite-backed tests (real Postgres) incl. concurrent-drain no-double-process. Wired only at the composition root (later in HT-43); no engine-core import. Co-Authored-By: Claude Opus 4.8 --- src/db/migrate.test.ts | 2 + src/db/migrate.ts | 122 ++++++ src/db/postgres.test.ts | 1 + .../adapters/postgres-queue/index.test.ts | 328 ++++++++++++++ .../adapters/postgres-queue/index.ts | 405 ++++++++++++++++++ 5 files changed, 858 insertions(+) create mode 100644 src/providers/adapters/postgres-queue/index.test.ts create mode 100644 src/providers/adapters/postgres-queue/index.ts diff --git a/src/db/migrate.test.ts b/src/db/migrate.test.ts index d6790d6..f9af5d3 100644 --- a/src/db/migrate.test.ts +++ b/src/db/migrate.test.ts @@ -52,6 +52,7 @@ describe('migrate', () => { { id: 10, name: 'mailbox_oauth_tokens' }, { id: 11, name: 'gmail_watch_state' }, { id: 12, name: 'inbound_deliveries' }, + { id: 13, name: 'queue_jobs' }, ]) }) @@ -74,6 +75,7 @@ describe('migrate', () => { { id: 10 }, { id: 11 }, { id: 12 }, + { id: 13 }, ]) }) diff --git a/src/db/migrate.ts b/src/db/migrate.ts index 3f9119e..9a04a35 100644 --- a/src/db/migrate.ts +++ b/src/db/migrate.ts @@ -562,6 +562,123 @@ CREATE TABLE inbound_deliveries ( CREATE UNIQUE INDEX inbound_deliveries_mailbox_id_provider_message_id_key ON inbound_deliveries (mailbox_id, provider_message_id); ` +/** + * Migration 013 — `queue_jobs`, the durable Postgres-backed queue behind + * `createPostgresQueue` (HT-43; `src/providers/adapters/postgres-queue/`). + * This is the production `QueueProvider` for the RIQ dogfood: the Gmail + * push webhook (`src/api/gmail-webhook.ts`) enqueues a "reconcile" job here, + * and a Vercel Cron tick drains and processes a bounded batch + * (`PostgresQueue.drainOnce`) — chosen over Vercel Queues (still beta) + * because it reuses the Supabase Postgres every deployment already + * provisions, rather than adding a second durable-work dependency. + * + * One row per enqueued job, simultaneously the **dedupe record**, the + * **claim/lease**, and the **retry/dead-letter bookkeeping** — the same + * three-way framing migration 012's doc comment uses for + * `inbound_deliveries`, applied here to outbound queue work instead of + * inbound delivery ledgering. + * + * ## Dedupe: a partial unique index, mirroring migration 003's precedent + * + * `queue_jobs_topic_dedupe_key` constrains `(topic, dedupe_key)` only when + * `dedupe_key IS NOT NULL` — the same partial-index shape migration 003's + * `threads_conversation_idempotency_key_idx` established for `threads`: + * only rows that opted into dedup (a caller-supplied key) constrain each + * other, and every `NULL`-key row is invisible to the index, so ordinary + * (non-deduped) enqueues never collide with one another. The adapter's + * `enqueue` targets this exact index with `INSERT ... ON CONFLICT (topic, + * dedupe_key) WHERE dedupe_key IS NOT NULL AND dead_lettered_at IS NULL DO + * NOTHING` — a retried enqueue call sharing the same `(topic, dedupeKey)` + * as a still-live job is silently suppressed, matching + * `EnqueueOptions.dedupeKey`'s "SHOULD suppress duplicate enqueues" + * contract (`src/providers/queue.ts`). + * + * The `AND dead_lettered_at IS NULL` arm is a deliberate WIDENING beyond + * migration 003's precedent, not a copy-paste: a `threads` row is never + * reprocessed after it reaches a terminal send state, so 003's index needed + * no such arm. A queue job's dedupe key, by contrast, must become reusable + * once the job it protected reaches ITS OWN terminal failure + * (dead-lettered) — otherwise a poison job's dedupe key would permanently + * block every future enqueue attempt for that same key, even after an + * operator fixes the root cause and wants to try again. Excluding + * dead-lettered rows from the constraint is what makes that re-enqueue + * possible while still retaining the dead-lettered row itself (see below). + * + * ## `run_after` + `locked_until`: "eligible" and "leased" are separate axes + * + * `run_after` is the earliest time a job may be claimed — `now()` for an + * immediate enqueue, later for `delaySeconds` or a backed-off retry. + * `locked_until` is a lease: `drainOnce`'s claim sets it to a near-future + * expiry so a crashed or timed-out worker's claim eventually lapses and the + * job becomes reclaimable, rather than stuck forever behind a lock nobody + * will release — the same lease shape migration 003's `threads.claimed_until` + * uses for outbound-send delivery claims, applied here to queue jobs + * instead. A job is claimable exactly when BOTH are satisfied: `run_after + * <= now()` (eligible) AND `locked_until IS NULL OR locked_until < now()` + * (unleased) — two independent conditions kept as two columns rather than + * folded into one, because "eligible but currently leased" (another worker + * has it) is a real, common state that a single combined timestamp could + * not distinguish from "not yet eligible." + * + * ## `dead_lettered_at` rows are retained forever — never silently dropped + * + * A job that exhausts its retry ceiling is dead-lettered, not deleted: + * `dead_lettered_at` is stamped and the row stays in the table permanently, + * queryable via `PostgresQueue.getStats()`'s `deadLettered` count or a + * direct `SELECT ... WHERE dead_lettered_at IS NOT NULL`. This is + * CHARTER.md invariant #1 ("never silently drop") applied to queue work — + * the same retention discipline migration 012 uses for + * `inbound_deliveries.status = 'dead-letter'`: a poison job is parked and + * visible for manual review, never erased. Deleting it on terminal failure + * would make "did this job ever run, and why did it fail?" an unanswerable + * question during an incident. + * + * ## The two indexes + * + * - `queue_jobs_topic_dedupe_key` — the dedupe constraint above; also + * doubles as the lookup a future admin tool would use to find "the live + * job for key X." + * - `queue_jobs_ready_idx` — `(topic, run_after) WHERE dead_lettered_at IS + * NULL`, sized for the drain hot path: `drainOnce`'s claim filters to a + * specific topic set and orders by `run_after`, and excluding + * dead-lettered rows keeps the index from accumulating entries for jobs + * that can never be claimed again. `locked_until` is deliberately NOT + * part of this index — it churns on every claim/release, and the "find + * the oldest eligible, unleased jobs" read pattern is already served by + * ordering on `(topic, run_after)` and re-checking `locked_until` in the + * claim query's `WHERE` clause, rather than indexing a column that + * changes this fast. + * + * No CHECK constraint ties `dead_lettered_at` to `attempts`/`max_attempts` + * (unlike, say, migration 002's direction-tied CHECKs): the + * attempts-vs-ceiling decision is adapter-level policy + * (`PostgresQueue.drainOnce`'s own `maxAttempts` option — see that + * module's doc comment for why it reads as a call-level knob rather than + * this row's `max_attempts` column), not a database-level invariant the + * schema should enforce. `max_attempts` is retained as per-row schema + * head-room for a future per-job ceiling override, matching migration + * 010's "the column is reserved, the logic lands in a later ticket" + * precedent. + */ +const MIGRATION_013_QUEUE_JOBS = ` +CREATE TABLE queue_jobs ( + id uuid PRIMARY KEY DEFAULT gen_random_uuid(), + topic text NOT NULL, + payload jsonb NOT NULL, + dedupe_key text, + attempts integer NOT NULL DEFAULT 0, + max_attempts integer NOT NULL DEFAULT 5, + run_after timestamptz NOT NULL DEFAULT now(), + locked_until timestamptz, + last_error text, + dead_lettered_at timestamptz, + created_at timestamptz NOT NULL DEFAULT now(), + updated_at timestamptz NOT NULL DEFAULT now() +); +CREATE UNIQUE INDEX queue_jobs_topic_dedupe_key ON queue_jobs (topic, dedupe_key) WHERE dedupe_key IS NOT NULL AND dead_lettered_at IS NULL; +CREATE INDEX queue_jobs_ready_idx ON queue_jobs (topic, run_after) WHERE dead_lettered_at IS NULL; +` + /** * 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 @@ -624,6 +741,11 @@ const MIGRATIONS: Migration[] = [ name: 'inbound_deliveries', sql: MIGRATION_012_INBOUND_DELIVERIES, }, + { + id: 13, + name: 'queue_jobs', + sql: MIGRATION_013_QUEUE_JOBS, + }, ] /** diff --git a/src/db/postgres.test.ts b/src/db/postgres.test.ts index 854c028..e9c0845 100644 --- a/src/db/postgres.test.ts +++ b/src/db/postgres.test.ts @@ -301,6 +301,7 @@ describe('createPostgresDb with a schema option', () => { 'inbound_deliveries', 'mailbox_oauth_tokens', 'mailboxes', + 'queue_jobs', 'threads', ]) diff --git a/src/providers/adapters/postgres-queue/index.test.ts b/src/providers/adapters/postgres-queue/index.test.ts new file mode 100644 index 0000000..890d4d3 --- /dev/null +++ b/src/providers/adapters/postgres-queue/index.test.ts @@ -0,0 +1,328 @@ +import { afterEach, describe, expect, it } from 'vitest' +import { createPgliteDb, type Db } from '../../../db/client.js' +import { migrate } from '../../../db/migrate.js' +import type { QueueHandlerResult, QueueMessage, QueueMessageHandler } from '../../queue.js' +import { createPostgresQueue, type PostgresQueue } from './index.js' + +// --- fixtures ---------------------------------------------------------------- + +/** A loose stand-in for the real Gmail reconcile job shape — this suite tests the QUEUE, not reconcile (brief). */ +interface ReconcileJob { + mailboxId: string + historyId: string +} + +const TOPIC = 'gmail-reconcile' + +function reconcileJob(n: number): ReconcileJob { + return { mailboxId: `mailbox-${n}`, historyId: String(n) } +} + +/** Build a handler that always returns `result` and records every message it was invoked with. */ +function fakeHandler(result: QueueHandlerResult): { + handler: QueueMessageHandler + calls: QueueMessage[] +} { + const calls: QueueMessage[] = [] + const handler: QueueMessageHandler = async (message) => { + calls.push(message) + return result + } + return { handler, calls } +} + +interface RawQueueJobRow { + id: string + topic: string + payload: unknown + dedupe_key: string | null + attempts: number + max_attempts: number + run_after: string + locked_until: string | null + last_error: string | null + dead_lettered_at: string | null + created_at: string + updated_at: string +} + +async function allJobRows(db: Db): Promise { + return db.query('SELECT * FROM queue_jobs ORDER BY created_at, id') +} + +async function countRows(db: Db, table: string): Promise { + const rows = await db.query<{ count: number }>(`SELECT count(*)::int AS count FROM ${table}`) + return rows[0].count +} + +/** Move every row's `run_after` into the past — a deterministic stand-in for waiting out a delay/backoff, avoiding a real sleep in the test. */ +async function forceAllDue(db: Db): Promise { + await db.query("UPDATE queue_jobs SET run_after = now() - interval '1 second'") +} + +// --- suite --------------------------------------------------------------------- + +describe('createPostgresQueue', () => { + let db: Db | undefined + + afterEach(async () => { + await db?.close() + db = undefined + }) + + async function freshQueue(): Promise<{ db: Db; queue: PostgresQueue }> { + db = await createPgliteDb() + await migrate(db) + return { db, queue: createPostgresQueue(db) } + } + + it('enqueue inserts a row whose payload round-trips as jsonb', async () => { + const { db, queue } = await freshQueue() + + await queue.enqueue(TOPIC, reconcileJob(1)) + + const rows = await allJobRows(db) + expect(rows).toHaveLength(1) + expect(rows[0]).toMatchObject({ + topic: TOPIC, + payload: reconcileJob(1), + dedupe_key: null, + attempts: 0, + max_attempts: 5, + locked_until: null, + last_error: null, + dead_lettered_at: null, + }) + }) + + it('dedupeKey suppresses a duplicate enqueue on the same topic, but a different key (or no key) still inserts', async () => { + const { db, queue } = await freshQueue() + + await queue.enqueue(TOPIC, reconcileJob(1), { dedupeKey: 'mailbox-1:1' }) + await queue.enqueue(TOPIC, reconcileJob(1), { dedupeKey: 'mailbox-1:1' }) + expect(await countRows(db, 'queue_jobs')).toBe(1) + + await queue.enqueue(TOPIC, reconcileJob(1), { dedupeKey: 'mailbox-1:2' }) + expect(await countRows(db, 'queue_jobs')).toBe(2) + + // Omitting the key entirely never dedupes against anything, including itself. + await queue.enqueue(TOPIC, reconcileJob(1)) + await queue.enqueue(TOPIC, reconcileJob(1)) + expect(await countRows(db, 'queue_jobs')).toBe(4) + }) + + it('delaySeconds sets run_after in the future, and the job is not claimed until due', async () => { + const { db, queue } = await freshQueue() + + await queue.enqueue(TOPIC, reconcileJob(1), { delaySeconds: 3600 }) + + const [row] = await allJobRows(db) + const delayMs = new Date(row.run_after).getTime() - new Date(row.created_at).getTime() + expect(delayMs).toBeGreaterThan(3500 * 1000) + + const { handler, calls } = fakeHandler({ kind: 'ack' }) + const report = await queue.drainOnce({ handlers: { [TOPIC]: handler } }) + + expect(report).toEqual({ claimed: 0, acked: 0, retried: 0, deadLettered: 0 }) + expect(calls).toHaveLength(0) + expect(await countRows(db, 'queue_jobs')).toBe(1) + }) + + it('drainOnce claims a ready job, invokes its handler with attempts: 1, and ack deletes the row', async () => { + const { db, queue } = await freshQueue() + await queue.enqueue(TOPIC, reconcileJob(1)) + + const { handler, calls } = fakeHandler({ kind: 'ack' }) + const report = await queue.drainOnce({ handlers: { [TOPIC]: handler } }) + + expect(report).toEqual({ claimed: 1, acked: 1, retried: 0, deadLettered: 0 }) + expect(calls).toHaveLength(1) + expect(calls[0]).toMatchObject({ topic: TOPIC, payload: reconcileJob(1), attempts: 1 }) + expect(calls[0].id).toEqual(expect.any(String)) + expect(calls[0].enqueuedAt).toBeInstanceOf(Date) + expect(await countRows(db, 'queue_jobs')).toBe(0) + }) + + it('retry reschedules into the future and clears the lease; the job is re-drained once due', async () => { + const { db, queue } = await freshQueue() + await queue.enqueue(TOPIC, reconcileJob(1)) + + let invocation = 0 + const handler: QueueMessageHandler = async () => { + invocation++ + return invocation === 1 ? { kind: 'retry', backoffSeconds: 30 } : { kind: 'ack' } + } + + const first = await queue.drainOnce({ handlers: { [TOPIC]: handler } }) + expect(first).toEqual({ claimed: 1, acked: 0, retried: 1, deadLettered: 0 }) + + const [afterRetry] = await allJobRows(db) + expect(afterRetry.attempts).toBe(1) + expect(afterRetry.locked_until).toBeNull() + expect(afterRetry.dead_lettered_at).toBeNull() + expect(new Date(afterRetry.run_after).getTime()).toBeGreaterThan(Date.now()) + + // Not yet due — a drain right now claims nothing. + const tooSoon = await queue.drainOnce({ handlers: { [TOPIC]: handler } }) + expect(tooSoon.claimed).toBe(0) + + // Time-travel run_after into the past rather than sleeping out the backoff. + await forceAllDue(db) + + const second = await queue.drainOnce({ handlers: { [TOPIC]: handler } }) + expect(second).toEqual({ claimed: 1, acked: 1, retried: 0, deadLettered: 0 }) + expect(invocation).toBe(2) + }) + + it('a handler that throws is treated as retry: attempts incremented, rescheduled, error message recorded', async () => { + const { db, queue } = await freshQueue() + await queue.enqueue(TOPIC, reconcileJob(1)) + + const handler: QueueMessageHandler = async () => { + throw new Error('boom') + } + + const report = await queue.drainOnce({ handlers: { [TOPIC]: handler } }) + expect(report).toEqual({ claimed: 1, acked: 0, retried: 1, deadLettered: 0 }) + + const [row] = await allJobRows(db) + expect(row.attempts).toBe(1) + expect(row.last_error).toBe('boom') + expect(row.dead_lettered_at).toBeNull() + expect(row.locked_until).toBeNull() + }) + + it('retry past maxAttempts dead-letters the job: row retained, NOT re-claimed on the next drain', async () => { + const { db, queue } = await freshQueue() + await queue.enqueue(TOPIC, reconcileJob(1)) + + const { handler } = fakeHandler({ kind: 'retry' }) + + // maxAttempts: 2 — first drain retries (attempts -> 1, below ceiling). + const first = await queue.drainOnce({ handlers: { [TOPIC]: handler } }, { maxAttempts: 2 }) + expect(first).toEqual({ claimed: 1, acked: 0, retried: 1, deadLettered: 0 }) + + await forceAllDue(db) + + // Second drain: attempts -> 2, at the ceiling -> dead-letter instead of retry. + const second = await queue.drainOnce({ handlers: { [TOPIC]: handler } }, { maxAttempts: 2 }) + expect(second).toEqual({ claimed: 1, acked: 0, retried: 0, deadLettered: 1 }) + + const [row] = await allJobRows(db) + expect(row.attempts).toBe(2) + expect(row.dead_lettered_at).not.toBeNull() + expect(row.locked_until).toBeNull() + expect(await countRows(db, 'queue_jobs')).toBe(1) // retained, never deleted + + // Force due again — a dead-lettered row must never be re-claimed. + await forceAllDue(db) + const third = await queue.drainOnce({ handlers: { [TOPIC]: handler } }, { maxAttempts: 2 }) + expect(third).toEqual({ claimed: 0, acked: 0, retried: 0, deadLettered: 0 }) + }) + + it('an explicit deadLetter result dead-letters the job immediately and records the reason', async () => { + const { db, queue } = await freshQueue() + await queue.enqueue(TOPIC, reconcileJob(1)) + + const { handler } = fakeHandler({ kind: 'deadLetter', reason: 'malformed payload' }) + const report = await queue.drainOnce({ handlers: { [TOPIC]: handler } }) + + expect(report).toEqual({ claimed: 1, acked: 0, retried: 0, deadLettered: 1 }) + + const [row] = await allJobRows(db) + expect(row.attempts).toBe(1) + expect(row.dead_lettered_at).not.toBeNull() + expect(row.locked_until).toBeNull() + expect(row.last_error).toBe('malformed payload') + expect(await countRows(db, 'queue_jobs')).toBe(1) // retained, never deleted + }) + + it('a job whose topic has no registered handler is not claimed', async () => { + const { db, queue } = await freshQueue() + await queue.enqueue('some-other-topic', reconcileJob(1)) + + const { handler, calls } = fakeHandler({ kind: 'ack' }) + const report = await queue.drainOnce({ handlers: { [TOPIC]: handler } }) + + expect(report).toEqual({ claimed: 0, acked: 0, retried: 0, deadLettered: 0 }) + expect(calls).toHaveLength(0) + expect(await countRows(db, 'queue_jobs')).toBe(1) + }) + + it('batchSize bounds how many jobs one drainOnce call claims', async () => { + const { db, queue } = await freshQueue() + for (let i = 0; i < 5; i++) { + await queue.enqueue(TOPIC, reconcileJob(i)) + } + + const { handler, calls } = fakeHandler({ kind: 'ack' }) + const report = await queue.drainOnce({ handlers: { [TOPIC]: handler } }, { batchSize: 2 }) + + expect(report).toEqual({ claimed: 2, acked: 2, retried: 0, deadLettered: 0 }) + expect(calls).toHaveLength(2) + expect(await countRows(db, 'queue_jobs')).toBe(3) + }) + + it('getStats reports the ready count, oldest-ready age, and dead-letter count from a single query', async () => { + const { queue } = await freshQueue() + + expect(await queue.getStats()).toEqual({ + ready: 0, + oldestReadyAgeSeconds: null, + deadLettered: 0, + }) + + await queue.enqueue(TOPIC, reconcileJob(1)) + await queue.enqueue(TOPIC, reconcileJob(2), { delaySeconds: 3600 }) + + const withOneReady = await queue.getStats() + expect(withOneReady.ready).toBe(1) + expect(withOneReady.deadLettered).toBe(0) + expect(withOneReady.oldestReadyAgeSeconds).not.toBeNull() + expect(withOneReady.oldestReadyAgeSeconds as number).toBeGreaterThanOrEqual(0) + + // Dead-letter the one ready job; the delayed one is still not ready. + const { handler } = fakeHandler({ kind: 'deadLetter', reason: 'x' }) + await queue.drainOnce({ handlers: { [TOPIC]: handler } }) + + expect(await queue.getStats()).toEqual({ + ready: 0, + oldestReadyAgeSeconds: null, + deadLettered: 1, + }) + }) + + it('two concurrent drainOnce calls never process the same job twice (FOR UPDATE SKIP LOCKED)', async () => { + const { db, queue } = await freshQueue() + const jobCount = 10 + for (let i = 0; i < jobCount; i++) { + await queue.enqueue(TOPIC, reconcileJob(i)) + } + + // PGlite is single-connection/in-process, so these two `drainOnce` calls + // are not necessarily racing on separate backend connections the way two + // real Supabase-backed Vercel Cron invocations would (see + // src/db/migrate.ts's `migrate()` doc comment on the same PGlite + // limitation for true concurrent-lock coverage). What this DOES prove + // unconditionally, regardless of how the two calls actually interleave: + // the claim query's WHERE clause (`locked_until IS NULL OR locked_until + // < now()`, re-checked inside the same atomic UPDATE the FOR UPDATE SKIP + // LOCKED subquery drives) never lets two calls claim the same row. + const processedIds: string[] = [] + const handler: QueueMessageHandler = async (message) => { + processedIds.push(message.id) + return { kind: 'ack' } + } + + const [a, b] = await Promise.all([ + queue.drainOnce({ handlers: { [TOPIC]: handler } }, { batchSize: jobCount }), + queue.drainOnce({ handlers: { [TOPIC]: handler } }, { batchSize: jobCount }), + ]) + + expect(a.claimed + b.claimed).toBe(jobCount) + expect(processedIds).toHaveLength(jobCount) + // No id appears twice — the union of what each call processed has no overlap. + expect(new Set(processedIds).size).toBe(jobCount) + expect(await countRows(db, 'queue_jobs')).toBe(0) + }) +}) diff --git a/src/providers/adapters/postgres-queue/index.ts b/src/providers/adapters/postgres-queue/index.ts new file mode 100644 index 0000000..5096547 --- /dev/null +++ b/src/providers/adapters/postgres-queue/index.ts @@ -0,0 +1,405 @@ +/** + * `createPostgresQueue` — the production `QueueProvider` for the RIQ + * dogfood deployment (HT-43): a durable queue built on `queue_jobs` + * (migration 013, `src/db/migrate.ts` — see that migration's doc comment + * for the full schema rationale), reusing the Supabase Postgres every + * deployment already provisions rather than adding Vercel Queues (still + * beta) as a second durable-work dependency. `enqueue` implements the + * `QueueProvider` interface (`src/providers/queue.ts`); `drainOnce` is the + * poll-drain side that interface deliberately doesn't model (its own module + * doc: "there is no dequeue/poll method... this interface models [push + * delivery] directly") — here, a Vercel Cron tick calls `drainOnce` to pull + * and process one bounded batch, the shape a Postgres-backed queue actually + * needs, since Postgres itself has no way to push. + * + * Per `src/providers/README.md`'s adapter-boundary rule, this module is + * wired in only at the composition root (a later part of HT-43) — engine + * code (`src/mail`, `src/api`, `src/store`) never imports it directly, only + * the `QueueProvider` interface type. + * + * ## The enqueue-commits-before-ack invariant + * + * `src/api/gmail-webhook.ts` acks the inbound Pub/Sub push only after its + * `deps.queue.enqueue(...)` call resolves. `enqueue` here is a single + * durable `INSERT` — under `src/db/client.ts`'s `Queryable` contract a + * single statement is its own implicitly-committed unit, so the returned + * promise resolves once Postgres has durably committed the row, never + * before. If the webhook process dies before `enqueue` resolves, nothing + * was durably enqueued and the caller never observed success (so Pub/Sub + * redelivers the push) — there is no window in which an ack could be sent + * for a job that did not actually commit. + * + * ## Lease model: `run_after` + `locked_until` = "eligible" + "leased" + * + * A job is claimable when `dead_lettered_at IS NULL` (not terminal), + * `run_after <= now()` (its delay/backoff has elapsed), AND + * `locked_until IS NULL OR locked_until < now()` (unleased, or a prior + * lease expired — e.g. a worker crashed mid-run). `drainOnce`'s claim is + * one atomic `UPDATE ... WHERE id IN (SELECT ... FOR UPDATE SKIP LOCKED + * LIMIT $batch) RETURNING *`: `FOR UPDATE SKIP LOCKED` means two concurrent + * `drainOnce` calls (overlapping cron invocations, a retry racing a slow + * run) never claim the same row — the second simply skips whatever the + * first has already locked and claims the next eligible rows instead, the + * Postgres-native substitute for a platform queue's visibility timeout. + * The same statement bumps `attempts` — counted at CLAIM time, not outcome + * time, so a handler that crashes or times out mid-run still counts + * against the retry ceiling instead of retrying forever for free. + * + * ## Dedupe + * + * `enqueue`'s `INSERT ... ON CONFLICT (topic, dedupe_key) WHERE dedupe_key + * IS NOT NULL AND dead_lettered_at IS NULL DO NOTHING` targets migration + * 013's `queue_jobs_topic_dedupe_key` partial unique index — see that + * migration's doc comment for the full reasoning. A duplicate enqueue + * sharing `(topic, dedupeKey)` with a still-live job is silently + * suppressed, matching `EnqueueOptions.dedupeKey`'s "SHOULD suppress + * duplicate enqueues" contract. `dedupeKey` omitted binds `NULL`, which + * never conflicts against a unique index (ordinary Postgres NULL + * semantics) — every no-key enqueue always inserts. + * + * ## Backoff + * + * A `retry` outcome (explicit, or a caught throw — see below) reschedules + * with exponential backoff: `min((result.backoffSeconds ?? + * baseBackoffSeconds) * 2 ^ (attempts - 1), maxBackoffSeconds)`. A + * handler's own `backoffSeconds` hint (`QueueHandlerResult`) becomes the + * exponential series' STARTING point instead of the adapter's configured + * default, so a handler signaling "rate-limited, try again soon" with a + * small hint still backs off further on each subsequent failure rather + * than retrying at the same short delay forever. + * + * ## A throw is a retry with no hint + * + * Per `QueueMessageHandler`'s contract ("a handler that throws is treated + * as equivalent to retry by adapters"), every handler invocation is + * wrapped in try/catch. A caught throw becomes `{ kind: 'retry' }` with no + * `backoffSeconds` (falls back to `baseBackoffSeconds`) and `last_error` + * set to the caught error's message; an explicit `{ kind: 'retry' }` + * return carries no message of its own, so `last_error` is left `null` in + * that case. + * + * ## Dead-lettering: retried-out or explicit, but always retained + * + * A `deadLetter` result, OR a `retry` whose `attempts` has reached the + * effective ceiling, sets `dead_lettered_at` and clears `locked_until` — + * the row is NEVER deleted (invariant #1: never silently drop a job). This + * mirrors `inbound_deliveries.status = 'dead-letter'` (migration 012): a + * poison job is parked, visible via {@link PostgresQueue.getStats}'s + * `deadLettered` count, and available for manual review. Migration 013's + * `queue_jobs_ready_idx` excludes dead-lettered rows, so a dead-lettered + * job is never re-claimed by a later `drainOnce`. + * + * ## The retry ceiling is a call-level knob, not the row's `max_attempts` + * + * `queue_jobs.max_attempts` (migration 013) exists in the schema and + * defaults to 5 on every row, but neither `enqueue` nor the claim query + * below ever write anything else to it. The dead-letter-vs-retry decision + * instead compares a claimed row's `attempts` against `drainOnce`'s own + * `maxAttempts` option (falling back to this factory's + * `options.maxAttempts`, falling back to 5) — NOT against the row's + * column. Flagged here as a deliberate judgment call, not an oversight: + * this makes the ceiling an operational knob adjustable for one drain pass + * without a backfill, and keeps "retry past maxAttempts dead-letters" + * simply testable (call `drainOnce` with a small override rather than + * grinding through 5 real retries), at the cost of `max_attempts` being + * schema head-room rather than a live per-job override today. A future + * ticket wiring a per-job ceiling through `EnqueueOptions` would make the + * column authoritative and this option its fallback for jobs that didn't + * specify one. + * + * ## `topic IN (...)`, not `topic = ANY($array)` + * + * The claim query restricts to topics with a registered handler via a + * dynamically-sized `IN ($n, $n+1, ...)` list — one bind parameter per + * topic, pushed onto `params` and referenced by `$${params.length}`, the + * same style `src/store/conversations.ts`'s `listConversations` already + * uses for its dynamic `WHERE`. `src/db/client.ts`'s `SqlValue` — the type + * every bound parameter in this codebase must satisfy — is deliberately + * narrow and does not include arrays; widening that shared seam for this + * one call site was judged out of scope for this adapter. `IN (...)` is + * semantically identical to `= ANY(array)` for a non-empty list, which the + * caller always has here: `drainOnce` returns early when no handlers are + * registered, before this query is ever built. + */ + +import type { Db, SqlValue } from '../../../db/client.js' +import type { + EnqueueOptions, + QueueHandlerResult, + QueueMessage, + QueueMessageHandler, + QueueProvider, +} from '../../queue.js' + +/** + * Tunable defaults for {@link createPostgresQueue}. Every field except the + * backoff base/cap is also overridable per `drainOnce` call via + * {@link DrainOnceOptions} — see the module doc's "retry ceiling" section + * for why `maxAttempts` in particular is a runtime knob rather than a + * column read back off the claimed row. + */ +export interface PostgresQueueOptions { + /** Default claim-lease duration, milliseconds. Defaults to 60 000 (60s). */ + leaseMs?: number + /** Default max jobs claimed per `drainOnce` call. Defaults to 20. */ + batchSize?: number + /** Default retry ceiling compared against a claimed job's `attempts`. Defaults to 5. */ + maxAttempts?: number + /** Base backoff, seconds, for a job's first retry (module doc's "Backoff" section). Defaults to 10. */ + baseBackoffSeconds?: number + /** Backoff cap, seconds — exponential growth never schedules a retry further out than this. Defaults to 3600 (1h). */ + maxBackoffSeconds?: number +} + +/** Dependencies {@link PostgresQueue.drainOnce} needs for one drain pass. */ +export interface DrainDeps { + /** Topic to handler. Only jobs whose topic has a registered handler here are claimed (module doc). */ + handlers: Record> +} + +/** Per-call overrides for {@link PostgresQueue.drainOnce}; each falls back to the factory's {@link PostgresQueueOptions}. */ +export interface DrainOnceOptions { + batchSize?: number + leaseMs?: number + maxAttempts?: number +} + +/** Outcome tally for one {@link PostgresQueue.drainOnce} call. */ +export interface DrainReport { + /** Jobs claimed (leased) this pass — the batch actually obtained, which may be smaller than requested. */ + claimed: number + /** Claimed jobs whose handler returned `{ kind: 'ack' }` — row deleted. */ + acked: number + /** Claimed jobs rescheduled for a later attempt (explicit `retry`, or a caught throw) — row updated, not deleted. */ + retried: number + /** Claimed jobs that reached a terminal failure this pass (explicit `deadLetter`, or `retry` past the ceiling) — row retained, never reprocessed. */ + deadLettered: number +} + +/** Point-in-time queue health — see {@link PostgresQueue.getStats}. */ +export interface QueueStats { + /** Live (not dead-lettered) jobs eligible for claim right now: due (`run_after <= now()`) and unleased. */ + ready: number + /** Age, in seconds, of the OLDEST ready job's `run_after` — how long the longest-waiting ready job has been eligible. `null` when nothing is ready. */ + oldestReadyAgeSeconds: number | null + /** Jobs in the terminal dead-lettered state, retained for manual review. */ + deadLettered: number +} + +/** The `QueueProvider` this module builds, plus the poll-drain method the interface deliberately doesn't model (module doc). */ +export interface PostgresQueue extends QueueProvider { + /** Claim and process one bounded batch of ready jobs. See the module doc for the full claim/apply-outcome contract. */ + drainOnce(deps: DrainDeps, opts?: DrainOnceOptions): Promise + /** Point-in-time queue health, for a smoke checklist or alerting. */ + getStats(): Promise +} + +const DEFAULT_LEASE_MS = 60_000 +const DEFAULT_BATCH_SIZE = 20 +const DEFAULT_MAX_ATTEMPTS = 5 +const DEFAULT_BASE_BACKOFF_SECONDS = 10 +const DEFAULT_MAX_BACKOFF_SECONDS = 3600 + +const QUEUE_JOB_COLUMNS = + 'id, topic, payload, dedupe_key, attempts, max_attempts, run_after, locked_until, last_error, dead_lettered_at, created_at, updated_at' + +/** Raw `queue_jobs` row shape (migration 013, `src/db/migrate.ts`), before mapping into a `QueueMessage`. */ +interface QueueJobRow { + id: string + topic: string + payload: unknown + dedupe_key: string | null + attempts: number + max_attempts: number + run_after: Date | string + locked_until: Date | string | null + last_error: string | null + dead_lettered_at: Date | string | null + created_at: Date | string + updated_at: Date | string +} + +/** Coerce a `timestamptz` column value into a `Date` — see `src/store/inbound-deliveries.ts`'s `toDate` for the same defensive reasoning (PGlite hands back real `Date`s; a future wire-protocol `Db` may not). */ +function toDate(value: Date | string): Date { + return value instanceof Date ? value : new Date(value) +} + +/** + * Claim up to `batchSize` ready jobs whose topic is in `topics`, atomically + * bumping `attempts` and setting a `leaseSeconds`-long lease (module doc's + * "Lease model" section). Builds a dynamically-sized `topic IN (...)` + * clause rather than binding a single array parameter — see the module + * doc's closing section for why. + */ +async function claimBatch( + db: Db, + topics: string[], + leaseSeconds: number, + batchSize: number, +): Promise { + const params: SqlValue[] = [leaseSeconds, batchSize] + const topicPlaceholders = topics.map((topic) => { + params.push(topic) + return `$${params.length}` + }) + return db.query( + `UPDATE queue_jobs + SET locked_until = now() + make_interval(secs => $1::float8), attempts = attempts + 1, updated_at = now() + WHERE id IN ( + SELECT id FROM queue_jobs + WHERE dead_lettered_at IS NULL + AND run_after <= now() + AND (locked_until IS NULL OR locked_until < now()) + AND topic IN (${topicPlaceholders.join(', ')}) + ORDER BY run_after + FOR UPDATE SKIP LOCKED + LIMIT $2::int + ) + RETURNING ${QUEUE_JOB_COLUMNS}`, + params, + ) +} + +/** Mark `id` dead-lettered: terminal, retained, never reclaimed (module doc's "Dead-lettering" section). */ +async function deadLetterJob(db: Db, id: string, reason: string): Promise { + await db.query( + `UPDATE queue_jobs + SET dead_lettered_at = now(), locked_until = NULL, last_error = $2, updated_at = now() + WHERE id = $1`, + [id, reason], + ) +} + +/** Build the Postgres-backed `QueueProvider` + drain adapter. See the module doc for the full contract. */ +export function createPostgresQueue(db: Db, options?: PostgresQueueOptions): PostgresQueue { + const defaultLeaseMs = options?.leaseMs ?? DEFAULT_LEASE_MS + const defaultBatchSize = options?.batchSize ?? DEFAULT_BATCH_SIZE + const defaultMaxAttempts = options?.maxAttempts ?? DEFAULT_MAX_ATTEMPTS + const baseBackoffSeconds = options?.baseBackoffSeconds ?? DEFAULT_BASE_BACKOFF_SECONDS + const maxBackoffSeconds = options?.maxBackoffSeconds ?? DEFAULT_MAX_BACKOFF_SECONDS + + return { + async enqueue(topic: string, payload: T, opts?: EnqueueOptions): Promise { + const delaySeconds = opts?.delaySeconds ?? 0 + const dedupeKey = opts?.dedupeKey ?? null + await db.query( + `INSERT INTO queue_jobs (topic, payload, dedupe_key, run_after) + VALUES ($1, $2::jsonb, $3, now() + make_interval(secs => $4::float8)) + ON CONFLICT (topic, dedupe_key) WHERE dedupe_key IS NOT NULL AND dead_lettered_at IS NULL DO NOTHING`, + [topic, JSON.stringify(payload), dedupeKey, delaySeconds], + ) + }, + + async drainOnce(deps: DrainDeps, opts?: DrainOnceOptions): Promise { + const topics = Object.keys(deps.handlers) + const report: DrainReport = { claimed: 0, acked: 0, retried: 0, deadLettered: 0 } + if (topics.length === 0) { + // Nothing registered — nothing to claim. Also sidesteps building a + // claim query with an empty `IN ()` list, which is a syntax error. + return report + } + + const leaseSeconds = (opts?.leaseMs ?? defaultLeaseMs) / 1000 + const batchSize = opts?.batchSize ?? defaultBatchSize + const maxAttempts = opts?.maxAttempts ?? defaultMaxAttempts + + const claimed = await claimBatch(db, topics, leaseSeconds, batchSize) + report.claimed = claimed.length + + for (const row of claimed) { + const handler = deps.handlers[row.topic] + if (handler === undefined) { + // Structurally unreachable: claimBatch's `topic IN (...)` list is + // built from exactly `Object.keys(deps.handlers)`, so every + // claimed row's topic has a registered handler. Thrown rather + // than silently skipping a claimed (leased) job. + throw new Error( + `createPostgresQueue: claimed job ${row.id} has topic '${row.topic}' with no registered handler`, + ) + } + + const message: QueueMessage = { + id: row.id, + topic: row.topic, + payload: row.payload, + attempts: row.attempts, + enqueuedAt: toDate(row.created_at), + } + + let result: QueueHandlerResult + let caughtErrorMessage: string | null = null + try { + result = await handler(message) + } catch (err) { + // A throw is a retry with no hint (module doc). + caughtErrorMessage = err instanceof Error ? err.message : String(err) + result = { kind: 'retry' } + } + + if (result.kind === 'ack') { + await db.query('DELETE FROM queue_jobs WHERE id = $1', [row.id]) + report.acked++ + continue + } + + if (result.kind === 'deadLetter') { + await deadLetterJob(db, row.id, result.reason) + report.deadLettered++ + continue + } + + // result.kind === 'retry': dead-letter once the effective ceiling is + // reached, otherwise reschedule with exponential backoff (module doc). + if (row.attempts >= maxAttempts) { + await deadLetterJob( + db, + row.id, + caughtErrorMessage ?? `createPostgresQueue: exceeded maxAttempts (${maxAttempts})`, + ) + report.deadLettered++ + continue + } + + const base = result.backoffSeconds ?? baseBackoffSeconds + const backoffSeconds = Math.min( + base * 2 ** Math.max(0, row.attempts - 1), + maxBackoffSeconds, + ) + await db.query( + `UPDATE queue_jobs + SET locked_until = NULL, run_after = now() + make_interval(secs => $2::float8), last_error = $3, updated_at = now() + WHERE id = $1`, + [row.id, backoffSeconds, caughtErrorMessage], + ) + report.retried++ + } + + return report + }, + + async getStats(): Promise { + const [row] = await db.query<{ + ready: number + oldest_ready_age_seconds: number | null + dead_lettered: number + }>( + `SELECT + (count(*) FILTER ( + WHERE dead_lettered_at IS NULL AND run_after <= now() + AND (locked_until IS NULL OR locked_until < now()) + ))::int AS ready, + (EXTRACT(EPOCH FROM (now() - min(run_after) FILTER ( + WHERE dead_lettered_at IS NULL AND run_after <= now() + AND (locked_until IS NULL OR locked_until < now()) + ))))::float8 AS oldest_ready_age_seconds, + (count(*) FILTER (WHERE dead_lettered_at IS NOT NULL))::int AS dead_lettered + FROM queue_jobs`, + ) + return { + ready: row.ready, + oldestReadyAgeSeconds: row.oldest_ready_age_seconds, + deadLettered: row.dead_lettered, + } + }, + } +} From 7ea9f35bd1e3bc36ad936848b24772e1c8991483 Mon Sep 17 00:00:00 2001 From: TJ Baker <1617679+zaridan@users.noreply.github.com> Date: Wed, 15 Jul 2026 13:17:16 -0700 Subject: [PATCH 3/4] feat(deploy): composition root + Vercel routes + Supabase blob adapter (HT-43) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Wire the framework-agnostic engine into a deployable Vercel app — the linchpin that turns the merged HT-34..HT-42 engine into a running inbound-mail deployment. - src/composition/config.ts — eager env-contract validation; aggregates all problems into one secret-free boot error (never echoes a value). - src/composition/root.ts — the composition root: constructs every concrete adapter (PostgresDb, Gmail sender/push-verifier/watch/history, the PG queue, Supabase blob) and wires them into createInboxApi (gmailPush + gmailConnect PRESENT here — absent by default on the engine) plus the two cron closures. Per-instance memoized. The refresh-token encryption key is threaded to the token store; JWKS source built once; no secret logged. - src/composition/app.ts — unified handler routing the CRON_SECRET-guarded internal cron endpoints (queue drain, watch maintenance) vs the inbox API; reuses authenticateRequest for the Bearer cron-secret check. - src/providers/adapters/supabase-storage — BlobStore over Supabase Storage (private bucket, signed reads only, service_role server-only). - api/[...path].ts + vercel.json — one catch-all Vercel Node function using the fetch Web Standard export (no node:http bridge needed; Node runtime, not Edge); crons: drain every minute, watch-maintenance daily 06:00 UTC. - scripts/migrate.ts — one-shot migration runner against DATABASE_URL. - runbook: note the Vercel Pro requirement for the sub-daily drain cron and the minted HELPTHREAD_SIGNING_SECRET. Adapter boundary held: engine core imports only interfaces; concretes are wired only at this composition root. Fake-backed tests for the internal endpoints, the blob adapter, and config validation, plus a PGlite integration test driving real requests through the whole composition end to end. Gates green: typecheck + lint + test (40 files / 744 tests). Co-Authored-By: Claude Opus 4.8 --- api/[...path].ts | 48 +++ package-lock.json | 94 ++++++ package.json | 6 +- scripts/migrate.ts | 46 +++ specs/deploy/gmail-inbound-runbook.md | 15 +- src/composition/app.test.ts | 133 +++++++++ src/composition/app.ts | 117 ++++++++ src/composition/config.test.ts | 137 +++++++++ src/composition/config.ts | 234 +++++++++++++++ src/composition/root.test.ts | 167 +++++++++++ src/composition/root.ts | 273 ++++++++++++++++++ .../adapters/supabase-storage/index.test.ts | 184 ++++++++++++ .../adapters/supabase-storage/index.ts | 168 +++++++++++ tsconfig.json | 2 +- vercel.json | 18 ++ 15 files changed, 1638 insertions(+), 4 deletions(-) create mode 100644 api/[...path].ts create mode 100644 scripts/migrate.ts create mode 100644 src/composition/app.test.ts create mode 100644 src/composition/app.ts create mode 100644 src/composition/config.test.ts create mode 100644 src/composition/config.ts create mode 100644 src/composition/root.test.ts create mode 100644 src/composition/root.ts create mode 100644 src/providers/adapters/supabase-storage/index.test.ts create mode 100644 src/providers/adapters/supabase-storage/index.ts create mode 100644 vercel.json diff --git a/api/[...path].ts b/api/[...path].ts new file mode 100644 index 0000000..836d14e --- /dev/null +++ b/api/[...path].ts @@ -0,0 +1,48 @@ +/** + * The single Vercel Function fronting the whole Helpthread engine (HT-43) — a + * catch-all under `/api` that hands every request to the composition root's + * unified handler (`src/composition/root.ts`). Vercel's Node runtime is the + * target (NOT Edge): the engine needs `node:crypto` (HMAC reply tokens, + * AES-GCM token encryption), which the Edge runtime lacks. + * + * ## Why one catch-all + the `fetch` Web Standard export + * + * Vercel's Node runtime supports the `fetch` Web Standard export + * (`export default { fetch(request: Request): Response }`), which handles ALL + * HTTP methods in one function and hands us a web-standard `Request` directly + * — so `createInboxApi`'s framework-agnostic `Request => Response` shape wires + * in with no `node:http` bridge at all (the dev harness's bridge, + * `src/dev/http-adapter.ts`, exists only because a bare `node:http` server + * gives `(req, res)`; Vercel does not). A catch-all `[...path]` file receives + * every `/api/v1/...` path with `request.url` intact, so the engine's own + * router (and the composition root's internal-cron routing) does all path + * dispatch — no per-route function files duplicating that knowledge. + * + * This file is deliberately thin: all wiring lives in the typechecked + * `src/composition/**`. It only awaits the memoized handler and guards against + * a construction/handler failure with a generic 500 (never leaking the error's + * text, which could name a missing env var). + */ + +import { getApp } from '../src/composition/root.js' + +export default { + async fetch(request: Request): Promise { + try { + const handler = await getApp() + return await handler(request) + } catch (err) { + // A thrown error here is a build/config failure (getApp rejected) or a + // bug that escaped the handler's own catch-alls. Log server-side; answer + // with the standard, detail-free error envelope. + console.error('[api] failed to build or run the app handler', err) + return new Response( + JSON.stringify({ error: { code: 'server_error', message: 'Internal server error.' } }), + { + status: 500, + headers: { 'Content-Type': 'application/json', 'Cache-Control': 'no-store' }, + }, + ) + } + }, +} diff --git a/package-lock.json b/package-lock.json index f59e100..6a1529c 100644 --- a/package-lock.json +++ b/package-lock.json @@ -13,6 +13,7 @@ ], "dependencies": { "@electric-sql/pglite": "^0.5.4", + "@supabase/supabase-js": "^2.110.6", "jose": "^6.2.3", "mimetext": "^3.0.28", "pg": "^8.22.0", @@ -1876,6 +1877,90 @@ "dev": true, "license": "MIT" }, + "node_modules/@supabase/auth-js": { + "version": "2.110.6", + "resolved": "https://registry.npmjs.org/@supabase/auth-js/-/auth-js-2.110.6.tgz", + "integrity": "sha512-20v99wV4dodllbDOsD8AUVkQ7Ie+vwcgPeORY9YlC9YyJSmVgWff9gRTdgmyfE6GJHoSMJuQ4HZokSLHA/7YUg==", + "license": "MIT", + "dependencies": { + "tslib": "2.8.1" + }, + "engines": { + "node": ">=22.0.0" + } + }, + "node_modules/@supabase/functions-js": { + "version": "2.110.6", + "resolved": "https://registry.npmjs.org/@supabase/functions-js/-/functions-js-2.110.6.tgz", + "integrity": "sha512-Ih/I766vc579WfNzXSv1ERssCHKzaPj9rpST/j9wsMuUv+AfmRteKrJn3sVmrJIwQU5qJya2+aWTlvhf4WoMsg==", + "license": "MIT", + "dependencies": { + "tslib": "2.8.1" + }, + "engines": { + "node": ">=22.0.0" + } + }, + "node_modules/@supabase/phoenix": { + "version": "0.4.4", + "resolved": "https://registry.npmjs.org/@supabase/phoenix/-/phoenix-0.4.4.tgz", + "integrity": "sha512-Gt0pqoXuIqX/8dvG0OKp/wMCobXNH3klNbUPBNyOfN0YA1IswrM3HyWFMOPk1Jy+BRaIyDPcFx4jLBwHNmlyfQ==", + "license": "MIT" + }, + "node_modules/@supabase/postgrest-js": { + "version": "2.110.6", + "resolved": "https://registry.npmjs.org/@supabase/postgrest-js/-/postgrest-js-2.110.6.tgz", + "integrity": "sha512-kDNKncLtYBI/QT8HuVGSui/F3KrC3iWy7KM5M3EvWbEujf1cL+fdwMr2eeA8PEZUIPBeNbDxRzdFZo5t86iqUQ==", + "license": "MIT", + "dependencies": { + "tslib": "2.8.1" + }, + "engines": { + "node": ">=22.0.0" + } + }, + "node_modules/@supabase/realtime-js": { + "version": "2.110.6", + "resolved": "https://registry.npmjs.org/@supabase/realtime-js/-/realtime-js-2.110.6.tgz", + "integrity": "sha512-F2BhRmbC1dJlNA6cdjtbIWQJNV112TYbzEc0/0UFZYvKL52mL0PR0ZrlpIPEludV85WEmjA1RpjW2H5tNQka9g==", + "license": "MIT", + "dependencies": { + "@supabase/phoenix": "0.4.4", + "tslib": "2.8.1" + }, + "engines": { + "node": ">=22.0.0" + } + }, + "node_modules/@supabase/storage-js": { + "version": "2.110.6", + "resolved": "https://registry.npmjs.org/@supabase/storage-js/-/storage-js-2.110.6.tgz", + "integrity": "sha512-DHS7Jy8MCu2sltDBjANu6oGjeUBUY+bowBnga6CySwtAF2BBKCTJGULnt7wyKPSLSXsoiCyzTwyU1r3SAx++Rg==", + "license": "MIT", + "dependencies": { + "iceberg-js": "^0.8.1", + "tslib": "2.8.1" + }, + "engines": { + "node": ">=22.0.0" + } + }, + "node_modules/@supabase/supabase-js": { + "version": "2.110.6", + "resolved": "https://registry.npmjs.org/@supabase/supabase-js/-/supabase-js-2.110.6.tgz", + "integrity": "sha512-UJTAz1NUiSRI2mQYhUPvNMwqfkSucV1iSCcMJz8jgsSUTOfic9C3D6LGNOrH6KTvYUhxRvnf3ktq2Sd3IXIQzQ==", + "license": "MIT", + "dependencies": { + "@supabase/auth-js": "2.110.6", + "@supabase/functions-js": "2.110.6", + "@supabase/postgrest-js": "2.110.6", + "@supabase/realtime-js": "2.110.6", + "@supabase/storage-js": "2.110.6" + }, + "engines": { + "node": ">=22.0.0" + } + }, "node_modules/@swc/helpers": { "version": "0.5.15", "resolved": "https://registry.npmjs.org/@swc/helpers/-/helpers-0.5.15.tgz", @@ -2675,6 +2760,15 @@ "dev": true, "license": "MIT" }, + "node_modules/iceberg-js": { + "version": "0.8.1", + "resolved": "https://registry.npmjs.org/iceberg-js/-/iceberg-js-0.8.1.tgz", + "integrity": "sha512-1dhVQZXhcHje7798IVM+xoo/1ZdVfzOMIc8/rgVSijRK38EDqOJoGula9N/8ZI5RD8QTxNQtK/Gozpr+qUqRRA==", + "license": "MIT", + "engines": { + "node": ">=20.0.0" + } + }, "node_modules/istanbul-lib-coverage": { "version": "3.2.2", "resolved": "https://registry.npmjs.org/istanbul-lib-coverage/-/istanbul-lib-coverage-3.2.2.tgz", diff --git a/package.json b/package.json index 472f14f..1353995 100644 --- a/package.json +++ b/package.json @@ -4,7 +4,7 @@ "private": false, "type": "module", "license": "AGPL-3.0-only", - "description": "Open-source, serverless helpdesk engine \u2014 shared inbox, threaded email, knowledge base \u2014 for teams who live on Vercel and Supabase.", + "description": "Open-source, serverless helpdesk engine — shared inbox, threaded email, knowledge base — for teams who live on Vercel and Supabase.", "engines": { "node": ">=20" }, @@ -16,10 +16,12 @@ "test": "vitest run", "test:watch": "vitest", "test:coverage": "vitest run --coverage", - "dev:api": "tsx scripts/dev-api.ts" + "dev:api": "tsx scripts/dev-api.ts", + "migrate": "tsx scripts/migrate.ts" }, "dependencies": { "@electric-sql/pglite": "^0.5.4", + "@supabase/supabase-js": "^2.110.6", "jose": "^6.2.3", "mimetext": "^3.0.28", "pg": "^8.22.0", diff --git a/scripts/migrate.ts b/scripts/migrate.ts new file mode 100644 index 0000000..bbf2903 --- /dev/null +++ b/scripts/migrate.ts @@ -0,0 +1,46 @@ +/** + * One-shot database migration runner (HT-43; specs/deploy/gmail-inbound-runbook.md + * Part B2). Applies every migration (`src/db/migrate.ts`) against + * `DATABASE_URL`. + * + * Run ONCE after provisioning the Supabase database, and again whenever new + * migrations are added. The composition root (`src/composition/root.ts`) + * deliberately does NOT migrate on cold start — schema changes are an operator + * step, not something every serverless instance re-runs. + * + * Usage: + * DATABASE_URL='postgres://...' npx tsx scripts/migrate.ts + * # or: npm run migrate (with DATABASE_URL in the environment) + * + * For the one-time DDL you may use the direct (5432) connection string instead + * of the 6543 transaction-mode pooler — either works, since `migrate()`'s + * advisory lock is transaction-scoped and pooler-safe (`src/db/postgres.ts`). + * + * Like `scripts/dev-api.ts`, this lives outside the checked TypeScript project + * (tsconfig `include` covers `src`/`tests`); it is operator tooling run via + * `tsx`, not engine code that ships. + */ + +import { migrate } from '../src/db/migrate.js' +import { createPostgresDb } from '../src/db/postgres.js' + +async function main(): Promise { + const connectionString = process.env.DATABASE_URL + if (connectionString === undefined || connectionString.trim().length === 0) { + console.error('scripts/migrate: DATABASE_URL is required (the Postgres connection string).') + process.exit(1) + } + + const db = await createPostgresDb({ connectionString }) + try { + await migrate(db) + console.log('scripts/migrate: all migrations applied.') + } finally { + await db.close() + } +} + +main().catch((err: unknown) => { + console.error('scripts/migrate: migration failed', err) + process.exit(1) +}) diff --git a/specs/deploy/gmail-inbound-runbook.md b/specs/deploy/gmail-inbound-runbook.md index 70cb29a..32b50f7 100644 --- a/specs/deploy/gmail-inbound-runbook.md +++ b/specs/deploy/gmail-inbound-runbook.md @@ -133,15 +133,28 @@ privilege). Encrypts refresh tokens at rest; **losing/rotating it orphans every stored token** (mailboxes must reconnect). - `HELPTHREAD_API_TOKEN` — the Agent-inbox Bearer token (`openssl rand -base64 24`; ≥16 chars). - - `CRON_SECRET` — guards the internal cron/drain endpoints (`openssl rand -base64 24`). + - `HELPTHREAD_SIGNING_SECRET` — the HMAC keyring backing reply/state/view + tokens (`openssl rand -base64 32`; ≥32 chars). Rotating it breaks + threading of replies to already-sent mail (single-secret dogfood limit). + - `CRON_SECRET` — guards the internal cron/drain endpoints (`openssl rand -base64 24`; ≥16 chars). 2. `PUBLIC_BASE_URL` = your production URL (e.g. `https://desk.resonantiq.app`), matching the OAuth redirect URI (A2.3) and the Pub/Sub push endpoint (A3.4). + No trailing slash (the composition root strips one defensively either way). 3. Deploy. `vercel.json` (in the repo) declares the two Vercel Cron jobs: - `*/1 * * * *` → `GET /api/v1/internal/queue/drain` (drain the job queue). - `0 6 * * *` → `GET /api/v1/internal/cron/watch-maintenance` (daily renewal + sweep; UTC). Vercel Cron invokes these as HTTP GETs; the handlers require the `CRON_SECRET` (Vercel sends it as a bearer via the `Authorization` header on cron requests) and are idempotent + lease-bounded. + > **Plan requirement:** the once-a-minute drain needs a **Vercel Pro** (or + > higher) plan. On Hobby, cron jobs may only run **once per day**, and a + > more-frequent expression *fails deployment* — so the ~1-minute delivery + > latency this design targets is a Pro-tier feature. + +All engine code is served by a single catch-all Vercel Function +(`api/[...path].ts`, the Node runtime — NOT Edge, since the engine needs +`node:crypto`) that hands every request to the composition root; no per-route +function files. The cron paths above resolve through that same function. ## Env reference diff --git a/src/composition/app.test.ts b/src/composition/app.test.ts new file mode 100644 index 0000000..05cc6c7 --- /dev/null +++ b/src/composition/app.test.ts @@ -0,0 +1,133 @@ +import { describe, expect, it, vi } from 'vitest' +import { createAppHandler, QUEUE_DRAIN_PATH, WATCH_MAINTENANCE_PATH } from './app.js' + +const CRON_SECRET = 'test-cron-secret-0123456789' +const ORIGIN = 'https://desk.example.test' + +/** Build a handler over spy deps; the inbox API spy returns a recognizable 299 so delegation is observable. */ +function makeHandler(cronSecret: string = CRON_SECRET) { + const inboxApi = vi.fn(async () => new Response('inbox', { status: 299 })) + const drainQueue = vi.fn(async () => ({ claimed: 3, acked: 3 })) + const runWatchMaintenance = vi.fn(async () => ({ total: 1, renewed: 1 })) + const handler = createAppHandler({ inboxApi, cronSecret, drainQueue, runWatchMaintenance }) + return { handler, inboxApi, drainQueue, runWatchMaintenance } +} + +/** A request to `path`; attaches `Authorization: Bearer ` unless `secret` is null. */ +function req( + path: string, + { + method = 'GET', + secret = CRON_SECRET as string | null, + }: { method?: string; secret?: string | null } = {}, +): Request { + const headers: Record = {} + if (secret !== null) headers.Authorization = `Bearer ${secret}` + return new Request(`${ORIGIN}${path}`, { method, headers }) +} + +describe('createAppHandler — non-cron delegation', () => { + it('delegates a normal inbox path to the inbox API unchanged', async () => { + const { handler, inboxApi, drainQueue, runWatchMaintenance } = makeHandler() + const request = req('/api/v1/conversations', { secret: null }) + + const res = await handler(request) + + expect(res.status).toBe(299) + expect(inboxApi).toHaveBeenCalledOnce() + expect(inboxApi).toHaveBeenCalledWith(request) + expect(drainQueue).not.toHaveBeenCalled() + expect(runWatchMaintenance).not.toHaveBeenCalled() + }) + + it('delegates the Gmail webhook path (also under /api/v1/inbound) to the inbox API', async () => { + const { handler, inboxApi } = makeHandler() + const res = await handler(req('/api/v1/inbound/gmail', { method: 'POST', secret: null })) + expect(res.status).toBe(299) + expect(inboxApi).toHaveBeenCalledOnce() + }) +}) + +describe('createAppHandler — queue drain endpoint', () => { + it('runs the drain and returns its report on a GET with the correct cron secret', async () => { + const { handler, drainQueue, inboxApi } = makeHandler() + + const res = await handler(req(QUEUE_DRAIN_PATH)) + + expect(res.status).toBe(200) + expect(await res.json()).toEqual({ ok: true, report: { claimed: 3, acked: 3 } }) + expect(res.headers.get('Cache-Control')).toBe('no-store') + expect(drainQueue).toHaveBeenCalledOnce() + expect(inboxApi).not.toHaveBeenCalled() + }) + + it('rejects a wrong cron secret with 401 and never runs the work', async () => { + const { handler, drainQueue } = makeHandler() + const res = await handler(req(QUEUE_DRAIN_PATH, { secret: 'wrong-secret-9999999999' })) + expect(res.status).toBe(401) + expect(drainQueue).not.toHaveBeenCalled() + }) + + it('rejects a missing Authorization header with 401', async () => { + const { handler, drainQueue } = makeHandler() + const res = await handler(req(QUEUE_DRAIN_PATH, { secret: null })) + expect(res.status).toBe(401) + expect(drainQueue).not.toHaveBeenCalled() + }) + + it('checks auth BEFORE method — a wrong-secret POST is 401, not 405 (no method oracle for an unauthenticated caller)', async () => { + const { handler, drainQueue } = makeHandler() + const res = await handler(req(QUEUE_DRAIN_PATH, { method: 'POST', secret: 'wrong-9999999999' })) + expect(res.status).toBe(401) + expect(drainQueue).not.toHaveBeenCalled() + }) + + it('rejects a non-GET method (authenticated) with 405', async () => { + const { handler, drainQueue } = makeHandler() + const res = await handler(req(QUEUE_DRAIN_PATH, { method: 'POST' })) + expect(res.status).toBe(405) + expect(drainQueue).not.toHaveBeenCalled() + }) + + it('answers a generic 500 (never the error text) when the work throws', async () => { + const inboxApi = vi.fn(async () => new Response(null, { status: 299 })) + const drainQueue = vi.fn(async () => { + throw new Error('secret-internal-detail-should-not-leak') + }) + const runWatchMaintenance = vi.fn(async () => ({})) + const handler = createAppHandler({ + inboxApi, + cronSecret: CRON_SECRET, + drainQueue, + runWatchMaintenance, + }) + + const res = await handler(req(QUEUE_DRAIN_PATH)) + const bodyText = await res.text() + + expect(res.status).toBe(500) + expect(bodyText).not.toContain('secret-internal-detail-should-not-leak') + expect(JSON.parse(bodyText)).toEqual({ + error: { code: 'server_error', message: 'Internal server error.' }, + }) + }) +}) + +describe('createAppHandler — watch-maintenance endpoint', () => { + it('runs the maintenance sweep and returns its report on an authenticated GET', async () => { + const { handler, runWatchMaintenance } = makeHandler() + + const res = await handler(req(WATCH_MAINTENANCE_PATH)) + + expect(res.status).toBe(200) + expect(await res.json()).toEqual({ ok: true, report: { total: 1, renewed: 1 } }) + expect(runWatchMaintenance).toHaveBeenCalledOnce() + }) + + it('rejects a wrong cron secret with 401', async () => { + const { handler, runWatchMaintenance } = makeHandler() + const res = await handler(req(WATCH_MAINTENANCE_PATH, { secret: 'nope-9999999999999' })) + expect(res.status).toBe(401) + expect(runWatchMaintenance).not.toHaveBeenCalled() + }) +}) diff --git a/src/composition/app.ts b/src/composition/app.ts new file mode 100644 index 0000000..3cb30ab --- /dev/null +++ b/src/composition/app.ts @@ -0,0 +1,117 @@ +/** + * The composition root's unified request handler (HT-43): one + * `(request: Request) => Promise` that fronts BOTH the Agent Inbox + * API (`createInboxApi`, `src/api/index.ts`) and the two internal cron + * endpoints Vercel Cron invokes (specs/deploy/gmail-inbound-runbook.md Part + * C). A single Vercel function (`api/[...path].ts`) delegates every request + * here; this module decides internal-cron-vs-inbox by pathname. + * + * ## Why the cron endpoints live here, not inside `createInboxApi` + * + * The runbook (and this ticket's plan) deliberately keep the drain/maintenance + * endpoints OUT of the Agent Inbox API's route table: they are deploy-infra + * (cron plumbing), not part of the Agent-inbox product surface + * (specs/api/agent-inbox-v1.md), and they authenticate with a DIFFERENT + * credential — the `CRON_SECRET` Vercel attaches as `Authorization: Bearer` + * (Vercel's own "Securing cron jobs" mechanism), not the service Bearer token + * every inbox route checks. Handling them here keeps `createInboxApi`'s + * surface exactly the spec'd one, with no cron-only routes or a second + * credential leaking into it. + * + * ## The cron contract (Vercel Cron) + * + * Vercel invokes a cron `path` with an HTTP **GET**, attaching + * `Authorization: Bearer ` when that env var is set. Vercel does + * NOT retry a failed invocation and may occasionally miss OR duplicate a run, + * so both endpoints' work must be idempotent and reconciliation-based — which + * the drain (lease-based `FOR UPDATE SKIP LOCKED`, `createPostgresQueue`) and + * the maintenance sweep (re-reads each mailbox's stored cursor; ingest dedups) + * both already are. A failed run simply retries on the next tick. + */ + +import { authenticateRequest } from '../api/auth.js' +import { apiError, json } from '../api/responses.js' + +/** `GET` (Vercel Cron) → drain one bounded batch of the durable job queue (runbook Part C: every minute). */ +export const QUEUE_DRAIN_PATH = '/api/v1/internal/queue/drain' + +/** `GET` (Vercel Cron) → daily Gmail `watch()` re-arm + reconciliation sweep (runbook Part C: daily at 06:00 UTC). */ +export const WATCH_MAINTENANCE_PATH = '/api/v1/internal/cron/watch-maintenance' + +/** Dependencies {@link createAppHandler} closes over. */ +export interface AppHandlerDeps { + /** The Agent Inbox API handler (`createInboxApi`) — every non-cron request is delegated here unchanged. */ + inboxApi: (request: Request) => Promise + /** The `CRON_SECRET` both internal endpoints require as `Authorization: Bearer ` (constant-time compared). */ + cronSecret: string + /** Drain one bounded batch of the job queue; returns a JSON-serializable report for the response body + logs. */ + drainQueue: () => Promise + /** Run one daily watch-renewal + reconciliation-sweep pass; returns a JSON-serializable report. */ + runWatchMaintenance: () => Promise +} + +/** + * Build the unified handler. Routes the two internal cron paths to their + * `CRON_SECRET`-guarded handlers and delegates everything else — the whole + * Agent Inbox API surface, including the Gmail webhook, connect, and callback + * — to `deps.inboxApi` unchanged. + */ +export function createAppHandler(deps: AppHandlerDeps): (request: Request) => Promise { + return async (request: Request): Promise => { + const { pathname } = new URL(request.url) + + if (pathname === QUEUE_DRAIN_PATH) { + return handleCronEndpoint(request, deps.cronSecret, 'queue-drain', deps.drainQueue) + } + if (pathname === WATCH_MAINTENANCE_PATH) { + return handleCronEndpoint( + request, + deps.cronSecret, + 'watch-maintenance', + deps.runWatchMaintenance, + ) + } + + return deps.inboxApi(request) + } +} + +/** + * The shared shape of an internal cron endpoint: GET-only, `CRON_SECRET` + * Bearer-gated, running an idempotent unit of work and returning its report. + * + * Auth is checked with `authenticateRequest` (`src/api/auth.ts`) — the exact + * constant-time `Authorization: Bearer ` comparison the inbox API + * uses for its service token, reused here against the cron secret since Vercel + * attaches the `CRON_SECRET` in exactly that header shape. The auth failure is + * a generic `401` that reveals nothing about which check failed — an + * unauthenticated caller can't even tell a wrong secret from a wrong method + * (the method check runs only AFTER auth), so this endpoint is not a probe for + * whether Gmail push/cron is configured. + * + * A thrown error from the work is logged server-side and answered with a + * generic `500` (never the error's own text — a store/queue/Gmail error could + * carry internal detail): safe because Vercel Cron retries on the next tick + * and the work is idempotent + lease-bounded. + */ +async function handleCronEndpoint( + request: Request, + cronSecret: string, + label: string, + work: () => Promise, +): Promise { + if (!authenticateRequest(request, cronSecret)) { + return apiError(401, 'unauthorized', 'Missing or invalid credentials.') + } + if (request.method !== 'GET') { + return apiError(405, 'method_not_allowed', 'This method is not supported here.') + } + + try { + const report = await work() + return json(200, { ok: true, report }) + } catch (err) { + console.error(`[composition] internal cron endpoint '${label}' failed`, err) + return apiError(500, 'server_error', 'Internal server error.') + } +} diff --git a/src/composition/config.test.ts b/src/composition/config.test.ts new file mode 100644 index 0000000..2f18ca6 --- /dev/null +++ b/src/composition/config.test.ts @@ -0,0 +1,137 @@ +import { describe, expect, it } from 'vitest' +import { loadConfig } from './config.js' + +/** A complete, valid env for the whole contract; individual cases override/delete one key. */ +function validEnv(): Record { + return { + DATABASE_URL: 'postgres://user:pass@db.pooler.supabase.com:6543/postgres', + SUPABASE_URL: 'https://abcdefgh.supabase.co', + SUPABASE_SERVICE_ROLE_KEY: 'service-role-key-value', + HELPTHREAD_BLOB_BUCKET: 'helpthread-blobs', + GMAIL_OAUTH_CLIENT_ID: 'client-id.apps.googleusercontent.com', + GMAIL_OAUTH_CLIENT_SECRET: 'gmail-oauth-client-secret', + GMAIL_PUBSUB_TOPIC: 'projects/resonantiq-helpthread/topics/gmail-push', + GMAIL_PUBSUB_SUBSCRIPTION: 'projects/resonantiq-helpthread/subscriptions/gmail-push-sub', + GMAIL_PUSH_SERVICE_ACCOUNT: 'gmail-push-invoker@resonantiq-helpthread.iam.gserviceaccount.com', + // base64 of exactly 32 bytes. + HELPTHREAD_TOKEN_ENC_KEY: Buffer.alloc(32, 7).toString('base64'), + HELPTHREAD_API_TOKEN: 'api-token-at-least-16-chars', + HELPTHREAD_SIGNING_SECRET: 'signing-secret-at-least-32-characters-long', + CRON_SECRET: 'cron-secret-at-least-16', + PUBLIC_BASE_URL: 'https://desk.resonantiq.app', + HELPTHREAD_MAIL_DOMAIN: 'mail.resonantiq.app', + HELPTHREAD_SUPPORT_ADDRESS: 'support@resonantiq.app', + } +} + +describe('loadConfig — happy path', () => { + it('parses a complete valid env into an AppConfig', () => { + const config = loadConfig(validEnv()) + + expect(config.databaseUrl).toBe('postgres://user:pass@db.pooler.supabase.com:6543/postgres') + expect(config.gmailPubsubTopic).toBe('projects/resonantiq-helpthread/topics/gmail-push') + expect(config.supportAddress).toBe('support@resonantiq.app') + expect(config.mailDomain).toBe('mail.resonantiq.app') + }) + + it('decodes HELPTHREAD_TOKEN_ENC_KEY to a 32-byte Buffer', () => { + const config = loadConfig(validEnv()) + expect(Buffer.isBuffer(config.tokenEncryptionKey)).toBe(true) + expect(config.tokenEncryptionKey.length).toBe(32) + }) + + it('strips a trailing slash from PUBLIC_BASE_URL so URL concatenation never double-slashes', () => { + const config = loadConfig({ ...validEnv(), PUBLIC_BASE_URL: 'https://desk.resonantiq.app/' }) + expect(config.publicBaseUrl).toBe('https://desk.resonantiq.app') + }) +}) + +describe('loadConfig — missing / malformed values', () => { + it('throws naming a single missing required variable', () => { + const env = validEnv() + delete (env as Record).DATABASE_URL + expect(() => loadConfig(env)).toThrow(/DATABASE_URL/) + }) + + it('treats a whitespace-only value as missing', () => { + expect(() => loadConfig({ ...validEnv(), GMAIL_PUBSUB_TOPIC: ' ' })).toThrow( + /GMAIL_PUBSUB_TOPIC/, + ) + }) + + it('aggregates ALL problems into one error, not just the first', () => { + const env = validEnv() + delete (env as Record).DATABASE_URL + delete (env as Record).SUPABASE_URL + delete (env as Record).HELPTHREAD_MAIL_DOMAIN + + let message = '' + try { + loadConfig(env) + } catch (err) { + message = err instanceof Error ? err.message : String(err) + } + expect(message).toContain('DATABASE_URL') + expect(message).toContain('SUPABASE_URL') + expect(message).toContain('HELPTHREAD_MAIL_DOMAIN') + }) + + it('rejects a too-short HELPTHREAD_API_TOKEN', () => { + expect(() => loadConfig({ ...validEnv(), HELPTHREAD_API_TOKEN: 'short' })).toThrow( + /HELPTHREAD_API_TOKEN/, + ) + }) + + it('rejects a too-short HELPTHREAD_SIGNING_SECRET (below the 32-char keyring floor)', () => { + expect(() => loadConfig({ ...validEnv(), HELPTHREAD_SIGNING_SECRET: 'too-short' })).toThrow( + /HELPTHREAD_SIGNING_SECRET/, + ) + }) + + it('rejects a too-short CRON_SECRET', () => { + expect(() => loadConfig({ ...validEnv(), CRON_SECRET: 'short' })).toThrow(/CRON_SECRET/) + }) + + it('rejects a HELPTHREAD_TOKEN_ENC_KEY that is not base64 of 32 bytes', () => { + // base64 of 16 bytes — decodes fine, wrong length. + const shortKey = Buffer.alloc(16, 1).toString('base64') + expect(() => loadConfig({ ...validEnv(), HELPTHREAD_TOKEN_ENC_KEY: shortKey })).toThrow( + /HELPTHREAD_TOKEN_ENC_KEY/, + ) + }) + + it('rejects a PUBLIC_BASE_URL that is not an absolute http(s) URL', () => { + expect(() => loadConfig({ ...validEnv(), PUBLIC_BASE_URL: 'not a url' })).toThrow( + /PUBLIC_BASE_URL/, + ) + expect(() => loadConfig({ ...validEnv(), PUBLIC_BASE_URL: 'ftp://desk.example.com' })).toThrow( + /PUBLIC_BASE_URL/, + ) + }) +}) + +describe('loadConfig — never leaks a secret value', () => { + it('reports a too-short token by LENGTH, never echoing the secret value', () => { + const secretValue = 'sekret' + let message = '' + try { + loadConfig({ ...validEnv(), HELPTHREAD_API_TOKEN: secretValue }) + } catch (err) { + message = err instanceof Error ? err.message : String(err) + } + expect(message).toContain('HELPTHREAD_API_TOKEN') + expect(message).not.toContain(secretValue) + }) + + it('reports a bad encryption key without echoing the (secret) raw value', () => { + const badKey = 'this-is-not-a-valid-key-value-at-all' + let message = '' + try { + loadConfig({ ...validEnv(), HELPTHREAD_TOKEN_ENC_KEY: badKey }) + } catch (err) { + message = err instanceof Error ? err.message : String(err) + } + expect(message).toContain('HELPTHREAD_TOKEN_ENC_KEY') + expect(message).not.toContain(badKey) + }) +}) diff --git a/src/composition/config.ts b/src/composition/config.ts new file mode 100644 index 0000000..6e6c00a --- /dev/null +++ b/src/composition/config.ts @@ -0,0 +1,234 @@ +/** + * Deploy-time environment configuration for the Helpthread engine's + * composition root (HT-43; specs/deploy/gmail-inbound-runbook.md's env + * reference). {@link loadConfig} reads the full env contract, validates every + * value eagerly, and returns a typed {@link AppConfig} — or throws ONE error + * listing every problem at once, so a misconfigured deploy fails loudly at + * boot rather than on a mailbox's first push (the same fail-fast discipline + * `createGmailConnectService`/`createGmailOAuthTokenService` already apply to + * their own required fields). + * + * ## Never leaks a secret value + * + * Validation errors name the offending VARIABLE and the nature of the problem + * ("missing", "must be at least N characters", "must be base64 of a 32-byte + * key") — never the value itself. A too-short secret's length is a structural + * fact, not the secret; the bytes never appear in a thrown message or a log + * line (matching `token-crypto.ts`'s and `gmail-oauth.ts`'s discipline). + * + * ## This module reads env; the composition root wires adapters + * + * `loadConfig` is pure over its `env` argument (defaulting to `process.env`) + * and constructs no adapters, opens no connections, and imports no platform + * SDK — it only parses and validates. Turning an `AppConfig` into concrete + * providers wired into `createInboxApi` is `./root.ts`'s job. + */ + +import { decodeEncryptionKey } from '../store/token-crypto.js' + +/** + * Minimum service Bearer token length — mirrors `createInboxApi`'s own + * `MIN_API_TOKEN_LENGTH` (`src/api/index.ts`) so a token this module accepts + * is never one the API then refuses to start with. + */ +const MIN_API_TOKEN_LENGTH = 16 + +/** + * Minimum HMAC signing-secret length — mirrors `reply-token.ts`'s + * `MIN_SECRET_LENGTH` (32), the floor `assertValidKeyring` enforces on the + * keyring `./root.ts` builds from {@link AppConfig.signingSecret}. + */ +const MIN_SIGNING_SECRET_LENGTH = 32 + +/** + * Minimum `CRON_SECRET` length — Vercel's own guidance for the value it + * auto-attaches as `Authorization: Bearer ` on cron invocations + * ("a random string of at least 16 characters"). + */ +const MIN_CRON_SECRET_LENGTH = 16 + +/** + * The fully-validated deploy configuration `./root.ts` builds concrete + * adapters from. Every field is present and well-formed by construction — a + * missing or malformed value is a {@link loadConfig} throw, never a + * `undefined` slot a downstream adapter has to re-check. + */ +export interface AppConfig { + /** Supabase transaction-mode pooler URI (port 6543) — `PostgresDb`'s connection string. */ + databaseUrl: string + /** Supabase project URL — the Storage `BlobStore` adapter's base. */ + supabaseUrl: string + /** Supabase `service_role` key — server-only; grants full Storage access. */ + supabaseServiceRoleKey: string + /** Private Storage bucket name attachment/oversized-raw blobs are namespaced within. */ + blobBucket: string + /** The Internal OAuth app's client id (connect flow + token refresh). */ + gmailOAuthClientId: string + /** The Internal OAuth app's client secret. */ + gmailOAuthClientSecret: string + /** Cloud Pub/Sub topic `watch()` arms notifications to (`projects/{project}/topics/{topic}`). */ + gmailPubsubTopic: string + /** The exact push subscription the webhook accepts (`projects/{project}/subscriptions/{name}`). */ + gmailPubsubSubscription: string + /** The push subscription's OIDC service-account email (the JWT `email` claim the webhook matches). */ + gmailPushServiceAccount: string + /** The 32-byte AES-256 key decoded from `HELPTHREAD_TOKEN_ENC_KEY` — encrypts stored refresh tokens at rest. */ + tokenEncryptionKey: Buffer + /** The Agent-inbox service Bearer token every API request is checked against. */ + apiToken: string + /** The HMAC signing secret backing the reply/state/view-token keyring. */ + signingSecret: string + /** The secret guarding the internal cron/drain endpoints (Vercel Cron's `Authorization: Bearer` value). */ + cronSecret: string + /** The deployment's public origin, trailing slash stripped — the base for the OAuth redirect, the push `aud`, and any absolute URL. */ + publicBaseUrl: string + /** Domain minted into outbound `Message-ID`s. */ + mailDomain: string + /** The connected support mailbox's address — the `from` on every Agent reply, and the mailbox outbound sends resolve their token from. */ + supportAddress: string +} + +/** Accumulates human-readable, secret-free validation problems for a single combined throw. */ +class ConfigErrors { + readonly #problems: string[] = [] + + add(message: string): void { + this.#problems.push(message) + } + + /** A present, non-empty (after trim) string, or `null` with a recorded "missing" problem. */ + requireString(env: NodeJS.ProcessEnv, name: string): string | null { + const raw = env[name] + if (raw === undefined || raw.trim().length === 0) { + this.add(`${name} is required but missing or empty`) + return null + } + return raw + } + + /** {@link requireString} plus a minimum-length floor (length only — never the value — is reported). */ + requireMinLength(env: NodeJS.ProcessEnv, name: string, min: number): string | null { + const value = this.requireString(env, name) + if (value === null) return null + if (value.length < min) { + this.add(`${name} must be at least ${min} characters (got ${value.length})`) + return null + } + return value + } + + throwIfAny(): void { + if (this.#problems.length > 0) { + throw new Error( + `loadConfig: invalid deployment configuration — fix the following environment ${ + this.#problems.length === 1 ? 'variable' : 'variables' + } and redeploy:\n - ${this.#problems.join('\n - ')}`, + ) + } + } +} + +/** + * Read and validate the whole env contract into an {@link AppConfig}. Throws + * one aggregated, secret-free error (see the module doc) if anything is + * missing or malformed. `env` defaults to `process.env`; injectable purely + * for tests. + */ +export function loadConfig(env: NodeJS.ProcessEnv = process.env): AppConfig { + const errors = new ConfigErrors() + + const databaseUrl = errors.requireString(env, 'DATABASE_URL') + const supabaseUrl = errors.requireString(env, 'SUPABASE_URL') + const supabaseServiceRoleKey = errors.requireString(env, 'SUPABASE_SERVICE_ROLE_KEY') + const blobBucket = errors.requireString(env, 'HELPTHREAD_BLOB_BUCKET') + const gmailOAuthClientId = errors.requireString(env, 'GMAIL_OAUTH_CLIENT_ID') + const gmailOAuthClientSecret = errors.requireString(env, 'GMAIL_OAUTH_CLIENT_SECRET') + const gmailPubsubTopic = errors.requireString(env, 'GMAIL_PUBSUB_TOPIC') + const gmailPubsubSubscription = errors.requireString(env, 'GMAIL_PUBSUB_SUBSCRIPTION') + const gmailPushServiceAccount = errors.requireString(env, 'GMAIL_PUSH_SERVICE_ACCOUNT') + const apiToken = errors.requireMinLength(env, 'HELPTHREAD_API_TOKEN', MIN_API_TOKEN_LENGTH) + const signingSecret = errors.requireMinLength( + env, + 'HELPTHREAD_SIGNING_SECRET', + MIN_SIGNING_SECRET_LENGTH, + ) + const cronSecret = errors.requireMinLength(env, 'CRON_SECRET', MIN_CRON_SECRET_LENGTH) + const mailDomain = errors.requireString(env, 'HELPTHREAD_MAIL_DOMAIN') + const supportAddress = errors.requireString(env, 'HELPTHREAD_SUPPORT_ADDRESS') + + const tokenEncryptionKey = resolveEncryptionKey(env, errors) + const publicBaseUrl = resolvePublicBaseUrl(env, errors) + + errors.throwIfAny() + + // Every value above is non-null here: throwIfAny() would have thrown + // otherwise. The non-null assertions make that guarantee explicit to the + // type system rather than defeating it with a cast on the whole object. + return { + databaseUrl: databaseUrl as string, + supabaseUrl: supabaseUrl as string, + supabaseServiceRoleKey: supabaseServiceRoleKey as string, + blobBucket: blobBucket as string, + gmailOAuthClientId: gmailOAuthClientId as string, + gmailOAuthClientSecret: gmailOAuthClientSecret as string, + gmailPubsubTopic: gmailPubsubTopic as string, + gmailPubsubSubscription: gmailPubsubSubscription as string, + gmailPushServiceAccount: gmailPushServiceAccount as string, + tokenEncryptionKey: tokenEncryptionKey as Buffer, + apiToken: apiToken as string, + signingSecret: signingSecret as string, + cronSecret: cronSecret as string, + publicBaseUrl: publicBaseUrl as string, + mailDomain: mailDomain as string, + supportAddress: supportAddress as string, + } +} + +/** + * Decode + length-validate `HELPTHREAD_TOKEN_ENC_KEY` via + * `decodeEncryptionKey` (`src/store/token-crypto.ts`), folding its throw into + * the aggregated error set rather than aborting the rest of the validation. + * Its message names the variable and the required shape, never the bytes. + */ +function resolveEncryptionKey(env: NodeJS.ProcessEnv, errors: ConfigErrors): Buffer | null { + const raw = errors.requireString(env, 'HELPTHREAD_TOKEN_ENC_KEY') + if (raw === null) return null + try { + return decodeEncryptionKey(raw) + } catch { + // decodeEncryptionKey's own message is safe (length-only), but re-phrase + // for this env var by name; never echo the (secret) raw value. + errors.add( + 'HELPTHREAD_TOKEN_ENC_KEY must be the base64 encoding of a 32-byte key (e.g. `openssl rand -base64 32`)', + ) + return null + } +} + +/** + * Validate `PUBLIC_BASE_URL` as an absolute http(s) origin and normalize away + * a trailing slash, so `${publicBaseUrl}/api/...` concatenations (the OAuth + * redirect URI, the push `aud`) never double-slash — a byte mismatch that + * would silently break the redirect-URI / audience equality checks Google and + * the webhook enforce. + */ +function resolvePublicBaseUrl(env: NodeJS.ProcessEnv, errors: ConfigErrors): string | null { + const raw = errors.requireString(env, 'PUBLIC_BASE_URL') + if (raw === null) return null + let parsed: URL + try { + parsed = new URL(raw) + } catch { + errors.add( + `PUBLIC_BASE_URL must be an absolute URL (e.g. https://desk.example.com), got ${JSON.stringify(raw)}`, + ) + return null + } + if (parsed.protocol !== 'https:' && parsed.protocol !== 'http:') { + errors.add( + `PUBLIC_BASE_URL must be an http(s) URL, got protocol ${JSON.stringify(parsed.protocol)}`, + ) + return null + } + return raw.replace(/\/+$/, '') +} diff --git a/src/composition/root.test.ts b/src/composition/root.test.ts new file mode 100644 index 0000000..eb81a4e --- /dev/null +++ b/src/composition/root.test.ts @@ -0,0 +1,167 @@ +/** + * Integration test for the composition root (HT-43): build the WHOLE app over + * an in-memory PGlite `Db` + a fake `BlobStore` (no real Postgres, Supabase, + * or network) and drive real `Request`s through the unified handler. This is + * the end-to-end proof that every adapter is wired correctly — the inbox API, + * the two CRON_SECRET-guarded cron endpoints, and the Gmail connect/webhook + * surfaces all respond as expected through one `buildApp` call. + * + * buildApp is network-free at construction: `createPostgresDb` is skipped (a + * PGlite `Db` is injected), and `createGooglePushKeySource` only fetches + * Google's JWKS lazily on first verify — which none of these cases triggers + * (the webhook case fails the pre-verify checks first). + */ + +import { afterEach, beforeEach, describe, expect, it } from 'vitest' +import type { Db } from '../db/client.js' +import { createPgliteDb } from '../db/client.js' +import { migrate } from '../db/migrate.js' +import type { BlobStore } from '../providers/index.js' +import type { AppConfig } from './config.js' +import { buildApp } from './root.js' + +const API_TOKEN = 'test-api-token-16-plus-chars' +const CRON_SECRET = 'test-cron-secret-16-plus' +const ORIGIN = 'https://desk.example.test' + +/** A fully-valid AppConfig for the injected-infra build (databaseUrl/supabase* are unused when db/blobStore are injected). */ +function testConfig(): AppConfig { + return { + databaseUrl: 'postgres://unused', + supabaseUrl: 'https://unused.supabase.co', + supabaseServiceRoleKey: 'unused', + blobBucket: 'unused', + gmailOAuthClientId: 'test-client-id', + gmailOAuthClientSecret: 'test-client-secret', + gmailPubsubTopic: 'projects/p/topics/gmail-push', + gmailPubsubSubscription: 'projects/p/subscriptions/gmail-push-sub', + gmailPushServiceAccount: 'push@p.iam.gserviceaccount.com', + tokenEncryptionKey: Buffer.alloc(32, 5), + apiToken: API_TOKEN, + signingSecret: 'signing-secret-at-least-32-characters-long!', + cronSecret: CRON_SECRET, + publicBaseUrl: ORIGIN, + mailDomain: 'mail.example.test', + supportAddress: 'support@example.test', + } +} + +/** In-memory BlobStore fake, mirroring `src/mail/ingest.test.ts`'s. */ +function fakeBlobStore(): BlobStore { + const store = new Map() + return { + async put(key, data) { + store.set(key, data) + }, + async get(key) { + const data = store.get(key) + if (data === undefined) throw new Error(`fakeBlobStore: no object at key ${key}`) + return data + }, + async getSignedUrl(key) { + return `https://blob.example.test/${key}` + }, + async delete(key) { + store.delete(key) + }, + async exists(key) { + return store.has(key) + }, + } +} + +describe('buildApp — end-to-end wiring over PGlite', () => { + let db: Db + let handler: (request: Request) => Promise + + beforeEach(async () => { + db = await createPgliteDb() + await migrate(db) + handler = await buildApp(testConfig(), { db, blobStore: fakeBlobStore() }) + }) + + afterEach(async () => { + await db.close() + }) + + it('serves the Agent Inbox API: GET /conversations with the service Bearer → 200', async () => { + const res = await handler( + new Request(`${ORIGIN}/api/v1/conversations`, { + headers: { Authorization: `Bearer ${API_TOKEN}` }, + }), + ) + expect(res.status).toBe(200) + const body = (await res.json()) as { conversations: unknown[] } + expect(Array.isArray(body.conversations)).toBe(true) + }) + + it('rejects an inbox request with a wrong service Bearer → 401', async () => { + const res = await handler( + new Request(`${ORIGIN}/api/v1/conversations`, { + headers: { Authorization: 'Bearer wrong-token-0000' }, + }), + ) + expect(res.status).toBe(401) + }) + + it('drives the queue-drain cron endpoint: authenticated GET drains the (empty) queue → 200 report', async () => { + const res = await handler( + new Request(`${ORIGIN}/api/v1/internal/queue/drain`, { + headers: { Authorization: `Bearer ${CRON_SECRET}` }, + }), + ) + expect(res.status).toBe(200) + const body = (await res.json()) as { ok: boolean; report: { claimed: number } } + expect(body.ok).toBe(true) + expect(body.report.claimed).toBe(0) + }) + + it('drives the watch-maintenance cron endpoint: authenticated GET → 200 report (0 active mailboxes)', async () => { + const res = await handler( + new Request(`${ORIGIN}/api/v1/internal/cron/watch-maintenance`, { + headers: { Authorization: `Bearer ${CRON_SECRET}` }, + }), + ) + expect(res.status).toBe(200) + const body = (await res.json()) as { ok: boolean; report: { total: number } } + expect(body.ok).toBe(true) + expect(body.report.total).toBe(0) + }) + + it('guards the cron endpoints with CRON_SECRET, not the service token → 401 on the wrong secret', async () => { + const res = await handler( + new Request(`${ORIGIN}/api/v1/internal/queue/drain`, { + headers: { Authorization: `Bearer ${API_TOKEN}` }, + }), + ) + expect(res.status).toBe(401) + }) + + it('wires gmailConnect: POST /inbound/gmail/connect (Bearer) → 200 with a Google consent URL', async () => { + const res = await handler( + new Request(`${ORIGIN}/api/v1/inbound/gmail/connect`, { + method: 'POST', + headers: { Authorization: `Bearer ${API_TOKEN}` }, + }), + ) + expect(res.status).toBe(200) + const body = (await res.json()) as { consentUrl: string } + expect(body.consentUrl).toContain('https://accounts.google.com/o/oauth2/v2/auth') + expect(body.consentUrl).toContain('client_id=test-client-id') + // redirect_uri is PUBLIC_BASE_URL + the callback path, URL-encoded. + expect(body.consentUrl).toContain(encodeURIComponent(`${ORIGIN}/api/v1/inbound/gmail/callback`)) + }) + + it('wires the Gmail push webhook: POST /inbound/gmail with no valid OIDC JWT → uniform 403 (route handled, not 404)', async () => { + const res = await handler( + new Request(`${ORIGIN}/api/v1/inbound/gmail`, { + method: 'POST', + headers: { 'Content-Type': 'application/json' }, + body: '{}', + }), + ) + expect(res.status).toBe(403) + const body = (await res.json()) as { error: { code: string } } + expect(body.error.code).toBe('gmail_push_rejected') + }) +}) diff --git a/src/composition/root.ts b/src/composition/root.ts new file mode 100644 index 0000000..b94e6ee --- /dev/null +++ b/src/composition/root.ts @@ -0,0 +1,273 @@ +/** + * The composition root (HT-43) — the ONE place concrete platform adapters are + * constructed from config and wired into the framework-agnostic engine. Per + * `src/providers/README.md`: "an adapter is selected at the composition + * root... engine modules never `import` an adapter themselves; they only ever + * see the interface type." Every `import` of a concrete adapter + * (`@supabase/*`, the Gmail adapters, the Postgres queue, `PostgresDb`) lives + * here and nowhere in `src/api/**`, `src/mail/**`, or `src/store/**`. + * + * ## What this builds + * + * {@link buildApp} constructs, from an {@link AppConfig}: + * - the `PostgresDb` (Supabase pooler) + every store over it, + * - the token encryption seam (`createMailboxTokenStore` with the decoded key), + * - the Gmail OAuth token service + the outbound `EmailSender`, + * - the Gmail push signature verifier (JWKS source built ONCE — see below), + * - the durable Postgres job queue, + * - the Gmail connect/consent service, + * - `createInboxApi` with `gmailPush` + `gmailConnect` PRESENT (they are + * absent-by-default on the engine; this root is where they get wired), and + * - the two internal cron closures (queue drain, watch maintenance), + * + * then hands them to {@link createAppHandler} (`./app.ts`) as one + * `(request) => Promise`. + * + * ## Per-instance singleton + * + * On Vercel each warm function instance loads this module once; {@link getApp} + * memoizes the built handler so the `pg.Pool` and the Gmail JWKS cache + * (`createGooglePushKeySource`, whose fetch cache only caches if the source is + * reused across requests) are constructed once per instance, not per request. + * + * ## Secrets in, never out + * + * The refresh-token encryption key, OAuth client secret, and every OAuth token + * are threaded to the exact adapter that needs them and nowhere else; none is + * ever logged (the adapters enforce that themselves — see `token-crypto.ts`, + * `gmail-oauth.ts`, `sender.ts`). This module adds no logging of config at all. + */ + +import type { GmailConnectDeps } from '../api/gmail-connect.js' +import { + GMAIL_RECONCILE_TOPIC, + type GmailPushDeps, + type GmailReconcileJob, +} from '../api/gmail-webhook.js' +import { createInboxApi } from '../api/index.js' +import type { Db } from '../db/client.js' +import { createPostgresDb } from '../db/postgres.js' +import { createGmailConnectService } from '../mail/gmail-connect.js' +import { createGmailOAuthTokenService } from '../mail/gmail-oauth.js' +import { createGmailReconcileHandler } from '../mail/gmail-reconcile.js' +import { + type GmailWatchMaintenanceDeps, + runGmailWatchMaintenance, +} from '../mail/gmail-watch-maintenance.js' +import { ingestInboundMessage } from '../mail/ingest.js' +import type { Keyring } from '../mail/reply-token.js' +import { + createGmailEmailSender, + createGmailHistoryClient, + createGmailPushSignatureVerifier, + createGmailWatchClient, + createGooglePushKeySource, +} from '../providers/adapters/gmail/index.js' +import { createPostgresQueue } from '../providers/adapters/postgres-queue/index.js' +import { createSupabaseStorageBlobStore } from '../providers/adapters/supabase-storage/index.js' +import type { BlobStore } from '../providers/blob.js' +import type { QueueMessage, QueueMessageHandler } from '../providers/queue.js' +import { + createConversationStore, + createGmailWatchStateStore, + createInboundDeliveryStore, + createMailboxStore, + createMailboxTokenStore, +} from '../store/index.js' +import { createAppHandler } from './app.js' +import { type AppConfig, loadConfig } from './config.js' + +/** + * The OAuth scopes the connect flow requests (gmail-connect.md §3, least + * privilege for the dogfood): read inbound mail + send replies. `watch()` + * needs `gmail.readonly`; `users.messages.send` needs `gmail.send`. + */ +const GMAIL_SCOPES = [ + 'https://www.googleapis.com/auth/gmail.readonly', + 'https://www.googleapis.com/auth/gmail.send', +] + +/** + * The keyId stamped into every minted reply/state/view token (`Keyring`, + * `src/mail/reply-token.ts`). FIXED and stable across deploys — it is embedded + * in every outbound `Message-ID`, so changing it would break threading of + * replies to already-sent mail. This deploy runs a single signing secret + * (`HELPTHREAD_SIGNING_SECRET`); rotating that secret while keeping this keyId + * invalidates tokens minted under the old secret (the keyring's `retired` keys + * — a future multi-secret enhancement — is how a non-breaking rotation would + * work, and is deliberately not built for the single-mailbox dogfood). + */ +const SIGNING_KEY_ID = 'ht1' + +/** Overrides for {@link buildApp}, injected only by tests so the wiring is exercised without real Postgres/Supabase or any network. */ +export interface BuildAppOverrides { + /** A `Db` to use instead of constructing a `PostgresDb` from `config.databaseUrl` (e.g. an in-memory PGlite `Db`). */ + db?: Db + /** A `BlobStore` to use instead of the Supabase Storage adapter (e.g. an in-memory fake). */ + blobStore?: BlobStore +} + +/** + * Construct every concrete adapter from `config` and wire them into the + * unified request handler. `overrides` lets a test substitute an in-memory + * `Db`/`BlobStore`; production passes none. Async because constructing the + * `PostgresDb` is (it validates any configured schema before returning — + * though the dogfood uses none, so no network round-trip happens at build). + * + * Does NOT run migrations: schema creation is a separate one-shot + * (`scripts/migrate.ts`, runbook Part B2), never something every cold start + * re-runs. + */ +export async function buildApp( + config: AppConfig, + overrides?: BuildAppOverrides, +): Promise<(request: Request) => Promise> { + const db = overrides?.db ?? (await createPostgresDb({ connectionString: config.databaseUrl })) + const blobStore = + overrides?.blobStore ?? + createSupabaseStorageBlobStore({ + url: config.supabaseUrl, + serviceRoleKey: config.supabaseServiceRoleKey, + bucket: config.blobBucket, + }) + + // --- Stores (all over the one Db). --- + const store = createConversationStore(db) + const mailboxStore = createMailboxStore(db) + const tokenStore = createMailboxTokenStore(db, config.tokenEncryptionKey) + const watchStateStore = createGmailWatchStateStore(db) + const inboundDeliveryStore = createInboundDeliveryStore(db) + + // --- The HMAC keyring backing reply/state/view tokens (single current key). --- + const keyring: Keyring = { current: { keyId: SIGNING_KEY_ID, secret: config.signingSecret } } + + // --- Durable job queue — ONE instance shared by the webhook enqueue, the + // maintenance sweep enqueue, and the drain, so they share tunables. --- + const queue = createPostgresQueue(db) + + // --- Gmail OAuth token service (reads the encrypted refresh token, refreshes). --- + const tokenService = createGmailOAuthTokenService({ + tokenStore, + mailboxStore, + clientId: config.gmailOAuthClientId, + clientSecret: config.gmailOAuthClientSecret, + }) + + // --- Outbound EmailSender. Gmail sends are per-mailbox (per access token); + // for the single-mailbox dogfood, resolve the support mailbox by address at + // SEND time (it is created dynamically at connect, so it may not exist when + // this root is first built) and bind its live token. --- + const sender = createGmailEmailSender({ + getAccessToken: async () => { + const mailbox = await mailboxStore.getMailboxByAddress(config.supportAddress) + if (mailbox === null) { + throw new Error( + `composition: no connected mailbox for support address ${config.supportAddress} — connect it via the OAuth flow first`, + ) + } + return tokenService.getAccessToken(mailbox.id) + }, + }) + + // --- Gmail push webhook deps. The JWKS key source is built ONCE here and + // reused across every request (its fetch cache only caches if reused — see + // createGooglePushKeySource's doc). --- + const gmailPush: GmailPushDeps = { + verifySignature: createGmailPushSignatureVerifier( + { + endpointUrl: `${config.publicBaseUrl}/api/v1/inbound/gmail`, + serviceAccountEmail: config.gmailPushServiceAccount, + }, + createGooglePushKeySource(), + ), + subscription: config.gmailPubsubSubscription, + mailboxes: mailboxStore, + queue, + } + + // --- Gmail connect/consent service. --- + const connectService = createGmailConnectService({ + db, + clientId: config.gmailOAuthClientId, + clientSecret: config.gmailOAuthClientSecret, + redirectUri: `${config.publicBaseUrl}/api/v1/inbound/gmail/callback`, + topicName: config.gmailPubsubTopic, + scopes: GMAIL_SCOPES, + keyring, + mailboxStore, + tokenStore, + watchStateStore, + createWatchClient: (getAccessToken) => createGmailWatchClient({ getAccessToken }), + }) + const gmailConnect: GmailConnectDeps = { service: connectService } + + // --- The Agent Inbox API, with gmailPush + gmailConnect PRESENT (the engine + // leaves them absent by default; this root is the one place they are wired). + // openTracking is intentionally OMITTED — the shipped privacy default is OFF + // (v1.1 designed contract). --- + const inboxApi = createInboxApi({ + store, + apiToken: config.apiToken, + sender, + keyring, + mailDomain: config.mailDomain, + supportAddress: config.supportAddress, + gmailPush, + gmailConnect, + }) + + // --- The reconcile handler the queue drain dispatches to. --- + const reconcileHandler = createGmailReconcileHandler({ + tokenService, + mailboxStore, + watchStateStore, + blobStore, + ingest: (raw) => ingestInboundMessage(raw, { db, inboundDeliveryStore, blobStore, keyring }), + createHistoryClient: (getAccessToken) => createGmailHistoryClient({ getAccessToken }), + }) + + // The drain's handler map is typed `QueueMessageHandler` (the queue + // stores arbitrary JSON payloads); the topic string is what guarantees the + // payload is a GmailReconcileJob, so the narrowing cast at this single wiring + // point is the honest boundary between "any queued job" and "this topic's + // job shape". + const drainHandlers: Record> = { + [GMAIL_RECONCILE_TOPIC]: (message) => + reconcileHandler(message as QueueMessage), + } + + // --- Watch-maintenance deps (daily re-arm + sweep). --- + const watchMaintenanceDeps: GmailWatchMaintenanceDeps = { + tokenService, + mailboxStore, + watchStateStore, + queue, + createWatchClient: (getAccessToken) => createGmailWatchClient({ getAccessToken }), + topicName: config.gmailPubsubTopic, + } + + return createAppHandler({ + inboxApi, + cronSecret: config.cronSecret, + drainQueue: () => queue.drainOnce({ handlers: drainHandlers }), + runWatchMaintenance: () => runGmailWatchMaintenance(watchMaintenanceDeps), + }) +} + +/** + * The per-instance memoized handler the Vercel entry (`api/[...path].ts`) + * calls. Reads + validates `process.env` via `loadConfig` and builds the app + * once; a config error becomes a rejected (cached) promise the entry maps to a + * generic 500 — a misconfigured instance stays down loudly rather than + * rebuilding the pool on every request. + */ +let appPromise: Promise<(request: Request) => Promise> | undefined + +export function getApp(): Promise<(request: Request) => Promise> { + if (appPromise === undefined) { + // The IIFE turns even loadConfig's synchronous throw into a rejected + // promise, so the entry's single try/await/catch covers both. + appPromise = (async () => buildApp(loadConfig()))() + } + return appPromise +} diff --git a/src/providers/adapters/supabase-storage/index.test.ts b/src/providers/adapters/supabase-storage/index.test.ts new file mode 100644 index 0000000..231d723 --- /dev/null +++ b/src/providers/adapters/supabase-storage/index.test.ts @@ -0,0 +1,184 @@ +import { describe, expect, it } from 'vitest' +import { createSupabaseStorageBlobStore, type SupabaseStorageBucket } from './index.js' + +/** Records the last options `upload` was called with, for asserting upsert/contentType are forwarded. */ +interface UploadCall { + path: string + body: Uint8Array + options?: { contentType?: string; upsert?: boolean } +} + +/** An in-memory fake of the narrow bucket-client slice the adapter uses, mirroring the SDK's `{ data, error }` convention. */ +function fakeBucket(initial: Record = {}): { + bucket: SupabaseStorageBucket + store: Map + uploads: UploadCall[] +} { + const store = new Map(Object.entries(initial)) + const uploads: UploadCall[] = [] + const bucket: SupabaseStorageBucket = { + async upload(path, body, options) { + uploads.push({ path, body, options }) + store.set(path, body) + return { data: { path }, error: null } + }, + async download(path) { + const data = store.get(path) + if (data === undefined) return { data: null, error: { message: 'Object not found' } } + // Copy into a fresh ArrayBuffer so the element satisfies the DOM lib's + // BlobPart (a generic Uint8Array does not). + return { data: new Blob([new Uint8Array(data).buffer as ArrayBuffer]), error: null } + }, + async createSignedUrl(path, expiresIn) { + if (!store.has(path)) return { data: null, error: { message: 'Object not found' } } + return { + data: { signedUrl: `https://signed.example.test/${path}?exp=${expiresIn}` }, + error: null, + } + }, + async remove(paths) { + for (const p of paths) store.delete(p) + return { data: [], error: null } + }, + async exists(path) { + return { data: store.has(path), error: null } + }, + } + return { bucket, store, uploads } +} + +describe('createSupabaseStorageBlobStore — round trips', () => { + it('put then get returns the same bytes', async () => { + const { bucket } = fakeBucket() + const blob = createSupabaseStorageBlobStore({ + url: 'u', + serviceRoleKey: 'k', + bucket: 'b', + bucketClient: bucket, + }) + const bytes = new Uint8Array([1, 2, 3, 250, 0, 128]) + + await blob.put('inbound/raw/mbox-1/msg-1', bytes, { contentType: 'message/rfc822' }) + const got = await blob.get('inbound/raw/mbox-1/msg-1') + + expect(Array.from(got)).toEqual(Array.from(bytes)) + }) + + it('forwards upsert:true and contentType to the underlying upload (put is create-OR-overwrite)', async () => { + const { bucket, uploads } = fakeBucket() + const blob = createSupabaseStorageBlobStore({ + url: 'u', + serviceRoleKey: 'k', + bucket: 'b', + bucketClient: bucket, + }) + + await blob.put('k1', new Uint8Array([9]), { contentType: 'application/octet-stream' }) + + expect(uploads).toHaveLength(1) + expect(uploads[0].options).toEqual({ upsert: true, contentType: 'application/octet-stream' }) + }) + + it('get rejects for a missing key (BlobStore.get contract)', async () => { + const { bucket } = fakeBucket() + const blob = createSupabaseStorageBlobStore({ + url: 'u', + serviceRoleKey: 'k', + bucket: 'b', + bucketClient: bucket, + }) + await expect(blob.get('does-not-exist')).rejects.toThrow(/get failed/) + }) + + it('getSignedUrl returns the signed URL and forwards the expiry', async () => { + const { bucket } = fakeBucket({ k1: new Uint8Array([1]) }) + const blob = createSupabaseStorageBlobStore({ + url: 'u', + serviceRoleKey: 'k', + bucket: 'b', + bucketClient: bucket, + }) + const url = await blob.getSignedUrl('k1', 300) + expect(url).toBe('https://signed.example.test/k1?exp=300') + }) + + it('delete removes the object; a subsequent get rejects', async () => { + const { bucket } = fakeBucket({ k1: new Uint8Array([1]) }) + const blob = createSupabaseStorageBlobStore({ + url: 'u', + serviceRoleKey: 'k', + bucket: 'b', + bucketClient: bucket, + }) + await blob.delete('k1') + await expect(blob.get('k1')).rejects.toThrow() + }) + + it('exists reflects presence', async () => { + const { bucket } = fakeBucket({ here: new Uint8Array([1]) }) + const blob = createSupabaseStorageBlobStore({ + url: 'u', + serviceRoleKey: 'k', + bucket: 'b', + bucketClient: bucket, + }) + expect(await blob.exists('here')).toBe(true) + expect(await blob.exists('gone')).toBe(false) + }) +}) + +describe('createSupabaseStorageBlobStore — error propagation', () => { + /** A bucket whose every method returns a Storage error, to prove the adapter surfaces (never swallows) them. */ + function erroringBucket(message: string): SupabaseStorageBucket { + const err = { data: null, error: { message } } + return { + async upload() { + return { data: null, error: { message } } + }, + async download() { + return err + }, + async createSignedUrl() { + return err + }, + async remove() { + return { data: null, error: { message } } + }, + async exists() { + return { data: false, error: { message } } + }, + } + } + + it('put surfaces an upload error (never a silent success)', async () => { + const blob = createSupabaseStorageBlobStore({ + url: 'u', + serviceRoleKey: 'k', + bucket: 'b', + bucketClient: erroringBucket('quota exceeded'), + }) + await expect(blob.put('k', new Uint8Array([1]), { contentType: 'text/plain' })).rejects.toThrow( + /put failed.*quota exceeded/, + ) + }) + + it('exists surfaces an infrastructure error rather than reporting false', async () => { + const blob = createSupabaseStorageBlobStore({ + url: 'u', + serviceRoleKey: 'k', + bucket: 'b', + bucketClient: erroringBucket('storage unreachable'), + }) + await expect(blob.exists('k')).rejects.toThrow(/exists check failed.*storage unreachable/) + }) + + it('delete surfaces a genuine error (but the fake here proves the throw path)', async () => { + const blob = createSupabaseStorageBlobStore({ + url: 'u', + serviceRoleKey: 'k', + bucket: 'b', + bucketClient: erroringBucket('permission denied'), + }) + await expect(blob.delete('k')).rejects.toThrow(/delete failed.*permission denied/) + }) +}) diff --git a/src/providers/adapters/supabase-storage/index.ts b/src/providers/adapters/supabase-storage/index.ts new file mode 100644 index 0000000..79d1740 --- /dev/null +++ b/src/providers/adapters/supabase-storage/index.ts @@ -0,0 +1,168 @@ +/** + * Supabase Storage `BlobStore` adapter (HT-43; CHARTER.md §4 names Supabase + * Storage as the first blob adapter target). Implements the `BlobStore` + * interface (`src/providers/blob.ts`) over a Supabase Storage bucket via + * `@supabase/supabase-js`'s Storage client. + * + * Per `src/providers/README.md`'s adapter-boundary rule, this is wired in + * ONLY at the composition root (`../../../composition/root.ts`) — engine code + * (`src/mail/ingest.ts`, `src/mail/gmail-reconcile.ts`) only ever sees the + * `BlobStore` interface type, never this module or the `@supabase/*` SDK. + * + * ## Private bucket, signed reads only + * + * The bucket MUST be private (runbook Part B3). `BlobStore`'s contract + * (`src/providers/blob.ts`) is explicit that objects are never public: the + * only read path this adapter exposes outward is {@link BlobStore.getSignedUrl}, + * a time-limited URL for one object. This adapter never configures the bucket + * for public read and never mints a permanent public URL. + * + * ## Key namespacing is the caller's job + * + * A `key` is passed to the Storage API verbatim as the object path (its `/` + * separators become Storage "folders", which is fine). Per `blob.ts`'s + * key-namespacing contract, callers — not this adapter — are responsible for + * choosing per-mailbox/per-conversation keys so one tenant's objects can't + * collide with or be enumerated from another's; this adapter treats `key` as + * an opaque string. + * + * ## The service_role key is a server-only secret + * + * The Supabase client is built with the `service_role` key, which bypasses + * Row-Level Security and grants full Storage access — it must only ever live + * server-side (the runbook is explicit). This adapter never logs it, and + * `auth.persistSession`/`autoRefreshToken` are disabled: there is no browser + * session to persist and no interactive user to refresh a token for, and a + * background refresh timer would be a resource leak in a serverless function. + */ + +import { createClient, type SupabaseClientOptions } from '@supabase/supabase-js' +import type { BlobStore } from '../../blob.js' + +/** + * The narrow slice of a Supabase Storage bucket client + * (`SupabaseClient.storage.from(bucket)`) this adapter uses. Declared + * structurally rather than importing the SDK's `StorageFileApi` type so a + * test can pass a hand-rolled fake with no SDK client at all — the real + * `.from(bucket)` result is structurally assignable to this. Every method + * mirrors the SDK's `{ data, error }` result convention (an operation either + * yields `data` or a non-null `error`, never throws for an expected failure). + */ +export interface SupabaseStorageBucket { + upload( + path: string, + body: Uint8Array, + options?: { contentType?: string; upsert?: boolean }, + ): Promise<{ data: unknown; error: { message: string } | null }> + download(path: string): Promise<{ data: Blob | null; error: { message: string } | null }> + createSignedUrl( + path: string, + expiresIn: number, + ): Promise<{ data: { signedUrl: string } | null; error: { message: string } | null }> + remove(paths: string[]): Promise<{ data: unknown; error: { message: string } | null }> + exists(path: string): Promise<{ data: boolean; error: { message: string } | null }> +} + +/** Options for {@link createSupabaseStorageBlobStore}. */ +export interface SupabaseStorageBlobStoreOptions { + /** Supabase project URL (`SUPABASE_URL`). */ + url: string + /** Supabase `service_role` key (`SUPABASE_SERVICE_ROLE_KEY`) — server-only secret. */ + serviceRoleKey: string + /** The private Storage bucket name (`HELPTHREAD_BLOB_BUCKET`). */ + bucket: string + /** + * A pre-built bucket client, injected for tests so the adapter's own + * mapping logic is exercised without a real Supabase client or network. + * Omitted in production — the adapter builds one from `url`/`serviceRoleKey`/ + * `bucket`. + */ + bucketClient?: SupabaseStorageBucket +} + +/** Extract a safe, non-secret message from a Storage `{ error }` result. */ +function errorMessage(error: { message: string } | null): string { + return error?.message ?? 'unknown error' +} + +/** + * Build a `BlobStore` backed by a Supabase Storage bucket. See the module doc + * for the private-bucket / signed-read / server-only-secret contract. + */ +export function createSupabaseStorageBlobStore( + options: SupabaseStorageBlobStoreOptions, +): BlobStore { + const bucket = options.bucketClient ?? buildBucketClient(options) + + return { + async put(key, data, opts): Promise { + // upsert: true — `BlobStore.put` is "create OR overwrite"; without it + // Supabase errors when an object already exists at the key. + const { error } = await bucket.upload(key, data, { + upsert: true, + ...(opts.contentType !== undefined ? { contentType: opts.contentType } : {}), + }) + if (error !== null) { + throw new Error( + `supabase-storage: put failed for key ${JSON.stringify(key)}: ${errorMessage(error)}`, + ) + } + }, + + async get(key): Promise { + const { data, error } = await bucket.download(key) + if (error !== null || data === null) { + // `BlobStore.get` rejects if no object exists at the key. + throw new Error( + `supabase-storage: get failed for key ${JSON.stringify(key)}: ${ + error !== null ? errorMessage(error) : 'no object at key' + }`, + ) + } + return new Uint8Array(await data.arrayBuffer()) + }, + + async getSignedUrl(key, expiresInSeconds): Promise { + const { data, error } = await bucket.createSignedUrl(key, expiresInSeconds) + if (error !== null || data === null) { + throw new Error( + `supabase-storage: getSignedUrl failed for key ${JSON.stringify(key)}: ${ + error !== null ? errorMessage(error) : 'no signed URL returned' + }`, + ) + } + return data.signedUrl + }, + + async delete(key): Promise { + // `BlobStore.delete` of a missing key is a no-op, not an error: Supabase + // `remove` of a nonexistent path returns an empty result WITHOUT an + // error, so only a genuine infrastructure error surfaces here. + const { error } = await bucket.remove([key]) + if (error !== null) { + throw new Error( + `supabase-storage: delete failed for key ${JSON.stringify(key)}: ${errorMessage(error)}`, + ) + } + }, + + async exists(key): Promise { + const { data, error } = await bucket.exists(key) + if (error !== null) { + throw new Error( + `supabase-storage: exists check failed for key ${JSON.stringify(key)}: ${errorMessage(error)}`, + ) + } + return data + }, + } +} + +/** Construct the real Supabase Storage bucket client from connection config. See the module doc on the disabled auth options. */ +function buildBucketClient(options: SupabaseStorageBlobStoreOptions): SupabaseStorageBucket { + const clientOptions: SupabaseClientOptions<'public'> = { + auth: { persistSession: false, autoRefreshToken: false }, + } + const client = createClient(options.url, options.serviceRoleKey, clientOptions) + return client.storage.from(options.bucket) +} diff --git a/tsconfig.json b/tsconfig.json index ed02166..7132417 100644 --- a/tsconfig.json +++ b/tsconfig.json @@ -11,5 +11,5 @@ "forceConsistentCasingInFileNames": true, "types": ["node"] }, - "include": ["src/**/*.ts", "tests/**/*.ts"] + "include": ["src/**/*.ts", "tests/**/*.ts", "api/**/*.ts"] } diff --git a/vercel.json b/vercel.json new file mode 100644 index 0000000..63a4cf9 --- /dev/null +++ b/vercel.json @@ -0,0 +1,18 @@ +{ + "$schema": "https://openapi.vercel.sh/vercel.json", + "functions": { + "api/**/*.ts": { + "maxDuration": 60 + } + }, + "crons": [ + { + "path": "/api/v1/internal/queue/drain", + "schedule": "*/1 * * * *" + }, + { + "path": "/api/v1/internal/cron/watch-maintenance", + "schedule": "0 6 * * *" + } + ] +} From 243811cabf49c3accc0b0027ab8fb347528efd48 Mon Sep 17 00:00:00 2001 From: TJ Baker <1617679+zaridan@users.noreply.github.com> Date: Wed, 15 Jul 2026 13:41:18 -0700 Subject: [PATCH 4/4] fix(deploy): address CodeRabbit review on HT-43 (origin-only base URL, race defense, docs) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - config.ts: PUBLIC_BASE_URL is now validated as a bare origin — reject a path, query, fragment, or embedded credentials (not silently strip them) and return URL.origin, so the redirect_uri / push `aud` concatenations can't be corrupted by a stray path. + rejection tests for each non-origin form. - vercel.json: cap function maxDuration at 50s — below the 60s job lease and the 60s cron interval — so a drain is always killed before its own leases expire and consecutive drains never overlap (defense-in-depth for the queue's lease-reclaim race; the SQL-level claim-generation fence is a tracked follow-up, needing a real two-connection Postgres race test). - tsconfig: typecheck-cover scripts/migrate.ts (deploy-critical operator tool). - supabase-storage: add download + createSignedUrl error-path tests. - runbook: dead-letter smoke check reworded (retained dead-letters are by design — check growth/age/rate, not pass/fail); failed-cron alerting note; the maxDuration --- specs/deploy/gmail-inbound-runbook.md | 21 ++++++++++--- src/composition/config.test.ts | 17 ++++++++++ src/composition/config.ts | 31 +++++++++++++++---- .../adapters/supabase-storage/index.test.ts | 22 +++++++++++++ tsconfig.json | 2 +- vercel.json | 2 +- 6 files changed, 83 insertions(+), 12 deletions(-) diff --git a/specs/deploy/gmail-inbound-runbook.md b/specs/deploy/gmail-inbound-runbook.md index 32b50f7..9324e0a 100644 --- a/specs/deploy/gmail-inbound-runbook.md +++ b/specs/deploy/gmail-inbound-runbook.md @@ -3,7 +3,7 @@ Status: draft (HT-43). The one-time operator steps to take the merged engine code (HT-34…HT-42) live: a deployed Vercel environment where **RIQ's own inbound Gmail flows end-to-end** into a Helpthread conversation. This is the -"deployed, end to end" acceptance HT-43 owns; the actual **real Google +"deployed, end-to-end" acceptance HT-43 owns; the actual **real Google consent** that connects the mailbox is the last step and is tracked as **HT-44**. @@ -19,7 +19,7 @@ come to exist. ## 0. Architecture being deployed -``` +```text Gmail mailbox ──watch()──▶ Cloud Pub/Sub topic ──push sub (OIDC JWT)──▶ POST /api/v1/inbound/gmail (webhook: verify JWT → enqueue reconcile job → 2xx) │ enqueue (durable INSERT into the PG job queue — commits BEFORE the 2xx) @@ -150,6 +150,17 @@ privilege). > higher) plan. On Hobby, cron jobs may only run **once per day**, and a > more-frequent expression *fails deployment* — so the ~1-minute delivery > latency this design targets is a Pro-tier feature. +4. **Vercel does not retry a failed cron invocation** — a transient non-2xx is + simply retried on the *next* scheduled tick. The queue drain self-heals on + the following minute; but the **daily** watch-maintenance job would go a full + day between attempts, so **alert on its non-2xx responses** (Vercel's cron + logs, or your log drain) rather than waiting to notice a stale mailbox. +5. **`maxDuration` must stay below the queue lease.** `vercel.json` caps the + function at **50s**, under both the 60s job lease (`DEFAULT_LEASE_MS`, + `src/providers/adapters/postgres-queue/`) and the 60s cron interval: the + function is always killed *before* any lease it holds expires, so a + still-running drain can never race a concurrent drain that reclaimed one of + its rows. If you raise the lease, keep `maxDuration` comfortably under it. All engine code is served by a single catch-all Vercel Function (`api/[...path].ts`, the Node runtime — NOT Edge, since the engine needs @@ -199,8 +210,10 @@ With the deploy live and env set: - [ ] Send a test email **to** the connected mailbox → within ~1 min (the drain tick) a new conversation appears (`GET /api/v1/conversations`). - [ ] Pub/Sub subscription **oldest-unacked-message age** stays low (no backlog). -- [ ] The job-queue table: no rows stuck `dead_lettered_at IS NOT NULL`; - oldest `ready` job age stays under a minute or two. +- [ ] The job-queue table: no *unexpected* dead-letter growth (retained + `dead_lettered_at IS NOT NULL` rows are by design — inspect them by + age/count/rate, not as a pass/fail), and oldest `ready` job age stays + under a minute or two. - [ ] Reply from the Agent inbox → the reply arrives at the customer, and a reply back **threads** into the same conversation (the sacred outbound-token check — HT-44's live proof). diff --git a/src/composition/config.test.ts b/src/composition/config.test.ts index 2f18ca6..2b0dab8 100644 --- a/src/composition/config.test.ts +++ b/src/composition/config.test.ts @@ -108,6 +108,23 @@ describe('loadConfig — missing / malformed values', () => { /PUBLIC_BASE_URL/, ) }) + + it('rejects a PUBLIC_BASE_URL that is not a bare origin (path/query/fragment/credentials)', () => { + for (const bad of [ + 'https://desk.example.com/base', // path + 'https://desk.example.com/api/v1', // deeper path + 'https://desk.example.com?x=1', // query + 'https://desk.example.com/#frag', // fragment + 'https://user:pass@desk.example.com', // credentials + ]) { + expect(() => loadConfig({ ...validEnv(), PUBLIC_BASE_URL: bad })).toThrow(/PUBLIC_BASE_URL/) + } + }) + + it('returns the canonical origin (bare host, no trailing slash) for a valid origin with a port', () => { + const config = loadConfig({ ...validEnv(), PUBLIC_BASE_URL: 'https://desk.example.com:8443/' }) + expect(config.publicBaseUrl).toBe('https://desk.example.com:8443') + }) }) describe('loadConfig — never leaks a secret value', () => { diff --git a/src/composition/config.ts b/src/composition/config.ts index 6e6c00a..5acc675 100644 --- a/src/composition/config.ts +++ b/src/composition/config.ts @@ -206,11 +206,15 @@ function resolveEncryptionKey(env: NodeJS.ProcessEnv, errors: ConfigErrors): Buf } /** - * Validate `PUBLIC_BASE_URL` as an absolute http(s) origin and normalize away - * a trailing slash, so `${publicBaseUrl}/api/...` concatenations (the OAuth - * redirect URI, the push `aud`) never double-slash — a byte mismatch that - * would silently break the redirect-URI / audience equality checks Google and - * the webhook enforce. + * Validate `PUBLIC_BASE_URL` as a bare http(s) **origin** and return its + * canonical form (`URL.origin` — scheme + host + optional port, no trailing + * slash). It must be origin-ONLY: a path, query, fragment, or embedded + * credentials are rejected, not silently dropped, because `${publicBaseUrl}` is + * concatenated with fixed paths (`/api/v1/inbound/gmail`, `.../callback`) to + * form the OAuth redirect URI and the push `aud` — values Google and the + * webhook byte-compare. A stray path/query in the base would corrupt those + * (`https://x/foo` + `/api/...` → `https://x/foo/api/...`, a mismatch), and + * silently stripping it could hide a real operator misconfiguration. */ function resolvePublicBaseUrl(env: NodeJS.ProcessEnv, errors: ConfigErrors): string | null { const raw = errors.requireString(env, 'PUBLIC_BASE_URL') @@ -230,5 +234,20 @@ function resolvePublicBaseUrl(env: NodeJS.ProcessEnv, errors: ConfigErrors): str ) return null } - return raw.replace(/\/+$/, '') + // Origin-only: `new URL('https://x')`/`new URL('https://x/')` both have + // pathname `/`; anything else (a real path), or a query/fragment/credentials, + // means the value is not a bare origin. + if ( + (parsed.pathname !== '/' && parsed.pathname !== '') || + parsed.search !== '' || + parsed.hash !== '' || + parsed.username !== '' || + parsed.password !== '' + ) { + errors.add( + 'PUBLIC_BASE_URL must be a bare origin with no path, query, fragment, or credentials (e.g. https://desk.example.com)', + ) + return null + } + return parsed.origin } diff --git a/src/providers/adapters/supabase-storage/index.test.ts b/src/providers/adapters/supabase-storage/index.test.ts index 231d723..3a7dcd6 100644 --- a/src/providers/adapters/supabase-storage/index.test.ts +++ b/src/providers/adapters/supabase-storage/index.test.ts @@ -162,6 +162,28 @@ describe('createSupabaseStorageBlobStore — error propagation', () => { ) }) + it('get surfaces a download infrastructure error (distinct from a plain missing object)', async () => { + const blob = createSupabaseStorageBlobStore({ + url: 'u', + serviceRoleKey: 'k', + bucket: 'b', + bucketClient: erroringBucket('storage 503'), + }) + await expect(blob.get('k')).rejects.toThrow(/get failed.*storage 503/) + }) + + it('getSignedUrl surfaces a signing error rather than returning a bad URL', async () => { + const blob = createSupabaseStorageBlobStore({ + url: 'u', + serviceRoleKey: 'k', + bucket: 'b', + bucketClient: erroringBucket('signing key rotated'), + }) + await expect(blob.getSignedUrl('k', 300)).rejects.toThrow( + /getSignedUrl failed.*signing key rotated/, + ) + }) + it('exists surfaces an infrastructure error rather than reporting false', async () => { const blob = createSupabaseStorageBlobStore({ url: 'u', diff --git a/tsconfig.json b/tsconfig.json index 7132417..b335974 100644 --- a/tsconfig.json +++ b/tsconfig.json @@ -11,5 +11,5 @@ "forceConsistentCasingInFileNames": true, "types": ["node"] }, - "include": ["src/**/*.ts", "tests/**/*.ts", "api/**/*.ts"] + "include": ["src/**/*.ts", "tests/**/*.ts", "api/**/*.ts", "scripts/migrate.ts"] } diff --git a/vercel.json b/vercel.json index 63a4cf9..8300e51 100644 --- a/vercel.json +++ b/vercel.json @@ -2,7 +2,7 @@ "$schema": "https://openapi.vercel.sh/vercel.json", "functions": { "api/**/*.ts": { - "maxDuration": 60 + "maxDuration": 50 } }, "crons": [