diff --git a/specs/deploy/gmail-inbound-runbook.md b/specs/deploy/gmail-inbound-runbook.md index b596cc8..bad7130 100644 --- a/specs/deploy/gmail-inbound-runbook.md +++ b/specs/deploy/gmail-inbound-runbook.md @@ -43,7 +43,8 @@ Gmail mailbox ──watch──▶ Cloud Pub/Sub topic ──push sub (OIDC JWT) ▼ 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 + Vercel Cron ──GET /api/v1/internal/cron/reconcile-sweep (every minute)──▶ enqueue reconcile per mailbox + Vercel Cron ──GET /api/v1/internal/cron/watch-maintenance (daily)──▶ re-arm watch() [push only] Operator connect: POST /api/v1/inbound/gmail/connect (Bearer) → consentUrl → browser → Google consent → GET /callback → mailbox connected @@ -71,13 +72,34 @@ the row commits) is what protects invariant #1. --- -## Part A — Google Cloud: OAuth app + Gmail + Pub/Sub - -Do this in the Google Cloud project that will own the push topic. +## Part A — Google Cloud: OAuth app (+ optional Pub/Sub) + +> **Read this before starting.** As of HT-94, only **A1 and A2** are required. +> A3 and A4 configure Gmail **push**, which is now optional: inbound mail +> arrives either by push webhook or by the bounded scheduled fetch that runs +> every minute (CHARTER.md §2, amended 2026-07-20). +> +> **Skipping A3/A4 is the recommended path for most operators.** It removes six +> setup steps — including the two that fail *silently*, the +> domain-restricted-sharing org-policy block and the missing +> `serviceAccountTokenCreator` grant — and removes the requirement that the +> Cloud project have **billing enabled**, which Pub/Sub forces and the Gmail API +> alone does not. +> +> What you give up is latency: push delivers in seconds, the sweep within 60 +> seconds. For a support inbox that difference is not usually worth ten console +> steps. Push remains fully supported and can be added later without a +> reconnect — set the three env vars and redeploy. + +Do this in the Google Cloud project that will own the OAuth app. ### 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`.) +Console → *APIs & Services → Enable APIs* → enable **Gmail API**. + +Also enable **Cloud Pub/Sub API** *only if* you are doing the optional A3. + +(CLI: `gcloud services enable gmail.googleapis.com`, adding +`pubsub.googleapis.com` only when you want push.) ### A2. The Internal OAuth app + client credentials 1. *APIs & Services → OAuth consent screen* → **Internal** user type. Fill @@ -98,7 +120,17 @@ 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 +### A3. The Pub/Sub topic + push subscription — **OPTIONAL** + +> Skip this whole section (and A4) unless you specifically want sub-minute +> latency. Without it the engine ingests through the every-minute reconcile +> sweep, and `GMAIL_PUBSUB_TOPIC` / `GMAIL_PUBSUB_SUBSCRIPTION` / +> `GMAIL_PUSH_SERVICE_ACCOUNT` are all left unset. +> +> **All three or none.** Setting some but not all is rejected at boot with an +> error naming the missing ones — a half-configured push is a push you believe +> works and doesn't, which is exactly the failure this optionality exists to +> remove. 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 @@ -120,8 +152,16 @@ privilege). > 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. +> When push is not configured, connect skips the arm entirely and seeds the +> baseline cursor from `getProfile()` instead; nothing else about connect +> changes. -### A4. Console/CLI gotchas hit during live provisioning (2026-07-17) +### A4. Console/CLI gotchas hit during live provisioning (2026-07-17) — **OPTIONAL, applies only to A3** + +> Both gotchas below fail **silently**: the grant or the subscription looks +> created, and push simply never arrives. They are the strongest single +> argument for skipping A3 entirely — the scheduled sweep has no equivalent +> failure mode, because there is nothing to provision. 1. **Domain-restricted sharing blocks the Gmail publisher grant.** If the org enforces `constraints/iam.allowedPolicyMemberDomains`, granting @@ -179,13 +219,20 @@ privilege). 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 three Vercel Cron jobs: +3. Deploy. `vercel.json` (in the repo) declares **five** Vercel Cron jobs: - `*/1 * * * *` → `GET /api/v1/internal/queue/drain` (drain the job queue — also delivers webhooks, : `WEBHOOK_DELIVERY_TOPIC` is handled here). - `*/1 * * * *` → `GET /api/v1/internal/outbox/drain` (turn `event_outbox` rows into webhook-delivery queue jobs — a SEPARATE tick from the queue drain above; that one then actually sends them). - - `0 6 * * *` → `GET /api/v1/internal/cron/watch-maintenance` (daily renewal + sweep; UTC). + - `*/1 * * * *` → `GET /api/v1/internal/cron/snooze-wake` (HT-77: flip due + `pending`+snoozed conversations back to `active`). + - `*/1 * * * *` → `GET /api/v1/internal/cron/reconcile-sweep` (HT-94: enqueue + a reconcile job per active mailbox — **this is the inbound transport**. + Runs whether or not push is configured; with push it is a backstop, without + it, it is how mail arrives at all). + - `0 6 * * *` → `GET /api/v1/internal/cron/watch-maintenance` (daily `watch()` + renewal; UTC). Reports a skip when push is not configured. 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. @@ -194,10 +241,15 @@ privilege). > 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. + simply retried on the *next* scheduled tick. The every-minute jobs self-heal + 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. + + The reconcile sweep deserves its own alert for a different reason: it + self-heals on the next tick, but a *persistently* failing sweep on a + push-free deployment means **no mail is arriving at all**, silently. Alert on + sustained non-2xx, not on a single one. 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 @@ -221,9 +273,13 @@ function files. The cron paths above resolve through that same function. | `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) | +| `GMAIL_PUBSUB_TOPIC` | Google A3.1 | **OPTIONAL** — `projects/…/topics/…` | +| `GMAIL_PUBSUB_SUBSCRIPTION` | Google A3.4 | **OPTIONAL** — `projects/…/subscriptions/…` | +| `GMAIL_PUSH_SERVICE_ACCOUNT` | Google A3.3 | **OPTIONAL** — the push SA email (JWT `email` claim) | + +> The three `GMAIL_PUBSUB*` / `GMAIL_PUSH*` vars are **all-or-nothing**. Set all +> three to enable push, or none to run on the scheduled sweep alone. Any partial +> combination fails at boot with an error naming what's missing. | `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 | @@ -303,9 +359,10 @@ Each `alerts[]` entry is `: `. The codes are stable: | `ingest-dead-letter-growth` | An inbound delivery exhausted its retry budget in the last 24h — a message an Agent has NOT seen | `SELECT provider_message_id, last_error, attempts FROM inbound_deliveries WHERE status = 'dead-letter' ORDER BY updated_at DESC`; the raw mail is still in Gmail — reprocess after fixing the cause | | `forged-token-burst` | ≥ threshold (default 5) stored deliveries in 24h carried reply tokens that FAILED signature verification — someone is guessing/tampering with threading tokens (threading.md §5) | Search Vercel logs for `forged_token_detected` (WARN); review `senderAddress`/`conversationId` across events. The mail itself threaded safely (a forged token never appends) | | `mailbox-needs-attention` | A mailbox is `paused` (cursor expired — gmail-push.md §5 rebaseline) or `needs_reconnect` (dead OAuth grant) — **inbound mail is not flowing** | `needs_reconnect`: re-run the Part E consent. `paused`: reconnect to rebaseline the cursor, then check for a gap | -| `watch-expiring` | An active mailbox's Gmail `watch` expires in < 72h (or was never armed) — the daily renewal has been failing for days | Function logs for `/internal/cron/watch-maintenance` (`gmail_watch_maintenance` events); a manual `GET` of that endpoint with the cron secret re-arms immediately | -| `webhook-endpoint-auto-disabled` | : a webhook endpoint hit 20 consecutive delivery failures and auto-disabled — a module (or an operator's own integration) has silently stopped receiving events | `SELECT id, url, consecutive_failures FROM webhook_endpoints WHERE status = 'auto_disabled'`; fix the receiving side, then `PATCH /api/v1/webhooks/{id}` with `{"status":"active"}` to re-enable (resets the counter) | -| `webhook-delivery-dead-letter-growth` | : a webhook delivery exhausted its retries in the last 24h (`WEBHOOK_DELIVERY_TOPIC` on `queue_jobs`) | `SELECT payload, last_error FROM queue_jobs WHERE topic = 'webhook.delivery' AND dead_lettered_at IS NOT NULL ORDER BY dead_lettered_at DESC` — `payload.endpointId` names the endpoint; this can precede (or accompany) an eventual auto-disable | +| `watch-expiring` | An active mailbox's Gmail `watch()` expires in < 72h (or was never armed) — the daily renewal has been failing for days. **Only ever raised when push is configured** (HT-94): with no `GMAIL_PUBSUB_*` vars there is no `watch()` to arm, a NULL expiration is the designed steady state, and this alert is suppressed | Function logs for `/internal/cron/watch-maintenance` (`gmail_watch_maintenance` events); a manual `GET` of that endpoint with the cron secret re-arms immediately. On a push-free deployment that GET is a no-op returning `{"skipped":"push-not-configured"}` — if you see this alert there at all, it is a bug, not a mailbox problem | +| `queue-drain-stalled` / `queue-dead-letter-growth` (on a push-free deployment) | The reconcile sweep is the sole inbound transport (HT-94), so sustained queue trouble here means **mail is not arriving at all** | Function logs for `/internal/cron/reconcile-sweep` (`gmail_reconcile_sweep` per-mailbox events, `reconcile_sweep` per-tick summary). Check `SELECT count(*) FROM queue_jobs WHERE topic = 'gmail.reconcile' AND dead_lettered_at IS NULL` — the sweep dedupes on `mailboxId`, so a healthy desk holds at most one live job per mailbox; more than that means the drain is not keeping up | +| `webhook-endpoint-auto-disabled` | HT-69: a webhook endpoint hit 20 consecutive delivery failures and auto-disabled — a module (or an operator's own integration) has silently stopped receiving events | `SELECT id, url, consecutive_failures FROM webhook_endpoints WHERE status = 'auto_disabled'`; fix the receiving side, then `PATCH /api/v1/webhooks/{id}` with `{"status":"active"}` to re-enable (resets the counter) | +| `webhook-delivery-dead-letter-growth` | HT-69: a webhook delivery exhausted its retries in the last 24h (`WEBHOOK_DELIVERY_TOPIC` on `queue_jobs`) | `SELECT payload, last_error FROM queue_jobs WHERE topic = 'webhook.delivery' AND dead_lettered_at IS NOT NULL ORDER BY dead_lettered_at DESC` — `payload.endpointId` names the endpoint; this can precede (or accompany) an eventual auto-disable | ### G3. Structured log events (Vercel log search) @@ -320,9 +377,15 @@ handled by the SAME drain), `outbox_drain` (per outbox-drain tick that claimed at least one `event_outbox` row — claimed/enqueued/dispatched; quiet ticks don't log, same convention as `queue_drain`), `gmail_reconcile` (per reconcile job: cursor positions, skip/retry/ack reasons), and -`gmail_watch_maintenance` (the daily renewal + sweep). Correlate transport -events to ingest events on `(mailboxId, providerMessageId)` -(inbound-ingestion.md §6). +`gmail_watch_maintenance` (the daily `watch()` renewal — push deployments +only), `gmail_reconcile_sweep` (per-mailbox sweep decisions: swept, skipped +for no baseline cursor, failed) and `reconcile_sweep` (the per-tick summary +`{total, swept, skipped, failed}`). Correlate transport events to ingest +events on `(mailboxId, providerMessageId)` (inbound-ingestion.md §6). + +Unlike the drains, the sweep logs **every** tick including quiet ones: on a +push-free deployment it is the only inbound transport, so its silence is the +sole signal that intake has stopped. ## What this runbook does not cover diff --git a/src/composition/app.test.ts b/src/composition/app.test.ts index 9332a28..41d13e6 100644 --- a/src/composition/app.test.ts +++ b/src/composition/app.test.ts @@ -4,6 +4,7 @@ import { HEALTH_PATH, OUTBOX_DRAIN_PATH, QUEUE_DRAIN_PATH, + RECONCILE_SWEEP_PATH, SNOOZE_WAKE_PATH, WATCH_MAINTENANCE_PATH, } from './app.js' @@ -35,6 +36,7 @@ function makeHandler(opts: { cronSecret?: string; uiBaseUrl?: string } = {}) { const drainOutbox = vi.fn(async () => ({ claimed: 2, enqueued: 2, dispatched: 2 })) const runSnoozeWake = vi.fn(async () => ({ due: 1, woken: 1 })) const runWatchMaintenance = vi.fn(async () => ({ total: 1, renewed: 1 })) + const runReconcileSweep = vi.fn(async () => ({ total: 1, swept: 1, skipped: 0, failed: 0 })) const runHealthCheck = vi.fn(async (): Promise => HEALTHY_REPORT) const handler = createAppHandler({ inboxApi, @@ -44,6 +46,7 @@ function makeHandler(opts: { cronSecret?: string; uiBaseUrl?: string } = {}) { drainOutbox, runSnoozeWake, runWatchMaintenance, + runReconcileSweep, runHealthCheck, }) return { @@ -53,6 +56,7 @@ function makeHandler(opts: { cronSecret?: string; uiBaseUrl?: string } = {}) { drainOutbox, runSnoozeWake, runWatchMaintenance, + runReconcileSweep, runHealthCheck, } } @@ -146,6 +150,7 @@ describe('createAppHandler — queue drain endpoint', () => { drainOutbox: vi.fn(async () => ({})), runSnoozeWake: vi.fn(async () => ({})), runWatchMaintenance, + runReconcileSweep: vi.fn(async () => ({})), runHealthCheck: vi.fn(async () => HEALTHY_REPORT), }) @@ -202,6 +207,7 @@ describe('createAppHandler — outbox drain endpoint (HT-69)', () => { }), runSnoozeWake: vi.fn(async () => ({})), runWatchMaintenance: vi.fn(async () => ({})), + runReconcileSweep: vi.fn(async () => ({})), runHealthCheck: vi.fn(async () => HEALTHY_REPORT), }) @@ -256,6 +262,7 @@ describe('createAppHandler — snooze wake endpoint (HT-77)', () => { throw new Error('secret-internal-detail-should-not-leak') }), runWatchMaintenance: vi.fn(async () => ({})), + runReconcileSweep: vi.fn(async () => ({})), runHealthCheck: vi.fn(async () => HEALTHY_REPORT), }) @@ -289,6 +296,78 @@ describe('createAppHandler — watch-maintenance endpoint', () => { }) }) +describe('createAppHandler — reconcile-sweep endpoint (HT-94, the primary inbound transport)', () => { + it('runs the sweep and returns its report on a GET with the correct cron secret', async () => { + const { handler, runReconcileSweep, inboxApi } = makeHandler() + + const res = await handler(req(RECONCILE_SWEEP_PATH)) + + expect(res.status).toBe(200) + expect(await res.json()).toEqual({ + ok: true, + report: { total: 1, swept: 1, skipped: 0, failed: 0 }, + }) + expect(res.headers.get('Cache-Control')).toBe('no-store') + expect(runReconcileSweep).toHaveBeenCalledOnce() + expect(inboxApi).not.toHaveBeenCalled() + }) + + it('rejects a wrong cron secret with 401 and never runs the work', async () => { + const { handler, runReconcileSweep } = makeHandler() + const res = await handler(req(RECONCILE_SWEEP_PATH, { secret: 'wrong-secret-9999999999' })) + expect(res.status).toBe(401) + expect(runReconcileSweep).not.toHaveBeenCalled() + }) + + it('rejects a missing Authorization header with 401', async () => { + const { handler, runReconcileSweep } = makeHandler() + const res = await handler(req(RECONCILE_SWEEP_PATH, { secret: null })) + expect(res.status).toBe(401) + expect(runReconcileSweep).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, runReconcileSweep } = makeHandler() + const res = await handler( + req(RECONCILE_SWEEP_PATH, { method: 'POST', secret: 'wrong-9999999999' }), + ) + expect(res.status).toBe(401) + expect(runReconcileSweep).not.toHaveBeenCalled() + }) + + it('rejects a non-GET method (authenticated) with 405', async () => { + const { handler, runReconcileSweep } = makeHandler() + const res = await handler(req(RECONCILE_SWEEP_PATH, { method: 'POST' })) + expect(res.status).toBe(405) + expect(runReconcileSweep).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 handler = createAppHandler({ + inboxApi, + cronSecret: CRON_SECRET, + drainQueue: vi.fn(async () => ({})), + drainOutbox: vi.fn(async () => ({})), + runSnoozeWake: vi.fn(async () => ({})), + runWatchMaintenance: vi.fn(async () => ({})), + runReconcileSweep: vi.fn(async () => { + throw new Error('secret-internal-detail-should-not-leak') + }), + runHealthCheck: vi.fn(async () => HEALTHY_REPORT), + }) + + const res = await handler(req(RECONCILE_SWEEP_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 — health endpoint (HT-44)', () => { it('answers 200 with the report VERBATIM (not {ok, report}-wrapped) when healthy', async () => { const { handler, runHealthCheck, inboxApi } = makeHandler() diff --git a/src/composition/app.ts b/src/composition/app.ts index ac0fdf6..ed03b2e 100644 --- a/src/composition/app.ts +++ b/src/composition/app.ts @@ -50,9 +50,24 @@ import type { HealthReport } from './health.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). */ +/** `GET` (Vercel Cron) → daily Gmail `watch()` re-arm (runbook Part C: daily at 06:00 UTC). Only meaningful when push is configured; reports a skip otherwise (HT-94). The reconciliation sweep this endpoint used to also perform now lives at {@link RECONCILE_SWEEP_PATH}. */ export const WATCH_MAINTENANCE_PATH = '/api/v1/internal/cron/watch-maintenance' +/** + * `GET` (Vercel Cron) → one bounded reconciliation sweep (HT-94; + * `src/mail/gmail-reconcile-sweep.ts`): enqueue a reconcile job per active + * mailbox with a baseline cursor. + * + * **This is the primary inbound transport** (CHARTER.md §2 as amended + * 2026-07-20), not a backstop — a deployment with no Pub/Sub ingests mail + * entirely through this endpoint, so it runs at the same every-minute cadence + * as {@link QUEUE_DRAIN_PATH} rather than the daily cadence + * {@link WATCH_MAINTENANCE_PATH} uses for infrastructure upkeep. Split out of + * that endpoint precisely so the two cadences could diverge, and so the sweep + * would stop paying for a per-mailbox token refresh it never needs. + */ +export const RECONCILE_SWEEP_PATH = '/api/v1/internal/cron/reconcile-sweep' + /** `GET` (Vercel Cron) → drain one bounded batch of `event_outbox` into `queue_jobs` webhook-delivery fan-out (HT-69; `src/webhooks/outbox-drain.ts`; runbook Part C: every minute, same cadence as {@link QUEUE_DRAIN_PATH}). A SEPARATE endpoint from the queue drain — this one turns outbox rows into queue jobs; the queue drain is what then delivers them. */ export const OUTBOX_DRAIN_PATH = '/api/v1/internal/outbox/drain' @@ -80,8 +95,10 @@ export interface AppHandlerDeps { drainOutbox: () => Promise /** Run one snooze wake pass (HT-77, {@link SNOOZE_WAKE_PATH}); returns a JSON-serializable report for the response body + logs. */ runSnoozeWake: () => Promise - /** Run one daily watch-renewal + reconciliation-sweep pass; returns a JSON-serializable report. */ + /** Run one daily watch-renewal pass; returns a JSON-serializable report. */ runWatchMaintenance: () => Promise + /** Run one bounded reconciliation sweep (HT-94, {@link RECONCILE_SWEEP_PATH}) — the primary inbound transport; returns a JSON-serializable report. */ + runReconcileSweep: () => Promise /** Assemble the health report (`./health.ts`) — the {@link HEALTH_PATH} endpoint's work. */ runHealthCheck: () => Promise /** @@ -114,6 +131,9 @@ export function createAppHandler(deps: AppHandlerDeps): (request: Request) => Pr if (pathname === SNOOZE_WAKE_PATH) { return handleCronEndpoint(request, deps.cronSecret, 'snooze-wake', deps.runSnoozeWake) } + if (pathname === RECONCILE_SWEEP_PATH) { + return handleCronEndpoint(request, deps.cronSecret, 'reconcile-sweep', deps.runReconcileSweep) + } if (pathname === WATCH_MAINTENANCE_PATH) { return handleCronEndpoint( request, diff --git a/src/composition/config.test.ts b/src/composition/config.test.ts index 4995da7..4605ba7 100644 --- a/src/composition/config.test.ts +++ b/src/composition/config.test.ts @@ -29,7 +29,7 @@ describe('loadConfig — happy path', () => { 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.gmailPush?.topic).toBe('projects/resonantiq-helpthread/topics/gmail-push') expect(config.supportAddress).toBe('support@resonantiq.app') expect(config.mailDomain).toBe('mail.resonantiq.app') }) @@ -53,10 +53,20 @@ describe('loadConfig — missing / malformed values', () => { 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('treats a whitespace-only GMAIL_PUBSUB_TOPIC as missing — triggers the Gmail-push partial-config error (the other two push vars are still set)', () => { + // Since HT-94 made the push trio optional-but-all-or-nothing, this is no + // longer a plain "required var missing" case: validEnv() still has the + // other two push vars set, so a whitespace-only topic lands in + // resolveGmailPush's PARTIAL branch, not the "all three unset" happy path. + let message = '' + try { + loadConfig({ ...validEnv(), GMAIL_PUBSUB_TOPIC: ' ' }) + } catch (err) { + message = err instanceof Error ? err.message : String(err) + } + expect(message).toContain('GMAIL_PUBSUB_TOPIC') + expect(message).toContain('partially configured') + expect(message).toContain('is unset') }) it('aggregates ALL problems into one error, not just the first', () => { @@ -127,6 +137,75 @@ describe('loadConfig — missing / malformed values', () => { }) }) +describe('loadConfig — gmailPush / GMAIL_PUBSUB_* trio (HT-94, optional-but-all-or-nothing)', () => { + /** `validEnv()` with all three push vars removed — the push-free base every case here starts from. */ + function envWithoutPush(): Record { + const env = validEnv() + delete (env as Record).GMAIL_PUBSUB_TOPIC + delete (env as Record).GMAIL_PUBSUB_SUBSCRIPTION + delete (env as Record).GMAIL_PUSH_SERVICE_ACCOUNT + return env + } + + it('all three unset: config.gmailPush is undefined and loadConfig SUCCEEDS — the push-free happy path', () => { + const config = loadConfig(envWithoutPush()) + expect(config.gmailPush).toBeUndefined() + }) + + it('only GMAIL_PUBSUB_TOPIC set (two missing): throws naming exactly the two missing vars, plural "are unset"', () => { + const env = { ...envWithoutPush(), GMAIL_PUBSUB_TOPIC: 'projects/x/topics/y' } + let message = '' + try { + loadConfig(env) + } catch (err) { + message = err instanceof Error ? err.message : String(err) + } + expect(message).toContain('partially configured') + expect(message).toContain('GMAIL_PUBSUB_SUBSCRIPTION') + expect(message).toContain('GMAIL_PUSH_SERVICE_ACCOUNT') + expect(message).toContain('are unset') + }) + + it('two of three set (GMAIL_PUSH_SERVICE_ACCOUNT missing): throws naming just that one var, singular "is unset"', () => { + const env = { + ...envWithoutPush(), + GMAIL_PUBSUB_TOPIC: 'projects/x/topics/y', + GMAIL_PUBSUB_SUBSCRIPTION: 'projects/x/subscriptions/y', + } + let message = '' + try { + loadConfig(env) + } catch (err) { + message = err instanceof Error ? err.message : String(err) + } + expect(message).toContain('partially configured') + expect(message).toContain('GMAIL_PUSH_SERVICE_ACCOUNT') + expect(message).toContain('is unset') + expect(message).not.toContain('GMAIL_PUBSUB_TOPIC is unset') + expect(message).not.toContain('GMAIL_PUBSUB_SUBSCRIPTION is unset') + }) + + it('a whitespace-only GMAIL_PUBSUB_SUBSCRIPTION counts as missing, same as unset — throws naming it, others treated present', () => { + const env = { + ...envWithoutPush(), + GMAIL_PUBSUB_TOPIC: 'projects/x/topics/y', + GMAIL_PUBSUB_SUBSCRIPTION: ' ', + GMAIL_PUSH_SERVICE_ACCOUNT: 'invoker@x.iam.gserviceaccount.com', + } + let message = '' + try { + loadConfig(env) + } catch (err) { + message = err instanceof Error ? err.message : String(err) + } + expect(message).toContain('partially configured') + expect(message).toContain('GMAIL_PUBSUB_SUBSCRIPTION') + expect(message).toContain('is unset') + expect(message).not.toContain('GMAIL_PUBSUB_TOPIC is unset') + expect(message).not.toContain('GMAIL_PUSH_SERVICE_ACCOUNT is unset') + }) +}) + describe('loadConfig — HELPTHREAD_UI_BASE_URL (HT-54, optional)', () => { it('is absent from AppConfig when unset — no error, invite deps simply absent', () => { const config = loadConfig(validEnv()) diff --git a/src/composition/config.ts b/src/composition/config.ts index f6a3852..bfbde6e 100644 --- a/src/composition/config.ts +++ b/src/composition/config.ts @@ -66,12 +66,31 @@ export interface AppConfig { 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 + /** + * Gmail push configuration — OPTIONAL as of HT-94. + * + * Inbound mail reaches the engine either by push webhook or by the bounded + * scheduled fetch (CHARTER.md §2, amended 2026-07-20). Push is the + * lower-latency option; the scheduled sweep is the transport that always + * runs. An operator who has not stood up a Pub/Sub topic — which is the + * majority of the Google Cloud setup burden, and the half that fails + * silently — leaves all three vars unset and the engine runs on the sweep + * alone. + * + * All three travel as ONE object rather than three optional strings so a + * half-configured push is unrepresentable: you cannot arm `watch()` against + * a topic without also being able to authenticate the resulting push, and a + * config that permits that shape invites exactly the silent-failure mode + * this amendment set out to remove. + */ + gmailPush?: { + /** Cloud Pub/Sub topic `watch()` arms notifications to (`projects/{project}/topics/{topic}`). */ + topic: string + /** The exact push subscription the webhook accepts (`projects/{project}/subscriptions/{name}`). */ + subscription: string + /** The push subscription's OIDC service-account email (the JWT `email` claim the webhook matches). */ + serviceAccount: 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. */ @@ -153,9 +172,7 @@ export function loadConfig(env: NodeJS.ProcessEnv = process.env): AppConfig { 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 gmailPush = resolveGmailPush(env, errors) const apiToken = errors.requireMinLength(env, 'HELPTHREAD_API_TOKEN', MIN_API_TOKEN_LENGTH) const signingSecret = errors.requireMinLength( env, @@ -182,9 +199,7 @@ export function loadConfig(env: NodeJS.ProcessEnv = process.env): AppConfig { blobBucket: blobBucket as string, gmailOAuthClientId: gmailOAuthClientId as string, gmailOAuthClientSecret: gmailOAuthClientSecret as string, - gmailPubsubTopic: gmailPubsubTopic as string, - gmailPubsubSubscription: gmailPubsubSubscription as string, - gmailPushServiceAccount: gmailPushServiceAccount as string, + ...(gmailPush !== undefined ? { gmailPush } : {}), tokenEncryptionKey: tokenEncryptionKey as Buffer, apiToken: apiToken as string, signingSecret: signingSecret as string, @@ -284,6 +299,55 @@ function isLoopbackHost(hostname: string): boolean { ) } +/** + * Resolve the OPTIONAL Gmail push trio, all-or-nothing (HT-94). + * + * Three outcomes, and only three: + * - all three unset → `undefined`; the engine runs on the scheduled sweep + * alone, and nothing in the Google Cloud Pub/Sub setup is required. + * - all three set → the configured object; push is armed and the webhook + * authenticates against it, exactly as before this change. + * - some subset set → a config ERROR naming the missing vars. A partially + * configured push is never silently treated as "off": an operator who set a + * topic and forgot the service account has a broken push they believe works, + * which is the precise failure this amendment exists to eliminate. Failing + * at boot is the whole point of this module (see `loadConfig`'s aggregation). + */ +function resolveGmailPush( + env: NodeJS.ProcessEnv, + errors: ConfigErrors, +): { topic: string; subscription: string; serviceAccount: string } | undefined { + const vars = { + topic: 'GMAIL_PUBSUB_TOPIC', + subscription: 'GMAIL_PUBSUB_SUBSCRIPTION', + serviceAccount: 'GMAIL_PUSH_SERVICE_ACCOUNT', + } as const + + const present: Partial> = {} + const missing: string[] = [] + for (const [key, name] of Object.entries(vars) as [keyof typeof vars, string][]) { + const raw = env[name] + if (raw === undefined || raw.trim().length === 0) missing.push(name) + else present[key] = raw + } + + if (missing.length === Object.keys(vars).length) return undefined + if (missing.length > 0) { + errors.add( + `Gmail push is partially configured: ${missing.join(', ')} ${ + missing.length === 1 ? 'is' : 'are' + } unset. Set all of ${Object.values(vars).join(', ')} to enable push, or none of them to run on the scheduled fetch alone.`, + ) + return undefined + } + + return { + topic: present.topic as string, + subscription: present.subscription as string, + serviceAccount: present.serviceAccount as string, + } +} + function resolveUiBaseUrl(env: NodeJS.ProcessEnv, errors: ConfigErrors): string | undefined { const raw = env.HELPTHREAD_UI_BASE_URL if (raw === undefined || raw.trim().length === 0) return undefined diff --git a/src/composition/health.test.ts b/src/composition/health.test.ts index a534b0a..2743aa0 100644 --- a/src/composition/health.test.ts +++ b/src/composition/health.test.ts @@ -19,7 +19,9 @@ describe('runHealthCheck', () => { db = database await migrate(database) const queue = createPostgresQueue(database) - const check = () => runHealthCheck({ db: database, queue }) + // Default to push CONFIGURED so every pre-HT-94 case keeps asserting the + // behavior it was written for; the push-free cases pass `false` explicitly. + const check = (pushConfigured = true) => runHealthCheck({ db: database, queue, pushConfigured }) return { database, check } } @@ -233,7 +235,7 @@ describe('runHealthCheck', () => { }) }) - it('watch expiry: a healthy 7-day watch is silent; near-expiry, a NULL expiration, and a missing state row each trip watch-expiring', async () => { + it('watch expiry: a healthy 7-day watch is silent; near-expiry, a NULL expiration, and a missing state row each trip watch-expiring (push CONFIGURED)', async () => { const { database, check } = await fresh() await seedMailbox(database, 'healthy@example.test', 'active', { expiration: new Date(Date.now() + 7 * 24 * 3600 * 1000), @@ -244,7 +246,10 @@ describe('runHealthCheck', () => { await seedMailbox(database, 'never-armed@example.test', 'active', { expiration: null }) await seedMailbox(database, 'no-state-row@example.test', 'active') - const report = await check() + // Explicit `true` (not the default) — this is the case HT-94's + // `pushConfigured` gate exists to still catch: watch-expiring alerts fire + // when push IS configured, never suppressed by the gate. + const report = await check(true) expect(report.ok).toBe(false) expect(report.alerts).toHaveLength(3) @@ -260,6 +265,40 @@ describe('runHealthCheck', () => { expect(byAddress.get('no-state-row@example.test')?.watchExpiresAt).toBeNull() }) + it('watch expiry is SILENT when push is NOT configured (HT-94) — an active mailbox with no watch_expiration is the designed steady state, not a fault', async () => { + const { database, check } = await fresh() + // Same shape as the "never-armed"/"no-state-row" cases above, which trip + // watch-expiring when push IS configured — this is the regression guard + // for the finding that the recommended push-free install path returned + // 503 permanently. + await seedMailbox(database, 'never-armed@example.test', 'active', { expiration: null }) + await seedMailbox(database, 'no-state-row@example.test', 'active') + + const report = await check(false) + + expect(report.ok).toBe(true) + expect(report.alerts).toEqual([]) + expect(report.alerts.some((a) => a.startsWith('watch-expiring: '))).toBe(false) + }) + + it('pushConfigured: false does NOT suppress mailbox-needs-attention — the gate is scoped to watch alerts only', async () => { + const { database, check } = await fresh() + await seedMailbox(database, 'paused@example.test', 'paused') + await seedMailbox(database, 'reconnect@example.test', 'needs_reconnect') + + const report = await check(false) + + expect(report.ok).toBe(false) + const attention = report.alerts.filter((a) => a.startsWith('mailbox-needs-attention: ')) + expect(attention).toHaveLength(2) + expect(attention.join('\n')).toContain('paused@example.test') + expect(attention.join('\n')).toContain('reconnect@example.test') + // No watch-expiring alerts leak in either — these mailboxes aren't even + // `active`, so the watch-expiring branch is unreachable regardless of + // the gate, but assert it explicitly since this is the push-free path. + expect(report.alerts).toHaveLength(2) + }) + it("mailbox statuses: paused and needs_reconnect trip mailbox-needs-attention; disconnected is silent (an operator's own action)", async () => { const { database, check } = await fresh() await seedMailbox(database, 'paused@example.test', 'paused') diff --git a/src/composition/health.ts b/src/composition/health.ts index b1da55d..0d69527 100644 --- a/src/composition/health.ts +++ b/src/composition/health.ts @@ -102,6 +102,21 @@ export const WATCH_EXPIRY_ALERT_HOURS = 72 export interface HealthCheckDeps { db: Db queue: { getStats(): Promise } + /** + * Whether Gmail push is configured for this deployment (HT-94) — i.e. + * whether `AppConfig.gmailPush` is present. + * + * Gates the `watch-expiring` alerts ONLY. With push unconfigured there is no + * `watch()` to arm, so `watch_expiration` is NULL by design for every active + * mailbox — alerting on it would return 503 permanently on the install path + * the runbook now recommends, and since this endpoint's contract is a single + * boolean, a permanent false alarm makes every REAL alert invisible. + * + * Deliberately a config fact rather than inferred from the data: a NULL + * expiration means "no push configured" on one deployment and "renewal cron + * is broken" on another, and only the config can tell those apart. + */ + pushConfigured: boolean } /** One mailbox's health row — see the module doc's Mailboxes section. */ @@ -263,7 +278,12 @@ export async function runHealthCheck(deps: HealthCheckDeps): Promise), } - // --- Watch-maintenance deps (daily re-arm + sweep). --- - const watchMaintenanceDeps: GmailWatchMaintenanceDeps = { - tokenService, + // --- Reconciliation-sweep deps (HT-94). Deliberately NO tokenService and no + // watch client: the sweep reads a cursor and enqueues, making no Gmail call + // of its own, which is what makes every-minute cadence affordable. --- + const reconcileSweepDeps: GmailReconcileSweepDeps = { mailboxStore, watchStateStore, queue, - createWatchClient: (getAccessToken) => createGmailWatchClient({ getAccessToken }), - topicName: config.gmailPubsubTopic, } + // --- Watch-maintenance deps (daily re-arm). Only meaningful when push is + // configured — with no topic there is no watch to re-arm. The reconciliation + // sweep is NOT part of this any more (HT-94): it runs on its own every-minute + // cron as the primary intake, independent of whether push exists. --- + const watchMaintenanceDeps: GmailWatchMaintenanceDeps | undefined = + config.gmailPush === undefined + ? undefined + : { + tokenService, + mailboxStore, + watchStateStore, + createWatchClient: (getAccessToken) => createGmailWatchClient({ getAccessToken }), + topicName: config.gmailPush.topic, + } + return createAppHandler({ inboxApi, cronSecret: config.cronSecret, @@ -438,7 +466,32 @@ export async function buildApp( } return report }, - runWatchMaintenance: () => runGmailWatchMaintenance(watchMaintenanceDeps), + // The primary inbound transport (HT-94) — runs regardless of whether push + // is configured, since push only makes the SAME reconcile job run sooner. + // Quiet ticks are logged unlike the drains': a sweep that stops sweeping is + // an intake outage, and its every-minute silence is the only signal. + runReconcileSweep: async () => { + const report = await runGmailReconcileSweep(reconcileSweepDeps) + console.info(JSON.stringify({ event: 'reconcile_sweep', ...report })) + return report + }, + // With push unconfigured there is no watch() to re-arm, so this cron has + // nothing to do. It stays ROUTED rather than 404-ing (HT-94): `vercel.json` + // is static, so a deployment without push would otherwise log a daily + // not-found that reads like a fault. Reporting a skip is the honest, + // greppable alternative — and it must never be silent, since a genuinely + // broken maintenance cron is the failure mode the runbook's external + // monitor exists to catch. + runWatchMaintenance: async () => { + if (watchMaintenanceDeps === undefined) { + const report = { skipped: 'push-not-configured' as const } + // Same event name the module itself logs under — one endpoint must not + // produce two event names, or a log filter finds half its own history. + console.info(JSON.stringify({ event: 'gmail_watch_maintenance', ...report })) + return report + } + return runGmailWatchMaintenance(watchMaintenanceDeps) + }, // Snooze wake pass (HT-77) — a SEPARATE cron tick from the two drains // above: flips due `pending`+snoozed conversations back to `active` // (`runSnoozeWake`, `src/mail/snooze-wake.ts`) via the SAME @@ -453,7 +506,8 @@ export async function buildApp( } return report }, - runHealthCheck: () => runHealthCheck({ db, queue }), + runHealthCheck: () => + runHealthCheck({ db, queue, pushConfigured: config.gmailPush !== undefined }), }) } diff --git a/src/mail/gmail-connect.ts b/src/mail/gmail-connect.ts index 94f2de9..f6f88dd 100644 --- a/src/mail/gmail-connect.ts +++ b/src/mail/gmail-connect.ts @@ -447,8 +447,16 @@ export interface GmailConnectServiceDeps { clientSecret: string /** Must exactly match `/api/v1/inbound/gmail/callback` on the deployment's public origin AND a redirect URI registered on the OAuth client (gmail-connect.md §3). */ redirectUri: string - /** The Cloud Pub/Sub topic `watch()` arms notifications to (`projects/{project}/topics/{topic}`, HT-43-provisioned) — injected config. */ - topicName: string + /** + * The Cloud Pub/Sub topic `watch()` arms notifications to + * (`projects/{project}/topics/{topic}`, HT-43-provisioned) — injected config. + * + * OPTIONAL as of HT-94. When absent, push is not configured for this + * deployment: connect skips the `watch()` arm entirely and seeds the + * baseline cursor from `getProfile()` instead, leaving the bounded + * scheduled fetch as the sole inbound transport. + */ + topicName?: string /** OAuth scopes requested on the consent screen (gmail-connect.md §3: `gmail.readonly` + `gmail.send` for the dogfood). */ scopes: string[] /** Signs/verifies the `state` CSRF token (module doc). */ @@ -524,7 +532,10 @@ export function createGmailConnectService(deps: GmailConnectServiceDeps): GmailC assertNonEmpty('clientId', clientId) assertNonEmpty('clientSecret', clientSecret) assertNonEmpty('redirectUri', redirectUri) - assertNonEmpty('topicName', topicName) + // Optional (HT-94): absent means push is not configured for this deployment. + // Present-but-blank is still a misconfiguration and still rejected — the + // all-or-nothing shape is enforced at the composition root (`resolveGmailPush`). + if (topicName !== undefined) assertNonEmpty('topicName', topicName) if (!Array.isArray(scopes) || scopes.length === 0) { throw new Error('createGmailConnectService: scopes must be a non-empty array') } @@ -570,15 +581,32 @@ export function createGmailConnectService(deps: GmailConnectServiceDeps): GmailC const watchClient = createWatchClient(() => Promise.resolve(exchanged.accessToken)) const profile = await watchClient.getProfile() - // --- Step 4: arm watch() BEFORE any persistence (module doc). --- - let armed: GmailWatchResult - try { - armed = await watchClient.watch({ topicName }) - } catch (err) { - throw new GmailConnectError( - 'watch_failed', - `Enabling Gmail push failed: ${errorMessage(err)}`, - ) + // --- Step 4: arm watch() BEFORE any persistence (module doc). + // + // SKIPPED ENTIRELY when push is not configured (HT-94, CHARTER.md §2 as + // amended 2026-07-20): there is no topic to arm against, and the bounded + // scheduled fetch is the transport. The baseline then comes from the + // `getProfile()` call step 3 ALREADY made. + // + // That substitution is safe for exactly the reason the module doc gives + // for rejecting it in the push case: getProfile's separately-read + // historyId "could straddle the arm." With no arm, there is nothing to + // straddle — one read, one baseline, and the sweep resumes from it. No + // extra API call is made either; step 3's response carries the value. --- + let baseline: { historyId: string; watchExpiration?: Date } + if (topicName === undefined) { + baseline = { historyId: profile.historyId } + } else { + let armed: GmailWatchResult + try { + armed = await watchClient.watch({ topicName }) + } catch (err) { + throw new GmailConnectError( + 'watch_failed', + `Enabling Gmail push failed: ${errorMessage(err)}`, + ) + } + baseline = { historyId: armed.historyId, watchExpiration: armed.expiration } } // --- Step 5: persist, now that the grant is proven usable — ONE atomic @@ -599,11 +627,7 @@ export function createGmailConnectService(deps: GmailConnectServiceDeps): GmailC }, tx, ) - await watchStateStore.seedBaseline( - created.id, - { historyId: armed.historyId, watchExpiration: armed.expiration }, - tx, - ) + await watchStateStore.seedBaseline(created.id, baseline, tx) return created }) diff --git a/src/mail/gmail-reconcile-sweep.test.ts b/src/mail/gmail-reconcile-sweep.test.ts new file mode 100644 index 0000000..0d096a2 --- /dev/null +++ b/src/mail/gmail-reconcile-sweep.test.ts @@ -0,0 +1,296 @@ +/** + * `runGmailReconcileSweep` against REAL PGlite-backed `MailboxStore`/ + * `GmailWatchStateStore` (so `listActiveMailboxes`/`getCursor` are genuinely + * exercised, not just mocked) plus a fake `QueueProvider`. Split out of + * `./gmail-watch-maintenance.test.ts` (HT-94) along with the sweep itself — + * see `./gmail-reconcile-sweep.ts`'s module doc for why. Exercises: the + * happy-path enqueue, the no-baseline-cursor skip, per-mailbox failure + * isolation, and propagation of a fault outside the per-mailbox loop with a + * fake queue; the bare-`mailboxId`-dedupe-key liveness contract (the most + * important behavior in this file — see the module doc's "The dedupe key is + * the bare mailboxId" section) against the REAL `createPostgresQueue`, since + * only the real adapter's partial unique index can actually prove it. + */ + +import { afterEach, describe, expect, it, vi } from 'vitest' +import { GMAIL_RECONCILE_TOPIC, type GmailReconcileJob } from '../api/gmail-webhook.js' +import { createPgliteDb, type Db } from '../db/client.js' +import { migrate } from '../db/migrate.js' +import { createPostgresQueue } from '../providers/adapters/postgres-queue/index.js' +import type { EnqueueOptions, QueueProvider } from '../providers/queue.js' +import { + createGmailWatchStateStore, + type GmailWatchStateStore, +} from '../store/gmail-watch-state.js' +import { createMailboxStore, type MailboxStore } from '../store/mailboxes.js' +import { type GmailReconcileSweepDeps, runGmailReconcileSweep } from './gmail-reconcile-sweep.js' + +type EnqueuedCall = { topic: string; payload: GmailReconcileJob; opts: EnqueueOptions | undefined } + +/** Records every `enqueue` call — the assertion surface for "swept" behavior and the no-dedupeKey rule (module doc). Optionally fails enqueue for specific mailboxIds, for the failure-isolation tests. */ +function fakeQueue(failingMailboxIds: string[] = []): { + queue: QueueProvider + enqueued: EnqueuedCall[] +} { + const enqueued: EnqueuedCall[] = [] + return { + queue: { + async enqueue(topic, payload, opts) { + const job = payload as GmailReconcileJob + if (failingMailboxIds.includes(job.mailboxId)) { + throw new Error('fake enqueue: queue write failed') + } + enqueued.push({ topic, payload: job, opts }) + }, + }, + enqueued, + } +} + +describe('runGmailReconcileSweep', () => { + let db: Db | undefined + + afterEach(async () => { + await db?.close() + db = undefined + }) + + async function freshStores(): Promise<{ + db: Db + mailboxStore: MailboxStore + watchStateStore: GmailWatchStateStore + }> { + db = await createPgliteDb() + await migrate(db) + return { + db, + mailboxStore: createMailboxStore(db), + watchStateStore: createGmailWatchStateStore(db), + } + } + + async function seedActiveMailboxWithCursor( + mailboxStore: MailboxStore, + watchStateStore: GmailWatchStateStore, + address: string, + cursor: string, + ): Promise { + const mailbox = await mailboxStore.upsertConnectedMailbox({ address, provider: 'gmail' }) + await watchStateStore.seedBaseline(mailbox.id, { + historyId: cursor, + watchExpiration: new Date('2026-01-01T00:00:00.000Z'), + }) + return mailbox.id + } + + function buildDeps( + mailboxStore: MailboxStore, + watchStateStore: GmailWatchStateStore, + queue: QueueProvider, + ): GmailReconcileSweepDeps { + return { mailboxStore, watchStateStore, queue } + } + + /** Count `queue_jobs` rows, optionally restricted to the live set (not dead-lettered) — the partial unique index's own predicate. */ + async function countQueueRows(db: Db, opts: { onlyLive?: boolean } = {}): Promise { + const where = opts.onlyLive ? 'WHERE dead_lettered_at IS NULL' : '' + const rows = await db.query<{ count: number }>( + `SELECT count(*)::int AS count FROM queue_jobs ${where}`, + ) + return rows[0].count + } + + it('two active mailboxes with cursors are both enqueued', async () => { + const { mailboxStore, watchStateStore } = await freshStores() + const mailboxA = await seedActiveMailboxWithCursor( + mailboxStore, + watchStateStore, + 'a@example.test', + 'cursor-a', + ) + const mailboxB = await seedActiveMailboxWithCursor( + mailboxStore, + watchStateStore, + 'b@example.test', + 'cursor-b', + ) + const { queue, enqueued } = fakeQueue() + + const report = await runGmailReconcileSweep(buildDeps(mailboxStore, watchStateStore, queue)) + + expect(report).toEqual({ total: 2, swept: 2, skipped: 0, failed: 0 }) + expect(enqueued).toHaveLength(2) + const byMailbox = new Map(enqueued.map((e) => [e.payload.mailboxId, e])) + // The enqueued job shape is { mailboxId, historyId: } on GMAIL_RECONCILE_TOPIC. + expect(byMailbox.get(mailboxA)).toMatchObject({ + topic: GMAIL_RECONCILE_TOPIC, + payload: { mailboxId: mailboxA, historyId: 'cursor-a' }, + }) + expect(byMailbox.get(mailboxB)).toMatchObject({ + topic: GMAIL_RECONCILE_TOPIC, + payload: { mailboxId: mailboxB, historyId: 'cursor-b' }, + }) + }) + + it('a mailbox with no baseline cursor (getCursor returns null) is skipped, not enqueued', async () => { + const { mailboxStore, watchStateStore } = await freshStores() + const mailbox = await mailboxStore.upsertConnectedMailbox({ + address: 'no-cursor@example.test', + provider: 'gmail', + }) + // No seedBaseline call — this mailbox has no gmail_watch_state row at all. + const { queue, enqueued } = fakeQueue() + + const report = await runGmailReconcileSweep(buildDeps(mailboxStore, watchStateStore, queue)) + + expect(report).toEqual({ total: 1, swept: 0, skipped: 1, failed: 0 }) + expect(enqueued).toHaveLength(0) + expect(await watchStateStore.getCursor(mailbox.id)).toBeNull() + }) + + it('a second sweep tick does NOT enqueue a second row while the mailbox job is still pending (live) — the bare-mailboxId dedupe key, against the REAL queue', async () => { + // The most important test in this file (module doc's "The dedupe key is + // the bare mailboxId" section): only the real adapter's partial unique + // index (`WHERE dedupe_key IS NOT NULL AND dead_lettered_at IS NULL`) can + // prove this — a fake queue would be tautological here and would not + // catch a regression to a composite `mailboxId:historyId` key. + const { db, mailboxStore, watchStateStore } = await freshStores() + const mailboxId = await seedActiveMailboxWithCursor( + mailboxStore, + watchStateStore, + 'live@example.test', + 'cursor-live', + ) + const queue = createPostgresQueue(db) + const deps = buildDeps(mailboxStore, watchStateStore, queue) + + await runGmailReconcileSweep(deps) + await runGmailReconcileSweep(deps) + + expect(await countQueueRows(db)).toBe(1) + const rows = await db.query<{ dedupe_key: string | null }>('SELECT dedupe_key FROM queue_jobs') + expect(rows[0].dedupe_key).toBe(mailboxId) + }) + + it('once the pending job is ACKED, the next sweep tick enqueues again — a quiet mailbox is not permanently suppressed', async () => { + const { db, mailboxStore, watchStateStore } = await freshStores() + await seedActiveMailboxWithCursor( + mailboxStore, + watchStateStore, + 'acked@example.test', + 'cursor-acked', + ) + const queue = createPostgresQueue(db) + const deps = buildDeps(mailboxStore, watchStateStore, queue) + + await runGmailReconcileSweep(deps) + expect(await countQueueRows(db)).toBe(1) + + const drainReport = await queue.drainOnce({ + handlers: { [GMAIL_RECONCILE_TOPIC]: async () => ({ kind: 'ack' }) }, + }) + expect(drainReport.acked).toBe(1) + expect(await countQueueRows(db)).toBe(0) + + await runGmailReconcileSweep(deps) + expect(await countQueueRows(db)).toBe(1) + }) + + it('once the pending job is DEAD-LETTERED, the next sweep tick enqueues again — a quiet mailbox is not permanently suppressed', async () => { + const { db, mailboxStore, watchStateStore } = await freshStores() + await seedActiveMailboxWithCursor( + mailboxStore, + watchStateStore, + 'dead-lettered@example.test', + 'cursor-dead-lettered', + ) + const queue = createPostgresQueue(db) + const deps = buildDeps(mailboxStore, watchStateStore, queue) + + await runGmailReconcileSweep(deps) + expect(await countQueueRows(db)).toBe(1) + + const drainReport = await queue.drainOnce({ + handlers: { + [GMAIL_RECONCILE_TOPIC]: async () => ({ kind: 'deadLetter', reason: 'test dead-letter' }), + }, + }) + expect(drainReport.deadLettered).toBe(1) + // Dead-lettered row is retained (never deleted — invariant #1), but it is + // no longer "live", so the unique index's predicate no longer covers it. + expect(await countQueueRows(db)).toBe(1) + expect(await countQueueRows(db, { onlyLive: true })).toBe(0) + + await runGmailReconcileSweep(deps) + // A new live row is inserted alongside the retained dead-lettered one — + // the dead-lettered row is excluded from the partial unique index, so it + // never conflicts with (and never suppresses) the fresh enqueue. + expect(await countQueueRows(db)).toBe(2) + expect(await countQueueRows(db, { onlyLive: true })).toBe(1) + }) + + it('a mailbox whose getCursor throws is counted failed — other mailboxes still enqueue', async () => { + const { mailboxStore, watchStateStore } = await freshStores() + const broken = await seedActiveMailboxWithCursor( + mailboxStore, + watchStateStore, + 'broken@example.test', + 'cursor-broken', + ) + const healthy = await seedActiveMailboxWithCursor( + mailboxStore, + watchStateStore, + 'healthy@example.test', + 'cursor-healthy', + ) + const originalGetCursor = watchStateStore.getCursor.bind(watchStateStore) + vi.spyOn(watchStateStore, 'getCursor').mockImplementation(async (mailboxId) => { + if (mailboxId === broken) { + throw new Error('fake getCursor: store read failed') + } + return originalGetCursor(mailboxId) + }) + const { queue, enqueued } = fakeQueue() + + const report = await runGmailReconcileSweep(buildDeps(mailboxStore, watchStateStore, queue)) + + expect(report).toEqual({ total: 2, swept: 1, skipped: 0, failed: 1 }) + expect(enqueued).toHaveLength(1) + expect(enqueued[0].payload.mailboxId).toBe(healthy) + }) + + it('a mailbox whose enqueue throws is counted failed — other mailboxes still enqueue', async () => { + const { mailboxStore, watchStateStore } = await freshStores() + const broken = await seedActiveMailboxWithCursor( + mailboxStore, + watchStateStore, + 'broken2@example.test', + 'cursor-broken2', + ) + const healthy = await seedActiveMailboxWithCursor( + mailboxStore, + watchStateStore, + 'healthy2@example.test', + 'cursor-healthy2', + ) + const { queue, enqueued } = fakeQueue([broken]) + + const report = await runGmailReconcileSweep(buildDeps(mailboxStore, watchStateStore, queue)) + + expect(report).toEqual({ total: 2, swept: 1, skipped: 0, failed: 1 }) + expect(enqueued).toHaveLength(1) + expect(enqueued[0].payload.mailboxId).toBe(healthy) + }) + + it('a fault outside the per-mailbox loop (listActiveMailboxes itself throwing) propagates rather than being swallowed', async () => { + const { mailboxStore, watchStateStore } = await freshStores() + vi.spyOn(mailboxStore, 'listActiveMailboxes').mockRejectedValue( + new Error('fake listActiveMailboxes: store read failed'), + ) + const { queue } = fakeQueue() + + await expect( + runGmailReconcileSweep(buildDeps(mailboxStore, watchStateStore, queue)), + ).rejects.toThrow('fake listActiveMailboxes: store read failed') + }) +}) diff --git a/src/mail/gmail-reconcile-sweep.ts b/src/mail/gmail-reconcile-sweep.ts new file mode 100644 index 0000000..a0e0ddf --- /dev/null +++ b/src/mail/gmail-reconcile-sweep.ts @@ -0,0 +1,182 @@ +/** + * `runGmailReconcileSweep` — the bounded scheduled fetch that is Helpthread's + * PRIMARY inbound transport (HT-94; CHARTER.md §2 as amended 2026-07-20: + * "push-based delivery where providers offer it, bounded scheduled fetches + * where they don't — and no resident process either way"). + * + * One pass enqueues a reconcile job per active mailbox that has a baseline + * cursor. The job is the SAME `GMAIL_RECONCILE_TOPIC` job the push webhook + * enqueues (`../api/gmail-webhook.ts`) and the same one `./gmail-reconcile.ts` + * consumes — this module changes *what triggers* reconciliation, never how + * reconciliation works. That equivalence is the whole point: a deployment + * without Pub/Sub ingests mail through exactly the code path a deployment with + * it does, just triggered by a clock instead of a notification. + * + * ## Split out of `./gmail-watch-maintenance.ts` (HT-94) + * + * This logic previously lived as "step 3" inside that module's daily + * per-mailbox pass, where it was framed as a *backstop* for push. Two reasons + * it had to become its own entry point rather than a flag on that one: + * + * 1. **Cadence.** As a backstop it ran daily. As the primary transport it runs + * every minute — a mailbox whose only intake is a daily sweep is not a + * helpdesk. Those cadences cannot share a cron. + * 2. **Cost, and this is the load-bearing one.** Watch renewal must acquire an + * access token per mailbox (it calls `users.watch()`); the sweep must not. + * All the sweep needs is a stored cursor and a queue write — no Gmail API + * call happens here at all. Keeping them welded together would have meant a + * token refresh per mailbox per MINUTE against Google's token endpoint, for + * a call the sweep never makes. The reconcile CONSUMER acquires its own + * token when it actually talks to Gmail. + * + * What remains in `./gmail-watch-maintenance.ts` is renewal alone, still + * daily, and now only scheduled when push is configured. + * + * ## The dedupe key is the bare `mailboxId` — corrected after review + * + * The inherited behavior was NO `dedupeKey`, on the reasoning that "a sweep of + * an already-current, quiet mailbox must still run rather than be suppressed + * as a duplicate." That reasoning is sound, and it argues against a COMPOSITE + * key like `mailboxId:historyId` — which would pin suppression to a cursor + * value and could wedge a quiet mailbox indefinitely. It does not argue + * against the bare `mailboxId`, because the queue's partial unique index only + * suppresses against jobs that are still LIVE + * (`../providers/adapters/postgres-queue/`: `WHERE dedupe_key IS NOT NULL AND + * dead_lettered_at IS NULL`). Once a mailbox's job completes, the next tick + * enqueues again. A quiet mailbox is still swept every minute. + * + * Carrying "no dedupeKey" from a DAILY cadence to an every-minute one was the + * actual mistake, and it was not benign: + * + * - **The consumer lease does not make contention free.** A failed claim + * returns `{ kind: 'retry' }` (`./gmail-reconcile.ts`), and the queue counts + * attempts and DEAD-LETTERS at the cap. A reconcile that runs longer than + * the retry window (a large history batch, or one multi-MB raw message + * through blob write + ingest) causes every tick behind it to burn its + * attempts and dead-letter — which then trips the `queue-dead-letter-growth` + * health alert. The lease prevents duplicated *work*; it does nothing about + * duplicated *rows*. + * - **There was no backpressure whatsoever.** Enqueue rate was one job per + * active mailbox per minute, unconditional; drain capacity is a bounded + * batch per tick, shared with webhook delivery. Past roughly that many + * mailboxes, `queue_jobs` grew monotonically and intake latency grew without + * bound. Keying on `mailboxId` collapses the redundant pending ticks that + * caused it. + * + * Note this also aligns the sweep with the push path, which has always + * enqueued with a dedupe key (`../api/gmail-webhook.ts`). + * + * ## Failure isolation + * + * Per-mailbox failures never stop the batch — one mailbox with an unreadable + * cursor must not stall intake for every other mailbox. A fault outside the + * per-mailbox loop (e.g. `listActiveMailboxes` itself failing) propagates, + * matching `./gmail-watch-maintenance.ts`'s discipline. + */ + +import { GMAIL_RECONCILE_TOPIC, type GmailReconcileJob } from '../api/gmail-webhook.js' +import type { QueueProvider } from '../providers/queue.js' +import type { GmailWatchStateStore } from '../store/gmail-watch-state.js' +import type { MailboxStore } from '../store/mailboxes.js' + +export interface GmailReconcileSweepDeps { + /** The per-mailbox source (`listActiveMailboxes`, `../store/mailboxes.ts`). */ + mailboxStore: MailboxStore + + /** The stored-cursor read (`../store/gmail-watch-state.ts`). */ + watchStateStore: GmailWatchStateStore + + /** Where each mailbox's reconcile job is enqueued — the SAME `GMAIL_RECONCILE_TOPIC` the push webhook enqueues onto. */ + queue: QueueProvider +} + +/** What one {@link runGmailReconcileSweep} pass did, for platform-log observability. */ +export interface GmailReconcileSweepReport { + /** Active mailboxes considered this pass. */ + total: number + /** + * Mailboxes an enqueue was ISSUED for (i.e. that had a baseline cursor). + * + * Not necessarily rows created: `QueueProvider.enqueue` returns `void`, so a + * dedupe-suppressed enqueue (a job for this mailbox already pending) is + * indistinguishable here from one that inserted. On a busy mailbox this + * counter therefore reads 1 whether or not the tick did anything — the + * queue's own depth metrics are the place to see that difference. + */ + swept: number + /** Mailboxes skipped for having no baseline cursor yet — connect seeds it, so this means a mailbox that never completed connect. */ + skipped: number + /** Mailboxes whose enqueue or cursor read threw this pass — retried on the next tick, a minute later. */ + failed: number +} + +/** + * Emit one structured, JSON-parseable log line — mirrors + * `./gmail-watch-maintenance.ts`'s `logMaintenanceEvent` and + * `./gmail-reconcile.ts`'s `logReconcileEvent`. Plain `console.*` of a + * JSON-serializable object is this codebase's logging convention (CHARTER.md + * §4: serverless, platform-log-aggregated). Never pass a raw caught error + * object; only its message, and only where that message is known token-free. + */ +function logSweepEvent(level: 'info' | 'warn' | 'error', record: Record): void { + const line = JSON.stringify({ event: 'gmail_reconcile_sweep', ...record }) + if (level === 'error') console.error(line) + else if (level === 'warn') console.warn(line) + else console.info(line) +} + +/** + * Run one bounded reconciliation sweep across every active mailbox. Never + * throws for an individual mailbox (see the module doc's failure isolation); + * a fault outside the per-mailbox loop propagates to the caller. + */ +export async function runGmailReconcileSweep( + deps: GmailReconcileSweepDeps, +): Promise { + const { mailboxStore, watchStateStore, queue } = deps + + const mailboxes = await mailboxStore.listActiveMailboxes() + const report: GmailReconcileSweepReport = { + total: mailboxes.length, + swept: 0, + skipped: 0, + failed: 0, + } + + for (const mailbox of mailboxes) { + try { + const cursor = await watchStateStore.getCursor(mailbox.id) + if (cursor === null) { + // No baseline yet. Connect (HT-40) seeds this, so reaching here means + // a mailbox row exists without a completed connect — worth a line, + // but not an error, and never a reason to stall the rest of the batch. + report.skipped++ + logSweepEvent('info', { + mailboxId: mailbox.id, + outcome: 'skipped', + reason: 'no-baseline-cursor', + }) + continue + } + + const job: GmailReconcileJob = { mailboxId: mailbox.id, historyId: cursor } + // Bare mailboxId, NOT `mailboxId:historyId` — see the module doc. This + // collapses a redundant tick against a still-pending job for the same + // mailbox, and stops suppressing as soon as that job leaves the live set. + await queue.enqueue(GMAIL_RECONCILE_TOPIC, job, { dedupeKey: mailbox.id }) + report.swept++ + } catch (err) { + // Safe to log the message: everything reachable here is a plain + // store/queue error. No access token is in scope in this module at all + // — the sweep makes no Gmail API call (module doc). + report.failed++ + logSweepEvent('error', { + mailboxId: mailbox.id, + outcome: 'failed', + error: err instanceof Error ? err.message : String(err), + }) + } + } + + return report +} diff --git a/src/mail/gmail-watch-maintenance.test.ts b/src/mail/gmail-watch-maintenance.test.ts index 31cd6df..5635ccc 100644 --- a/src/mail/gmail-watch-maintenance.test.ts +++ b/src/mail/gmail-watch-maintenance.test.ts @@ -3,19 +3,20 @@ * `GmailWatchStateStore` (so the SQL behind HT-42's new * `listActiveMailboxes`/`setWatchExpiration` methods is genuinely * exercised, not just mocked) plus fakes for the Gmail-API-facing seams - * (`createWatchClient`, `GmailOAuthTokenService`) and the `QueueProvider`. + * (`createWatchClient`, `GmailOAuthTokenService`). No queue seam any more — + * renewal never enqueues; the sweep that did moved out in HT-94. * Exercises the orchestration control flow documented in - * `gmail-watch-maintenance.ts`'s module doc: the renewal/sweep - * independence, the token-failure branches, the no-dedupeKey sweep, and - * failure-isolation per mailbox. + * `gmail-watch-maintenance.ts`'s module doc: renewal and the + * token-failure branches, and failure-isolation per mailbox. The + * reconciliation sweep itself moved to `./gmail-reconcile-sweep.ts` (HT-94) + * along with its own coverage (`./gmail-reconcile-sweep.test.ts`) — this + * file tests renewal only. */ import { afterEach, describe, expect, it, vi } from 'vitest' -import { GMAIL_RECONCILE_TOPIC, type GmailReconcileJob } from '../api/gmail-webhook.js' import { createPgliteDb, type Db } from '../db/client.js' import { migrate } from '../db/migrate.js' import type { GmailWatchClient } from '../providers/adapters/gmail/index.js' -import type { EnqueueOptions, QueueProvider } from '../providers/queue.js' import { createGmailWatchStateStore, type GmailWatchStateStore, @@ -30,21 +31,6 @@ import { const TOPIC_NAME = 'projects/helpthread-test/topics/gmail-push' const DEFAULT_RENEWAL_EXPIRATION = new Date('2026-02-01T00:00:00.000Z') -type EnqueuedCall = { topic: string; payload: GmailReconcileJob; opts: EnqueueOptions | undefined } - -/** Records every `enqueue` call — the assertion surface for "swept" behavior and the no-dedupeKey rule (module doc). */ -function fakeQueue(): { queue: QueueProvider; enqueued: EnqueuedCall[] } { - const enqueued: EnqueuedCall[] = [] - return { - queue: { - async enqueue(topic, payload, opts) { - enqueued.push({ topic, payload: payload as GmailReconcileJob, opts }) - }, - }, - enqueued, - } -} - /** * A fake `GmailOAuthTokenService` whose behavior is keyed by mailboxId. * `'needs_reconnect'` mirrors gmail-oauth.ts's real `invalid_grant` @@ -145,21 +131,19 @@ describe('runGmailWatchMaintenance', () => { function buildDeps( mailboxStore: MailboxStore, watchStateStore: GmailWatchStateStore, - queue: QueueProvider, overrides: Partial = {}, ): GmailWatchMaintenanceDeps { return { tokenService: fakeTokenService(mailboxStore, {}), mailboxStore, watchStateStore, - queue, createWatchClient: fakeCreateWatchClient([]), topicName: TOPIC_NAME, ...overrides, } } - it('happy path: two active mailboxes with cursors are both renewed and both swept', async () => { + it('happy path: two active mailboxes with cursors are both renewed', async () => { const { db: rawDb, mailboxStore, watchStateStore } = await freshStores() const mailboxA = await seedActiveMailboxWithCursor( mailboxStore, @@ -173,11 +157,10 @@ describe('runGmailWatchMaintenance', () => { 'b@example.test', 'cursor-b', ) - const { queue, enqueued } = fakeQueue() - const report = await runGmailWatchMaintenance(buildDeps(mailboxStore, watchStateStore, queue)) + const report = await runGmailWatchMaintenance(buildDeps(mailboxStore, watchStateStore)) - expect(report).toEqual({ total: 2, renewed: 2, swept: 2, needsReconnect: 0, failed: 0 }) + expect(report).toEqual({ total: 2, renewed: 2, needsReconnect: 0, failed: 0 }) expect((await readWatchExpiration(rawDb, mailboxA))?.toISOString()).toBe( DEFAULT_RENEWAL_EXPIRATION.toISOString(), ) @@ -187,17 +170,6 @@ describe('runGmailWatchMaintenance', () => { // setWatchExpiration must never touch history_id — the sacred cursor rule. expect(await watchStateStore.getCursor(mailboxA)).toBe('cursor-a') expect(await watchStateStore.getCursor(mailboxB)).toBe('cursor-b') - - expect(enqueued).toHaveLength(2) - const byMailbox = new Map(enqueued.map((e) => [e.payload.mailboxId, e])) - expect(byMailbox.get(mailboxA)).toMatchObject({ - topic: GMAIL_RECONCILE_TOPIC, - payload: { mailboxId: mailboxA, historyId: 'cursor-a' }, - }) - expect(byMailbox.get(mailboxB)).toMatchObject({ - topic: GMAIL_RECONCILE_TOPIC, - payload: { mailboxId: mailboxB, historyId: 'cursor-b' }, - }) }) it('acquires the access token exactly once per mailbox — reused for watch(), not re-fetched', async () => { @@ -214,13 +186,10 @@ describe('runGmailWatchMaintenance', () => { 'once-b@example.test', 'cursor-b', ) - const { queue } = fakeQueue() const tokenService = fakeTokenService(mailboxStore, {}) const getTokenSpy = vi.spyOn(tokenService, 'getAccessToken') - await runGmailWatchMaintenance( - buildDeps(mailboxStore, watchStateStore, queue, { tokenService }), - ) + await runGmailWatchMaintenance(buildDeps(mailboxStore, watchStateStore, { tokenService })) // Exactly one token-service call per mailbox: step 1 acquires the token and // the watch client reuses it, rather than fetching a second time. (Two @@ -230,23 +199,21 @@ describe('runGmailWatchMaintenance', () => { expect(getTokenSpy).toHaveBeenCalledWith(mailboxB) }) - it('a mailbox with no baseline cursor is renewed but NOT swept', async () => { + it('a mailbox with no baseline cursor is still renewed', async () => { const { mailboxStore, watchStateStore } = await freshStores() const mailbox = await mailboxStore.upsertConnectedMailbox({ address: 'no-cursor@example.test', provider: 'gmail', }) // No seedBaseline call — this mailbox has no gmail_watch_state row at all. - const { queue, enqueued } = fakeQueue() - const report = await runGmailWatchMaintenance(buildDeps(mailboxStore, watchStateStore, queue)) + const report = await runGmailWatchMaintenance(buildDeps(mailboxStore, watchStateStore)) - expect(report).toEqual({ total: 1, renewed: 1, swept: 0, needsReconnect: 0, failed: 0 }) - expect(enqueued).toHaveLength(0) + expect(report).toEqual({ total: 1, renewed: 1, needsReconnect: 0, failed: 0 }) expect(await watchStateStore.getCursor(mailbox.id)).toBeNull() }) - it('a token failure that leaves the mailbox needs_reconnect is counted needsReconnect, not renewed/swept — and other mailboxes still process', async () => { + it('a token failure that leaves the mailbox needs_reconnect is counted needsReconnect, not renewed — and other mailboxes still process', async () => { const { mailboxStore, watchStateStore } = await freshStores() const dead = await seedActiveMailboxWithCursor( mailboxStore, @@ -254,24 +221,21 @@ describe('runGmailWatchMaintenance', () => { 'dead@example.test', 'cursor-dead', ) - const healthy = await seedActiveMailboxWithCursor( + await seedActiveMailboxWithCursor( mailboxStore, watchStateStore, 'healthy@example.test', 'cursor-healthy', ) - const { queue, enqueued } = fakeQueue() const report = await runGmailWatchMaintenance( - buildDeps(mailboxStore, watchStateStore, queue, { + buildDeps(mailboxStore, watchStateStore, { tokenService: fakeTokenService(mailboxStore, { [dead]: 'needs_reconnect' }), }), ) - expect(report).toEqual({ total: 2, renewed: 1, swept: 1, needsReconnect: 1, failed: 0 }) + expect(report).toEqual({ total: 2, renewed: 1, needsReconnect: 1, failed: 0 }) expect((await mailboxStore.getMailboxById(dead))?.status).toBe('needs_reconnect') - expect(enqueued).toHaveLength(1) - expect(enqueued[0].payload.mailboxId).toBe(healthy) }) it('a transient token failure (mailbox stays active) is counted failed, not needsReconnect — and other mailboxes still process', async () => { @@ -282,27 +246,24 @@ describe('runGmailWatchMaintenance', () => { 'flaky@example.test', 'cursor-flaky', ) - const healthy = await seedActiveMailboxWithCursor( + await seedActiveMailboxWithCursor( mailboxStore, watchStateStore, 'healthy2@example.test', 'cursor-healthy2', ) - const { queue, enqueued } = fakeQueue() const report = await runGmailWatchMaintenance( - buildDeps(mailboxStore, watchStateStore, queue, { + buildDeps(mailboxStore, watchStateStore, { tokenService: fakeTokenService(mailboxStore, { [flaky]: 'transient' }), }), ) - expect(report).toEqual({ total: 2, renewed: 1, swept: 1, needsReconnect: 0, failed: 1 }) + expect(report).toEqual({ total: 2, renewed: 1, needsReconnect: 0, failed: 1 }) expect((await mailboxStore.getMailboxById(flaky))?.status).toBe('active') - expect(enqueued).toHaveLength(1) - expect(enqueued[0].payload.mailboxId).toBe(healthy) }) - it('a watch() throw is counted failed WITHOUT marking needs_reconnect, and the sweep still runs for that mailbox (valid token) — other mailboxes still process', async () => { + it('a watch() throw is counted failed WITHOUT marking needs_reconnect — other mailboxes still process', async () => { const { db: rawDb, mailboxStore, watchStateStore } = await freshStores() const glitchy = await seedActiveMailboxWithCursor( mailboxStore, @@ -310,59 +271,33 @@ describe('runGmailWatchMaintenance', () => { 'glitchy@example.test', 'cursor-glitchy', ) - const healthy = await seedActiveMailboxWithCursor( + await seedActiveMailboxWithCursor( mailboxStore, watchStateStore, 'healthy3@example.test', 'cursor-healthy3', ) const originalExpiration = new Date('2026-01-01T00:00:00.000Z') // written by seedBaseline above - const { queue, enqueued } = fakeQueue() const report = await runGmailWatchMaintenance( - buildDeps(mailboxStore, watchStateStore, queue, { + buildDeps(mailboxStore, watchStateStore, { createWatchClient: fakeCreateWatchClient([glitchy]), }), ) - expect(report).toEqual({ total: 2, renewed: 1, swept: 2, needsReconnect: 0, failed: 1 }) + expect(report).toEqual({ total: 2, renewed: 1, needsReconnect: 0, failed: 1 }) expect((await mailboxStore.getMailboxById(glitchy))?.status).toBe('active') // watch() failed, so the expiration seedBaseline wrote earlier is untouched. expect((await readWatchExpiration(rawDb, glitchy))?.toISOString()).toBe( originalExpiration.toISOString(), ) - // The sweep is independent of renewal — both mailboxes still get a job. - expect(enqueued).toHaveLength(2) - const sweptIds = enqueued.map((e) => e.payload.mailboxId).sort() - expect(sweptIds).toEqual([glitchy, healthy].sort()) - }) - - it('enqueues with no dedupeKey — a second run sweeps the same mailbox again rather than being suppressed as a duplicate', async () => { - const { mailboxStore, watchStateStore } = await freshStores() - await seedActiveMailboxWithCursor( - mailboxStore, - watchStateStore, - 'repeat@example.test', - 'cursor-repeat', - ) - const { queue, enqueued } = fakeQueue() - const deps = buildDeps(mailboxStore, watchStateStore, queue) - - await runGmailWatchMaintenance(deps) - await runGmailWatchMaintenance(deps) - - expect(enqueued).toHaveLength(2) - for (const call of enqueued) { - expect(call.opts?.dedupeKey).toBeUndefined() - } }) it('validates topicName is non-empty before doing any work', async () => { const { mailboxStore, watchStateStore } = await freshStores() const listSpy = vi.spyOn(mailboxStore, 'listActiveMailboxes') - const { queue } = fakeQueue() - const deps = buildDeps(mailboxStore, watchStateStore, queue, { topicName: '' }) + const deps = buildDeps(mailboxStore, watchStateStore, { topicName: '' }) await expect(runGmailWatchMaintenance(deps)).rejects.toThrow(/topicName/) expect(listSpy).not.toHaveBeenCalled() diff --git a/src/mail/gmail-watch-maintenance.ts b/src/mail/gmail-watch-maintenance.ts index b37e1ce..14a51c8 100644 --- a/src/mail/gmail-watch-maintenance.ts +++ b/src/mail/gmail-watch-maintenance.ts @@ -1,25 +1,30 @@ /** - * `runGmailWatchMaintenance` — the daily Gmail maintenance sweep (HT-42; - * specs/mail/gmail-push.md §6). Two jobs, run per active mailbox: + * `runGmailWatchMaintenance` — the daily Gmail `watch()` renewal (HT-42; + * specs/mail/gmail-push.md §6). One job, run per active mailbox: * - * 1. **Re-arm `watch()`.** Gmail push notifications stop — silently, with - * no error on either side — once a mailbox's `watch()` registration - * expires (~7 days out). This re-arms it and stores the fresh - * expiration. Daily (not every-6-days) buys a safety margin against a - * missed run; `watch()` is idempotent, so re-arming early is free. - * 2. **A bounded reconciliation sweep.** Push is best-effort (gmail-push.md - * §1): Gmail rate-limits and may drop or delay notifications. This - * enqueues one reconcile job per active mailbox — the SAME job the push - * webhook enqueues (`../api/gmail-webhook.ts`'s `GMAIL_RECONCILE_TOPIC`, - * `GmailReconcileJob`) — so a dropped or delayed *last* notification - * before a quiet spell never leaves a mailbox stale indefinitely. The - * reconcile consumer (`./gmail-reconcile.ts`, HT-41) re-reads the - * mailbox's STORED cursor itself and ignores the job's `historyId`, so a - * redundant sweep of an already-current mailbox is free — deduped by - * the idempotent ingest pipeline (inbound-ingestion.md §4), never - * doubled. + * **Re-arm `watch()`.** Gmail push notifications stop — silently, with no + * error on either side — once a mailbox's `watch()` registration expires (~7 + * days out). This re-arms it and stores the fresh expiration. Daily (not + * every-6-days) buys a safety margin against a missed run; `watch()` is + * idempotent, so re-arming early is free. * - * ## A plain sweep function, not a queue/cron adapter + * ## Renewal only, as of HT-94 + * + * This module also used to run a bounded reconciliation sweep, as a daily + * *backstop* for push being best-effort. That sweep is now the engine's + * PRIMARY inbound transport and lives in `./gmail-reconcile-sweep.ts`, running + * every minute (CHARTER.md §2, amended 2026-07-20). Two reasons it could not + * stay here, both in that module's doc: the cadences differ by three orders of + * magnitude, and renewal needs a per-mailbox access token while the sweep needs + * none — welding them would have meant refreshing a token every minute for a + * Gmail call the sweep never makes. + * + * Consequently this whole module is meaningful ONLY when push is configured. + * With no Pub/Sub topic there is no `watch()` to re-arm, and the composition + * root does not construct its deps at all; the cron endpoint stays routed and + * reports a skip (`../composition/root.ts`). + * + * ## A plain function, not a queue/cron adapter * * Exactly like `./delivery-worker.ts` (HT-16): `runGmailWatchMaintenance` * is a plain `async function` of injected dependencies, NOT built on a @@ -37,7 +42,7 @@ * (gmail-push.md §6). The whole per-mailbox unit of work * ({@link maintainOneMailbox}) is wrapped in its own try/catch inside * {@link runGmailWatchMaintenance}'s loop, so even a genuinely unexpected - * throw (a store or queue failure outside the two expected-failure + * throw (a store failure outside the two expected-failure * branches below) only counts that one mailbox `failed` and moves on to * the next — never aborting the batch. * @@ -64,14 +69,6 @@ * the ~7-day expiry leaves ample margin for a few missed runs) rather than * halting a healthy mailbox on a transient Gmail blip. * - * ## Re-arm and sweep are independent - * - * A `watch()` renewal failure does NOT skip the sweep for that mailbox - * (and a sweep is attempted even for a mailbox whose renewal just failed) - * — the two are unrelated Gmail API calls sharing only the mailbox's - * access token, so one failing is no reason to skip the other. Both are - * attempted for every mailbox with a valid token. - * * ## Never overwrite the cursor on renewal * * `watchStateStore.setWatchExpiration` (`../store/gmail-watch-state.ts`) @@ -80,25 +77,17 @@ * fresh `historyId` is AHEAD of the stored cursor, and overwriting the * cursor with it would silently skip un-reconciled mail). * - * ## The reconciliation lease lives in the CONSUMER, not here (HT-48) + * ## The reconciliation lease (HT-48) — now entirely the sweep's concern * - * Push-triggered reconciliation and this sweep both advance the same - * mailbox's cursor. gmail-push.md §6 calls serializing them a pure - * efficiency guard (avoiding redundant `history.list`/`messages.get` - * work), NOT a correctness requirement — the ingest pipeline's own dedup - * (inbound-ingestion.md §4) already makes either ordering safe. HT-48 - * implements that lease entirely in the reconcile job's CONSUMER - * (`./gmail-reconcile.ts`'s `claimReconcileLease`/`releaseReconcileLease` - * around `history.list`), not in this PRODUCER — this sweep still enqueues - * its reconcile job with NO `dedupeKey` on purpose (see - * {@link maintainOneMailbox}): a daily sweep of an already-current, quiet - * mailbox must still run, not be silently suppressed as a duplicate of an - * earlier job. Whether the resulting job actually does any Gmail work, or - * skips because another in-flight reconcile already holds the lease, is - * decided entirely on the consumer side, once the job is dequeued. + * The lease that serializes overlapping reconciliation lives in the reconcile + * job's CONSUMER (`./gmail-reconcile.ts`'s + * `claimReconcileLease`/`releaseReconcileLease` around `history.list`), never + * in a producer. Since this module no longer produces reconcile jobs, that + * discussion moved with the sweep — see `./gmail-reconcile-sweep.ts`'s doc, + * where it matters considerably more: at every-minute cadence the lease stops + * being an efficiency guard and becomes structural. */ -import { GMAIL_RECONCILE_TOPIC, type GmailReconcileJob } from '../api/gmail-webhook.js' // Type-only: engine modules never take a RUNTIME dependency on a concrete // adapter (src/providers/README.md's rule) — mirrors `./gmail-reconcile.ts`'s // identical `createHistoryClient` injection and `./gmail-connect.ts`'s own @@ -106,7 +95,6 @@ import { GMAIL_RECONCILE_TOPIC, type GmailReconcileJob } from '../api/gmail-webh // `createGmailWatchClient` (`../providers/adapters/gmail/watch.ts`); tests // pass a fake. import type { GmailWatchClient } from '../providers/adapters/gmail/index.js' -import type { QueueProvider } from '../providers/queue.js' import type { GmailWatchStateStore } from '../store/gmail-watch-state.js' import type { MailboxStore } from '../store/mailboxes.js' import type { GmailOAuthTokenService } from './gmail-oauth.js' @@ -122,9 +110,6 @@ export interface GmailWatchMaintenanceDeps { /** The per-mailbox `watch_expiration` write and stored-cursor read (`../store/gmail-watch-state.ts`). */ watchStateStore: GmailWatchStateStore - /** Where each mailbox's reconcile job is enqueued — the SAME `GMAIL_RECONCILE_TOPIC` the push webhook enqueues onto (`../api/gmail-webhook.ts`). */ - queue: QueueProvider - /** * Builds a {@link GmailWatchClient} bound to a per-mailbox * `getAccessToken`. REQUIRED and injected — `src/providers/README.md`'s @@ -151,8 +136,6 @@ export interface GmailWatchMaintenanceReport { total: number /** Mailboxes whose `watch()` was successfully re-armed and `watch_expiration` updated. */ renewed: number - /** Mailboxes for which a reconcile job was enqueued (had a stored cursor to sweep from). */ - swept: number /** Mailboxes found `needs_reconnect` after a token-acquisition failure — the token layer's own transition, not this cron's (see module doc). */ needsReconnect: number /** Mailboxes with a transient token or `watch()` failure this run — retried automatically on tomorrow's run. */ @@ -186,7 +169,6 @@ export async function runGmailWatchMaintenance( const counts: Omit = { renewed: 0, - swept: 0, needsReconnect: 0, failed: 0, } @@ -199,7 +181,7 @@ export async function runGmailWatchMaintenance( // expected-failure branches inside maintainOneMailbox. Never let one // mailbox stop the batch (module doc's "failure-isolated per // mailbox"). Safe to log err's message: everything reachable here - // (MailboxStore/GmailWatchStateStore/QueueProvider calls) is a plain + // (MailboxStore/GmailWatchStateStore calls) is a plain // store/queue error, never a token — see gmail-oauth.ts's and // watch.ts's module docs for why the token itself never surfaces in // a thrown error message. @@ -229,7 +211,7 @@ async function maintainOneMailbox( deps: GmailWatchMaintenanceDeps, counts: Omit, ): Promise { - const { tokenService, mailboxStore, watchStateStore, queue, createWatchClient, topicName } = deps + const { tokenService, mailboxStore, watchStateStore, createWatchClient, topicName } = deps // --- Step 1: acquire a token ONCE — this both probes the grant (to // distinguish a dead grant from a transient failure, the classification @@ -292,29 +274,12 @@ async function maintainOneMailbox( }) } - // --- Step 3: bounded reconciliation sweep — independent of re-arm - // above. Skips only when there's no baseline cursor yet (nothing to - // reconcile from — watch() at connect time, HT-40, seeds it; this cron - // only renews the expiration). NO dedupeKey (module doc): a daily sweep - // of an already-current, quiet mailbox must still run, never be - // suppressed as a duplicate of an earlier job — redundant reconcile work - // here is exactly what the consumer's lease (HT-48, ./gmail-reconcile.ts) - // now optimizes away when this job lands while another reconcile of the - // same mailbox is already in flight, and is otherwise safe because - // ingest dedups on (mailboxId, providerMessageId). --- - const cursor = await watchStateStore.getCursor(mailboxId) - if (cursor === null) { - logMaintenanceEvent('info', { - mailboxId, - outcome: 'skipped-sweep', - reason: 'no-baseline-cursor', - }) - return - } - - const job: GmailReconcileJob = { mailboxId, historyId: cursor } - await queue.enqueue(GMAIL_RECONCILE_TOPIC, job, {}) - counts.swept++ + // The bounded reconciliation sweep that used to be step 3 here moved to + // `./gmail-reconcile-sweep.ts` (HT-94). It is no longer a daily backstop for + // push but the primary inbound transport, running every minute — and it + // needs no access token, unlike the renewal above, so keeping the two welded + // would have meant a token refresh per mailbox per minute for a Gmail call + // the sweep never makes. See that module's doc for the full rationale. } /** diff --git a/src/store/gmail-watch-state.ts b/src/store/gmail-watch-state.ts index a8ba6c7..8d53f36 100644 --- a/src/store/gmail-watch-state.ts +++ b/src/store/gmail-watch-state.ts @@ -57,6 +57,12 @@ export interface GmailWatchStateStore { * a new cursor, or vice versa — the two values always land together, from * the same `watch()` call, in one write. * + * `watchExpiration` is OPTIONAL as of HT-94: when push is not configured for + * the deployment, connect arms no `watch()`, so there is no expiration to + * record and the (already nullable) column is written NULL. The cursor still + * seeds — from `getProfile()` — because the bounded scheduled fetch resumes + * from it exactly as the push path's reconcile does. + * * Optionally runs against a caller-supplied `tx` (`Db.transaction`'s * `Queryable`) instead of the bound `db`, so the connect flow can commit * this seed together with the mailbox row and token write as one atomic @@ -64,7 +70,7 @@ export interface GmailWatchStateStore { */ seedBaseline( mailboxId: string, - input: { historyId: string; watchExpiration: Date }, + input: { historyId: string; watchExpiration?: Date }, tx?: Queryable, ): Promise @@ -239,7 +245,10 @@ export function createGmailWatchStateStore(db: Db): GmailWatchStateStore { history_id = EXCLUDED.history_id, watch_expiration = EXCLUDED.watch_expiration, updated_at = now()`, - [mailboxId, input.historyId, input.watchExpiration], + // `?? null` is explicit rather than relying on the driver's + // undefined→NULL coercion: with push unconfigured (HT-94) there is no + // watch to expire, and the column is nullable precisely for that case. + [mailboxId, input.historyId, input.watchExpiration ?? null], ) }, diff --git a/vercel.json b/vercel.json index 876a6c4..f986629 100644 --- a/vercel.json +++ b/vercel.json @@ -28,6 +28,10 @@ "path": "/api/v1/internal/cron/snooze-wake", "schedule": "*/1 * * * *" }, + { + "path": "/api/v1/internal/cron/reconcile-sweep", + "schedule": "*/1 * * * *" + }, { "path": "/api/v1/internal/cron/watch-maintenance", "schedule": "0 6 * * *"