diff --git a/README.md b/README.md index 97e2911..534784f 100644 --- a/README.md +++ b/README.md @@ -92,6 +92,25 @@ extension outranks the pit; `moshpit` lets a registered name win anyway, see [docs/moshpit-dns.md](docs/moshpit-dns.md)). Tables are created automatically on first request. +### Outbound webhooks + +Parked domains and projects send events through +[@profullstack/webhooks](https://github.com/profullstack/webhooks): CloudEvents +bodies, Standard Webhooks signatures (`webhook-id`, `webhook-timestamp`, +`webhook-signature`), a durable queue in the database with retries over about +three days, and a delivery log. Domain events: `waitlist.signup`, `bid.placed`, +`bid.won`, `bid.accepted`, `inbound.*` (whatever arrived at +`/api/webhooks/`, relayed). Manage targets in the dashboard +(Webhooks tab) or over HTTP with an API key: + + curl -H "Authorization: Bearer $KEY" "https://moshcoding.com/api/v1/webhooks?dn=example.com" + curl -H "Authorization: Bearer $KEY" -H 'content-type: application/json' \ + -d '{"url":"https://example.com/hook","events":["bid.*"]}' \ + "https://moshcoding.com/api/v1/webhooks?dn=example.com" + +Also `/:id` (GET/PATCH/DELETE), `/:id/secret`, `/:id/rotate`, `/:id/test`, +`/:id/deliveries`, `/deliveries`, `/deliveries/:id/retry`. + ### Deploy (Railway) Builds from the **Dockerfile** (`oven/bun` → `bun run build` → `bun run start`); diff --git a/app/api/domain-webhooks/route.ts b/app/api/domain-webhooks/route.ts index 19319d0..3619160 100644 --- a/app/api/domain-webhooks/route.ts +++ b/app/api/domain-webhooks/route.ts @@ -1,17 +1,21 @@ import { NextRequest, NextResponse } from "next/server"; import { readSession, authConfigured, SESSION_COOKIE } from "@/lib/session"; import { safeDomain } from "@/lib/config"; +import { accountIdForCoinpay, accountOwnsDomain, listInboundEvents } from "@/lib/db"; import { - accountIdForCoinpay, - accountOwnsDomain, - listDomainWebhooks, addDomainWebhook, deleteDomainWebhook, - setDomainWebhookActive, - listInboundEvents, distinctWebhookUrls, -} from "@/lib/db"; -import { fireDomainEvent, newSecret, isInternalUrl, isSelfWebhookUrl } from "@/lib/webhooks"; + domainDeliveries, + isInternalUrl, + isSelfWebhookUrl, + listDomainWebhooks, + rotateDomainWebhookSecret, + setDomainWebhookActive, + testDomainWebhooks, + WebhookError, + WEBHOOK_EVENT_TYPES, +} from "@/lib/webhooks"; import { listAccessibleProjectIds } from "@/lib/authz"; export const runtime = "nodejs"; @@ -46,19 +50,21 @@ export async function GET(req: NextRequest) { try { knownUrls = await distinctWebhookUrls(await listAccessibleProjectIds(s.sub)); } catch { /* best-effort */ } } - const [webhooks, events] = await Promise.all([listDomainWebhooks(dn), listInboundEvents(dn, 25)]); + const [webhooks, events, deliveries] = await Promise.all([listDomainWebhooks(dn), listInboundEvents(dn, 25), domainDeliveries(dn, 25)]); // Don't re-suggest URLs already added as targets for this domain. const have = new Set(webhooks.map((w) => w.url)); return NextResponse.json({ dn, inboundUrl: inboundUrl(req, dn), - webhooks: webhooks.map((w) => ({ id: w.id, url: w.url, secret: w.secret, active: w.active, created_at: w.created_at })), + webhooks: webhooks.map((w) => ({ id: w.id, url: w.url, secret: w.secret, events: w.events, active: w.active, failures: w.failures, disabled_reason: w.disabled_reason, created_at: w.created_at })), events, + deliveries: deliveries.map(({ payload: _p, ...d }) => d), + eventTypes: WEBHOOK_EVENT_TYPES, knownUrls: knownUrls.filter((u) => !have.has(u)), }); } -// POST /api/domain-webhooks — owner: add, pause/resume, delete, or test a target. +// POST /api/domain-webhooks — owner: add, pause/resume, rotate, delete, or test a target. export async function POST(req: NextRequest) { const accountId = await resolveAccountId(req); if (!accountId) return NextResponse.json({ error: "Sign in to manage webhooks." }, { status: 401 }); @@ -84,9 +90,21 @@ export async function POST(req: NextRequest) { return NextResponse.json({ ok: true, id, active: body.active }); } + if (body.action === "rotate") { + const secret = await rotateDomainWebhookSecret(String(body.id || ""), dn); + if (!secret) return NextResponse.json({ error: "Webhook target not found." }, { status: 404 }); + return NextResponse.json({ ok: true, id: String(body.id), secret }); + } + + // Sends webhook.test now: to one target (`id`) or every active one. if (body.action === "test") { - await fireDomainEvent(dn, "test.ping", { message: "Test event from moshcoding", dn }); - return NextResponse.json({ ok: true, sent: true }); + try { + const results = await testDomainWebhooks(dn, body.id ? String(body.id) : undefined); + return NextResponse.json({ ok: true, sent: results.length > 0, results }); + } catch (e) { + if (e instanceof WebhookError) return NextResponse.json({ error: e.message }, { status: e.status }); + throw e; + } } // Default: add an outbound target. @@ -100,6 +118,11 @@ export async function POST(req: NextRequest) { if (isSelfWebhookUrl(url)) { return NextResponse.json({ error: "That's an inbound receiver URL — sending events there would loop back into this dashboard. Use the URL of your own server." }, { status: 400 }); } - const wh = await addDomainWebhook(dn, url, newSecret()); - return NextResponse.json({ ok: true, webhook: { id: wh.id, url: wh.url, secret: wh.secret, active: wh.active } }, { status: 201 }); + try { + const wh = await addDomainWebhook(dn, url, body.events); + return NextResponse.json({ ok: true, webhook: { id: wh.id, url: wh.url, secret: wh.secret, events: wh.events, active: wh.active } }, { status: 201 }); + } catch (e) { + if (e instanceof WebhookError) return NextResponse.json({ error: e.message }, { status: e.status }); + throw e; + } } diff --git a/app/api/projects/[id]/webhooks/[endpointId]/route.ts b/app/api/projects/[id]/webhooks/[endpointId]/route.ts index db53991..7f025a5 100644 --- a/app/api/projects/[id]/webhooks/[endpointId]/route.ts +++ b/app/api/projects/[id]/webhooks/[endpointId]/route.ts @@ -1,7 +1,7 @@ import { NextRequest, NextResponse } from "next/server"; import { requireUser, unauthorized, bad } from "@/lib/api"; import { authorizeProject } from "@/lib/authz"; -import { deleteProjectWebhook, setProjectWebhookActive } from "@/lib/db"; +import { deleteProjectWebhook, setProjectWebhookActive } from "@/lib/webhooks"; export const runtime = "nodejs"; export const dynamic = "force-dynamic"; diff --git a/app/api/projects/[id]/webhooks/route.ts b/app/api/projects/[id]/webhooks/route.ts index 2ce7db0..68cef6f 100644 --- a/app/api/projects/[id]/webhooks/route.ts +++ b/app/api/projects/[id]/webhooks/route.ts @@ -1,8 +1,7 @@ import { NextRequest, NextResponse } from "next/server"; -import { db } from "@/lib/db"; import { requireUser, unauthorized, bad } from "@/lib/api"; import { authorizeProject } from "@/lib/authz"; -import { newSecret, isInternalUrl } from "@/lib/webhooks"; +import { createProjectWebhook, isInternalUrl, listProjectWebhooks, WebhookError } from "@/lib/webhooks"; import { normalizeWebhookEventSubscriptions } from "@/lib/webhook-events"; export const runtime = "nodejs"; @@ -15,11 +14,7 @@ export async function GET(req: NextRequest, ctx: { params: Promise<{ id: string const { id: projectId } = await ctx.params; const az = await authorizeProject(u.sub, projectId, "project.read"); if (!az.ok) return bad(az.error, az.status); - const { rows } = await db().execute({ - sql: `SELECT id, url, events, active, created_at FROM webhook_endpoints WHERE project_id = ? ORDER BY created_at`, - args: [projectId], - }); - return NextResponse.json({ endpoints: rows }); + return NextResponse.json({ endpoints: await listProjectWebhooks(projectId) }); } // create an outbound endpoint (needs webhook.manage). Returns the signing secret ONCE. @@ -39,11 +34,11 @@ export async function POST(req: NextRequest, ctx: { params: Promise<{ id: string return bad("events must be an array of names using letters, digits, dot, underscore, colon, or dash"); } - const secret = newSecret(); - const res = await db().execute({ - sql: `INSERT INTO webhook_endpoints (project_id, url, secret, events) VALUES (?, ?, ?, ?) - RETURNING id, url, events, active, created_at`, - args: [projectId, url, secret, JSON.stringify(events)], - }); - return NextResponse.json({ endpoint: res.rows[0], secret }, { status: 201 }); + try { + const { endpoint, secret } = await createProjectWebhook(projectId, url, events); + return NextResponse.json({ endpoint, secret }, { status: 201 }); + } catch (e) { + if (e instanceof WebhookError) return bad(e.message, e.status); + throw e; + } } diff --git a/app/api/projects/[id]/webhooks/test/route.ts b/app/api/projects/[id]/webhooks/test/route.ts index 3324684..c00c83b 100644 --- a/app/api/projects/[id]/webhooks/test/route.ts +++ b/app/api/projects/[id]/webhooks/test/route.ts @@ -1,12 +1,13 @@ import { NextRequest, NextResponse } from "next/server"; import { requireUser, unauthorized, bad } from "@/lib/api"; import { authorizeProject } from "@/lib/authz"; -import { dispatchEvent } from "@/lib/webhooks"; +import { testProjectWebhooks } from "@/lib/webhooks"; export const runtime = "nodejs"; export const dynamic = "force-dynamic"; -// fire a test event to all of a project's active endpoints (needs webhook.manage) +// send a webhook.test event now to all of a project's active endpoints, +// whatever they subscribe to (needs webhook.manage) export async function POST(req: NextRequest, ctx: { params: Promise<{ id: string }> }) { const u = await requireUser(req); if (!u) return unauthorized(); @@ -14,11 +15,6 @@ export async function POST(req: NextRequest, ctx: { params: Promise<{ id: string const az = await authorizeProject(u.sub, projectId, "webhook.manage"); if (!az.ok) return bad(az.error, az.status); - const results = await dispatchEvent( - projectId, - "ping", - { message: "🤘 moshcoding test event", at: new Date().toISOString() }, - { bypassEventFilter: true }, - ); + const results = await testProjectWebhooks(projectId); return NextResponse.json({ dispatched: results.length, results }); } diff --git a/app/api/v1/webhooks/[[...path]]/route.ts b/app/api/v1/webhooks/[[...path]]/route.ts new file mode 100644 index 0000000..6decd27 --- /dev/null +++ b/app/api/v1/webhooks/[[...path]]/route.ts @@ -0,0 +1,35 @@ +import { NextRequest, NextResponse } from "next/server"; +import { webhooksApi } from "@profullstack/webhooks"; +import { resolveAccountIdOrToken } from "@/lib/api"; +import { safeDomain } from "@/lib/config"; +import { accountOwnsDomain } from "@/lib/db"; +import { domainOwner, hooks } from "@/lib/webhooks"; + +export const runtime = "nodejs"; +export const dynamic = "force-dynamic"; + +// The @profullstack/webhooks API for a parked domain's outbound targets: +// GET/POST /api/v1/webhooks?dn=, GET/PATCH/DELETE /:id, +// GET /:id/secret, POST /:id/rotate, POST /:id/test, GET /:id/deliveries, +// GET /deliveries, POST /deliveries/:id/retry +// Auth: the dashboard session or `Authorization: Bearer `; the +// account must own the domain (pass it as ?dn= or an x-domain header). +let api: ReturnType | undefined; +const owners = new WeakMap(); + +async function handle(req: NextRequest): Promise { + const accountId = await resolveAccountIdOrToken(req); + if (!accountId) return NextResponse.json({ error: "sign in or send an API key" }, { status: 401 }); + const dn = safeDomain(req.nextUrl.searchParams.get("dn") || req.headers.get("x-domain")); + if (!dn) return NextResponse.json({ error: "name the domain: ?dn= or an x-domain header" }, { status: 400 }); + if (!(await accountOwnsDomain(accountId, dn))) return NextResponse.json({ error: "You don't own this domain." }, { status: 403 }); + owners.set(req, domainOwner(dn)); + api ??= webhooksApi(hooks(), { basePath: "/api/v1/webhooks", owner: (r) => owners.get(r) ?? null }); + return (await api(req)) ?? NextResponse.json({ error: "not found" }, { status: 404 }); +} + +export const GET = handle; +export const POST = handle; +export const PATCH = handle; +export const PUT = handle; +export const DELETE = handle; diff --git a/app/dashboard/[[...tab]]/page.tsx b/app/dashboard/[[...tab]]/page.tsx index 25adffe..174c148 100644 --- a/app/dashboard/[[...tab]]/page.tsx +++ b/app/dashboard/[[...tab]]/page.tsx @@ -963,10 +963,13 @@ function DomainWebhooksPanel({ onError, onOk }: { onError: (m: string) => void; const [active, setActive] = useState(null); const [data, setData] = useState(null); const [url, setUrl] = useState(""); + const [events, setEvents] = useState(""); const [busy, setBusy] = useState(false); + // A new or rotated secret, shown once with a copy button. + const [fresh, setFresh] = useState<{ id: string; secret: string } | null>(null); const load = async (dn: string) => { - setActive(dn); setData(null); + setActive(dn); setData(null); setFresh(null); try { const r = await fetch(`/api/domain-webhooks?dn=${encodeURIComponent(dn)}`); const d = await r.json(); @@ -995,7 +998,14 @@ function DomainWebhooksPanel({ onError, onOk }: { onError: (m: string) => void; }); const d = await r.json(); if (!r.ok) throw new Error(d.error); + if (payload.action === "test" && Array.isArray(d.results)) { + const bad = d.results.filter((x: any) => !x.ok); + if (!d.results.length) ok = "No active targets to test."; + else if (bad.length) ok = `Test sent: ${d.results.length - bad.length}/${d.results.length} answered 2xx (${bad[0].error || `HTTP ${bad[0].status}`}).`; + } onOk(ok); await load(active); + const secret = d.secret || d.webhook?.secret; + if (secret) setFresh({ id: d.id || d.webhook?.id, secret }); } catch (e: any) { onError(e.message || "Failed."); } finally { setBusy(false); } }; @@ -1007,7 +1017,7 @@ function DomainWebhooksPanel({ onError, onOk }: { onError: (m: string) => void; return (

