diff --git a/apps/hub/src/inbox-unsnooze-sweep.ts b/apps/hub/src/inbox-unsnooze-sweep.ts new file mode 100644 index 000000000..ae5f44796 --- /dev/null +++ b/apps/hub/src/inbox-unsnooze-sweep.ts @@ -0,0 +1,132 @@ +// A minimal periodic loop that reopens a snoozed inbox item once its +// `until` has passed (CL-7208) — mirroring `credential-expiry-sweep.ts` +// and `routine-scheduler.ts`, the only other periodic loops in this hub, +// rather than standing up a new generic scheduling primitive. Neither +// `@corbits/notify`'s `createNotifyDispatcher` (shaped around sink +// delivery — attempts, backoff, a `SinkRegistry` — not a status flip) nor +// the routine scheduler (`RoutineStore` is routine-domain-specific) fits +// "reopen a mailbox row when its snooze timer elapses" without bending +// their semantics, so this repeats the same small `setInterval` + own +// due-scan shape those two already use. +// +// `claimAndReopenSnooze` (in `@corbits/inbox`) does the actual claim: a +// single transaction that deletes the due snooze row and flips the +// message back to `open` together, so a throw here rolls back both and +// leaves the row for the next tick to retry rather than orphaning it — +// the mail-then-claim lesson CL-7209 applied to the credential-expiry +// sweep, applied here to the claim itself. +import { getLogger } from "@intx/log"; +import { reportError } from "@corbits/error-sink"; +import { + claimAndReopenSnooze, + findDueSnoozes, + type DueSnooze, +} from "@corbits/inbox"; +import type { MailboxDb, MailboxEventBus } from "@corbits/mailbox"; + +export type InboxUnsnoozeSweepStore = { + findDueSnoozes(now: Date): Promise; + /** Atomically claims one due snooze row and reopens its message if it's + * still snoozed. Returns whether it actually reopened something — see + * `claimAndReopenSnooze`'s own doc comment for the false cases. */ + claimAndReopen(row: DueSnooze, now: Date): Promise; +}; + +export function createDrizzleInboxUnsnoozeSweepStore( + db: MailboxDb, +): InboxUnsnoozeSweepStore { + return { + findDueSnoozes: (now) => findDueSnoozes(db, now), + claimAndReopen: (row, now) => claimAndReopenSnooze(db, row, now), + }; +} + +export type InboxUnsnoozeSweepDeps = { + store: InboxUnsnoozeSweepStore; + bus: Pick; + /** Injectable for deterministic tests; defaults to `Date.now`-backed wall time. */ + now?: () => Date; +}; + +const POLL_INTERVAL_MS = 60 * 1000; +const publishLog = getLogger(["hub", "inbox-unsnooze-sweep"]); + +function publishReopened( + bus: Pick, + row: DueSnooze, +): void { + // Best-effort, matching every other mailbox event publish in this repo + // (see packages/inbox/src/routes.ts's own `publish` helper): a bus + // failure here must never turn an already-committed reopen into a + // reported sweep failure. + try { + bus.publish( + { tenantId: row.tenantId, principalId: row.principalId }, + { type: "mailbox", id: row.messageId, op: "enrich" }, + ); + } catch (error) { + publishLog.error( + "mailbox reopen event publish failed for {id} on tenant {tenantId}, principal {principalId}: {error}", + { + id: row.messageId, + tenantId: row.tenantId, + principalId: row.principalId, + error: error instanceof Error ? error.message : String(error), + }, + ); + } +} + +/** + * One sweep: reopen every snooze due at `at`. Exported (rather than kept + * as a closure) so a test can drive a single, deterministic pass without + * waiting on `setInterval`. + */ +export async function tickInboxUnsnoozeSweep( + deps: Pick, + at: Date, +): Promise { + const due = await deps.store.findDueSnoozes(at); + for (const row of due) { + try { + const reopened = await deps.store.claimAndReopen(row, at); + if (reopened) publishReopened(deps.bus, row); + } catch (error) { + reportError(error, { + operation: "inbox_unsnooze_sweep", + tenantId: row.tenantId, + extra: { messageId: row.messageId }, + }); + // The claim is transactional (delete + reopen together): a throw + // means neither happened, so the row is still there and the next + // tick retries it instead of leaving the item snoozed forever. + } + } +} + +export function createInboxUnsnoozeSweep(deps: InboxUnsnoozeSweepDeps) { + const now = deps.now ?? (() => new Date()); + let tickInFlight = false; + + async function tick(): Promise { + if (tickInFlight) return; + tickInFlight = true; + try { + await tickInboxUnsnoozeSweep({ store: deps.store, bus: deps.bus }, now()); + } catch (err) { + reportError(err, { operation: "inbox_unsnooze_sweep_tick" }); + } finally { + tickInFlight = false; + } + } + + const interval = setInterval(() => void tick(), POLL_INTERVAL_MS); + if (typeof interval.unref === "function") interval.unref(); + + return { + tick, + stop(): void { + clearInterval(interval); + }, + }; +} diff --git a/apps/hub/src/index.ts b/apps/hub/src/index.ts index c277a9f73..8913fd023 100644 --- a/apps/hub/src/index.ts +++ b/apps/hub/src/index.ts @@ -331,6 +331,10 @@ import { createCredentialExpirySweep, createDrizzleCredentialExpirySweepStore, } from "./credential-expiry-sweep"; +import { + createDrizzleInboxUnsnoozeSweepStore, + createInboxUnsnoozeSweep, +} from "./inbox-unsnooze-sweep"; import { type } from "arktype"; import { betterAuth } from "better-auth"; @@ -2547,6 +2551,14 @@ export async function createHub(config: HubConfig) { }, }); + // Reopen a snoozed inbox item once its `until` has passed (CL-7208) — a + // light periodic sweep over `@corbits/inbox`'s own snooze table, on the + // same mailboxDb/mailboxBus every other mailbox consumer here shares. + const inboxUnsnoozeSweep = createInboxUnsnoozeSweep({ + store: createDrizzleInboxUnsnoozeSweepStore(mailboxDb), + bus: mailboxBus, + }); + // Shared `FoldedRunsDeps` for every one-shot Myra prompt below (routine // drafting, agent-definition drafting): a real one-shot inference call // that launches a folded run, awaits its single reply, and tears the run @@ -3496,6 +3508,7 @@ export async function createHub(config: HubConfig) { chatOrchestrator.dispose(); routineScheduler.stop(); credentialExpirySweep.stop(); + inboxUnsnoozeSweep.stop(); benchProvisioner.stop(); await insightsUsage.close(); await insightsLatency.close(); diff --git a/apps/hub/test/inbox-unsnooze-sweep.test.ts b/apps/hub/test/inbox-unsnooze-sweep.test.ts new file mode 100644 index 000000000..309a21007 --- /dev/null +++ b/apps/hub/test/inbox-unsnooze-sweep.test.ts @@ -0,0 +1,100 @@ +// The sweep's own orchestration — find due snoozes, claim-and-reopen each, +// publish only on an actual reopen, and never drop a row on a claim +// failure — against an in-memory fake store. Which rows are due, and the +// transactional delete+reopen claim itself, are `@corbits/inbox`'s own +// concern (`findDueSnoozes`/`claimAndReopenSnooze`); this only checks the +// tick loop calls them correctly and behaves on what comes back. +import { describe, expect, test } from "bun:test"; +import type { MailboxEvent, MailboxEventScope } from "@corbits/mailbox"; +import { + tickInboxUnsnoozeSweep, + type InboxUnsnoozeSweepStore, +} from "../src/inbox-unsnooze-sweep"; +import type { DueSnooze } from "@corbits/inbox"; + +function row(overrides: Partial = {}): DueSnooze { + return { + tenantId: "tnt_1", + principalId: "prn_1", + messageId: "msg_1", + ...overrides, + }; +} + +function fakeBus() { + const published: { scope: MailboxEventScope; event: MailboxEvent }[] = []; + return { + published, + bus: { + publish(scope: MailboxEventScope, event: MailboxEvent) { + published.push({ scope, event }); + }, + }, + }; +} + +describe("tickInboxUnsnoozeSweep", () => { + test("reopens every due row and publishes a reopened event for each", async () => { + const due = [row({ messageId: "msg_1" }), row({ messageId: "msg_2" })]; + const claimed: DueSnooze[] = []; + const store: InboxUnsnoozeSweepStore = { + findDueSnoozes: async () => due, + claimAndReopen: async (candidate) => { + claimed.push(candidate); + return true; + }, + }; + const { bus, published } = fakeBus(); + + await tickInboxUnsnoozeSweep({ store, bus }, new Date()); + + expect(claimed.map((c) => c.messageId)).toEqual(["msg_1", "msg_2"]); + expect(published).toHaveLength(2); + expect(published[0]?.event).toEqual({ + type: "mailbox", + id: "msg_1", + op: "enrich", + }); + }); + + test("does not publish when claimAndReopen reports it reopened nothing", async () => { + const store: InboxUnsnoozeSweepStore = { + findDueSnoozes: async () => [row()], + // False: another replica already claimed it, or the message had + // already left `snoozed` — either way, no event to publish. + claimAndReopen: async () => false, + }; + const { bus, published } = fakeBus(); + + await tickInboxUnsnoozeSweep({ store, bus }, new Date()); + + expect(published).toHaveLength(0); + }); + + test("a throw from claimAndReopen is swallowed per-row and does not publish", async () => { + const due = [row({ messageId: "msg_bad" }), row({ messageId: "msg_ok" })]; + const claimed: string[] = []; + const store: InboxUnsnoozeSweepStore = { + findDueSnoozes: async () => due, + claimAndReopen: async (candidate) => { + claimed.push(candidate.messageId); + if (candidate.messageId === "msg_bad") { + throw new Error("connection reset mid-claim"); + } + return true; + }, + }; + const { bus, published } = fakeBus(); + + // Must not throw out of the tick — one bad row shouldn't abort the rest. + await tickInboxUnsnoozeSweep({ store, bus }, new Date()); + + expect(claimed).toEqual(["msg_bad", "msg_ok"]); + // Only the row that actually reopened publishes; the thrown claim is + // left for the next tick to retry (claimAndReopenSnooze's transaction + // rolls back on a throw, so the snooze row is still there) rather than + // being dropped. + expect(published).toHaveLength(1); + expect(published[0]?.event.id).toBe("msg_ok"); + }); +}); diff --git a/bun.lock b/bun.lock index cc2eb879b..e793f0386 100644 --- a/bun.lock +++ b/bun.lock @@ -849,13 +849,14 @@ "@intx/hub-api": "workspace:*", "@intx/log": "0.3.0", "arktype": "catalog:", + "drizzle-orm": "catalog:", "hono": "catalog:", + "postgres": "catalog:", }, "devDependencies": { "@intx/db": "workspace:*", "@intx/hub-common": "0.3.0", "@types/bun": "catalog:", - "postgres": "catalog:", "typescript": "catalog:", }, }, diff --git a/packages/inbox/package.json b/packages/inbox/package.json index eaeb0caea..56301860a 100644 --- a/packages/inbox/package.json +++ b/packages/inbox/package.json @@ -7,7 +7,8 @@ "type": "module", "exports": { ".": "./src/index.ts", - "./client": "./src/client.ts" + "./client": "./src/client.ts", + "./migrations": "./src/migrations.ts" }, "scripts": { "typecheck": "tsc --noEmit", @@ -20,13 +21,14 @@ "@intx/hub-api": "workspace:*", "@intx/log": "0.3.0", "arktype": "catalog:", - "hono": "catalog:" + "drizzle-orm": "catalog:", + "hono": "catalog:", + "postgres": "catalog:" }, "devDependencies": { "@intx/db": "workspace:*", "@intx/hub-common": "0.3.0", "@types/bun": "catalog:", - "postgres": "catalog:", "typescript": "catalog:" } } diff --git a/packages/inbox/src/index.ts b/packages/inbox/src/index.ts index 984f47c21..75816bc8e 100644 --- a/packages/inbox/src/index.ts +++ b/packages/inbox/src/index.ts @@ -28,7 +28,14 @@ export { type CreateWorkbenchMailboxDeliveryOpts, } from "./delivery"; export { + applyInboxMigrations, applyMailboxMigrations, + type ApplyInboxMigrationsReport, type ApplyMailboxMigrationsReport, } from "./migrations"; export { createInboxRoutes, type CreateInboxRoutesDeps } from "./routes"; +export { + claimAndReopenSnooze, + findDueSnoozes, + type DueSnooze, +} from "./snooze-store"; diff --git a/packages/inbox/src/migrations.ts b/packages/inbox/src/migrations.ts index c1d8497e6..ce26d334d 100644 --- a/packages/inbox/src/migrations.ts +++ b/packages/inbox/src/migrations.ts @@ -1,8 +1,19 @@ -// URL-shaped wrapper around `@corbits/mailbox`'s `runMailboxMigrations` so -// scripts/db-setup.ts can apply it the same way it applies every other -// installed package's migrations. +// Two independent migration stories for `@corbits/inbox`: +// +// - `applyMailboxMigrations`: a URL-shaped wrapper around +// `@corbits/mailbox`'s own `runMailboxMigrations`, so scripts/db-setup.ts +// can apply it the same way it applies every other installed package's +// migrations. That package keeps its own ledger inside the `mailbox` +// schema. +// - `applyInboxMigrations`: this package's own product table (CL-7208's +// `inbox.snooze` — see `./schema.ts` for why `@corbits/mailbox`'s +// enrichment has no column for a snooze's `until`). Bookkeeping uses its +// own ledger table, following the same pattern as +// `@workbench/onboarding`'s `migrations.ts`, so this package's migration +// history stays extractable on its own. import { createMailboxDb, runMailboxMigrations } from "@corbits/mailbox"; +import postgres from "postgres"; export interface ApplyMailboxMigrationsReport { applied: string[]; @@ -28,3 +39,92 @@ export async function applyMailboxMigrations( await close(); } } + +export interface InboxMigration { + name: string; + sql: string; +} + +const SCHEMA = "inbox"; +const LEDGER_TABLE = "inbox_migrations"; + +export const inboxMigrations: readonly InboxMigration[] = [ + { + name: "0001_snooze", + sql: ` + CREATE TABLE IF NOT EXISTS "inbox"."snooze" ( + "tenant_id" text NOT NULL, + "principal_id" text NOT NULL, + "message_id" text NOT NULL, + "until" timestamptz NOT NULL, + PRIMARY KEY ("tenant_id", "principal_id", "message_id") + ); + CREATE INDEX IF NOT EXISTS "inbox_snooze_until_idx" ON "inbox"."snooze" ("until"); + `, + }, +]; + +function quoteIdentifier(name: string): string { + return `"${name.replace(/"/g, '""')}"`; +} + +function quoteQualified(schema: string, name: string): string { + return `${quoteIdentifier(schema)}.${quoteIdentifier(name)}`; +} + +export interface ApplyInboxMigrationsReport { + applied: string[]; + alreadyApplied: string[]; +} + +/** + * Apply `inboxMigrations` against `databaseUrl`, idempotently: a migration + * already recorded in the ledger is skipped, never re-run. Failures are + * loud — the migration name and the underlying error are both surfaced. + */ +export async function applyInboxMigrations( + databaseUrl: string, +): Promise { + const sql = postgres(databaseUrl, { max: 1, onnotice: () => undefined }); + try { + await sql.unsafe(`CREATE SCHEMA IF NOT EXISTS ${quoteIdentifier(SCHEMA)}`); + await sql.unsafe( + `CREATE TABLE IF NOT EXISTS ${quoteQualified(SCHEMA, LEDGER_TABLE)} (` + + `name text PRIMARY KEY, applied_at timestamptz NOT NULL DEFAULT now())`, + ); + + const applied: string[] = []; + const alreadyApplied: string[] = []; + + for (const migration of inboxMigrations) { + const existing = await sql.unsafe( + `SELECT 1 FROM ${quoteQualified(SCHEMA, LEDGER_TABLE)} WHERE name = $1`, + [migration.name], + ); + if (existing.length > 0) { + alreadyApplied.push(migration.name); + continue; + } + try { + await sql.begin(async (tx) => { + await tx.unsafe(migration.sql); + await tx.unsafe( + `INSERT INTO ${quoteQualified(SCHEMA, LEDGER_TABLE)} (name) VALUES ($1)`, + [migration.name], + ); + }); + applied.push(migration.name); + } catch (error) { + throw new Error( + `@corbits/inbox migration ${JSON.stringify(migration.name)} failed: ` + + `${error instanceof Error ? error.message : String(error)}`, + { cause: error }, + ); + } + } + + return { applied, alreadyApplied }; + } finally { + await sql.end({ timeout: 5 }); + } +} diff --git a/packages/inbox/src/routes.ts b/packages/inbox/src/routes.ts index 10d0bdd2c..0977a800c 100644 --- a/packages/inbox/src/routes.ts +++ b/packages/inbox/src/routes.ts @@ -20,6 +20,7 @@ import { import { reportError } from "@corbits/error-sink"; import type { TenantEnv } from "@intx/hub-api"; import { getLogger } from "@intx/log"; +import { type } from "arktype"; import { Hono, type Context } from "hono"; import { @@ -35,9 +36,20 @@ import { type InboxCounts, type InboxItem, } from "./project"; +import { setSnoozeUntil } from "./snooze-store"; import { WORKBENCH_INBOX_PRIORITIES } from "./vocabulary"; import { walkAllOpen } from "./walk"; +// Parsed at the boundary, per AGENTS.md: `until` is untrusted request +// input, never `as`-cast. Required — a snooze with no `until` is exactly +// the bug this ticket fixes (CL-7208). +const SnoozeBodySchema = type({ until: "string" }); + +/** Thrown inside `/:id/snooze`'s transaction to roll back a snooze-until + * insert for a message that turned out not to be in scope, and caught + * outside to answer 404 instead of a 500. */ +class SnoozeTargetNotFound extends Error {} + // Page size for bulk product ops (mark-all-read, clear-done, counts). Large // enough that a normal inbox finishes in one round-trip; anything past it // walks with the package's cursor. @@ -389,23 +401,54 @@ export function createInboxRoutes( const tenant = c.get("tenant"); const principal = c.get("principal"); const id = c.req.param("id"); - const body: unknown = await c.req.json().catch(() => ({})); - // `until` is accepted for forward-compat with a scheduled unsnooze; the - // product store today only records the status flip. - let until: string | undefined; - if (typeof body === "object" && body !== null && "until" in body) { - const rawUntil = (body as { until: unknown }).until; - if (rawUntil !== undefined && typeof rawUntil !== "string") { - return c.json({ error: "until must be a string" }, 400); - } - if (typeof rawUntil === "string") until = rawUntil; + const rawBody: unknown = await c.req.json().catch(() => null); + const parsedBody = SnoozeBodySchema(rawBody); + if (parsedBody instanceof type.errors) { + return c.json( + { error: `invalid snooze body: ${parsedBody.summary}` }, + 400, + ); + } + const until = new Date(parsedBody.until); + if (Number.isNaN(until.getTime())) { + return c.json({ error: "until must be a valid timestamp" }, 400); + } + const now = new Date(); + if (until.getTime() <= now.getTime()) { + return c.json({ error: "until must be in the future" }, 400); + } + + const scope = { tenantId: tenant.id, principalId: principal.id, id }; + // Both writes run in one transaction: an uncommitted insert is + // invisible to any other transaction under read-committed isolation, + // so the sweep's own `claimAndReopenSnooze` can never see this snooze + // row until the status flip has *also* committed alongside it. Without + // this, a sweep tick landing between two separate statements could + // find the row before the status flip lands, no-op (the message isn't + // `snoozed` yet) and delete the row as "cleanup," and then the status + // flip would still land afterward — leaving the message stuck + // `snoozed` with the row that was supposed to reopen it already gone. + // That is exactly this ticket's bug, reintroduced by an un-transacted + // write order (CL-7208). Throwing (rather than returning false) on a + // missing message rolls the snooze insert back too, via the sentinel + // below caught outside the transaction. + let notFound = false; + await db + .transaction(async (tx) => { + await setSnoozeUntil(tx, scope, until); + const ok = await enrichMailboxMessage(tx, scope, { status: "snoozed" }); + if (!ok) throw new SnoozeTargetNotFound(); + }) + .catch((error) => { + if (error instanceof SnoozeTargetNotFound) { + notFound = true; + return; + } + throw error; + }); + if (notFound) { + return c.json({ error: "not found" }, 404); } - const ok = await enrichMailboxMessage( - db, - { tenantId: tenant.id, principalId: principal.id, id }, - { status: "snoozed" }, - ); - if (!ok) return c.json({ error: "not found" }, 404); publish( bus, { tenantId: tenant.id, principalId: principal.id }, @@ -414,7 +457,7 @@ export function createInboxRoutes( ); return c.json({ ok: true, - until, + until: until.toISOString(), }); }); diff --git a/packages/inbox/src/schema.ts b/packages/inbox/src/schema.ts new file mode 100644 index 000000000..4fddf2c5b --- /dev/null +++ b/packages/inbox/src/schema.ts @@ -0,0 +1,28 @@ +// The one product table `@corbits/inbox` owns: `@corbits/mailbox`'s +// enrichment only supports `priority`/`classification`/`status` (see +// `enrichMailboxMessage`'s `MailboxEnrichment`), so there is no column +// there to hold a snooze's `until` timestamp. This table is that column, +// keyed to the same (tenantId, principalId, messageId) scope every +// mailbox mutation uses. Lives in its own `inbox` Postgres schema, never +// `public` — see docs/package-migrations.md and the same pattern in +// `@workbench/onboarding`'s `schema.ts`. +import { pgSchema, primaryKey, text, timestamp } from "drizzle-orm/pg-core"; + +export const inboxSchema = pgSchema("inbox"); + +export const inboxSnooze = inboxSchema.table( + "snooze", + { + tenantId: text("tenant_id").notNull(), + principalId: text("principal_id").notNull(), + messageId: text("message_id").notNull(), + until: timestamp("until", { withTimezone: true }).notNull(), + }, + (table) => [ + primaryKey({ + columns: [table.tenantId, table.principalId, table.messageId], + }), + ], +); + +export type InboxSnoozeRow = typeof inboxSnooze.$inferSelect; diff --git a/packages/inbox/src/snooze-store.ts b/packages/inbox/src/snooze-store.ts new file mode 100644 index 000000000..631b4ddac --- /dev/null +++ b/packages/inbox/src/snooze-store.ts @@ -0,0 +1,150 @@ +// Persistence for CL-7208's snooze-until: `@corbits/mailbox`'s enrichment +// has no column for it (see `./schema.ts`), so this module owns the +// package's one product table and the two operations built on it — set at +// snooze time, and the atomic claim-and-reopen a periodic sweep drives +// once `until` has passed. +import { and, eq, lte } from "drizzle-orm"; +import { + enrichMailboxMessage, + getMailboxMessage, + type MailboxDb, +} from "@corbits/mailbox"; + +import { inboxSnooze } from "./schema"; + +export interface SnoozeScope { + tenantId: string; + principalId: string; + id: string; +} + +/** + * Record when a snoozed message should reopen. `routes.ts`'s `/:id/snooze` + * handler calls this and the status flip to `snoozed` inside one + * `db.transaction`, not as two independent statements: an uncommitted + * insert is invisible to any other transaction under read-committed + * isolation, so a concurrent sweep tick's `claimAndReopenSnooze` can never + * observe this row before the status flip has *also* committed alongside + * it. Two separate statements would let a sweep tick land in the gap, see + * the row before the flip, no-op (the message isn't `snoozed` yet), and + * delete the row as harmless-looking cleanup — and then the status flip + * would still land afterward, reproducing this ticket's exact bug: a + * message stuck `snoozed` with nothing left to ever reopen it. + */ +export async function setSnoozeUntil( + db: MailboxDb, + scope: SnoozeScope, + until: Date, +): Promise { + await db + .insert(inboxSnooze) + .values({ + tenantId: scope.tenantId, + principalId: scope.principalId, + messageId: scope.id, + until, + }) + .onConflictDoUpdate({ + target: [ + inboxSnooze.tenantId, + inboxSnooze.principalId, + inboxSnooze.messageId, + ], + set: { until }, + }); +} + +export async function clearSnoozeUntil( + db: MailboxDb, + scope: SnoozeScope, +): Promise { + await db + .delete(inboxSnooze) + .where( + and( + eq(inboxSnooze.tenantId, scope.tenantId), + eq(inboxSnooze.principalId, scope.principalId), + eq(inboxSnooze.messageId, scope.id), + ), + ); +} + +export interface DueSnooze { + tenantId: string; + principalId: string; + messageId: string; +} + +/** Every snooze row due at or before `now`. A bare `SELECT` — claims + * nothing, so two hub replicas can see the same row; `claimAndReopenSnooze` + * is what makes acting on it race-safe. */ +export async function findDueSnoozes( + db: MailboxDb, + now: Date, +): Promise { + const rows = await db + .select({ + tenantId: inboxSnooze.tenantId, + principalId: inboxSnooze.principalId, + messageId: inboxSnooze.messageId, + }) + .from(inboxSnooze) + .where(lte(inboxSnooze.until, now)); + return rows; +} + +/** + * Atomically claim one due snooze row and reopen its message if it's still + * `snoozed` — all inside one transaction, so the delete-claim and the + * status flip either both happen or neither does. Two replicas racing the + * same row: the second's `DELETE ... RETURNING` simply returns no rows + * (the first already deleted it), so only one replica ever flips the + * status or is told to publish a reopened event. + * + * Returns whether this call actually reopened the message — `false` means + * it lost the race, the row was no longer due, or the message had already + * left `snoozed` (e.g. the user manually reopened it) and the row was just + * cleaned up. A throw rolls the whole transaction back, so a failure here + * leaves the snooze row exactly as it was — retried on the sweep's next + * tick rather than dropped (the same mail-then-claim ordering lesson + * CL-7209 applied to the credential-expiry sweep, applied here to the + * claim itself). + */ +export async function claimAndReopenSnooze( + db: MailboxDb, + row: DueSnooze, + now: Date, +): Promise { + return db.transaction(async (tx) => { + const claimed = await tx + .delete(inboxSnooze) + .where( + and( + eq(inboxSnooze.tenantId, row.tenantId), + eq(inboxSnooze.principalId, row.principalId), + eq(inboxSnooze.messageId, row.messageId), + lte(inboxSnooze.until, now), + ), + ) + .returning({ messageId: inboxSnooze.messageId }); + if (claimed.length === 0) return false; + + const message = await getMailboxMessage(tx, { + tenantId: row.tenantId, + principalId: row.principalId, + id: row.messageId, + }); + if (message === null || message.status !== "snoozed") return false; + + await enrichMailboxMessage( + tx, + { + tenantId: row.tenantId, + principalId: row.principalId, + id: row.messageId, + }, + { status: "open" }, + ); + return true; + }); +} diff --git a/packages/inbox/test/routes.test.ts b/packages/inbox/test/routes.test.ts index ac1bab718..ae105dd57 100644 --- a/packages/inbox/test/routes.test.ts +++ b/packages/inbox/test/routes.test.ts @@ -83,6 +83,52 @@ describe("GET / cursor/filter cross-check", () => { }); }); +describe("POST /:id/snooze rejects before touching the database (CL-7208)", () => { + test("rejects a missing until", async () => { + const app = mount(); + const response = await app.request("/msg_1/snooze", { + method: "POST", + headers: { "content-type": "application/json" }, + body: JSON.stringify({}), + }); + expect(response.status).toBe(400); + }); + + test("rejects an until that isn't a string", async () => { + const app = mount(); + const response = await app.request("/msg_1/snooze", { + method: "POST", + headers: { "content-type": "application/json" }, + body: JSON.stringify({ until: 12345 }), + }); + expect(response.status).toBe(400); + }); + + test("rejects an until that doesn't parse as a timestamp", async () => { + const app = mount(); + const response = await app.request("/msg_1/snooze", { + method: "POST", + headers: { "content-type": "application/json" }, + body: JSON.stringify({ until: "not-a-date" }), + }); + expect(response.status).toBe(400); + const body = (await response.json()) as { error: string }; + expect(body.error).toBe("until must be a valid timestamp"); + }); + + test("rejects an until that has already passed, rather than snoozing forever", async () => { + const app = mount(); + const response = await app.request("/msg_1/snooze", { + method: "POST", + headers: { "content-type": "application/json" }, + body: JSON.stringify({ until: "2020-01-01T00:00:00.000Z" }), + }); + expect(response.status).toBe(400); + const body = (await response.json()) as { error: string }; + expect(body.error).toBe("until must be in the future"); + }); +}); + describe("bulk ops surface a failed inbox walk instead of silently truncating it (CL-7207)", () => { test("GET /counts reports and 500s rather than swallowing a walk failure", async () => { const app = mount(throwingDb("connection reset mid-walk")); diff --git a/packages/inbox/test/snooze-store.test.ts b/packages/inbox/test/snooze-store.test.ts new file mode 100644 index 000000000..c0d2ce958 --- /dev/null +++ b/packages/inbox/test/snooze-store.test.ts @@ -0,0 +1,327 @@ +// DB-gated integration test for CL-7208's snooze-until persistence and +// claim-and-reopen: proves `setSnoozeUntil`/`findDueSnoozes` write and read +// a real row in the package's own `inbox.snooze` table (applied through +// `scripts/db-setup.ts`, the same path `apps/hub` boots with — see +// `delivery.test.ts` for the pattern this follows), and that +// `claimAndReopenSnooze` actually flips a real mailbox message back to +// `open` and removes the snooze row once `until` has passed. Runs against +// its own scratch database, never the developer's or the walking-skeleton +// suite's. +import { afterAll, beforeAll, describe, expect, test } from "bun:test"; +import postgres from "postgres"; +import { Hono } from "hono"; +import type { TenantEnv } from "@intx/hub-api"; + +import { createDB, schema } from "@intx/db"; +import { generateId } from "@intx/hub-common"; +import { + createInMemoryMailboxEventBus, + createMailboxDb, + enrichMailboxMessage, + getMailboxMessage, + writeMailboxMessage, +} from "@corbits/mailbox"; + +import { setupDatabase } from "../../../scripts/db-setup"; +import { e2eDatabaseUrl } from "../../../scripts/e2e/harness"; +import { createInboxRoutes } from "../src/routes"; +import { + claimAndReopenSnooze, + clearSnoozeUntil, + findDueSnoozes, + setSnoozeUntil, +} from "../src/snooze-store"; + +function scratchUrlFor(e2eUrl: string): string { + const url = new URL(e2eUrl); + const database = url.pathname.replace(/^\//, ""); + url.pathname = `/${database}_inbox_snooze_test`; + return url.toString(); +} + +function dbConfigFromUrl(databaseUrl: string) { + const url = new URL(databaseUrl); + return { + host: url.hostname, + port: url.port === "" ? 5432 : Number(url.port), + user: decodeURIComponent(url.username), + password: decodeURIComponent(url.password), + database: url.pathname.replace(/^\//, ""), + }; +} + +const databaseUrl = e2eDatabaseUrl(); +const describeIfDb = databaseUrl === undefined ? describe.skip : describe; + +describeIfDb("snooze-store against a real inbox.snooze table", () => { + const scratchUrl = scratchUrlFor( + databaseUrl ?? "postgres://localhost:5432/unused", + ); + const scratchDatabase = new URL(scratchUrl).pathname.replace(/^\//, ""); + + const tenantId = generateId("tenant"); + const principalId = generateId("principal"); + + beforeAll(async () => { + const maintenanceUrl = new URL(scratchUrl); + maintenanceUrl.pathname = "/postgres"; + const maintenance = postgres(maintenanceUrl.toString(), { + max: 1, + onnotice: () => undefined, + }); + try { + await maintenance.unsafe(`DROP DATABASE IF EXISTS "${scratchDatabase}"`); + await maintenance.unsafe(`CREATE DATABASE "${scratchDatabase}"`); + } finally { + await maintenance.end(); + } + + // Platform migrations plus every installed package's, including + // @corbits/inbox's own `inbox.snooze` table (CL-7208). + await setupDatabase(scratchUrl); + + const { db, close } = createDB(dbConfigFromUrl(scratchUrl)); + try { + await db.insert(schema.tenant).values({ + id: tenantId, + name: "Snooze Test Bench", + slug: `snooze-${tenantId}`, + domain: `snooze-${tenantId}.localhost`, + }); + await db.insert(schema.principal).values({ + id: principalId, + tenantId, + kind: "agent", + refId: "not-a-real-agent-instance", + status: "active", + }); + } finally { + await close(); + } + }, 30000); + + afterAll(async () => { + const maintenanceUrl = new URL(scratchUrl); + maintenanceUrl.pathname = "/postgres"; + const maintenance = postgres(maintenanceUrl.toString(), { + max: 1, + onnotice: () => undefined, + }); + try { + await maintenance.unsafe(`DROP DATABASE IF EXISTS "${scratchDatabase}"`); + } finally { + await maintenance.end(); + } + }); + + test("setSnoozeUntil persists a row findDueSnoozes reads back once due", async () => { + const mailboxDb = createMailboxDb(scratchUrl); + try { + const written = await writeMailboxMessage(mailboxDb.db, { + tenantId, + principalId, + address: `${principalId}@inbox.test`, + fromAddress: "routine:test", + subject: "Snooze me", + body: "test body", + status: "snoozed", + }); + expect(written).not.toBeNull(); + const messageId = written?.id; + if (messageId === undefined) throw new Error("message not written"); + const scope = { tenantId, principalId, id: messageId }; + + const future = new Date(Date.now() + 60_000); + await setSnoozeUntil(mailboxDb.db, scope, future); + + // Not due yet. + const notYetDue = await findDueSnoozes(mailboxDb.db, new Date()); + expect(notYetDue.find((r) => r.messageId === messageId)).toBeUndefined(); + + // Due once `now` passes `until`. + const afterUntil = new Date(future.getTime() + 1000); + const due = await findDueSnoozes(mailboxDb.db, afterUntil); + expect(due.find((r) => r.messageId === messageId)).toEqual({ + tenantId, + principalId, + messageId, + }); + + const reopened = await claimAndReopenSnooze( + mailboxDb.db, + { tenantId, principalId, messageId }, + afterUntil, + ); + expect(reopened).toBe(true); + + const message = await getMailboxMessage(mailboxDb.db, scope); + expect(message?.status).toBe("open"); + + // The claim deleted the snooze row: a second claim attempt finds + // nothing left to reopen. + const secondClaim = await claimAndReopenSnooze( + mailboxDb.db, + { tenantId, principalId, messageId }, + afterUntil, + ); + expect(secondClaim).toBe(false); + } finally { + await mailboxDb.close(); + } + }); + + test("claimAndReopenSnooze no-ops and cleans up a row whose message is no longer snoozed", async () => { + const mailboxDb = createMailboxDb(scratchUrl); + try { + const written = await writeMailboxMessage(mailboxDb.db, { + tenantId, + principalId, + address: `${principalId}@inbox.test`, + fromAddress: "routine:test", + subject: "Snooze then manually reopen", + body: "test body", + status: "snoozed", + }); + const messageId = written?.id; + if (messageId === undefined) throw new Error("message not written"); + const scope = { tenantId, principalId, id: messageId }; + + const past = new Date(Date.now() - 1000); + await setSnoozeUntil(mailboxDb.db, scope, past); + + // Simulate the user manually reopening it before the sweep ran — + // exactly the race `claimAndReopenSnooze`'s status check guards. + await enrichMailboxMessage(mailboxDb.db, scope, { status: "open" }); + + const due = await findDueSnoozes(mailboxDb.db, new Date()); + expect(due.some((r) => r.messageId === messageId)).toBe(true); + + const reopened = await claimAndReopenSnooze( + mailboxDb.db, + { tenantId, principalId, messageId }, + new Date(), + ); + // The row was due, so the claim wins, but the message was already + // `open` — nothing to reopen, so this reports false while still + // cleaning up the now-stale snooze row. + expect(reopened).toBe(false); + + const stillDue = await findDueSnoozes(mailboxDb.db, new Date()); + expect(stillDue.some((r) => r.messageId === messageId)).toBe(false); + } finally { + await mailboxDb.close(); + } + }); + + test("clearSnoozeUntil removes the row without touching the message", async () => { + const mailboxDb = createMailboxDb(scratchUrl); + try { + const written = await writeMailboxMessage(mailboxDb.db, { + tenantId, + principalId, + address: `${principalId}@inbox.test`, + fromAddress: "routine:test", + subject: "Snooze then cancel", + body: "test body", + }); + const messageId = written?.id; + if (messageId === undefined) throw new Error("message not written"); + const scope = { tenantId, principalId, id: messageId }; + + await setSnoozeUntil(mailboxDb.db, scope, new Date(Date.now() - 1000)); + await clearSnoozeUntil(mailboxDb.db, scope); + + const due = await findDueSnoozes(mailboxDb.db, new Date()); + expect(due.some((r) => r.messageId === messageId)).toBe(false); + } finally { + await mailboxDb.close(); + } + }); + + test("a sweep claim racing the route's own snooze write never strands the message `snoozed` with no way to reopen it", async () => { + // Regression test for a race a Critique pass on CL-7208 found and + // reproduced: `POST /:id/snooze` used to persist `until` and flip + // `status` to `snoozed` as two separate statements. A sweep tick's + // `claimAndReopenSnooze` landing in the gap between them could see the + // snooze row before the status flip landed, no-op (the message wasn't + // `snoozed` yet) and delete the row as harmless cleanup — and then the + // route's own status flip would still land afterward, leaving the + // message stuck `snoozed` with the row that was supposed to reopen it + // already gone. `routes.ts` now runs both writes in one transaction so + // the row is invisible to a concurrent claim until the flip commits + // with it; this drives the same interleaving the bug needed and + // asserts the message is never left `snoozed` with zero snooze rows. + const mailboxDb = createMailboxDb(scratchUrl); + try { + const written = await writeMailboxMessage(mailboxDb.db, { + tenantId, + principalId, + address: `${principalId}@inbox.test`, + fromAddress: "routine:test", + subject: "Race the sweep", + body: "test body", + }); + const messageId = written?.id; + if (messageId === undefined) throw new Error("message not written"); + const scope = { tenantId, principalId, id: messageId }; + + const app = new Hono(); + app.use("*", async (c, next) => { + c.set("tenant", { id: tenantId } as never); + c.set("principal", { id: principalId } as never); + await next(); + }); + app.route( + "/", + createInboxRoutes({ + db: mailboxDb.db, + bus: createInMemoryMailboxEventBus(), + }), + ); + + const until = new Date(Date.now() + 50); + // A "due" check has to compare against a fixed point safely past + // `until`, not real wall-clock time at check time — the whole + // request/claim/verify round trip below runs well under 50ms, so a + // `new Date()` taken after it finishes can still be *before* + // `until`, which would make an already-correct row look "not due + // yet" rather than reopened. `wellPastUntil` is what the real sweep + // would eventually pass once `until` has genuinely elapsed. + const wellPastUntil = new Date(until.getTime() + 60_000); + + const [response] = await Promise.all([ + app.request(`/${messageId}/snooze`, { + method: "POST", + headers: { "content-type": "application/json" }, + body: JSON.stringify({ until: until.toISOString() }), + }), + // Fired concurrently with the route's own write, already past + // `until` — exactly the timing the bug needed: the sweep + // considers the row due from the moment it can see it at all. + claimAndReopenSnooze( + mailboxDb.db, + { tenantId, principalId, messageId }, + wellPastUntil, + ), + ]); + expect(response.status).toBe(200); + + const message = await getMailboxMessage(mailboxDb.db, scope); + const due = await findDueSnoozes(mailboxDb.db, wellPastUntil); + const hasSnoozeRow = due.some((r) => r.messageId === messageId); + + // The invariant the bug violated: never `snoozed` with nothing left + // to ever reopen it. Whichever side of the race won, the message + // must end up either genuinely `open` (the claim won) or `snoozed` + // with its row still there to be claimed on a later tick (the route + // won) — never `snoozed` with the row already gone. + if (message?.status === "snoozed") { + expect(hasSnoozeRow).toBe(true); + } else { + expect(message?.status).toBe("open"); + } + } finally { + await mailboxDb.close(); + } + }); +}); diff --git a/scripts/checks/no-product-tenancy.ts b/scripts/checks/no-product-tenancy.ts index ea4632ef9..3d6392f05 100644 --- a/scripts/checks/no-product-tenancy.ts +++ b/scripts/checks/no-product-tenancy.ts @@ -143,6 +143,18 @@ const ALLOWLIST: readonly { maxOccurrences: 1, tables: ["pending_seed"], }, + { + // `@corbits/mailbox`'s enrichment carries only priority, + // classification and status — there is no column for a snooze's + // `until` timestamp, so the reopen sweep has nothing to scan. This + // table is that one column, keyed to the same + // (tenantId, principalId, messageId) scope every mailbox mutation + // already uses, in its own `inbox` schema. Tenancy, principals and + // grants stay native (CL-7208). + relPath: "packages/inbox/src/schema.ts", + maxOccurrences: 1, + tables: ["snooze"], + }, { relPath: "packages/slack-tag/src/schema.ts", maxOccurrences: 1, diff --git a/scripts/db-setup.ts b/scripts/db-setup.ts index ad28cf56a..f8189d2d2 100644 --- a/scripts/db-setup.ts +++ b/scripts/db-setup.ts @@ -39,7 +39,10 @@ import { import { applyWebhookTriggersMigrations } from "../packages/webhook-triggers/src/migrations"; import { applyNotifyMigrations } from "../packages/notify/src/migrations"; import { applyRoutineMigrations } from "../packages/routines/src/migrations"; -import { applyMailboxMigrations } from "../packages/inbox/src/migrations"; +import { + applyInboxMigrations, + applyMailboxMigrations, +} from "../packages/inbox/src/migrations"; import { applyInsightsMigrations } from "../packages/insights/src/migrations"; import { applyPreferencesMigrations } from "../packages/preferences/src/migrations"; import { applyBenchMigrations } from "../packages/bench/src/migrations"; @@ -76,6 +79,8 @@ const INSTALLED_PACKAGE_MIGRATIONS: readonly { { name: "@corbits/routines", apply: applyRoutineMigrations }, { name: "@corbits/notify", apply: applyNotifyMigrations }, { name: "@corbits/mailbox", apply: applyMailboxMigrations }, + // CL-7208's snooze-until table, own schema — see packages/inbox/src/schema.ts. + { name: "@corbits/inbox", apply: applyInboxMigrations }, { name: "@corbits/insights", apply: applyInsightsMigrations }, { name: "@corbits/preferences", apply: applyPreferencesMigrations }, { name: "@corbits/bench", apply: applyBenchMigrations },