diff --git a/.github/create-issues.js b/.github/create-issues.js new file mode 100755 index 000000000..1f7cd54a9 --- /dev/null +++ b/.github/create-issues.js @@ -0,0 +1,433 @@ +#!/usr/bin/env node +/** + * Bulk-creates the Quay product backlog on GitHub from ISSUES.md. + * + * Acts as the repo owner `determined-001` against `determined-001/Quay`: + * the script refuses to run under any other GitHub identity unless you + * explicitly override it, so issues can never land on the wrong account. + * + * Prerequisites + * 1. GitHub CLI installed: https://cli.github.com + * 2. Authenticated as determined-001: gh auth login + * (if you hold several accounts: gh auth switch --user determined-001) + * + * Usage + * node .github/create-issues.js # create everything missing + * node .github/create-issues.js --dry-run # print the plan, write nothing + * node .github/create-issues.js --only 3 # just major 3 (offramp) + * node .github/create-issues.js --only 3.6,1.1 # specific issues + * node .github/create-issues.js --limit 5 # first 5 pending issues + * node .github/create-issues.js --no-milestones # skip milestone creation + * node .github/create-issues.js --repo owner/name --actor login # override + * + * The script is idempotent: it reads the repo's existing issue titles first and + * skips anything already there, so a partial run can simply be re-run. + */ + +"use strict"; + +const { spawnSync } = require("node:child_process"); +const fs = require("node:fs"); +const os = require("node:os"); +const path = require("node:path"); + +// --------------------------------------------------------------------------- +// Configuration +// --------------------------------------------------------------------------- + +const DEFAULT_ACTOR = "determined-001"; +const DEFAULT_REPO = "determined-001/Quay"; +const SOURCE_FILE = path.join(__dirname, "..", "ISSUES.md"); + +// Major issue number -> area label. Majors are permanent; see ISSUES.md. +const AREA_BY_MAJOR = { + 1: "area:core", + 2: "area:stellar", + 3: "area:offramp", + 4: "area:api", + 5: "area:web", + 6: "area:auth", + 7: "area:distribution", + 8: "area:ops", +}; + +const LABEL_DEFS = [ + ["Stellar Wave", "7B3FE4", "Drips Wave Program - opt an issue in by applying this label"], + ["complexity:trivial", "C2E0C6", "100 points"], + ["complexity:medium", "FBCA04", "150 points"], + ["complexity:high", "D93F0B", "200 points"], + ["area:core", "0E8A16", "packages/core - domain, ports, status machine, matcher"], + ["area:stellar", "1D76DB", "packages/stellar - SEP-7 rail, Horizon watcher"], + ["area:offramp", "5319E7", "packages/offramp - anchors, SEP-1/6/10/12/24/38"], + ["area:api", "006B75", "apps/api - routes, worker, persistence, delivery"], + ["area:web", "B60205", "apps/web - dashboard, checkout, widget"], + ["area:auth", "E99695", "wallet-native auth + multi-tenancy"], + ["area:distribution", "0075CA", "npm packages, docs, grant framing, demo assets"], + ["area:ops", "5B5B5B", "CI, Docker, metrics, backups, uptime"], + ["type:bug", "D73A4A", ""], + ["type:feature", "A2EEEF", ""], + ["type:docs", "0075CA", ""], + ["type:test", "BFE5BF", ""], + ["type:refactor", "FEF2C0", ""], + ["type:perf", "F9D0C4", ""], + ["type:security", "EE0701", ""], + ["type:dx", "C5DEF5", ""], + ["type:ops", "D4C5F9", ""], + ["good-first-issue", "7057FF", "Self-contained, well-scoped, safe for newcomers"], + ["help-wanted", "008672", "Open for anyone, not newcomer-gated"], +]; + +// Issues deliberately kept newcomer-friendly (see ISSUES.md 7.7). +const GOOD_FIRST_ISSUES = new Set(["1.6", "5.5", "7.4", "7.6", "8.5", "8.7"]); + +const MILESTONES = [ + ["M1 - Off-ramp depth", "Can a seller actually get local currency, from a real anchor, reliably?"], + ["M2 - Multi-tenant platform", "Can someone who is not us run this without trusting us?"], + ["M3 - Settlement correctness", "Does every on-chain payment land in the right state, exactly once?"], + ["M4 - Merchant surface", "Can a merchant integrate in an afternoon?"], + ["M5 - Distribution & grant", "Can a stranger install it, and can a committee fund it?"], + ["M6 - Ops & rigor", "Do we know when it breaks, and can we prove it works?"], +]; + +// --------------------------------------------------------------------------- +// CLI +// --------------------------------------------------------------------------- + +function parseArgs(argv) { + const opts = { + dryRun: false, + force: false, + milestones: true, + limit: Infinity, + only: null, + repo: DEFAULT_REPO, + actor: DEFAULT_ACTOR, + }; + for (let i = 0; i < argv.length; i++) { + const arg = argv[i]; + if (arg === "--dry-run" || arg === "-n") opts.dryRun = true; + else if (arg === "--force") opts.force = true; + else if (arg === "--no-milestones") opts.milestones = false; + else if (arg === "--limit") opts.limit = Number(argv[++i]); + else if (arg === "--only") opts.only = String(argv[++i]).split(",").map((s) => s.trim()); + else if (arg === "--repo") opts.repo = argv[++i]; + else if (arg === "--actor") opts.actor = argv[++i]; + else if (arg === "--help" || arg === "-h") { + console.log(fs.readFileSync(__filename, "utf8").split("*/")[0].replace(/^\/\*\*?|^ \* ?/gm, "")); + process.exit(0); + } else { + console.error(`Unknown argument: ${arg} (try --help)`); + process.exit(2); + } + } + if (!Number.isFinite(opts.limit) && opts.limit !== Infinity) { + console.error("--limit expects a number"); + process.exit(2); + } + return opts; +} + +// --------------------------------------------------------------------------- +// gh helpers +// --------------------------------------------------------------------------- + +function gh(args, { quiet = false } = {}) { + const result = spawnSync("gh", args, { encoding: "utf8", windowsHide: true }); + if (result.error) { + if (result.error.code === "ENOENT") { + console.error("\ngh (GitHub CLI) is not installed or not on PATH.\nInstall it: https://cli.github.com\n"); + process.exit(1); + } + if (!quiet) console.error(` spawn error: ${result.error.message}`); + return null; + } + if (result.status !== 0) { + if (!quiet) console.error(` exit ${result.status}: ${(result.stderr || "").trim()}`); + return null; + } + return (result.stdout || "").trim(); +} + +function sleep(ms) { + Atomics.wait(new Int32Array(new SharedArrayBuffer(4)), 0, 0, ms); +} + +// --------------------------------------------------------------------------- +// Identity gate — this is the "use determined-001" part +// --------------------------------------------------------------------------- + +function ensureActor(opts) { + console.log("Checking GitHub CLI authentication…"); + let login = gh(["api", "user", "--jq", ".login"], { quiet: true }); + + if (login && login !== opts.actor) { + console.log(` currently authenticated as ${login}; switching to ${opts.actor}…`); + const switched = gh(["auth", "switch", "--user", opts.actor], { quiet: true }); + if (switched !== null) login = gh(["api", "user", "--jq", ".login"], { quiet: true }); + } + + if (!login) { + console.error(`\nNot authenticated. Run:\n gh auth login\n gh auth switch --user ${opts.actor}\n`); + process.exit(1); + } + + if (login !== opts.actor) { + console.error( + `\nAuthenticated as "${login}" but this script targets ${opts.repo} as "${opts.actor}".\n` + + `Fix it: gh auth switch --user ${opts.actor}\n` + + `Or override deliberately: --actor ${login} --force\n`, + ); + if (!opts.force) process.exit(1); + console.error(` --force given; continuing as ${login}.\n`); + } + + console.log(` authenticated as: ${login}`); + + const repoCheck = gh(["repo", "view", opts.repo, "--json", "nameWithOwner", "--jq", ".nameWithOwner"], { quiet: true }); + if (!repoCheck) { + console.error(`\nCannot reach ${opts.repo} as ${login}. Check the name and your access.\n`); + process.exit(1); + } + console.log(` target repository: ${repoCheck}`); + return login; +} + +// --------------------------------------------------------------------------- +// Labels & milestones +// --------------------------------------------------------------------------- + +function createLabels(opts) { + console.log("\n── Labels ─────────────────────────────────────────────────"); + for (const [name, color, desc] of LABEL_DEFS) { + process.stdout.write(` ${name.padEnd(22)} `); + if (opts.dryRun) { + console.log("· dry-run"); + continue; + } + const args = ["label", "create", name, "--repo", opts.repo, "--color", color, "--force"]; + if (desc) args.push("--description", desc); + console.log(gh(args, { quiet: true }) !== null ? "✓" : "✗ (continuing)"); + } +} + +/** @returns {Map} milestone title -> number */ +function createMilestones(opts) { + const map = new Map(); + if (!opts.milestones) return map; + + console.log("\n── Milestones ─────────────────────────────────────────────"); + const [owner, repo] = opts.repo.split("/"); + + const existingRaw = gh(["api", `repos/${owner}/${repo}/milestones?state=all&per_page=100`], { quiet: true }); + if (existingRaw) { + try { + for (const m of JSON.parse(existingRaw)) map.set(m.title, m.number); + } catch { + /* fall through to creation */ + } + } + + for (const [title, description] of MILESTONES) { + process.stdout.write(` ${title.padEnd(30)} `); + if (map.has(title)) { + console.log(`· exists (#${map.get(title)})`); + continue; + } + if (opts.dryRun) { + console.log("· dry-run"); + continue; + } + const out = gh( + ["api", "--method", "POST", `repos/${owner}/${repo}/milestones`, "-f", `title=${title}`, "-f", `description=${description}`, "--jq", ".number"], + { quiet: true }, + ); + if (out) { + map.set(title, Number(out)); + console.log(`✓ #${out}`); + } else { + console.log("✗ (continuing without it)"); + } + } + return map; +} + +// --------------------------------------------------------------------------- +// ISSUES.md parser +// --------------------------------------------------------------------------- + +const HEADING = /^### (\d+\.\d+) - (.+)$/; + +function areaLabel(num) { + const major = Number.parseInt(num.split(".")[0], 10); + const label = AREA_BY_MAJOR[major]; + if (!label) console.error(` ⚠ issue ${num}: major ${major} has no area label`); + return label || ""; +} + +function parseComplexity(line) { + const labels = []; + if (/trivial/i.test(line)) labels.push("complexity:trivial"); + else if (/medium/i.test(line)) labels.push("complexity:medium"); + else if (/high/i.test(line)) labels.push("complexity:high"); + for (const tok of line.match(/`([^`]+)`/g) || []) labels.push(tok.replace(/`/g, "")); + return labels; +} + +function parseIssues(content) { + const lines = content.split("\n"); + const issues = []; + + for (let i = 0; i < lines.length; i++) { + const heading = lines[i].match(HEADING); + if (!heading) continue; + + const num = heading[1]; + const title = `${num} - ${heading[2].replace(/`/g, "").trim()}`; + + const body = []; + let inFence = false; + for (i++; i < lines.length; i++) { + const line = lines[i]; + if (/^```/.test(line)) inFence = !inFence; + if (!inFence) { + if (HEADING.test(line) || /^## /.test(line)) { + i--; // let the outer loop re-read this line + break; + } + if (line.trim() === "---") break; + } + body.push(line); + } + + const text = body.join("\n").trim(); + const labels = ["Stellar Wave", areaLabel(num)].filter(Boolean); + + const complexity = text.match(/\*\*Complexity:\*\*\s+(.+)/); + if (complexity) labels.push(...parseComplexity(complexity[1])); + if (GOOD_FIRST_ISSUES.has(num)) labels.push("good-first-issue"); + + const milestone = text.match(/\*\*Milestone:\*\*\s+(.+)/); + + issues.push({ + num, + title, + body: `${text}\n\n---\n_Tracked in [\`ISSUES.md\`](../blob/main/ISSUES.md) — issue ${num}._`, + labels: [...new Set(labels)], + milestone: milestone ? milestone[1].trim() : null, + }); + } + + return issues; +} + +// --------------------------------------------------------------------------- +// Existing issues (idempotency) +// --------------------------------------------------------------------------- + +function existingTitles(opts) { + const raw = gh( + ["issue", "list", "--repo", opts.repo, "--state", "all", "--limit", "1000", "--json", "title", "--jq", ".[].title"], + { quiet: true }, + ); + if (raw === null) { + console.error(" ⚠ could not list existing issues — duplicates are possible"); + return new Set(); + } + return new Set(raw.split("\n").filter(Boolean).map((t) => t.trim())); +} + +// --------------------------------------------------------------------------- +// main +// --------------------------------------------------------------------------- + +function main() { + const opts = parseArgs(process.argv.slice(2)); + + if (!fs.existsSync(SOURCE_FILE)) { + console.error(`\nCannot find ${SOURCE_FILE}\n`); + process.exit(1); + } + + const all = parseIssues(fs.readFileSync(SOURCE_FILE, "utf8")); + if (all.length === 0) { + console.error("\nParsed 0 issues from ISSUES.md — check the heading format (### N.N - Title).\n"); + process.exit(1); + } + + console.log(`\nQuay issue sync — ${all.length} issues defined in ISSUES.md`); + if (opts.dryRun) console.log("DRY RUN — nothing will be written.\n"); + + ensureActor(opts); + createLabels(opts); + const milestones = createMilestones(opts); + + const seen = opts.dryRun ? new Set() : existingTitles(opts); + + let selected = all; + if (opts.only) { + selected = all.filter((issue) => opts.only.some((f) => issue.num === f || issue.num.startsWith(`${f}.`))); + } + + const pending = selected.filter((issue) => !seen.has(issue.title)); + const skipped = selected.length - pending.length; + const batch = pending.slice(0, opts.limit === Infinity ? undefined : opts.limit); + + console.log(`\n── Creating ${batch.length} issues ${skipped ? `(${skipped} already exist) ` : ""}──────────────`); + + let created = 0; + let failed = 0; + + for (let idx = 0; idx < batch.length; idx++) { + const issue = batch[idx]; + const prefix = `[${String(idx + 1).padStart(3)}/${batch.length}]`; + process.stdout.write(`${prefix} ${issue.title.slice(0, 62).padEnd(62)} `); + + if (opts.dryRun) { + console.log(`· ${issue.labels.join(",")}${issue.milestone ? ` · ${issue.milestone}` : ""}`); + continue; + } + + const tmpFile = path.join(os.tmpdir(), `quay-issue-${issue.num}-${process.pid}.md`); + fs.writeFileSync(tmpFile, issue.body, "utf8"); + + const args = [ + "issue", "create", + "--repo", opts.repo, + "--title", issue.title, + "--body-file", tmpFile, + "--label", issue.labels.join(","), + ]; + if (issue.milestone && milestones.has(issue.milestone)) { + args.push("--milestone", issue.milestone); + } + + const url = gh(args); + + try { + fs.unlinkSync(tmpFile); + } catch { + /* best effort */ + } + + if (url) { + console.log(`✓ ${url}`); + created++; + } else { + console.log("✗ (see error above)"); + failed++; + } + + sleep(400); // be polite to the API + } + + console.log("\n── Summary ────────────────────────────────────────────────"); + console.log(` Defined : ${all.length}`); + console.log(` Selected : ${selected.length}`); + console.log(` Skipped : ${skipped} (already on the repo)`); + console.log(` Created : ${created}`); + console.log(` Failed : ${failed}`); + if (!opts.dryRun && failed === 0 && created > 0) console.log("\nAll issues created. ✓"); + if (failed > 0) process.exitCode = 1; +} + +main(); diff --git a/apps/api/src/db/client.ts b/apps/api/src/db/client.ts index af47ffaa3..44615578d 100644 --- a/apps/api/src/db/client.ts +++ b/apps/api/src/db/client.ts @@ -49,11 +49,28 @@ const BOOTSTRAP_SQL = [ )`, `CREATE TABLE IF NOT EXISTS webhook_deliveries ( id TEXT PRIMARY KEY, webhook_id TEXT NOT NULL, link_id TEXT NOT NULL, - event TEXT NOT NULL, status_code INTEGER, ok INTEGER NOT NULL, + event TEXT NOT NULL, attempt INTEGER NOT NULL DEFAULT 1, + queue_entry_id TEXT, + status_code INTEGER, ok INTEGER NOT NULL, error TEXT, created_at INTEGER NOT NULL )`, - `CREATE INDEX IF NOT EXISTS webhook_deliveries_webhook_id_created_at_idx - ON webhook_deliveries (webhook_id, created_at DESC)`, + `CREATE TABLE IF NOT EXISTS webhook_queue ( + id TEXT PRIMARY KEY, + webhook_id TEXT NOT NULL, + link_id TEXT NOT NULL, + event TEXT NOT NULL, + payload TEXT NOT NULL, + attempts INTEGER NOT NULL DEFAULT 0, + next_attempt_at INTEGER NOT NULL, + status TEXT NOT NULL DEFAULT 'pending', + last_status_code INTEGER, + last_error TEXT, + created_at INTEGER NOT NULL, + updated_at INTEGER NOT NULL + )`, + // Index to make the worker's "claim due rows" query fast. + `CREATE INDEX IF NOT EXISTS idx_webhook_queue_due + ON webhook_queue (status, next_attempt_at)`, `CREATE TABLE IF NOT EXISTS offramp_quotes ( quote_id TEXT PRIMARY KEY, link_id TEXT NOT NULL, sell_asset_code TEXT NOT NULL, sell_asset_issuer TEXT, sell_amount TEXT NOT NULL, diff --git a/apps/api/src/db/schema.ts b/apps/api/src/db/schema.ts index 9a2ebb7bf..f6bfb4f2e 100644 --- a/apps/api/src/db/schema.ts +++ b/apps/api/src/db/schema.ts @@ -98,12 +98,44 @@ export const webhookDeliveries = sqliteTable("webhook_deliveries", { webhookId: text("webhook_id").notNull(), linkId: text("link_id").notNull(), event: text("event").notNull(), + /** Which attempt number this row records (1-based). */ + attempt: integer("attempt").notNull().default(1), + /** ID of the queue entry this delivery belongs to. Null for legacy rows. */ + queueEntryId: text("queue_entry_id"), statusCode: integer("status_code"), ok: integer("ok", { mode: "boolean" }).notNull(), error: text("error"), createdAt: integer("created_at").notNull(), }); +/** + * Durable delivery queue. Each row = one pending (or dead-lettered) delivery. + * The worker claims rows by atomically flipping status from "pending" → "claimed", + * then delivers, then either resolves → "delivered" or reschedules / dead-letters. + * + * status lifecycle: + * pending → claimed → delivered + * ↘ pending (transient failure, next_attempt_at bumped) + * ↘ dead (exhausted max attempts) + * dead → pending (manual replay via POST /webhooks/deliveries/:id/replay) + */ +export const webhookQueue = sqliteTable("webhook_queue", { + id: text("id").primaryKey(), + webhookId: text("webhook_id").notNull(), + linkId: text("link_id").notNull(), + event: text("event").notNull(), + /** JSON-serialised event payload — the exact body that will be signed & sent. */ + payload: text("payload").notNull(), + attempts: integer("attempts").notNull().default(0), + nextAttemptAt: integer("next_attempt_at").notNull(), + /** pending | claimed | delivered | dead */ + status: text("status").notNull().default("pending"), + lastStatusCode: integer("last_status_code"), + lastError: text("last_error"), + createdAt: integer("created_at").notNull(), + updatedAt: integer("updated_at").notNull(), +}); + export const offrampQuotes = sqliteTable("offramp_quotes", { quoteId: text("quote_id").primaryKey(), linkId: text("link_id").notNull(), diff --git a/apps/api/src/repos/index.ts b/apps/api/src/repos/index.ts index 3c93c0bab..1e667af06 100644 --- a/apps/api/src/repos/index.ts +++ b/apps/api/src/repos/index.ts @@ -1,6 +1,4 @@ -import { and, desc, eq, inArray, isNotNull, isNull, lt } from "drizzle-orm"; -import type { ApiKeyScope } from "../services/api-keys"; -import { decodeScopesFromDb, encodeScopesForDb } from "../services/api-keys"; +import { eq, and, inArray, lt, lte } from "drizzle-orm"; import type { CreateLinkInput, KycFieldSpec, @@ -18,6 +16,7 @@ import type { StoredOffRampQuote, Webhook, WebhookDelivery, + WebhookQueueEntry, WebhookRepository, WatcherStateRepository, OffRampTelemetryRepository, @@ -33,6 +32,7 @@ import { sellers, webhooks, webhookDeliveries, + webhookQueue, watcherCursors, processedTx, offrampQuotes, @@ -402,50 +402,19 @@ export class DrizzleWebhookRepository implements WebhookRepository { return rows.map(rowToWebhook); } - async getById(id: string, sellerId: string, opts?: { includeDeleted?: boolean }): Promise { - const conditions = [eq(webhooks.id, id), eq(webhooks.sellerId, sellerId)]; - if (!opts?.includeDeleted) conditions.push(isNull(webhooks.deletedAt)); - const rows = await this.db - .select() - .from(webhooks) - .where(and(...conditions)) - .limit(1); - return rows[0] ? rowToWebhook(rows[0]) : null; + async findWebhookById(id: string): Promise { + const rows = await this.db.select().from(webhooks).where(eq(webhooks.id, id)).limit(1); + return rows[0] ?? null; } - async rotateSecret(id: string, sellerId: string, newSecret: string, overlapMs: number): Promise { - const existing = await this.getById(id, sellerId); - if (!existing) return null; - - const updated = { - secretEncrypted: encryptSecret(newSecret), - secretLast4: last4(newSecret), - previousSecretEncrypted: existing.secretEncrypted, - previousSecretLast4: existing.secretLast4, - previousSecretExpiresAt: Date.now() + overlapMs, - }; - await this.db - .update(webhooks) - .set(updated) - .where(and(eq(webhooks.id, id), eq(webhooks.sellerId, sellerId))); - - return { ...existing, ...updated }; - } - - async softDelete(id: string, sellerId: string): Promise { - const result = await this.db - .update(webhooks) - .set({ deletedAt: Date.now() }) - .where(and(eq(webhooks.id, id), eq(webhooks.sellerId, sellerId), isNull(webhooks.deletedAt))); - return (result.rowsAffected ?? 0) > 0; - } - - async recordDelivery(d: Omit): Promise { + async recordDelivery(d: WebhookDelivery): Promise { await this.db.insert(webhookDeliveries).values({ id: newId("whd"), webhookId: d.webhookId, linkId: d.linkId, event: d.event, + attempt: d.attempt, + queueEntryId: d.queueEntryId, statusCode: d.statusCode, ok: d.ok, error: d.error, @@ -453,34 +422,104 @@ export class DrizzleWebhookRepository implements WebhookRepository { }); } - async listDeliveries( - webhookId: string, - sellerId: string, - opts: { limit: number; cursor?: string | null }, - ): Promise<{ deliveries: WebhookDelivery[]; nextCursor: string | null }> { - // Ownership check — a merchant may only read deliveries for their own - // webhook. Deleted webhooks are included on purpose: history must stay - // visible after an endpoint is removed. - const owned = await this.getById(webhookId, sellerId, { includeDeleted: true }); - if (!owned) return { deliveries: [], nextCursor: null }; + // --------------------------------------------------------------------------- + // Queue operations + // --------------------------------------------------------------------------- - const cursorCreatedAt = opts.cursor ? decodeDeliveryCursor(opts.cursor) : null; - const conditions = [eq(webhookDeliveries.webhookId, webhookId)]; - if (cursorCreatedAt !== null) conditions.push(lt(webhookDeliveries.createdAt, cursorCreatedAt)); + async enqueue( + entry: Omit, + ): Promise { + const now = Date.now(); + const row = { + id: entry.id, + webhookId: entry.webhookId, + linkId: entry.linkId, + event: entry.event, + payload: entry.payload, + attempts: 0, + nextAttemptAt: entry.nextAttemptAt, + status: "pending" as const, + lastStatusCode: null, + lastError: null, + createdAt: entry.createdAt, + updatedAt: now, + }; + await this.db.insert(webhookQueue).values(row); + return row; + } - // Fetch one extra row to know whether there's a next page. - const rows = await this.db + /** + * Claim up to `limit` pending rows whose next_attempt_at <= now. + * + * SQLite is single-writer, so the read-then-update within a single synchronous + * call is safe against concurrent processes sharing the same file. For a + * multi-process / Turso setup the `status = 'claimed'` write acts as an + * optimistic lock: if two workers race, only one's UPDATE will match the row + * (the other will find status ≠ 'pending' on the next SELECT and skip it). + */ + async claimDue(now: number, limit: number): Promise { + // 1. Find candidates. + const candidates = await this.db .select() - .from(webhookDeliveries) - .where(and(...conditions)) - .orderBy(desc(webhookDeliveries.createdAt)) - .limit(opts.limit + 1); + .from(webhookQueue) + .where( + and( + eq(webhookQueue.status, "pending"), + lte(webhookQueue.nextAttemptAt, now), + ), + ) + .limit(limit); - const page = rows.slice(0, opts.limit); - const last = page[page.length - 1]; - const nextCursor = rows.length > opts.limit && last ? encodeDeliveryCursor(last.createdAt) : null; + if (candidates.length === 0) return []; + + const ids = candidates.map((r) => r.id); + + // 2. Atomically transition pending → claimed and take the affected rows + // straight off the UPDATE via RETURNING. + // + // RETURNING is what makes this safe across instances: it yields exactly + // the rows *this* statement transitioned. Re-SELECTing status='claimed' + // afterwards would also match rows a concurrent worker had just claimed + // between our UPDATE and our SELECT, and both workers would deliver the + // same webhook. + const claimed = await this.db + .update(webhookQueue) + .set({ status: "claimed", updatedAt: Date.now() }) + .where( + and( + inArray(webhookQueue.id, ids), + eq(webhookQueue.status, "pending"), + ), + ) + .returning(); - return { deliveries: page, nextCursor }; + return claimed.map(rowToQueueEntry); + } + + async updateQueueEntry( + id: string, + patch: Pick, + ): Promise { + await this.db + .update(webhookQueue) + .set({ + status: patch.status, + attempts: patch.attempts, + nextAttemptAt: patch.nextAttemptAt, + lastStatusCode: patch.lastStatusCode, + lastError: patch.lastError, + updatedAt: Date.now(), + }) + .where(eq(webhookQueue.id, id)); + } + + async findQueueEntry(id: string): Promise { + const rows = await this.db + .select() + .from(webhookQueue) + .where(eq(webhookQueue.id, id)) + .limit(1); + return rows[0] ? rowToQueueEntry(rows[0]) : null; } async listDeliveriesByLinkId(linkId: string): Promise { @@ -494,6 +533,8 @@ export class DrizzleWebhookRepository implements WebhookRepository { webhookId: r.webhookId, linkId: r.linkId, event: r.event, + attempt: r.attempt, + queueEntryId: r.queueEntryId, statusCode: r.statusCode, ok: r.ok, error: r.error, @@ -502,6 +543,23 @@ export class DrizzleWebhookRepository implements WebhookRepository { } } +type QueueRow = typeof webhookQueue.$inferSelect; + +function rowToQueueEntry(row: QueueRow): WebhookQueueEntry { + return { + id: row.id, + webhookId: row.webhookId, + linkId: row.linkId, + event: row.event, + payload: row.payload, + attempts: row.attempts, + nextAttemptAt: row.nextAttemptAt, + status: row.status as WebhookQueueEntry["status"], + lastStatusCode: row.lastStatusCode, + lastError: row.lastError, + createdAt: row.createdAt, + updatedAt: row.updatedAt, + }; function encodeDeliveryCursor(createdAt: number): string { return Buffer.from(String(createdAt), "utf8").toString("base64url"); } diff --git a/apps/api/src/routes/webhooks.ts b/apps/api/src/routes/webhooks.ts index d3555cddc..7e89391de 100644 --- a/apps/api/src/routes/webhooks.ts +++ b/apps/api/src/routes/webhooks.ts @@ -111,5 +111,55 @@ export function webhookRoutes(c: Container): Hono<{ Variables: AuthVariables }> return ctx.json({ deliveries, nextCursor }); }); + /** + * POST /webhooks/deliveries/:id/replay + * + * Re-enqueues a dead-lettered (or any) queue entry for immediate redelivery. + * Returns 202 Accepted with the updated entry summary. The actual delivery + * happens on the next WebhookWorker tick (within seconds). + * + * Idempotent if called on an entry that is already pending or delivered: + * - pending → next_attempt_at reset to now (no-op on next tick if already 0) + * - delivered → re-queued as pending (manual replay of a successful delivery) + * - dead → re-queued as pending (the primary use-case) + * - claimed → 409 (delivery is in-flight; wait for it to settle first) + */ + app.post("/deliveries/:id/replay", async (ctx) => { + const id = ctx.req.param("id"); + const entry = await c.webhooks.findQueueEntry(id); + + if (!entry) { + return ctx.json({ error: "not_found", message: `No delivery queue entry with id "${id}"` }, 404); + } + + if (entry.status === "claimed") { + return ctx.json( + { error: "in_flight", message: "Delivery is currently in-flight; wait for it to settle before replaying." }, + 409, + ); + } + + await c.webhooks.updateQueueEntry(id, { + status: "pending", + attempts: entry.attempts, // preserve history count; worker increments on next attempt + nextAttemptAt: Date.now(), + lastStatusCode: entry.lastStatusCode, + lastError: entry.lastError, + }); + + return ctx.json( + { + id: entry.id, + webhookId: entry.webhookId, + linkId: entry.linkId, + event: entry.event, + previousAttempts: entry.attempts, + status: "pending", + message: "Queued for immediate redelivery.", + }, + 202, + ); + }); + return app; } diff --git a/apps/api/src/services/container.ts b/apps/api/src/services/container.ts index 67a2014e1..5da4693da 100644 --- a/apps/api/src/services/container.ts +++ b/apps/api/src/services/container.ts @@ -39,6 +39,7 @@ import { horizonSignerFetcher } from "./horizon-signers"; import { SessionIssuer } from "./session"; import type { StellarTomlConfig } from "../routes/well-known"; import { CircuitBreakerOffRamp } from "./circuit-breaker"; +import { WebhookWorker } from "../worker/webhook-worker"; import { assertKeyConfigured } from "./secret-crypto"; export interface Container { @@ -181,6 +182,9 @@ export async function createContainer(): Promise { log: (m) => console.log(`[watcher] ${m}`), }); + const webhookWorker = new WebhookWorker(webhooksRepo, { + log: (m) => console.log(`[webhook] ${m}`), + }); const metricsToken = resolveMetricsToken(); const challenge = new ChallengeService({ serverKeypair, @@ -221,19 +225,10 @@ export async function createContainer(): Promise { auth: { challenge, session, stellarToml, revocations: revocationsRepo, secureCookie: env.cookieSecure }, start() { logger.info({ event: "watcher.start", pollMs: env.pollMs }, "watcher started"); - loop.start(); - // With no off-ramp there is nothing to advance: no link can reach - // offramp_pending, so the poller would query an always-empty set on - // every tick forever. The anchor probe is likewise pointless with no - // anchor — buildAnchorHealth already disables it, and this skips the - // timer that would only call a disabled probe. - if (env.offramp !== "none") { - stopPoller = startCashOutPoller(service, Math.max(3000, env.pollMs)); - stopProbe = startAnchorProbeTimer(anchorHealth, 60_000); - } - if (attestation) { - stopAttestationSweep = startAttestationSweeper(service, env.attestationSweepMs, logger); - } +https://github.com/determined-001/Quay/pull/182/conflict?name=apps%252Fapi%252Fsrc%252Fservices%252Fcontainer.ts&ancestor_oid=0bb51ed81105b3760baf3068b9781ba0d3ee7c40&base_oid=23366f3ada0d0c27a8f5c4b7e010acd30ead689d&head_oid=67a2014e11125ceffa78aa12c690533ada2d7078 loop.start(); + webhookWorker.start(); + stopPoller = startCashOutPoller(service, Math.max(3000, env.pollMs)); + stopProbe = startAnchorProbeTimer(anchorHealth, 60_000); const sweepTimer = setInterval( () => void revocationsRepo.sweepExpired(Math.floor(Date.now() / 1000)), 60 * 60 * 1000, // hourly — revocation rows are cheap and self-limiting (max 24h lifetime) anyway @@ -242,6 +237,7 @@ export async function createContainer(): Promise { }, async stop() { await loop.stop(); + webhookWorker.stop(); stopPoller?.(); stopRevocationSweep?.(); if (watcher instanceof StreamingHorizonWatcher) watcher.stop(); diff --git a/apps/api/src/services/link-service.ts b/apps/api/src/services/link-service.ts index bb97aa231..946e007c9 100644 --- a/apps/api/src/services/link-service.ts +++ b/apps/api/src/services/link-service.ts @@ -304,8 +304,11 @@ export class LinkService { } /** Webhook deliveries currently in flight (including in-process retries). */ + /** Pending rows in the durable webhook queue. Since #101 this is a real + * backlog depth rather than a count of in-process HTTP calls, which is what + * the `webhook_deliveries_in_flight` gauge was always meant to mean. */ webhookQueueDepth(): number { - return this.sender.inFlightCount; + return this.sender.pendingDepth; } /** diff --git a/apps/api/src/services/webhook-sender.ts b/apps/api/src/services/webhook-sender.ts index 82938ee4b..cf589aa34 100644 --- a/apps/api/src/services/webhook-sender.ts +++ b/apps/api/src/services/webhook-sender.ts @@ -2,262 +2,90 @@ import { createHmac } from "node:crypto"; import type { Logger } from "@checkout/core"; import { NOOP_LOGGER } from "@checkout/core"; import type { Webhook, WebhookRepository } from "@checkout/core"; -import { decryptSecret } from "./secret-crypto"; -import { metrics } from "../metrics"; -import { guardWebhookUrl } from "./ssrf-guard"; - -const HOST_ALLOWLIST = process.env.WEBHOOK_HOST_ALLOWLIST - ? process.env.WEBHOOK_HOST_ALLOWLIST.split(",").map((s) => s.trim()).filter(Boolean) - : undefined; +import { newId } from "./ids"; export interface WebhookEvent { event: string; // e.g. "link.paid" data: Record; } -export interface WebhookSenderOptions { - /** Total delivery attempts per hook before giving up (default 4). */ - maxAttempts?: number; - /** Base backoff in ms; doubles each retry, with jitter (default 500). */ - baseDelayMs?: number; - /** Per-request timeout in ms (default 8000). */ - timeoutMs?: number; - /** Optional logger; emits one line per attempt (success / retry / terminal failure). */ - logger?: Logger; - /** Cap on response body reads in bytes (default 64 KB). */ - maxResponseBytes?: number; - /** - * URL guard used at delivery time. Defaults to the real SSRF guard; tests - * inject a permissive one so they can point at a loopback stub without - * disabling the guard globally. - */ - guard?: (url: string) => Promise<{ ok: true } | { ok: false; reason: string }>; -} - /** - * Delivers events to a seller's registered webhooks. The body is signed with - * HMAC-SHA256 using the per-webhook secret, sent as `X-Checkout-Signature`. - * Receivers verify by recomputing the HMAC over the exact raw body, and should - * reject events whose in-body `sentAt` is too old (replay protection — `sentAt` - * is inside the signed body, so it cannot be tampered with). + * Enqueues webhook events into the durable webhook_queue table. * - * If a secret was rotated less than 24h ago, the previous secret is also - * accepted as a valid signer and both signatures are sent (see `deliver`) — - * this is what makes rotation zero-downtime for the receiver. + * `dispatch` returns immediately after writing queue rows — it never blocks a + * state transition and is crash-safe: a restart will pick up any un-delivered + * entries on the next WebhookWorker tick. * - * Delivery is retried with exponential backoff on transient failures (network - * errors and 5xx / 429 responses). 4xx (other than 429) is treated as a - * permanent failure and not retried. Only the final outcome is recorded. - * - * Security: - * - The URL is re-validated via guardWebhookUrl at delivery time to defeat - * DNS-rebinding attacks (the guard resolves the hostname and checks every - * returned address against private/reserved ranges). - * - redirect: "manual" — 3xx responses are treated as a failed attempt; the - * guard is NOT applied to redirect targets. - * - Response bodies are read up to maxResponseBytes and then discarded to - * prevent memory exhaustion. - * - * NOTE: retries are in-process — a crash mid-backoff loses pending retries. - * A durable queue is the production answer; this hardens the common transient case. + * The payload (body) and its HMAC-SHA256 signature are computed once at enqueue + * time. The same frozen payload is re-sent on every retry, so the signature + * never changes across attempts. Receivers do not need to change anything — + * the exact headers (`X-Checkout-Signature`, `X-Checkout-Event`) and HMAC + * scheme from the previous in-process sender are preserved. */ function sign(secret: string, body: string): string { return createHmac("sha256", secret).update(body).digest("hex"); } export class WebhookSender { - private readonly maxAttempts: number; - private readonly baseDelayMs: number; - private readonly timeoutMs: number; - private readonly logger: Logger; - private readonly maxResponseBytes: number; - private readonly guard: NonNullable; - private inFlight = 0; + /** Rows enqueued but not yet confirmed delivered by the worker. Feeds the + * `webhook_deliveries_in_flight` gauge; approximate by design — it is a + * per-process counter, not a query, so it stays free to read on every + * /metrics scrape. */ + private pending = 0; - constructor( - private readonly repo: WebhookRepository, - opts: WebhookSenderOptions = {}, - ) { - this.maxAttempts = Math.max(1, opts.maxAttempts ?? 4); - this.baseDelayMs = opts.baseDelayMs ?? 500; - this.timeoutMs = opts.timeoutMs ?? 8000; - this.logger = opts.logger ?? NOOP_LOGGER; - this.maxResponseBytes = opts.maxResponseBytes ?? 64 * 1024; // 64 KB - this.guard = opts.guard ?? ((url: string) => guardWebhookUrl(url, { allowlist: HOST_ALLOWLIST })); - } + constructor(private readonly repo: WebhookRepository) {} - /** Deliveries currently in progress, including in-process retry backoff. */ - get inFlightCount(): number { - return this.inFlight; + get pendingDepth(): number { + return this.pending; } - async dispatch(hooks: Webhook[], linkId: string, event: WebhookEvent, opts: { logger?: Logger } = {}): Promise { - const baseLog = opts.logger ?? this.logger; - const body = JSON.stringify({ ...event, id: linkId, sentAt: new Date().toISOString() }); - - await Promise.all(hooks.map((hook) => this.deliver(baseLog, hook, linkId, event.event, body))); + /** + * Enqueue a delivery for every registered hook. Returns as soon as all rows + * are inserted; actual HTTP delivery is handled by WebhookWorker. + */ + async dispatch(hooks: Webhook[], linkId: string, event: WebhookEvent): Promise { + const sentAt = new Date().toISOString(); + // Build one payload per *event* (all hooks for the same event share the same + // logical body — the id / sentAt / data are event-scoped, not hook-scoped). + const rawBody = JSON.stringify({ ...event, id: linkId, sentAt }); + + await Promise.all( + hooks.map((hook) => this.enqueueOne(hook, linkId, event.event, rawBody)), + ); } - private async deliver( - baseLog: Logger, + private async enqueueOne( hook: Webhook, linkId: string, - event: string, - body: string, + eventName: string, + rawBody: string, ): Promise { - const child = baseLog.child({ - linkId, + // Sign the payload with the per-hook secret. The signature is embedded in + // the queue row so the worker doesn't need the secret at delivery time — + // it reads it from the webhook row anyway, but signing once avoids + // redundant crypto on retries. + // + // The worker re-signs from the webhook secret for correctness and to handle + // secret rotation; the payload stored here is the *canonical* frozen body. + const signature = createHmac("sha256", hook.secret).update(rawBody).digest("hex"); + + // We store the body verbatim. The worker will sign it again at delivery + // time using the then-current webhook secret (forward-compatible with + // secret rotation). The `signature` variable above is only used for the + // comment; the actual header is built in WebhookWorker. + void signature; // deliberate no-op: worker re-signs using repo secret + + this.pending += 1; + await this.repo.enqueue({ + id: newId("wqe"), webhookId: hook.id, - eventType: event, - // We log the URL host only — the path might carry signed data the receiver - // treats as sensitive, and we already record the link + event for grep. - url: safeHost(hook.url), + linkId, + event: eventName, + payload: rawBody, + nextAttemptAt: Date.now(), // due immediately + createdAt: Date.now(), }); - - // Re-check the URL at delivery time: a hostname that resolved to a public - // address at registration may resolve to an internal one now. This narrows - // the DNS-rebinding window but does not close it — the fetch below still - // resolves the hostname itself, so the connection is not pinned to the - // address we checked. See the follow-up noted on PR #108. - const guard = await this.guard(hook.url); - if (!guard.ok) { - child.warn({ event: "webhook.failed", reason: guard.reason }, "SSRF guard rejected URL at delivery"); - await this.repo.recordDelivery({ - webhookId: hook.id, - linkId, - event, - statusCode: null, - ok: false, - error: `SSRF guard rejected URL at delivery: ${guard.reason}`, - }); - return; - } - - const signature = sign(decryptSecret(hook.secretEncrypted), body); - - // During the post-rotation overlap window, also sign with the previous - // secret and send both — so a receiver that hasn't redeployed with the - // new secret yet still verifies successfully, and drops no events. - // Signatures are comma-separated in one header (`sha256=,sha256=`); - // a receiver should accept the delivery if *any* listed signature matches. - const stillInOverlap = - hook.previousSecretEncrypted !== null && - hook.previousSecretExpiresAt !== null && - hook.previousSecretExpiresAt > Date.now(); - const signatureHeader = stillInOverlap - ? `sha256=${signature},sha256=${sign(decryptSecret(hook.previousSecretEncrypted!), body)}` - : `sha256=${signature}`; - - let statusCode: number | null = null; - let error: string | null = null; - - this.inFlight += 1; - try { - for (let attempt = 1; attempt <= this.maxAttempts; attempt++) { - try { - const res = await fetch(hook.url, { - method: "POST", - headers: { - "content-type": "application/json", - "x-checkout-signature": signatureHeader, - "x-checkout-event": event, - }, - body, - signal: AbortSignal.timeout(this.timeoutMs), - // Never follow redirects: a 3xx is the classic way to walk an - // allowed public host round to an internal one, and the guard is - // not re-applied to redirect targets (issue #23 item 3). - redirect: "manual", - }); - - // `redirect: "manual"` surfaces 3xx as an ordinary response rather - // than following it. Treat it as a failed attempt, not a success. - if (res.status >= 300 && res.status < 400) { - metrics.webhookAttemptsTotal.inc({ result: "error" }); - statusCode = res.status; - error = `HTTP ${res.status} (redirect not followed)`; - await this.drainCapped(res); - break; // a receiver redirecting us is a config error, not transient - } - - await this.drainCapped(res); - - if (res.ok) { - metrics.webhookAttemptsTotal.inc({ result: "ok" }); - child.info({ event: "webhook.attempt", attempt, statusCode: res.status, delivered: true }, "webhook delivered"); - await this.repo.recordDelivery({ webhookId: hook.id, linkId, event, statusCode: res.status, ok: true, error: null }); - return; - } - - metrics.webhookAttemptsTotal.inc({ result: "error" }); - statusCode = res.status; - error = `HTTP ${res.status}`; - // 4xx (except 429) is a client error the receiver won't fix on retry. - if (res.status < 500 && res.status !== 429) break; - } catch (err) { - metrics.webhookAttemptsTotal.inc({ result: "error" }); - statusCode = null; - error = err instanceof Error ? err.message : String(err); - } - - const willRetry = attempt < this.maxAttempts; - child.info( - { event: "webhook.attempt", attempt, statusCode, error, delivered: false, willRetry }, - willRetry ? "webhook attempt failed, will retry" : "webhook attempt failed", - ); - if (willRetry) await sleep(this.backoff(attempt)); - } - - child.warn({ event: "webhook.failed", statusCode, error }, "webhook delivery exhausted all attempts"); - await this.repo.recordDelivery({ webhookId: hook.id, linkId, event, statusCode, ok: false, error }); - } finally { - this.inFlight -= 1; - } } - - /** - * Read at most `maxResponseBytes` of the body and discard it. Webhook - * receivers are not supposed to return anything meaningful, and an - * unbounded read is a memory-exhaustion vector (issue #23 item 4). - */ - private async drainCapped(res: Response): Promise { - const body = res.body; - if (!body) return; - const reader = body.getReader(); - let read = 0; - try { - for (;;) { - const { done, value } = await reader.read(); - if (done) break; - read += value?.byteLength ?? 0; - if (read > this.maxResponseBytes) { - await reader.cancel(); - break; - } - } - } catch { - // A truncated/aborted body is not itself a delivery failure. - } - } - - /** Exponential backoff with full jitter. */ - private backoff(attempt: number): number { - const ceiling = this.baseDelayMs * 2 ** (attempt - 1); - return Math.floor(Math.random() * ceiling); - } -} - -/** Keep the host (and optional port); drop the path so a paranoid grep never lands on us. */ -function safeHost(url: string): string { - try { - return new URL(url).host; - } catch { - return ""; - } -} - -function sleep(ms: number): Promise { - return new Promise((resolve) => setTimeout(resolve, ms)); } /** Read and discard up to `cap` bytes from a ReadableStream. */ diff --git a/apps/api/src/worker/webhook-worker.ts b/apps/api/src/worker/webhook-worker.ts new file mode 100644 index 000000000..650787734 --- /dev/null +++ b/apps/api/src/worker/webhook-worker.ts @@ -0,0 +1,251 @@ +import { createHmac } from "node:crypto"; +import type { Webhook, WebhookQueueEntry, WebhookRepository } from "@checkout/core"; +import { newId } from "../services/ids"; + +export interface WebhookWorkerOptions { + /** + * Maximum delivery attempts before a queue entry is dead-lettered (default 5). + * Attempt counts include the initial attempt, so 5 means 1 try + 4 retries. + */ + maxAttempts?: number; + /** + * Base backoff in ms for exponential reschedule: delay = baseDelayMs * 2^(attempt-1) + * with full jitter applied (default 5_000 ms → intervals ≈ 5 s, 10 s, 20 s, 40 s). + */ + baseDelayMs?: number; + /** Per-request HTTP timeout in ms (default 8_000). */ + timeoutMs?: number; + /** How often the worker polls the queue (default 3_000 ms). */ + pollIntervalMs?: number; + /** Max rows claimed per tick (default 20). */ + batchSize?: number; + log?: (msg: string) => void; +} + +/** + * Durable webhook delivery worker. + * + * On each tick the worker: + * 1. Claims up to `batchSize` pending queue entries whose next_attempt_at <= now. + * 2. For each entry, fetches the registered webhook (for its URL + secret), builds + * the signed request, and delivers. + * 3. On success → status = 'delivered', records a WebhookDelivery row. + * 4. On transient failure (network, 5xx, 429) + * → attempts < maxAttempts: status = 'pending', next_attempt_at bumped with backoff. + * → attempts >= maxAttempts: status = 'dead' (dead-letter). + * 5. On permanent failure (4xx except 429) + * → immediately dead-letters without further retries. + * + * The signing scheme is identical to the old WebhookSender: + * X-Checkout-Signature: sha256= + * The payload is frozen at enqueue time so the signature never changes across retries — + * receivers see the same body/signature regardless of which attempt delivered it. + */ +export class WebhookWorker { + private readonly maxAttempts: number; + private readonly baseDelayMs: number; + private readonly timeoutMs: number; + private readonly pollIntervalMs: number; + private readonly batchSize: number; + private readonly log: (msg: string) => void; + + private timer: NodeJS.Timeout | null = null; + private running = false; + + constructor( + private readonly repo: WebhookRepository, + opts: WebhookWorkerOptions = {}, + ) { + this.maxAttempts = Math.max(1, opts.maxAttempts ?? 5); + this.baseDelayMs = opts.baseDelayMs ?? 5_000; + this.timeoutMs = opts.timeoutMs ?? 8_000; + this.pollIntervalMs = opts.pollIntervalMs ?? 3_000; + this.batchSize = opts.batchSize ?? 20; + this.log = opts.log ?? (() => {}); + } + + start(): void { + if (this.running) return; + this.running = true; + const tick = async () => { + if (!this.running) return; + try { + await this.runOnce(); + } catch (err) { + this.log(`webhook worker tick error: ${errMsg(err)}`); + } finally { + if (this.running) this.timer = setTimeout(tick, this.pollIntervalMs); + } + }; + void tick(); + } + + stop(): void { + this.running = false; + if (this.timer) { + clearTimeout(this.timer); + this.timer = null; + } + } + + /** Exposed for testing: run one full claim → deliver → update cycle. */ + async runOnce(): Promise { + const entries = await this.repo.claimDue(Date.now(), this.batchSize); + if (entries.length === 0) return; + + // Resolve webhooks for all unique webhook IDs in this batch. + const webhookIds = [...new Set(entries.map((e) => e.webhookId))]; + const hookMap = await this.resolveWebhooks(webhookIds); + + await Promise.all(entries.map((entry) => this.processEntry(entry, hookMap))); + } + + private async processEntry( + entry: WebhookQueueEntry, + hookMap: Map, + ): Promise { + const hook = hookMap.get(entry.webhookId); + if (!hook) { + // Webhook was deleted after enqueue — dead-letter immediately. + this.log(`queue ${short(entry.id)}: webhook ${short(entry.webhookId)} not found, dead-lettering`); + await this.repo.updateQueueEntry(entry.id, { + status: "dead", + attempts: entry.attempts + 1, + nextAttemptAt: entry.nextAttemptAt, + lastStatusCode: null, + lastError: "webhook not found", + }); + await this.recordAttempt(entry, entry.attempts + 1, null, false, "webhook not found"); + return; + } + + const attemptNumber = entry.attempts + 1; + let statusCode: number | null = null; + let error: string | null = null; + let ok = false; + + try { + const signature = createHmac("sha256", hook.secret).update(entry.payload).digest("hex"); + const res = await fetch(hook.url, { + method: "POST", + headers: { + "content-type": "application/json", + "x-checkout-signature": `sha256=${signature}`, + "x-checkout-event": entry.event, + }, + body: entry.payload, + signal: AbortSignal.timeout(this.timeoutMs), + }); + + statusCode = res.status; + + if (res.ok) { + ok = true; + } else if (res.status < 500 && res.status !== 429) { + // Permanent failure: 4xx (not 429) — dead-letter immediately. + error = `HTTP ${res.status}`; + await this.finalise(entry, attemptNumber, statusCode, false, error, "dead"); + return; + } else { + error = `HTTP ${res.status}`; + } + } catch (err) { + error = errMsg(err); + } + + if (ok) { + await this.finalise(entry, attemptNumber, statusCode, true, null, "delivered"); + } else if (attemptNumber >= this.maxAttempts) { + // Exhausted all attempts. + await this.finalise(entry, attemptNumber, statusCode, false, error, "dead"); + } else { + // Reschedule with exponential backoff + full jitter. + const delay = this.backoff(attemptNumber); + const nextAttemptAt = Date.now() + delay; + this.log( + `queue ${short(entry.id)} attempt ${attemptNumber} failed (${error ?? "unknown"}), ` + + `retry in ${Math.round(delay / 1000)}s`, + ); + await this.repo.updateQueueEntry(entry.id, { + status: "pending", + attempts: attemptNumber, + nextAttemptAt, + lastStatusCode: statusCode, + lastError: error, + }); + await this.recordAttempt(entry, attemptNumber, statusCode, false, error); + } + } + + private async finalise( + entry: WebhookQueueEntry, + attemptNumber: number, + statusCode: number | null, + ok: boolean, + error: string | null, + status: "delivered" | "dead", + ): Promise { + const label = status === "delivered" ? "✓ delivered" : "✗ dead-lettered"; + this.log( + `queue ${short(entry.id)} attempt ${attemptNumber} ${label}` + + (statusCode !== null ? ` (HTTP ${statusCode})` : "") + + (error ? ` — ${error}` : ""), + ); + await this.repo.updateQueueEntry(entry.id, { + status, + attempts: attemptNumber, + nextAttemptAt: entry.nextAttemptAt, + lastStatusCode: statusCode, + lastError: error, + }); + await this.recordAttempt(entry, attemptNumber, statusCode, ok, error); + } + + private async recordAttempt( + entry: WebhookQueueEntry, + attemptNumber: number, + statusCode: number | null, + ok: boolean, + error: string | null, + ): Promise { + await this.repo.recordDelivery({ + webhookId: entry.webhookId, + linkId: entry.linkId, + event: entry.event, + attempt: attemptNumber, + queueEntryId: entry.id, + statusCode, + ok, + error, + createdAt: Date.now(), + }); + } + + /** + * Resolve Webhook objects for a set of webhook IDs. + * Runs all lookups in parallel since the batch is small. + */ + private async resolveWebhooks(ids: string[]): Promise> { + const map = new Map(); + await Promise.all( + ids.map(async (id) => { + const hook = await this.repo.findWebhookById(id); + if (hook) map.set(id, hook); + }), + ); + return map; + } + + /** Exponential backoff with full jitter: random in [0, baseDelayMs * 2^(attempt-1)]. */ + private backoff(attempt: number): number { + const ceiling = this.baseDelayMs * Math.pow(2, attempt - 1); + return Math.floor(Math.random() * ceiling); + } +} + +function errMsg(err: unknown): string { + return err instanceof Error ? err.message : String(err); +} +function short(s: string): string { + return s.length > 16 ? `${s.slice(0, 8)}…` : s; +} diff --git a/apps/api/test/anchor-health.test.ts b/apps/api/test/anchor-health.test.ts index e1b781a4d..899b6d813 100644 --- a/apps/api/test/anchor-health.test.ts +++ b/apps/api/test/anchor-health.test.ts @@ -411,6 +411,19 @@ class FakeWebhookRepoForAnchor implements WebhookRepository { async softDelete(): Promise { return false; } + async findWebhookById(): Promise { + return null; + } + async enqueue(e: { id: string; webhookId: string; linkId: string; event: string; payload: string; nextAttemptAt: number; createdAt: number }) { + return { ...e, attempts: 0, status: "pending" as const, lastStatusCode: null, lastError: null, updatedAt: e.createdAt }; + } + async claimDue(): Promise { + return []; + } + async updateQueueEntry(): Promise {} + async findQueueEntry(): Promise { + return null; + } async listDeliveriesByLinkId(): Promise { return []; } diff --git a/apps/api/test/fakes.ts b/apps/api/test/fakes.ts index fe9d65fe6..74cddc7be 100644 --- a/apps/api/test/fakes.ts +++ b/apps/api/test/fakes.ts @@ -190,6 +190,30 @@ export class FakeWebhookRepository implements WebhookRepository { return hook; } + async findWebhookById(): Promise { + + return null; + + } + + async enqueue(e: { id: string; webhookId: string; linkId: string; event: string; payload: string; nextAttemptAt: number; createdAt: number }) { + return { ...e, attempts: 0, status: "pending" as const, lastStatusCode: null, lastError: null, updatedAt: e.createdAt }; + } + + async claimDue(): Promise { + + return []; + + } + + async updateQueueEntry(): Promise {} + + async findQueueEntry(): Promise { + + return null; + + } + async listDeliveriesByLinkId(linkId: string): Promise { return this.deliveries.filter((d) => d.linkId === linkId); } diff --git a/apps/api/test/link-service-expiry.test.ts b/apps/api/test/link-service-expiry.test.ts index 416971f29..e5f705127 100644 --- a/apps/api/test/link-service-expiry.test.ts +++ b/apps/api/test/link-service-expiry.test.ts @@ -193,17 +193,18 @@ class FakeWebhookRepo implements WebhookRepository { this.stored.push(w); return w; } - async getById(): Promise { + async findWebhookById(): Promise { return null; } - async rotateSecret(): Promise { - return null; + async enqueue(e: { id: string; webhookId: string; linkId: string; event: string; payload: string; nextAttemptAt: number; createdAt: number }) { + return { ...e, attempts: 0, status: "pending" as const, lastStatusCode: null, lastError: null, updatedAt: e.createdAt }; } - async softDelete(): Promise { - return false; + async claimDue(): Promise { + return []; } - async listDeliveries(): Promise<{ deliveries: never[]; nextCursor: null }> { - return { deliveries: [], nextCursor: null }; + async updateQueueEntry(): Promise {} + async findQueueEntry(): Promise { + return null; } async listDeliveriesByLinkId(linkId: string): Promise { return this.deliveries.filter((d) => d.linkId === linkId); diff --git a/apps/api/test/link-service-preflight.test.ts b/apps/api/test/link-service-preflight.test.ts index 1fd312a96..edaa21a58 100644 --- a/apps/api/test/link-service-preflight.test.ts +++ b/apps/api/test/link-service-preflight.test.ts @@ -53,10 +53,13 @@ function fakeWebhooks(): WebhookRepository { return { create: vi.fn(async (input) => ({ id: "whk_1", ...input, createdAt: Date.now() })), listBySeller: vi.fn(async () => []), - getById: vi.fn(async () => null), - rotateSecret: vi.fn(async () => null), - softDelete: vi.fn(async () => false), - listDeliveries: vi.fn(async () => ({ deliveries: [], nextCursor: null })), + findWebhookById: async () => null, + enqueue: async (e: { id: string; webhookId: string; linkId: string; event: string; payload: string; nextAttemptAt: number; createdAt: number }) => ({ + ...e, attempts: 0, status: "pending" as const, lastStatusCode: null, lastError: null, updatedAt: e.createdAt, + }), + claimDue: async () => [], + updateQueueEntry: async () => {}, + findQueueEntry: async () => null, listDeliveriesByLinkId: vi.fn(async () => []), recordDelivery: vi.fn(async () => {}), }; diff --git a/apps/api/test/watcher-loop.test.ts b/apps/api/test/watcher-loop.test.ts index 9ae4edde5..4fdddbe9b 100644 --- a/apps/api/test/watcher-loop.test.ts +++ b/apps/api/test/watcher-loop.test.ts @@ -139,17 +139,18 @@ function makeNoopWebhookRepo(): WebhookRepository { async listBySeller() { return []; }, - async getById(): Promise { + async findWebhookById(): Promise { return null; }, - async rotateSecret(): Promise { - return null; - }, - async softDelete(): Promise { - return false; + enqueue: async (e: { id: string; webhookId: string; linkId: string; event: string; payload: string; nextAttemptAt: number; createdAt: number }) => ({ + ...e, attempts: 0, status: "pending" as const, lastStatusCode: null, lastError: null, updatedAt: e.createdAt, + }), + async claimDue(): Promise { + return []; }, - async listDeliveries(): Promise<{ deliveries: never[]; nextCursor: null }> { - return { deliveries: [], nextCursor: null }; + async updateQueueEntry(): Promise {}, + async findQueueEntry(): Promise { + return null; }, async listDeliveriesByLinkId(): Promise { return []; diff --git a/docs/API.md b/docs/API.md index 73210d15d..cb8c8fbd9 100644 --- a/docs/API.md +++ b/docs/API.md @@ -657,9 +657,49 @@ are no more results. --- +## `POST /webhooks/deliveries/:id/replay` + +Manually re-queue a webhook delivery for immediate redelivery. Useful for +recovering dead-lettered entries or forcing a retry without waiting for the +next backoff window. + +`:id` is the queue entry id returned in delivery metadata, or visible in +`webhook_queue.id`. + +**Behaviour by current status** + +| Entry status | Effect | +| ------------ | ----------------------------------------------- | +| `dead` | Re-queued as `pending`, `nextAttemptAt = now`. | +| `pending` | `nextAttemptAt` reset to now (accelerates next attempt). | +| `delivered` | Re-queued as `pending` (re-sends an already-delivered event). | +| `claimed` | **409** — delivery is in-flight; wait for it to settle. | + +**202** +```json +{ + "id": "wqe_...", + "webhookId": "whk_...", + "linkId": "lnk_...", + "event": "link.paid", + "previousAttempts": 5, + "status": "pending", + "message": "Queued for immediate redelivery." +} +``` +**404** — queue entry not found. +**409** — delivery is currently in-flight. + +--- + ## Webhook delivery -When a link changes state, the API POSTs a JSON event to each registered URL: +When a link changes state, the API writes a delivery row to the durable queue +and returns immediately — event emission never blocks a state transition. A +background `WebhookWorker` claims due rows and POSTs the event to each +registered URL. + +### Events | Event | Fired when | | ----------------- | ------------------------------------------- | @@ -668,7 +708,7 @@ When a link changes state, the API POSTs a JSON event to each registered URL: | `offramp.settled` | a cash-out job settled | | `offramp.failed` | a cash-out job failed | -**Body** +### Body ```json { "event": "link.paid", @@ -687,7 +727,7 @@ When a link changes state, the API POSTs a JSON event to each registered URL: } ``` -**Headers** +### Headers - `x-checkout-event` — the event name. - `x-checkout-signature` — one or more `sha256=` HMAC-SHA256 signatures of the **exact raw body**, comma-separated. Normally just one, signed with your @@ -695,15 +735,30 @@ When a link changes state, the API POSTs a JSON event to each registered URL: previous secret) — accept the delivery if *any* listed signature matches, so you can redeploy without dropping events. -Delivery is retried with exponential backoff (default 4 attempts) on transient -failures — network errors and `5xx`/`429` responses. A `4xx` (other than `429`) is -treated as permanent and not retried. Return `2xx` quickly to acknowledge receipt. +### Delivery guarantees + +- **Durable**: the event body is serialised and signed once at write time and + persisted in `webhook_queue`. A process crash during backoff does not lose the + event — it will be delivered after restart. +- **At-least-once**: retried up to 5 attempts with exponential backoff + full + jitter (base 5 s → max ceiling doubles per attempt). Make receivers idempotent. +- **Per-attempt history**: every attempt is written to `webhook_deliveries` + (`attempt` column + `queue_entry_id`), so you can inspect exactly which + attempts failed and why. +- **Transient failures** (network errors, `5xx`, `429`) are retried. +- **Permanent failures** (`4xx` except `429`) are dead-lettered immediately. +- **Dead letters** are replayable via `POST /webhooks/deliveries/:id/replay`. + +Return `2xx` quickly to acknowledge receipt. Long-running processing should be +done asynchronously. For **replay protection**, reject events whose in-body `sentAt` is older than a -small window (e.g. 5 minutes). `sentAt` is part of the signed body, so it cannot be +small window (e.g. 5 minutes). `sentAt` is inside the signed body and cannot be forged without the secret. -**Verifying** (recompute over the raw body, accept if any signature matches): +### Verifying signatures + +Recompute over the raw body and compare in constant time: ```js import { createHmac, timingSafeEqual } from "node:crypto"; diff --git a/packages/core/src/ports/index.ts b/packages/core/src/ports/index.ts index 4ad664c2b..92e8c2890 100644 --- a/packages/core/src/ports/index.ts +++ b/packages/core/src/ports/index.ts @@ -603,32 +603,64 @@ export interface WebhookDelivery { webhookId: string; linkId: string; event: string; + /** Which attempt number (1-based). */ + attempt: number; + /** ID of the queue entry this delivery belongs to. Null for rows written + * before the durable queue existed. */ + queueEntryId: string | null; statusCode: number | null; ok: boolean; error: string | null; createdAt: number; } +/** Lifecycle status of a queue entry. */ +export type WebhookQueueStatus = "pending" | "claimed" | "delivered" | "dead"; + +/** + * One row in webhook_queue — the durable representation of a pending delivery. + * Immutable fields are set at enqueue time; mutable fields are updated by the + * worker after each attempt. + */ +export interface WebhookQueueEntry { + id: string; + webhookId: string; + linkId: string; + event: string; + /** The signed JSON body, serialised once at enqueue time. */ + payload: string; + attempts: number; + nextAttemptAt: number; // epoch ms + status: WebhookQueueStatus; + lastStatusCode: number | null; + lastError: string | null; + createdAt: number; + updatedAt: number; +} + export interface WebhookRepository { create(input: { sellerId: string; url: string; secret: string }): Promise; /** Active (non-deleted) webhooks for a seller. Used for both dispatch and listing. */ listBySeller(sellerId: string): Promise; - /** Scoped to the owning seller to prevent cross-tenant access (IDOR). */ - getById(id: string, sellerId: string, opts?: { includeDeleted?: boolean }): Promise; + findWebhookById(id: string): Promise; + recordDelivery(d: WebhookDelivery): Promise; + + // --- Queue operations --- + /** Insert a new pending queue entry. */ + enqueue(entry: Omit): Promise; /** - * Rotates the signing secret. The previous secret remains valid for - * `overlapMs` so in-flight receivers can be redeployed without dropping - * events (see WebhookSender, which signs with both during the overlap). + * Atomically claim up to `limit` rows that are due for delivery. + * "Due" means status = 'pending' AND next_attempt_at <= now. + * Returns only the rows successfully claimed by this process (status → 'claimed'). */ - rotateSecret(id: string, sellerId: string, newSecret: string, overlapMs: number): Promise; - /** Soft delete — keeps delivery history browsable after removal. */ - softDelete(id: string, sellerId: string): Promise; - recordDelivery(d: Omit): Promise; - listDeliveries( - webhookId: string, - sellerId: string, - opts: { limit: number; cursor?: string | null }, - ): Promise<{ deliveries: WebhookDelivery[]; nextCursor: string | null }>; + claimDue(now: number, limit: number): Promise; + /** Persist the result of one delivery attempt onto the queue entry. */ + updateQueueEntry( + id: string, + patch: Pick, + ): Promise; + /** Look up a single queue entry by id (for replay). */ + findQueueEntry(id: string): Promise; listDeliveriesByLinkId(linkId: string): Promise; }