Webhooks

-

Every parked domain gets hosted webhooks — no server to run. Inbound: POST events to the URL below. Outbound: we POST domain events (waitlist.signup, bid.placed/won/accepted) to your URLs, signed (Standard Webhooks).

+

Every parked domain gets hosted webhooks — no server to run. Inbound: POST events to the URL below. Outbound: we POST domain events (waitlist.signup, bid.placed/won/accepted, inbound.*) to your URLs as CloudEvents, signed (Standard Webhooks), retried with backoff. Also scriptable: /api/v1/webhooks?dn=<domain> with an API key.

{domains.map((d) => ( @@ -1025,16 +1035,32 @@ function DomainWebhooksPanel({ onError, onOk }: { onError: (m: string) => void; {data.knownUrls?.length ? ( {data.knownUrls.map((u: string) => ) : null} - - + setEvents(e.target.value)} /> + +
+ {fresh && ( +
+ + + +
+ )}
    {(!data.webhooks || data.webhooks.length === 0) &&
  • No outbound targets yet.
    • } {data.webhooks?.map((w: any) => (
  • - {w.url} · {w.active ? "active" : "paused"} · secret {String(w.secret).slice(0, 12)}… + {w.url} · {(w.events || ["*"]).join(", ") === "*" ? "all events" : (w.events || []).join(", ")} · {w.active ? "active" : w.disabled_reason ? `off: ${w.disabled_reason}` : "paused"} · secret {String(w.secret).slice(0, 12)}… + +
+

Recent deliveries ({data.deliveries?.length || 0})

+
    + {(!data.deliveries || data.deliveries.length === 0) &&
  • Nothing sent yet.
    • } + {data.deliveries?.map((d: any) => ( +
  • + {d.event_type} · {d.status} · {d.attempts} attempt{d.attempts === 1 ? "" : "s"}{d.last_status ? ` · HTTP ${d.last_status}` : ""}{d.last_error ? ` · ${String(d.last_error).slice(0, 60)}` : ""} + {d.created_at} +
    • + ))} +
+

Recent inbound events ({data.events?.length || 0})

    {(!data.events || data.events.length === 0) &&
  • Nothing received yet.
    • } diff --git a/bun.lock b/bun.lock index 3f85694..f9834a2 100644 --- a/bun.lock +++ b/bun.lock @@ -11,6 +11,7 @@ "@profullstack/profile": "0.1.6", "@profullstack/redeem-credits": "0.2.0", "@profullstack/stack": "0.5.0", + "@profullstack/webhooks": "0.2.0", "mailchecker": "^6.0.21", "next": "16.4.3", "pg": "^8.23.1", @@ -150,6 +151,8 @@ "@profullstack/stack": ["@profullstack/stack@0.5.0", "", { "dependencies": { "@profullstack/emailer": "^1.0.1", "@profullstack/form-guard": "^0.1.1", "@profullstack/referrals": "^0.1.0" }, "peerDependencies": { "@supabase/ssr": ">=0.5.0", "next": ">=13.0.0", "react": ">=18.0.0" }, "optionalPeers": ["@supabase/ssr", "next", "react"] }, "sha512-LbXovIhyzLEq+a+y3q8KWC+jN1tJjvSoUiFEhpiEBisQn5U+iutIGHDgg16Lul74kQaFZeFlbp7CMPAVJ8KL6g=="], + "@profullstack/webhooks": ["@profullstack/webhooks@0.2.0", "", {}, "sha512-lfUoJwN5GmxHzCOphI7fNQAsZ9uiC2C6BZxoPA7MSxM2ryr7MKwOi2YYszwvN6aPaPTRHLhHdqoHQJbkaP9mUg=="], + "@swc/helpers": ["@swc/helpers@0.5.23", "", { "dependencies": { "tslib": "^2.8.0" } }, "sha512-5lSsMOTXURePglDfvuAQUqkGek9Hg2kksOYay2m0+XR++b2NWYL/4sWyuvVBIs8oKnJaxkdi9whaL/sqN13afw=="], "@types/node": ["@types/node@26.6.5", "", { "dependencies": { "undici-types": "~8.9.0" } }, "sha512-uMzsv9stDdofIZBx4UGh4Z9ZgXnGC9seRDscilBcTHAeOFk67MZg/k9bWVfO+bk4qZmG+l52hRnLwUiH2LnJTg=="], diff --git a/instrumentation.ts b/instrumentation.ts new file mode 100644 index 0000000..2386bed --- /dev/null +++ b/instrumentation.ts @@ -0,0 +1,10 @@ +// Runs once when the Next server boots. Starts the webhook drainer: queued +// deliveries are sent every few seconds and retried with backoff, and they +// survive restarts because the queue is in the database. +export async function register() { + if (process.env.NEXT_RUNTIME !== "nodejs") return; + const env = (name: string) => process.env[name] ?? ""; + if (!env("DATABASE_URL") && !env("TURSO_DATABASE_URL")) return; + const { hooks } = await import("./lib/webhooks"); + hooks().start(); +} diff --git a/lib/db.ts b/lib/db.ts index d995f23..cb97c50 100644 --- a/lib/db.ts +++ b/lib/db.ts @@ -5,6 +5,8 @@ import { createRequire } from "node:module"; import type { WaitlistSort } from "./waitlist-filter"; import { sqlTime } from "./sql-time"; import { schemaSql } from "@profullstack/profile"; +import { schemaSql as webhookSchemaSql } from "@profullstack/webhooks"; +import { legacySecretToSpec } from "./webhook-legacy"; let _db: Client | undefined; const POSTGRES_URL = /^postgres(ql)?:\/\//i; @@ -534,6 +536,62 @@ async function initSchema(): Promise { `); await d.execute(`CREATE INDEX IF NOT EXISTS idx_media_dn ON media (dn, created_at)`); await ensureProfileTable(); + await ensureWebhookHub(); +} + +/** + * The @profullstack/webhooks tables (its own DDL, `hub_` prefix: the plain + * names are the pre-hub tables) and the one-time copy of the pre-hub + * endpoints into them. INTEGER becomes bigint on Postgres (libsql-pg), which + * the millisecond timestamps need. + */ +async function ensureWebhookHub(): Promise { + const d = db(); + for (const stmt of webhookSchemaSql("sqlite", "hub_")) await d.execute(stmt); + await migrateLegacyWebhooks(); +} + +const legacyMs = (v: unknown): number => { + if (v instanceof Date) return v.getTime(); + const s = String(v ?? ""); + const t = Date.parse(/^\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2}$/.test(s) ? `${s.replace(" ", "T")}Z` : s.replace(" ", "T").replace(/([+-]\d{2})$/, "$1:00")); + return Number.isFinite(t) ? t : Date.now(); +}; + +/** + * Copies endpoints made before the hub into its table, once each (the old row + * is stamped hub_migrated_at, so a target deleted later is not copied back). + * Each keeps its id (routes that address it by id still work), URL, + * subscription and on/off state. The signing key is kept byte for byte: the + * old sender used the whole secret string as the HMAC key, the hub reads a + * `whsec_` secret as base64, so the stored secret becomes `whsec_` + base64 of + * the old string; the same key, so every receiver still verifies. + */ +export async function migrateLegacyWebhooks(): Promise { + const d = db(); + await addColumnIfMissing("webhook_endpoints", "hub_migrated_at", "TEXT"); + await addColumnIfMissing("domain_webhooks", "hub_migrated_at", "TEXT"); + const sources = [ + { table: "webhook_endpoints", sql: `SELECT id, project_id AS scope, url, secret, events, active, created_at FROM webhook_endpoints WHERE hub_migrated_at IS NULL`, owner: (r: any) => `project:${r.scope}` }, + { table: "domain_webhooks", sql: `SELECT id, dn AS scope, url, secret, '["*"]' AS events, active, created_at FROM domain_webhooks WHERE hub_migrated_at IS NULL`, owner: (r: any) => `dn:${String(r.scope).toLowerCase()}` }, + ]; + let copied = 0; + for (const src of sources) { + const rows = (await d.execute(src.sql)).rows as any[]; + for (const r of rows) { + let events = '["*"]'; + try { const e = JSON.parse(String(r.events)); if (Array.isArray(e) && e.length) events = JSON.stringify(e.map(String)); } catch { /* all events */ } + const at = legacyMs(r.created_at); + await d.execute({ + sql: `INSERT INTO hub_webhook_endpoints (id, owner, url, secret, events, description, enabled, failures, disabled_reason, created_at, updated_at) + VALUES (?, ?, ?, ?, ?, '', ?, 0, NULL, ?, ?) ON CONFLICT (id) DO NOTHING`, + args: [String(r.id), src.owner(r), String(r.url), legacySecretToSpec(String(r.secret)), events, Number(r.active) === 1 ? 1 : 0, at, at], + }); + await d.execute({ sql: `UPDATE ${src.table} SET hub_migrated_at = ? WHERE id = ?`, args: [sqlTime(), String(r.id)] }); + copied++; + } + } + return copied; } /** @@ -897,38 +955,6 @@ export async function renameProject(id: string, name: string): Promise { await db().execute({ sql: `UPDATE projects SET name = ? WHERE id = ?`, args: [name, id] }); } -/** Pause or resume one project webhook without discarding its signing secret. */ -export async function setProjectWebhookActive( - id: string, - projectId: string, - active: boolean, -): Promise { - await ensureSchema(); - const r = await db().execute({ - sql: `UPDATE webhook_endpoints SET active = ? WHERE id = ? AND project_id = ?`, - args: [active ? 1 : 0, id, projectId], - }); - return Number(r.rowsAffected || 0) > 0; -} - -/** Delete one project webhook and its delivery history. */ -export async function deleteProjectWebhook(id: string, projectId: string): Promise { - await ensureSchema(); - const d = db(); - const owned = await d.execute({ - sql: `SELECT 1 FROM webhook_endpoints WHERE id = ? AND project_id = ? LIMIT 1`, - args: [id, projectId], - }); - if (!owned.rows.length) return false; - - await d.execute({ sql: `DELETE FROM webhook_deliveries WHERE endpoint_id = ?`, args: [id] }); - const r = await d.execute({ - sql: `DELETE FROM webhook_endpoints WHERE id = ? AND project_id = ?`, - args: [id, projectId], - }); - return Number(r.rowsAffected || 0) > 0; -} - /** Pause or resume one inbound receiver without rotating its shared secret. */ export async function setProjectInboundWebhookActive( id: string, @@ -968,6 +994,8 @@ export async function deleteProject(id: string): Promise { const d = db(); await d.execute({ sql: `DELETE FROM webhook_deliveries WHERE endpoint_id IN (SELECT id FROM webhook_endpoints WHERE project_id = ?)`, args: [id] }); await d.execute({ sql: `DELETE FROM webhook_endpoints WHERE project_id = ?`, args: [id] }); + await d.execute({ sql: `DELETE FROM hub_webhook_deliveries WHERE owner = ?`, args: [`project:${id}`] }); + await d.execute({ sql: `DELETE FROM hub_webhook_endpoints WHERE owner = ?`, args: [`project:${id}`] }); await d.execute({ sql: `DELETE FROM inbound_events WHERE inbound_id IN (SELECT id FROM inbound_webhooks WHERE project_id = ?)`, args: [id] }); await d.execute({ sql: `DELETE FROM inbound_webhooks WHERE project_id = ?`, args: [id] }); await d.execute({ sql: `DELETE FROM projects WHERE id = ?`, args: [id] }); @@ -1144,8 +1172,7 @@ export async function accountOwnsDomain(accountId: string, dn: string): Promise< return Number((a.rows[0] as any)?.is_admin || 0) === 1; } -// ---- per-domain webhooks (inbound + outbound) ----------------------------- -export type DomainWebhook = { id: string; dn: string; url: string; secret: string; active: boolean; created_at: string }; +// ---- per-domain webhooks (inbound; outbound targets live in lib/webhooks.ts) -- /** Is this a real domain on the platform (so we accept inbound events for it)? */ export async function isKnownDomain(dn: string): Promise { @@ -1156,51 +1183,12 @@ export async function isKnownDomain(dn: string): Promise { UNION SELECT 1 FROM accounts WHERE lower(domain) = ? UNION SELECT 1 FROM tenants WHERE domain = ? UNION SELECT 1 FROM auctions WHERE domain = ? - UNION SELECT 1 FROM domain_webhooks WHERE dn = ? LIMIT 1`, - args: [d, d, d, d, d], + UNION SELECT 1 FROM hub_webhook_endpoints WHERE owner = ? LIMIT 1`, + args: [d, d, d, d, `dn:${d}`], }); return r.rows.length > 0; } -export async function listDomainWebhooks(dn: string): Promise { - await ensureSchema(); - const r = await db().execute({ sql: `SELECT * FROM domain_webhooks WHERE dn = ? ORDER BY created_at`, args: [dn.toLowerCase()] }); - return r.rows.map((x: any) => ({ id: String(x.id), dn: String(x.dn), url: String(x.url), secret: String(x.secret), active: Number(x.active) === 1, created_at: String(x.created_at) })); -} - -/** Active {url,secret} targets for a domain — used by fireDomainEvent. */ -export async function activeDomainWebhooks(dn: string): Promise<{ url: string; secret: string }[]> { - await ensureSchema(); - const r = await db().execute({ sql: `SELECT url, secret FROM domain_webhooks WHERE dn = ? AND active = 1`, args: [dn.toLowerCase()] }); - return r.rows.map((x: any) => ({ url: String(x.url), secret: String(x.secret) })); -} - -export async function addDomainWebhook(dn: string, url: string, secret: string): Promise { - await ensureSchema(); - const r = await db().execute({ - sql: `INSERT INTO domain_webhooks (dn, url, secret) VALUES (?, ?, ?) RETURNING *`, - args: [dn.toLowerCase(), url, secret], - }); - const x: any = r.rows[0]; - return { id: String(x.id), dn: String(x.dn), url: String(x.url), secret: String(x.secret), active: true, created_at: String(x.created_at) }; -} - -export async function deleteDomainWebhook(id: string, dn: string): Promise { - await ensureSchema(); - const r = await db().execute({ sql: `DELETE FROM domain_webhooks WHERE id = ? AND dn = ?`, args: [id, dn.toLowerCase()] }); - return Number(r.rowsAffected || 0) > 0; -} - -/** Pause or resume one outbound target without discarding its signing secret. */ -export async function setDomainWebhookActive(id: string, dn: string, active: boolean): Promise { - await ensureSchema(); - const r = await db().execute({ - sql: `UPDATE domain_webhooks SET active = ? WHERE id = ? AND dn = ?`, - args: [active ? 1 : 0, id, dn.toLowerCase()], - }); - return Number(r.rowsAffected || 0) > 0; -} - /** Stores an inbound event, keeping only the most recent 200 per domain. */ export async function recordInboundEvent(opts: { dn: string; source?: string | null; eventType?: string | null; payload: string }): Promise { await ensureSchema(); @@ -1217,19 +1205,6 @@ export async function recordInboundEvent(opts: { dn: string; source?: string | n }); } -/** Distinct outbound webhook URLs across the given project ids — used to - * suggest/pre-fill a domain webhook target from the user's project webhooks. */ -export async function distinctWebhookUrls(projectIds: string[]): Promise { - await ensureSchema(); - if (!projectIds.length) return []; - const placeholders = projectIds.map(() => "?").join(","); - const r = await db().execute({ - sql: `SELECT DISTINCT url FROM webhook_endpoints WHERE project_id IN (${placeholders}) ORDER BY created_at DESC LIMIT 25`, - args: projectIds, - }); - return r.rows.map((x: any) => String(x.url)).filter(Boolean); -} - export async function listInboundEvents(dn: string, limit = 50): Promise<{ id: string; source: string | null; event_type: string | null; payload: string; created_at: string }[]> { await ensureSchema(); const r = await db().execute({ diff --git a/lib/webhook-legacy.ts b/lib/webhook-legacy.ts new file mode 100644 index 0000000..010d166 --- /dev/null +++ b/lib/webhook-legacy.ts @@ -0,0 +1,11 @@ +/** + * The pre-hub sender (lib/webhook-signing.ts) used the whole secret string, + * `whsec_` included, as the HMAC key. @profullstack/webhooks reads a `whsec_` + * secret the Standard Webhooks way: the key is the base64 after the prefix. + * Wrapping the old string's bytes in base64 gives the hub the very same key, + * so a receiver holding the old secret still verifies every delivery, and the + * new form also works with any Standard Webhooks library. + */ +export function legacySecretToSpec(secret: string): string { + return "whsec_" + Buffer.from(secret, "utf8").toString("base64"); +} diff --git a/lib/webhook-store.ts b/lib/webhook-store.ts new file mode 100644 index 0000000..857c3cb --- /dev/null +++ b/lib/webhook-store.ts @@ -0,0 +1,157 @@ +// A @profullstack/webhooks store on the app's own libSQL-shaped client. +// +// The package ships sqliteStore (bun:sqlite & co.) and postgresStore (a pg +// pool); this app talks to both engines through one @libsql/client surface +// (`db()`: Postgres via @profullstack/libsql-pg in production, a `file:` +// database locally and in the tests), so the store is written against that. +// The tables are the package's own DDL with a `hub_` prefix (the unprefixed +// names are the pre-hub `webhook_endpoints`/`webhook_deliveries`), created in +// initSchema (lib/db.ts) along with the one-time copy of the old rows. + +import type { Client } from "@libsql/client"; +import type { Delivery, Endpoint, WebhookStore } from "@profullstack/webhooks"; + +export const HUB_PREFIX = "hub_"; +export const HUB_ENDPOINTS = `${HUB_PREFIX}webhook_endpoints`; +export const HUB_DELIVERIES = `${HUB_PREFIX}webhook_deliveries`; + +const ENDPOINT_COLS = "id, owner, url, secret, events, description, enabled, failures, disabled_reason, created_at, updated_at"; +const DELIVERY_COLS = "id, endpoint_id, owner, event_id, event_type, payload, status, attempts, next_at, last_status, last_error, created_at, updated_at"; + +const ENDPOINT_FIELDS: Record = { + url: "url", secret: "secret", events: "events", description: "description", enabled: "enabled", + failures: "failures", disabledReason: "disabled_reason", updatedAt: "updated_at", +}; +const DELIVERY_FIELDS: Record = { + status: "status", attempts: "attempts", nextAt: "next_at", lastStatus: "last_status", lastError: "last_error", updatedAt: "updated_at", +}; + +const num = (v: unknown) => (v === null || v === undefined ? null : Number(v)); + +function toEndpoint(r: any): Endpoint | null { + if (!r) return null; + let events: string[] = ["*"]; + try { events = JSON.parse(String(r.events)); } catch { /* default */ } + return { + id: String(r.id), + owner: String(r.owner), + url: String(r.url), + secret: String(r.secret), + events, + description: r.description == null ? "" : String(r.description), + enabled: Number(r.enabled) === 1 || r.enabled === true, + failures: Number(r.failures ?? 0), + disabledReason: r.disabled_reason == null ? null : String(r.disabled_reason), + createdAt: Number(r.created_at), + updatedAt: Number(r.updated_at), + }; +} + +function toDelivery(r: any): Delivery | null { + if (!r) return null; + return { + id: String(r.id), + endpointId: String(r.endpoint_id), + owner: String(r.owner), + eventId: String(r.event_id), + eventType: String(r.event_type), + payload: String(r.payload), + status: String(r.status) as Delivery["status"], + attempts: Number(r.attempts ?? 0), + nextAt: Number(r.next_at), + lastStatus: num(r.last_status), + lastError: r.last_error == null ? null : String(r.last_error), + createdAt: Number(r.created_at), + updatedAt: Number(r.updated_at), + }; +} + +// libSQL's remote client throws on `undefined` (a local file binds it as null), +// so every value is normalised before it is bound. +function sqlValue(key: string, value: unknown): string | number | null { + if (key === "events") return JSON.stringify(value ?? ["*"]); + if (key === "enabled") return value ? 1 : 0; + if (value === undefined || value === null) return null; + return value as string | number; +} + +/** + * `ready` runs before every statement: the app's ensureSchema(), so the tables + * exist (and the old rows are copied) before the first read. + */ +export function libsqlStore(client: () => Client, ready: () => Promise = async () => {}): WebhookStore { + const all = async (sql: string, args: (string | number | null)[] = []) => { + await ready(); + return (await client().execute({ sql, args })).rows as any[]; + }; + const run = async (sql: string, args: (string | number | null)[] = []) => { + await ready(); + await client().execute({ sql, args }); + }; + const update = async (table: string, fields: Record, id: string, patch: Record) => { + const keys = Object.keys(patch).filter((k) => fields[k] && patch[k] !== undefined); + if (!keys.length) return; + await run( + `UPDATE ${table} SET ${keys.map((k) => `${fields[k]} = ?`).join(", ")} WHERE id = ?`, + [...keys.map((k) => sqlValue(k, patch[k])), id], + ); + }; + + return { + async insertEndpoint(e) { + await run(`INSERT INTO ${HUB_ENDPOINTS} (${ENDPOINT_COLS}) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`, [ + e.id, e.owner, e.url, e.secret, JSON.stringify(e.events), e.description ?? "", e.enabled ? 1 : 0, + e.failures ?? 0, e.disabledReason ?? null, e.createdAt, e.updatedAt, + ]); + }, + async getEndpoint(id) { + return toEndpoint((await all(`SELECT ${ENDPOINT_COLS} FROM ${HUB_ENDPOINTS} WHERE id = ?`, [id]))[0]); + }, + async listEndpoints(owner) { + return (await all(`SELECT ${ENDPOINT_COLS} FROM ${HUB_ENDPOINTS} WHERE owner = ? ORDER BY created_at, id`, [owner])) + .map((r) => toEndpoint(r)!); + }, + updateEndpoint: (id, patch) => update(HUB_ENDPOINTS, ENDPOINT_FIELDS, id, patch as Record), + async deleteEndpoint(id) { + await run(`DELETE FROM ${HUB_DELIVERIES} WHERE endpoint_id = ?`, [id]); + await run(`DELETE FROM ${HUB_ENDPOINTS} WHERE id = ?`, [id]); + }, + async insertDeliveries(rows) { + for (const d of rows) { + await run(`INSERT INTO ${HUB_DELIVERIES} (${DELIVERY_COLS}) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`, [ + d.id, d.endpointId, d.owner, d.eventId, d.eventType, d.payload, d.status, d.attempts, d.nextAt, + d.lastStatus ?? null, d.lastError ?? null, d.createdAt, d.updatedAt, + ]); + } + }, + async getDelivery(id) { + return toDelivery((await all(`SELECT ${DELIVERY_COLS} FROM ${HUB_DELIVERIES} WHERE id = ?`, [id]))[0]); + }, + async listDeliveries({ owner, endpointId, limit = 50 }) { + const where = endpointId ? "owner = ? AND endpoint_id = ?" : "owner = ?"; + const args = endpointId ? [owner, endpointId] : [owner]; + return (await all(`SELECT ${DELIVERY_COLS} FROM ${HUB_DELIVERIES} WHERE ${where} ORDER BY created_at DESC, id DESC LIMIT ?`, [...args, limit])) + .map((r) => toDelivery(r)!); + }, + updateDelivery: (id, patch) => update(HUB_DELIVERIES, DELIVERY_FIELDS, id, patch as Record), + /** + * Takes due rows by pushing their next_at past the lease in one statement. + * The outer `next_at <= ?` is re-checked against the row Postgres actually + * locks, so two drainers racing on one row cannot both take it. + */ + async claimDue(now, limit, leaseMs) { + const rows = await all( + `UPDATE ${HUB_DELIVERIES} SET next_at = ? + WHERE id IN (SELECT id FROM ${HUB_DELIVERIES} WHERE status = 'pending' AND next_at <= ? ORDER BY next_at LIMIT ?) + AND status = 'pending' AND next_at <= ? + RETURNING ${DELIVERY_COLS}`, + [now + leaseMs, now, limit, now], + ); + // RETURNING carries the new next_at; the hub only reads the rest. Oldest first. + return rows.map((r) => toDelivery(r)!).sort((a, b) => a.createdAt - b.createdAt); + }, + async prune(before) { + await run(`DELETE FROM ${HUB_DELIVERIES} WHERE status <> 'pending' AND updated_at < ?`, [before]); + }, + }; +} diff --git a/lib/webhooks.ts b/lib/webhooks.ts index b311269..8497c02 100644 --- a/lib/webhooks.ts +++ b/lib/webhooks.ts @@ -1,129 +1,212 @@ -import crypto from "node:crypto"; -import { db, activeDomainWebhooks } from "./db"; -import { sqlTime } from "./sql-time"; +// Outbound webhooks on @profullstack/webhooks: customers register endpoints, +// the app emits events, the hub signs (Standard Webhooks), queues, retries with +// backoff and logs every delivery in the database. +// +// Two kinds of owner share one hub: +// dn: a parked domain's targets (dashboard "Webhooks" tab, and +// /api/v1/webhooks?dn= with an API key) +// project: a project's outbound endpoints (/api/projects/:id/webhooks) +// +// Endpoints created before the hub were copied into its tables with their ids, +// URLs, subscriptions and signing keys (see migrateLegacyWebhooks in lib/db.ts), +// so receivers keep verifying without changing anything. + +import { Webhooks, WebhookError, type PublicEndpoint } from "@profullstack/webhooks"; +import { db, ensureSchema } from "./db"; +import { libsqlStore, HUB_ENDPOINTS } from "./webhook-store"; import { isInternalUrl, isSelfWebhookUrl } from "./url-guard"; import { RELAY_HOP_HEADER } from "./webhook-events"; export { newSecret, signWebhook, verifyWebhook } from "./webhook-signing"; -import { signWebhook } from "./webhook-signing"; -export { isInternalUrl, isSelfWebhookUrl }; +export { isInternalUrl, isSelfWebhookUrl, WebhookError }; + +/** The events a domain's targets can subscribe to. `inbound.*` relays whatever arrived at /api/webhooks/. */ +export const WEBHOOK_EVENT_TYPES = [ + "waitlist.signup", + "bid.placed", + "bid.won", + "bid.accepted", + "inbound.*", +]; + +export const domainOwner = (dn: string) => `dn:${dn.trim().toLowerCase()}`; +export const projectOwner = (projectId: string) => `project:${projectId}`; + +// process.env[name] with a computed key: Next.js inlines process.env.LITERAL at build time. +const env = (name: string) => process.env[name] ?? ""; + +/** fetch that refuses our own inbound receivers: delivering there would feed the relay loop. */ +const guardedFetch: typeof fetch = (async (input: any, init?: any) => { + const url = typeof input === "string" ? input : input instanceof URL ? input.href : input.url; + if (isSelfWebhookUrl(url)) throw new Error("blocked: self-referential url"); + return fetch(input, init); +}) as typeof fetch; -/* ---- per-domain outbound delivery (best-effort, no owner server needed) ---- */ /** - * Fires a parked-domain event to every active target URL for that domain, - * Standard-Webhooks-signed with each target's secret. Best-effort and - * SSRF-guarded; never throws (so it can't break the triggering request). + * Headers every delivery carried before the hub, kept so receivers that read + * them still work: the relay hop counter (receivers and our own inbound route + * use it to stop loops) and the user agent. + */ +function legacyHeaders({ body }: { body: string }): Record { + let hop = 1; + try { + const h = Number(JSON.parse(body)?.data?.relay_hop); + if (Number.isInteger(h) && h > 0) hop = h; + } catch { /* not JSON: first hop */ } + return { "user-agent": "moshcoding-webhooks/1", [RELAY_HOP_HEADER]: String(hop) }; +} + +export type HooksOptions = { fetchImpl?: typeof fetch; guard?: { allowHttp?: boolean; allowPrivate?: boolean; lookup?: (host: string) => Promise } }; + +export function createHooks(opts: HooksOptions = {}): Webhooks { + return new Webhooks({ + store: libsqlStore(db, ensureSchema), + source: env("APP_BASE_URL") || "https://moshcoding.com", + eventTypes: WEBHOOK_EVENT_TYPES, + maxEndpoints: 25, + extraHeaders: legacyHeaders, + fetchImpl: opts.fetchImpl ?? guardedFetch, + guard: opts.guard ?? {}, + log: (m) => console.warn(m), + }); +} + +let _hooks: Webhooks | undefined; +/** The process-wide hub. instrumentation.ts calls start() on it once at boot. */ +export function hooks(): Webhooks { + return (_hooks ??= createHooks()); +} +/** For tests: swap in a hub with a fake fetch / DNS guard. */ +export function setHooksForTests(h: Webhooks | undefined) { + _hooks = h; +} + +/* ---- domain events ------------------------------------------------------- */ + +/** + * Queues a parked-domain event for every subscribed target of that domain. + * Never throws, so it can't break the request that triggered it. * - * `opts.hop` is the hop count of the event that triggered this fan-out; each - * delivery carries hop+1 so a receiver can tell how far an event has travelled. + * `opts.hop` is the hop count of the event that triggered this fan-out; the + * delivery carries hop+1 (in data.relay_hop and the x-moshcoding-hop header) + * so a receiver can tell how far an event has travelled. */ export async function fireDomainEvent(dn: string, type: string, data: unknown, opts: { hop?: number } = {}): Promise { - let targets: { url: string; secret: string }[] = []; - try { targets = await activeDomainWebhooks(dn); } catch { return; } - if (!targets.length) return; - const id = "evt_" + crypto.randomBytes(12).toString("hex"); - const ts = Math.floor(Date.now() / 1000); - const hop = (opts.hop ?? 0) + 1; - const body = JSON.stringify({ id, type, dn, data, created_at: new Date().toISOString() }); - await Promise.allSettled( - targets.map(async (t) => { - if (isInternalUrl(t.url) || isSelfWebhookUrl(t.url)) return; - try { - await fetch(t.url, { - method: "POST", - headers: { - "content-type": "application/json", - "user-agent": "moshcoding-webhooks/1", - [RELAY_HOP_HEADER]: String(hop), - ...signWebhook(id, ts, body, t.secret), - }, - body, - signal: AbortSignal.timeout(10_000), - }); - } catch { /* best-effort */ } - }), - ); -} - -/* ---- outbound delivery ---- */ -export function eventEnvelope(type: string, data: unknown, projectId: string) { - const id = "evt_" + crypto.randomBytes(12).toString("hex"); - return { id, type, data, created_at: new Date().toISOString(), project_id: projectId }; -} - -/** Dispatch an event to active project endpoints, normally honoring subscriptions. */ -export async function dispatchEvent( - projectId: string, - type: string, - data: unknown, - opts: { bypassEventFilter?: boolean } = {}, -) { - const { rows } = await db().execute({ - sql: `SELECT id, url, secret, events FROM webhook_endpoints WHERE project_id = ? AND active = 1`, - args: [projectId], - }); - const results: any[] = []; - for (const ep of rows as any[]) { - let events: string[] = ["*"]; - try { events = JSON.parse(ep.events); } catch { /* default */ } - if (!opts.bypassEventFilter && !events.includes("*") && !events.includes(type)) continue; - results.push(await deliverToEndpoint(String(ep.id), String(ep.url), String(ep.secret), type, data, projectId)); + try { + const payload = opts.hop !== undefined && data && typeof data === "object" + ? { ...(data as Record), relay_hop: opts.hop + 1 } + : data; + await hooks().emit(domainOwner(dn), type, payload, { subject: dn.toLowerCase() }); + } catch (e: any) { + console.warn(`webhooks: emit ${type} for ${dn} failed: ${e?.message}`); } +} + +export type DomainWebhook = { id: string; dn: string; url: string; secret: string; events: string[]; active: boolean; failures: number; disabled_reason: string | null; created_at: string }; + +async function withSecret(owner: string, dn: string, e: PublicEndpoint): Promise { + const { secret } = await hooks().secret(owner, e.id); + return { id: e.id, dn, url: e.url, secret, events: e.events, active: e.enabled, failures: e.failures, disabled_reason: e.disabled_reason, created_at: e.created_at }; +} + +export async function listDomainWebhooks(dn: string): Promise { + const owner = domainOwner(dn); + const list = await hooks().listEndpoints(owner); + return Promise.all(list.map((e) => withSecret(owner, dn.toLowerCase(), e))); +} + +/** Adds a target; throws WebhookError (status 400/409) for a URL or event list the hub refuses. */ +export async function addDomainWebhook(dn: string, url: string, events?: string[] | string): Promise { + const owner = domainOwner(dn); + const { endpoint, secret } = await hooks().createEndpoint(owner, { url, events }); + return { id: endpoint.id, dn: dn.toLowerCase(), url: endpoint.url, secret, events: endpoint.events, active: endpoint.enabled, failures: 0, disabled_reason: null, created_at: endpoint.created_at }; +} + +const notFound = (e: unknown) => e instanceof WebhookError && e.status === 404; + +export async function deleteDomainWebhook(id: string, dn: string): Promise { + try { await hooks().deleteEndpoint(domainOwner(dn), id); return true; } catch (e) { if (notFound(e)) return false; throw e; } +} + +/** Pause or resume one outbound target without discarding its signing secret. */ +export async function setDomainWebhookActive(id: string, dn: string, active: boolean): Promise { + try { await hooks().updateEndpoint(domainOwner(dn), id, { enabled: active }); return true; } catch (e) { if (notFound(e)) return false; throw e; } +} + +/** A new signing secret for one target; the old one stops working at once. */ +export async function rotateDomainWebhookSecret(id: string, dn: string): Promise { + try { return (await hooks().rotateSecret(domainOwner(dn), id)).secret; } catch (e) { if (notFound(e)) return null; throw e; } +} + +/** Sends `webhook.test` now to one target, or to every enabled target of the domain. */ +export async function testDomainWebhooks(dn: string, id?: string) { + const owner = domainOwner(dn); + const ids = id ? [id] : (await hooks().listEndpoints(owner)).filter((e) => e.enabled).map((e) => e.id); + const results = []; + for (const one of ids) results.push({ id: one, ...(await hooks().test(owner, one)) }); return results; } -/** Deliver a single event to one endpoint, recording the attempt. */ -export async function deliverToEndpoint( - endpointId: string, url: string, secret: string, type: string, data: unknown, projectId: string -) { - const envelope = eventEnvelope(type, data, projectId); - const body = JSON.stringify(envelope); - const deliveryId = envelope.id; +export async function domainDeliveries(dn: string, limit = 25) { + return hooks().deliveries(domainOwner(dn), { limit }); +} - if (isInternalUrl(url)) { - await recordDelivery(endpointId, type, body, deliveryId, "dead_letter", 1, null, "blocked: internal url"); - return { ok: false, error: "internal url blocked" }; - } - if (isSelfWebhookUrl(url)) { - await recordDelivery(endpointId, type, body, deliveryId, "dead_letter", 1, null, "blocked: self-referential url"); - return { ok: false, error: "self-referential url blocked" }; - } - const ts = Math.floor(Date.now() / 1000); - const headers = { - "content-type": "application/json", - "user-agent": "moshcoding-webhooks/1", - ...signWebhook(deliveryId, ts, body, secret), - }; - try { - const res = await fetch(url, { method: "POST", headers, body, signal: AbortSignal.timeout(10_000) }); - const ok = res.ok; - await recordDelivery(endpointId, type, body, deliveryId, ok ? "delivered" : "failed", 1, res.status, ok ? null : `HTTP ${res.status}`); - return { ok, status: res.status }; - } catch (err: any) { - await recordDelivery(endpointId, type, body, deliveryId, "failed", 1, null, err?.message || "network error"); - return { ok: false, error: err?.message }; - } +/* ---- project endpoints --------------------------------------------------- */ + +/** The row shape /api/projects/:id/webhooks has always answered with. */ +export function projectEndpointRow(e: PublicEndpoint) { + return { id: e.id, url: e.url, events: JSON.stringify(e.events), active: e.enabled ? 1 : 0, created_at: e.created_at }; +} + +export async function listProjectWebhooks(projectId: string) { + return (await hooks().listEndpoints(projectOwner(projectId))).map(projectEndpointRow); } -async function recordDelivery( - endpointId: string, type: string, body: string, idem: string, - status: string, attempts: number, responseStatus: number | null, err: string | null -) { - // Schedule next retry for failed deliveries using exponential backoff (5m, 30m, 2h, 12h). - // Without this column set, the retry queue has no anchor to know when to re-attempt. - const backoffMinutes = [5, 30, 120, 720]; - const nextAttempt = status === "delivered" || status === "dead_letter" - ? null - : new Date(Date.now() + (backoffMinutes[Math.min(attempts - 1, backoffMinutes.length - 1)] || 5) * 60_000) - .toISOString().replace("T", " ").slice(0, 19); - await db().execute({ - sql: `INSERT INTO webhook_deliveries - (endpoint_id, event_type, payload, idempotency_key, status, attempts, response_status, last_error, delivered_at, next_attempt_at) - VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?) - ON CONFLICT (endpoint_id, idempotency_key) DO UPDATE SET - status=excluded.status, attempts=webhook_deliveries.attempts+1, - response_status=excluded.response_status, last_error=excluded.last_error, - next_attempt_at=excluded.next_attempt_at`, - args: [endpointId, type, body, idem, status, attempts, responseStatus, err, status === "delivered" ? sqlTime() : null, nextAttempt], +/** + * Creates a project endpoint. `events` was already normalised with the app's + * own grammar (normalizeWebhookEventSubscriptions), which is wider than the + * hub's list of domain events, so the subscription is written to the store + * after the hub has made (and validated the URL of) the endpoint. + */ +export async function createProjectWebhook(projectId: string, url: string, events: string[]) { + const { endpoint, secret } = await hooks().createEndpoint(projectOwner(projectId), { url }); + if (!(events.length === 1 && events[0] === "*")) await libsqlStore(db, ensureSchema).updateEndpoint(endpoint.id, { events, updatedAt: Date.now() }); + return { endpoint: projectEndpointRow({ ...endpoint, events }), secret }; +} + +/** Pause or resume one project webhook without discarding its signing secret. */ +export async function setProjectWebhookActive(id: string, projectId: string, active: boolean): Promise { + try { await hooks().updateEndpoint(projectOwner(projectId), id, { enabled: active }); return true; } catch (e) { if (notFound(e)) return false; throw e; } +} + +/** Delete one project webhook and its delivery history. */ +export async function deleteProjectWebhook(id: string, projectId: string): Promise { + try { await hooks().deleteEndpoint(projectOwner(projectId), id); return true; } catch (e) { if (notFound(e)) return false; throw e; } +} + +/** Queue an event for a project's subscribed, enabled endpoints. */ +export async function dispatchEvent(projectId: string, type: string, data: unknown) { + return hooks().emit(projectOwner(projectId), type, data, { subject: projectId }); +} + +/** Send `webhook.test` now to every enabled endpoint of a project, whatever it subscribes to. */ +export async function testProjectWebhooks(projectId: string) { + const owner = projectOwner(projectId); + const enabled = (await hooks().listEndpoints(owner)).filter((e) => e.enabled); + const results = []; + for (const e of enabled) results.push(await hooks().test(owner, e.id)); + return results; +} + +/** Distinct outbound URLs across the given projects, to pre-fill a domain target. */ +export async function distinctWebhookUrls(projectIds: string[]): Promise { + if (!projectIds.length) return []; + await ensureSchema(); + const placeholders = projectIds.map(() => "?").join(","); + const r = await db().execute({ + sql: `SELECT url, MAX(created_at) AS latest FROM ${HUB_ENDPOINTS} WHERE owner IN (${placeholders}) GROUP BY url ORDER BY latest DESC LIMIT 25`, + args: projectIds.map(projectOwner), }); + return r.rows.map((x: any) => String(x.url)).filter(Boolean); } + diff --git a/package.json b/package.json index 752b02a..53c8a24 100644 --- a/package.json +++ b/package.json @@ -1,6 +1,6 @@ { "name": "moshcoding", - "version": "0.3.0", + "version": "0.4.0", "private": true, "description": "Code hard. Mosh harder. Next.js + Bun. Landing + one-line metal coming-soon pages with email waitlist and CoinPayPortal login.", "scripts": { @@ -18,6 +18,7 @@ "@profullstack/profile": "0.1.6", "@profullstack/redeem-credits": "0.2.0", "@profullstack/stack": "0.5.0", + "@profullstack/webhooks": "0.2.0", "mailchecker": "^6.0.21", "next": "16.4.3", "pg": "^8.23.1", diff --git a/tests/domain-webhook-active.test.mjs b/tests/domain-webhook-active.test.mjs index 23aa310..4f928f4 100644 --- a/tests/domain-webhook-active.test.mjs +++ b/tests/domain-webhook-active.test.mjs @@ -1,39 +1,110 @@ import assert from "node:assert/strict"; import test from "node:test"; -process.env.TURSO_DATABASE_URL = "file::memory:"; +// TEST_PG_URL=postgres://... runs the same suite on Postgres (through libsql-pg). +if (process.env.TEST_PG_URL) process.env.DATABASE_URL = process.env.TEST_PG_URL; +else { + process.env.TURSO_DATABASE_URL = "file::memory:"; + delete process.env.DATABASE_URL; +} -const { - activeDomainWebhooks, - addDomainWebhook, - listDomainWebhooks, - setDomainWebhookActive, -} = await import("../lib/db.ts"); +const wh = await import("../lib/webhooks.ts"); +const { isKnownDomain } = await import("../lib/db.ts"); -test("domain webhook targets can pause and resume without losing their configuration", async () => { - const webhook = await addDomainWebhook("example.test", "https://hooks.example.test/events", "whsec_keep_me"); +const calls = []; +wh.setHooksForTests( + wh.createHooks({ + fetchImpl: async (url, init) => { + calls.push({ url: String(url), init }); + return new Response(null, { status: 204 }); + }, + guard: { lookup: async () => ["93.184.216.34"] }, + }), +); + +const active = async (dn) => (await wh.listDomainWebhooks(dn)).filter((w) => w.active).map(({ url, secret }) => ({ url, secret })); +test("domain webhook targets can pause and resume without losing their configuration", async () => { + const webhook = await wh.addDomainWebhook("example.test", "https://hooks.example.test/events"); assert.equal(webhook.active, true); - assert.deepEqual(await activeDomainWebhooks("example.test"), [ - { url: "https://hooks.example.test/events", secret: "whsec_keep_me" }, - ]); + assert.match(webhook.secret, /^whsec_/); + const keep = { url: "https://hooks.example.test/events", secret: webhook.secret }; - assert.equal(await setDomainWebhookActive(webhook.id, "example.test", false), true); - assert.deepEqual(await activeDomainWebhooks("example.test"), []); + assert.deepEqual(await active("example.test"), [keep]); + assert.equal(await isKnownDomain("example.test"), true); + + assert.equal(await wh.setDomainWebhookActive(webhook.id, "example.test", false), true); + assert.deepEqual(await active("example.test"), []); assert.deepEqual( - (await listDomainWebhooks("example.test")).map(({ url, secret, active }) => ({ url, secret, active })), - [{ url: "https://hooks.example.test/events", secret: "whsec_keep_me", active: false }], + (await wh.listDomainWebhooks("example.test")).map(({ url, secret, active }) => ({ url, secret, active })), + [{ ...keep, active: false }], ); - assert.equal(await setDomainWebhookActive(webhook.id, "example.test", true), true); - assert.deepEqual(await activeDomainWebhooks("example.test"), [ - { url: "https://hooks.example.test/events", secret: "whsec_keep_me" }, - ]); + assert.equal(await wh.setDomainWebhookActive(webhook.id, "example.test", true), true); + assert.deepEqual(await active("example.test"), [keep]); }); test("a target cannot be changed through another domain", async () => { - const webhook = await addDomainWebhook("owner.test", "https://hooks.example.test/owner", "whsec_owner"); + const webhook = await wh.addDomainWebhook("owner.test", "https://hooks.example.test/owner"); + + assert.equal(await wh.setDomainWebhookActive(webhook.id, "other.test", false), false); + assert.equal(await wh.deleteDomainWebhook(webhook.id, "other.test"), false); + assert.equal(await wh.rotateDomainWebhookSecret(webhook.id, "other.test"), null); + assert.equal((await wh.listDomainWebhooks("owner.test"))[0].active, true); +}); + +test("domain events are queued, delivered as signed CloudEvents, and filtered by subscription", async () => { + const all = await wh.addDomainWebhook("events.test", "https://hooks.example.test/all"); + const bids = await wh.addDomainWebhook("events.test", "https://hooks.example.test/bids", "bid.*"); + calls.length = 0; + + await wh.fireDomainEvent("events.test", "waitlist.signup", { email: "a@b.test", dn: "events.test" }); + await wh.fireDomainEvent("events.test", "bid.placed", { dn: "events.test", amountCents: 500 }); + await wh.hooks().drain(); + + const to = (u) => calls.filter((c) => c.url === u).map((c) => JSON.parse(c.init.body).type); + assert.deepEqual(to("https://hooks.example.test/all"), ["waitlist.signup", "bid.placed"]); + assert.deepEqual(to("https://hooks.example.test/bids"), ["bid.placed"]); + + const first = calls[0]; + const body = JSON.parse(first.init.body); + assert.equal(body.specversion, "1.0"); + assert.equal(body.subject, "events.test"); + assert.equal(first.init.headers["x-moshcoding-hop"], "1"); + assert.match(first.init.headers["webhook-signature"], /^v1,/); + + const log = await wh.domainDeliveries("events.test"); + assert.equal(log.length, 3); + assert.ok(log.every((d) => d.status === "delivered")); + assert.ok(all.id && bids.id); +}); + +test("a relayed inbound event carries its hop count", async () => { + await wh.addDomainWebhook("relay.test", "https://hooks.example.test/relay"); + calls.length = 0; + await wh.fireDomainEvent("relay.test", "inbound.push", { body: "{}" }, { hop: 1 }); + await wh.hooks().drain(); + assert.equal(calls.length, 1); + assert.equal(calls[0].init.headers["x-moshcoding-hop"], "2"); + assert.equal(JSON.parse(calls[0].init.body).data.relay_hop, 2); +}); + +test("rotating a secret returns a new one", async () => { + const webhook = await wh.addDomainWebhook("rotate.test", "https://hooks.example.test/rotate"); + const next = await wh.rotateDomainWebhookSecret(webhook.id, "rotate.test"); + assert.ok(next && next !== webhook.secret); + assert.equal((await wh.listDomainWebhooks("rotate.test"))[0].secret, next); +}); + +test("deliveries to our own inbound receiver are refused", async () => { + // The default hub (no fetchImpl override) blocks self-referential URLs at send time. + const h = wh.createHooks({ guard: { lookup: async () => ["93.184.216.34"] } }); + const { endpoint } = await h.createEndpoint("dn:self.test", { url: "https://moshcoding.com/api/webhooks/self.test" }); + const r = await h.test("dn:self.test", endpoint.id); + assert.equal(r.ok, false); + assert.match(String(r.error), /self-referential/); +}); - assert.equal(await setDomainWebhookActive(webhook.id, "other.test", false), false); - assert.equal((await listDomainWebhooks("owner.test"))[0].active, true); +test("an unknown event pattern is refused", async () => { + await assert.rejects(() => wh.addDomainWebhook("bad.test", "https://hooks.example.test/x", "nope.thing"), /no event matches/); }); diff --git a/tests/project-webhook-management.test.mjs b/tests/project-webhook-management.test.mjs index 9fbcf7a..7a1da54 100644 --- a/tests/project-webhook-management.test.mjs +++ b/tests/project-webhook-management.test.mjs @@ -1,134 +1,94 @@ import assert from "node:assert/strict"; import test from "node:test"; -process.env.TURSO_DATABASE_URL = "file::memory:"; +// TEST_PG_URL=postgres://... runs the same suite on Postgres (through libsql-pg). +if (process.env.TEST_PG_URL) process.env.DATABASE_URL = process.env.TEST_PG_URL; +else { + process.env.TURSO_DATABASE_URL = "file::memory:"; + delete process.env.DATABASE_URL; +} -const { - db, - deleteProjectWebhook, - ensureSchema, - setProjectWebhookActive, -} = await import("../lib/db.ts"); -const { dispatchEvent } = await import("../lib/webhooks.ts"); +const { db, ensureSchema } = await import("../lib/db.ts"); +const wh = await import("../lib/webhooks.ts"); await ensureSchema(); -async function insertEndpoint(projectId, suffix) { - const id = `endpoint-${suffix}`; - await db().execute({ - sql: `INSERT INTO webhook_endpoints (id, project_id, url, secret, events) - VALUES (?, ?, ?, ?, '["*"]')`, - args: [id, projectId, `https://hooks.example.test/${suffix}`, `whsec_${suffix}`], - }); - return id; +// A hub whose fetch is a recorder and whose DNS check accepts the test hosts. +const calls = []; +wh.setHooksForTests( + wh.createHooks({ + fetchImpl: async (url, init) => { + calls.push({ url: String(url), init }); + return new Response(null, { status: 204 }); + }, + guard: { lookup: async () => ["93.184.216.34"] }, + }), +); + +async function create(projectId, suffix, events = ["*"]) { + const { endpoint, secret } = await wh.createProjectWebhook(projectId, `https://hooks.example.test/${suffix}`, events); + return { id: endpoint.id, endpoint, secret }; } test("project webhook targets pause and resume without losing configuration", async () => { - const id = await insertEndpoint("project-pause", "pause"); - const originalFetch = globalThis.fetch; - const calls = []; - globalThis.fetch = async (...args) => { - calls.push(args); - return new Response(null, { status: 204 }); - }; - - try { - assert.equal(await setProjectWebhookActive(id, "project-pause", false), true); - await dispatchEvent("project-pause", "build.finished", { ok: true }); - assert.equal(calls.length, 0); - - const paused = await db().execute({ - sql: `SELECT url, secret, events, active FROM webhook_endpoints WHERE id = ?`, - args: [id], - }); - assert.deepEqual( - { - url: String(paused.rows[0].url), - secret: String(paused.rows[0].secret), - events: String(paused.rows[0].events), - active: Number(paused.rows[0].active), - }, - { - url: "https://hooks.example.test/pause", - secret: "whsec_pause", - events: '["*"]', - active: 0, - }, - ); - - assert.equal(await setProjectWebhookActive(id, "project-pause", true), true); - await dispatchEvent("project-pause", "build.finished", { ok: true }); - assert.equal(calls.length, 1); - } finally { - globalThis.fetch = originalFetch; - } + const { id, secret } = await create("project-pause", "pause"); + calls.length = 0; + + assert.equal(await wh.setProjectWebhookActive(id, "project-pause", false), true); + assert.equal((await wh.dispatchEvent("project-pause", "build.finished", { ok: true })).queued, 0); + + const [paused] = await wh.listProjectWebhooks("project-pause"); + assert.deepEqual( + { url: paused.url, events: paused.events, active: paused.active }, + { url: "https://hooks.example.test/pause", events: '["*"]', active: 0 }, + ); + assert.equal((await wh.hooks().secret("project:project-pause", id)).secret, secret); + + assert.equal(await wh.setProjectWebhookActive(id, "project-pause", true), true); + assert.equal((await wh.dispatchEvent("project-pause", "build.finished", { ok: true })).queued, 1); + await wh.hooks().drain(); + assert.equal(calls.length, 1); + assert.equal(calls[0].url, "https://hooks.example.test/pause"); }); test("test deliveries reach active endpoints regardless of event subscriptions", async () => { - const id = "endpoint-subscription-test"; - await db().execute({ - sql: `INSERT INTO webhook_endpoints (id, project_id, url, secret, events) - VALUES (?, ?, ?, ?, ?)`, - args: [id, "project-subscription-test", "https://hooks.example.test/subscription-test", "whsec_subscription_test", '["build.finished"]'], - }); - const originalFetch = globalThis.fetch; - const calls = []; - globalThis.fetch = async (...args) => { - calls.push(args); - return new Response(null, { status: 204 }); - }; - - try { - const filtered = await dispatchEvent("project-subscription-test", "ping", { ok: true }); - assert.equal(filtered.length, 0); - assert.equal(calls.length, 0); - - const testDeliveries = await dispatchEvent( - "project-subscription-test", - "ping", - { ok: true }, - { bypassEventFilter: true }, - ); - assert.equal(testDeliveries.length, 1); - assert.equal(calls.length, 1); - } finally { - globalThis.fetch = originalFetch; - } + await create("project-subscription-test", "subscription-test", ["build.finished"]); + calls.length = 0; + + assert.equal((await wh.dispatchEvent("project-subscription-test", "ping", { ok: true })).queued, 0); + await wh.hooks().drain(); + assert.equal(calls.length, 0); + + const results = await wh.testProjectWebhooks("project-subscription-test"); + assert.equal(results.length, 1); + assert.equal(results[0].ok, true); + assert.equal(calls.length, 1); }); -test("project scoping prevents cross-project changes", async () => { - const id = await insertEndpoint("project-owner", "scoped"); +test("app-grammar subscriptions are kept on create", async () => { + const { endpoint } = await create("project-grammar", "grammar", ["deploy:done", "build.finished"]); + assert.equal(endpoint.events, '["deploy:done","build.finished"]'); + const [listed] = await wh.listProjectWebhooks("project-grammar"); + assert.equal(listed.events, '["deploy:done","build.finished"]'); +}); - assert.equal(await setProjectWebhookActive(id, "project-other", false), false); - assert.equal(await deleteProjectWebhook(id, "project-other"), false); +test("project scoping prevents cross-project changes", async () => { + const { id } = await create("project-owner", "scoped"); - const endpoint = await db().execute({ - sql: `SELECT active FROM webhook_endpoints WHERE id = ?`, - args: [id], - }); - assert.equal(Number(endpoint.rows[0].active), 1); + assert.equal(await wh.setProjectWebhookActive(id, "project-other", false), false); + assert.equal(await wh.deleteProjectWebhook(id, "project-other"), false); + assert.equal((await wh.listProjectWebhooks("project-owner"))[0].active, 1); }); test("deleting a project webhook also deletes its delivery history", async () => { - const id = await insertEndpoint("project-delete", "delete"); - await db().execute({ - sql: `INSERT INTO webhook_deliveries - (id, endpoint_id, event_type, payload, idempotency_key, status) - VALUES (?, ?, ?, ?, ?, 'failed')`, - args: ["delivery-delete", id, "build.failed", "{}", "event-delete"], - }); - - assert.equal(await deleteProjectWebhook(id, "project-delete"), true); - assert.equal(await deleteProjectWebhook(id, "project-delete"), false); - - const endpoints = await db().execute({ - sql: `SELECT 1 FROM webhook_endpoints WHERE id = ?`, - args: [id], - }); - const deliveries = await db().execute({ - sql: `SELECT 1 FROM webhook_deliveries WHERE endpoint_id = ?`, - args: [id], - }); + const { id } = await create("project-delete", "delete"); + await wh.dispatchEvent("project-delete", "build.failed", {}); + + assert.equal(await wh.deleteProjectWebhook(id, "project-delete"), true); + assert.equal(await wh.deleteProjectWebhook(id, "project-delete"), false); + + const endpoints = await db().execute({ sql: `SELECT 1 FROM hub_webhook_endpoints WHERE id = ?`, args: [id] }); + const deliveries = await db().execute({ sql: `SELECT 1 FROM hub_webhook_deliveries WHERE endpoint_id = ?`, args: [id] }); assert.equal(endpoints.rows.length, 0); assert.equal(deliveries.rows.length, 0); }); diff --git a/tests/webhook-hub-legacy.test.mjs b/tests/webhook-hub-legacy.test.mjs new file mode 100644 index 0000000..54181f4 --- /dev/null +++ b/tests/webhook-hub-legacy.test.mjs @@ -0,0 +1,122 @@ +import assert from "node:assert/strict"; +import crypto from "node:crypto"; +import test from "node:test"; +import { verifySignature } from "@profullstack/webhooks"; + +// TEST_PG_URL=postgres://... runs the same suite on Postgres (through libsql-pg). +if (process.env.TEST_PG_URL) process.env.DATABASE_URL = process.env.TEST_PG_URL; +else { + process.env.TURSO_DATABASE_URL = "file::memory:"; + delete process.env.DATABASE_URL; +} + +const { db, ensureSchema, migrateLegacyWebhooks } = await import("../lib/db.ts"); +const wh = await import("../lib/webhooks.ts"); +const { verifyWebhook } = await import("../lib/webhook-signing.ts"); +const { legacySecretToSpec } = await import("../lib/webhook-legacy.ts"); + +await ensureSchema(); + +const calls = []; +wh.setHooksForTests( + wh.createHooks({ + fetchImpl: async (url, init) => { + calls.push({ url: String(url), init }); + return new Response(null, { status: 204 }); + }, + guard: { lookup: async () => ["93.184.216.34"] }, + }), +); + +// Exactly what a receiver written against the pre-hub sender does: HMAC-SHA256 +// keyed with the whole secret string over "..", base64, "v1,". +function oldReceiverVerifies(headers, body, secret) { + const expected = `v1,${crypto.createHmac("sha256", secret).update(`${headers["webhook-id"]}.${headers["webhook-timestamp"]}.${body}`).digest("base64")}`; + return String(headers["webhook-signature"]).split(" ").includes(expected); +} + +test("pre-hub endpoints are copied once, keep their ids, and old receivers still verify", async () => { + const oldProjectSecret = "whsec_" + crypto.randomBytes(24).toString("base64url"); + const oldDomainSecret = "whsec_" + crypto.randomBytes(24).toString("base64url"); + await db().execute({ + sql: `INSERT INTO webhook_endpoints (id, project_id, url, secret, events, active, created_at) VALUES (?, ?, ?, ?, ?, 1, '2026-07-10 01:52:32')`, + args: ["legacyproj1", "p-legacy", "https://hooks.example.test/legacy-project", oldProjectSecret, '["build.finished"]'], + }); + await db().execute({ + sql: `INSERT INTO domain_webhooks (id, dn, url, secret, active) VALUES (?, ?, ?, ?, 1)`, + args: ["legacydom1", "legacy.test", "https://hooks.example.test/legacy-domain", oldDomainSecret], + }); + await db().execute({ + sql: `INSERT INTO domain_webhooks (id, dn, url, secret, active) VALUES (?, ?, ?, ?, 0)`, + args: ["legacydom2", "legacy.test", "https://hooks.example.test/legacy-paused", "whsec_paused"], + }); + + assert.equal(await migrateLegacyWebhooks(), 3); + assert.equal(await migrateLegacyWebhooks(), 0, "a second run copies nothing"); + + const project = await wh.listProjectWebhooks("p-legacy"); + assert.deepEqual(project.map(({ id, events, active }) => ({ id, events, active })), [ + { id: "legacyproj1", events: '["build.finished"]', active: 1 }, + ]); + assert.equal(project[0].created_at, "2026-07-10T01:52:32.000Z"); + const domain = await wh.listDomainWebhooks("legacy.test"); + assert.deepEqual(domain.map(({ id, active }) => ({ id, active })), [ + { id: "legacydom1", active: true }, + { id: "legacydom2", active: false }, + ]); + + // A domain event to the migrated target: the old receiver, holding the old + // secret, verifies it; so does any Standard Webhooks library with the shown one. + calls.length = 0; + await wh.fireDomainEvent("legacy.test", "waitlist.signup", { email: "x@y.test", dn: "legacy.test" }); + await wh.dispatchEvent("p-legacy", "build.finished", { ok: true }); + await wh.hooks().drain(); + assert.equal(calls.length, 2); + for (const c of calls) { + const old = c.url.endsWith("legacy-domain") ? oldDomainSecret : oldProjectSecret; + assert.equal(oldReceiverVerifies(c.init.headers, c.init.body, old), true, `old receiver verifies ${c.url}`); + assert.equal(verifyWebhook(c.init.headers, c.init.body, old), true, "the app's own old verifier agrees"); + assert.equal(verifySignature({ headers: c.init.headers, body: c.init.body, secret: legacySecretToSpec(old) }).ok, true); + } + assert.equal(domain[0].secret, legacySecretToSpec(oldDomainSecret)); +}); + +test("a target deleted after the copy is not copied back", async () => { + await db().execute({ + sql: `INSERT INTO domain_webhooks (id, dn, url, secret, active) VALUES (?, ?, ?, ?, 1)`, + args: ["legacydom3", "gone.test", "https://hooks.example.test/gone", "whsec_gone"], + }); + await migrateLegacyWebhooks(); + assert.equal(await wh.deleteDomainWebhook("legacydom3", "gone.test"), true); + await migrateLegacyWebhooks(); + assert.deepEqual(await wh.listDomainWebhooks("gone.test"), []); +}); + +test("a failed delivery is retried later, then delivered", async () => { + let fail = true; + const sent = []; + let now = Date.now(); + const h = new (await import("@profullstack/webhooks")).Webhooks({ + store: (await import("../lib/webhook-store.ts")).libsqlStore(db, ensureSchema), + fetchImpl: async (url) => { sent.push(String(url)); return new Response(null, { status: fail ? 503 : 200 }); }, + guard: { lookup: async () => ["93.184.216.34"] }, + now: () => now, + }); + const { endpoint } = await h.createEndpoint("dn:retry.test", { url: "https://hooks.example.test/retry" }); + await h.emit("dn:retry.test", "bid.placed", { n: 1 }); + await h.drain(); + let [d] = await h.deliveries("dn:retry.test"); + assert.equal(d.status, "pending"); + assert.equal(d.attempts, 1); + assert.equal(d.last_status, 503); + + fail = false; + await h.drain(); + assert.equal(sent.length, 1, "not due yet"); + now += 31_000; + await h.drain(); + [d] = await h.deliveries("dn:retry.test"); + assert.equal(d.status, "delivered"); + assert.equal(d.attempts, 2); + assert.equal((await h.getEndpoint("dn:retry.test", endpoint.id)).failures, 0); +}); diff --git a/tests/webhooks-api-route.test.mjs b/tests/webhooks-api-route.test.mjs new file mode 100644 index 0000000..057fe46 --- /dev/null +++ b/tests/webhooks-api-route.test.mjs @@ -0,0 +1,53 @@ +import assert from "node:assert/strict"; +import test from "node:test"; + +// TEST_PG_URL=postgres://... runs the same suite on Postgres (through libsql-pg). +if (process.env.TEST_PG_URL) process.env.DATABASE_URL = process.env.TEST_PG_URL; +else { + process.env.TURSO_DATABASE_URL = "file::memory:"; + delete process.env.DATABASE_URL; +} +process.env.SESSION_SECRET = "test-secret"; + +const { NextRequest } = await import("next/server"); +const route = await import("../app/api/v1/webhooks/[[...path]]/route.ts"); +const { createApiKey } = await import("../lib/apikeys.ts"); +const { db, ensureSchema } = await import("../lib/db.ts"); +const wh = await import("../lib/webhooks.ts"); + +await ensureSchema(); +wh.setHooksForTests(wh.createHooks({ fetchImpl: async () => new Response(null, { status: 204 }), guard: { lookup: async () => ["93.184.216.34"] } })); + +const req = (path, init = {}) => new NextRequest(new URL(path, "https://moshcoding.com"), init); + +test("/api/v1/webhooks answers 401 without a key", async () => { + const r = await route.GET(req("/api/v1/webhooks?dn=example.test")); + assert.equal(r.status, 401); +}); + +test("an API key manages the targets of a domain its account owns, and only those", async () => { + await db().execute({ sql: `INSERT INTO accounts (id, email, password_hash, password_salt, domain) VALUES (?, ?, ?, ?, ?)`, args: ["acct-api", "api@example.test", "x", "x", "mine.test"] }); + const { token } = await createApiKey("acct-api", "test"); + const auth = { authorization: `Bearer ${token}` }; + + assert.equal((await route.GET(req("/api/v1/webhooks?dn=theirs.test", { headers: auth }))).status, 403); + assert.equal((await route.GET(req("/api/v1/webhooks", { headers: auth }))).status, 400); + + const created = await route.POST(req("/api/v1/webhooks?dn=mine.test", { + method: "POST", headers: { ...auth, "content-type": "application/json" }, + body: JSON.stringify({ url: "https://hooks.example.test/api", events: ["bid.*"] }), + })); + assert.equal(created.status, 201); + const { endpoint, secret } = await created.json(); + assert.match(secret, /^whsec_/); + + const list = await (await route.GET(req("/api/v1/webhooks?dn=mine.test", { headers: auth }))).json(); + assert.deepEqual(list.endpoints.map((e) => e.id), [endpoint.id]); + assert.ok(list.event_types.includes("bid.placed")); + + const tested = await route.POST(req(`/api/v1/webhooks/${endpoint.id}/test?dn=mine.test`, { method: "POST", headers: auth })); + assert.equal((await tested.json()).ok, true); + + // The same target is invisible through a domain the key does not own. + assert.equal((await route.GET(req(`/api/v1/webhooks/${endpoint.id}?dn=theirs.test`, { headers: auth }))).status, 403); +});