diff --git a/__tests__/lib/autoblogReceiver.test.ts b/__tests__/lib/autoblogReceiver.test.ts new file mode 100644 index 0000000..07e1914 --- /dev/null +++ b/__tests__/lib/autoblogReceiver.test.ts @@ -0,0 +1,58 @@ +// @vitest-environment node +import { describe, it, expect } from 'vitest' +import { buildEvent, signRequest } from '@profullstack/autoblog' +import { verifyAutoblog } from '@/lib/autoblogReceiver' + +// The receiver moved from @profullstack/autoblog's verifyAndParse to +// @profullstack/webhooks. These requests are built with the autoblog SENDER's +// own code, so they prove crawlproof's existing deliveries still verify. + +const post = { + id: 'p1', + title: 'Hello', + slug: 'hello', + html: '

hi

', + status: 'published', + published_at: '2026-10-28T00:00:00Z', + tags: ['a'], + categories: [], +} + +function request(secret: string, opts: { bearer?: string; body?: string; ts?: number } = {}) { + const event = buildEvent(post as any, { source: 'https://crawlproof.com', eventId: 'evt_1' }) + const body = opts.body ?? JSON.stringify(event) + const ts = opts.ts ?? Math.floor(Date.now() / 1000) + const headers = new Headers({ + 'content-type': 'application/cloudevents+json', + authorization: `Bearer ${opts.bearer ?? secret}`, + ...signRequest({ id: event.id, timestamp: ts, body, secret }), + }) + return { headers, body } +} + +describe('verifyAutoblog', () => { + // The prod secret is 64 hex characters; a whsec_ one is the trap: the + // webhooks package would base64-decode it unless given the bytes. + for (const secret of ['371f7a'.padEnd(64, '0'), 'whsec_dGhpcyBpcyBub3QgdGhlIGtleQ==']) { + it(`accepts a delivery signed by @profullstack/autoblog (${secret.slice(0, 6)}…)`, () => { + const r = verifyAutoblog({ ...request(secret), secret }) + expect(r.ok).toBe(true) + if (r.ok) expect(r.post).toMatchObject({ id: 'p1', slug: 'hello' }) + }) + } + + it('refuses a wrong bearer, a wrong signature, a stale timestamp and a bad envelope', () => { + const secret = 'a'.repeat(64) + expect(verifyAutoblog({ ...request(secret, { bearer: 'b'.repeat(64) }), secret })).toMatchObject({ ok: false, status: 401 }) + const signedOther = request('c'.repeat(64)) + signedOther.headers.set('authorization', `Bearer ${secret}`) + expect(verifyAutoblog({ ...signedOther, secret })).toMatchObject({ ok: false, status: 401 }) + expect(verifyAutoblog({ ...request(secret, { ts: Math.floor(Date.now() / 1000) - 3600 }), secret })).toMatchObject({ + ok: false, + status: 401, + }) + const bad = JSON.stringify({ specversion: '1.0', id: 'x', type: 't', source: 's', data: { post: { id: 1 } } }) + expect(verifyAutoblog({ ...request(secret, { body: bad }), secret })).toMatchObject({ ok: false, status: 400 }) + expect(verifyAutoblog({ ...request(secret), secret: '' })).toMatchObject({ ok: false, status: 401 }) + }) +}) diff --git a/__tests__/lib/mcpServer.test.ts b/__tests__/lib/mcpServer.test.ts index e3ee6b9..ddc2978 100644 --- a/__tests__/lib/mcpServer.test.ts +++ b/__tests__/lib/mcpServer.test.ts @@ -55,6 +55,12 @@ vi.mock('@/lib/credits', () => ({ createInviteFor: (...args: unknown[]) => createInviteFor(...args), })) +// Webhooks: the real hub on an in-memory store, so the tools run end to end. +const webhookHub = vi.hoisted(() => ({ current: null as any })) +vi.mock('@/lib/webhooks/hub', async (orig) => ({ ...(await orig()), getHooks: () => webhookHub.current })) +const { memoryStore } = await import('@profullstack/webhooks') +const { createHooks } = await import('@/lib/webhooks/hub') + const { handleMcpMessage, TOOLS, DEFAULT_PROTOCOL_VERSION } = await import( '@/lib/mcpServer' ) @@ -158,7 +164,9 @@ describe('tools/list', () => { expect(tools.map((t) => t.name).sort()).toEqual([ 'create_invite', 'create_listing', + 'create_webhook', 'delete_listing', + 'delete_webhook', 'get_my_listing', 'get_product', 'get_vocabulary', @@ -166,12 +174,17 @@ describe('tools/list', () => { 'karma_rules', 'list_categories', 'list_my_listings', + 'list_webhooks', 'my_invites', 'my_karma', 'redeem_credit_code', 'redeem_karma_credit', + 'rotate_webhook_secret', 'search_products', + 'test_webhook', 'update_listing', + 'update_webhook', + 'webhook_deliveries', ]) for (const tool of tools) expect(tool.inputSchema).toBeTruthy() }) @@ -482,4 +495,44 @@ describe('credit codes and invites over MCP', () => { ) expect(createInviteFor).toHaveBeenCalledWith(principal, 'for Carol', request, 'http://localhost:3000') }) + + describe('webhook tools', () => { + const call = (name: string, args: Record = {}) => + handleMcpMessage({ id: 1, method: 'tools/call', params: { name, arguments: args } }, { principal }) + + beforeEach(() => { + webhookHub.current = createHooks(memoryStore(), { log: () => {}, guard: { lookup: async () => ['93.184.216.34'] } }) + }) + + it('creates, lists, rotates and deletes an endpoint for the caller', async () => { + const created = parseToolText(await call('create_webhook', { url: 'https://hooks.example.com/x', events: ['product.*'] })) + expect(created.secret).toMatch(/^whsec_/) + const id = created.endpoint.id + + const listed = parseToolText(await call('list_webhooks')) + expect(listed.endpoints.map((e: { id: string }) => e.id)).toEqual([id]) + expect(listed.event_types.map((e: { type: string }) => e.type)).toContain('karma.changed') + expect(await webhookHub.current.listEndpoints('u1')).toHaveLength(1) + + const rotated = parseToolText(await call('rotate_webhook_secret', { id })) + expect(rotated.secret).not.toBe(created.secret) + + const updated = parseToolText(await call('update_webhook', { id, enabled: false })) + expect(updated.enabled).toBe(false) + + expect(parseToolText(await call('delete_webhook', { id }))).toEqual({ deleted: true }) + }) + + it("turns the hub's own errors into tool errors", async () => { + const bad = await call('create_webhook', { url: 'https://hooks.example.com/x', events: ['nope.event'] }) + expect(resultOf(bad).isError).toBe(true) + const missing = await call('test_webhook', { id: 'wh_missing' }) + expect(resultOf(missing).isError).toBe(true) + }) + + it('needs an account', async () => { + const res = await handleMcpMessage({ id: 1, method: 'tools/call', params: { name: 'list_webhooks', arguments: {} } }) + expect(resultOf(res).isError).toBe(true) + }) + }) }) diff --git a/__tests__/lib/webhooks/api.test.ts b/__tests__/lib/webhooks/api.test.ts new file mode 100644 index 0000000..ea86833 --- /dev/null +++ b/__tests__/lib/webhooks/api.test.ts @@ -0,0 +1,57 @@ +// @vitest-environment node +import { describe, it, expect, vi, beforeEach } from 'vitest' +import { memoryStore } from '@profullstack/webhooks' +import { createHooks } from '@/lib/webhooks/hub' + +const auth = vi.hoisted(() => ({ principal: null as null | { userId: string } })) +const hooks = vi.hoisted(() => ({ current: null as any })) + +vi.mock('@/lib/apiAuth', async () => { + const { ApiError } = await import('@/lib/publicApi') + return { + requirePrincipal: async () => { + if (!auth.principal) throw new ApiError(401, 'Authentication required.') + return auth.principal + }, + } +}) +vi.mock('@/lib/webhooks/hub', async (orig) => ({ ...(await orig()), getHooks: () => hooks.current })) + +const { GET, POST, DELETE } = await import('../../../app/api/v1/webhooks/[[...path]]/route') + +const req = (path: string, init: RequestInit = {}) => new Request(`https://saasrow.com/api/v1/webhooks${path}`, init) + +describe('/api/v1/webhooks', () => { + beforeEach(() => { + auth.principal = null + hooks.current = createHooks(memoryStore(), { log: () => {}, guard: { lookup: async () => ['93.184.216.34'] } }) + }) + + it('answers 401 without a credential', async () => { + const res = await GET(req('')) + expect(res.status).toBe(401) + expect(res.headers.get('access-control-allow-origin')).toBe('*') + }) + + it("scopes every call to the caller's account", async () => { + auth.principal = { userId: 'u1' } + const created = await POST( + req('', { method: 'POST', body: JSON.stringify({ url: 'https://hooks.example.com/x', events: ['product.*'] }) }), + ) + expect(created.status).toBe(201) + const { endpoint, secret } = await created.json() + expect(secret).toMatch(/^whsec_/) + + const list = await (await GET(req(''))).json() + expect(list.endpoints.map((e: { id: string }) => e.id)).toEqual([endpoint.id]) + expect(list.event_types).toContain('product.upvoted') + + auth.principal = { userId: 'u2' } + expect((await (await GET(req(''))).json()).endpoints).toEqual([]) + expect((await GET(req(`/${endpoint.id}`))).status).toBe(404) + expect((await DELETE(req(`/${endpoint.id}`, { method: 'DELETE' }))).status).toBe(404) + + auth.principal = { userId: 'u1' } + expect((await DELETE(req(`/${endpoint.id}`, { method: 'DELETE' }))).status).toBe(200) + }) +}) diff --git a/__tests__/lib/webhooks/fakeSupabase.ts b/__tests__/lib/webhooks/fakeSupabase.ts new file mode 100644 index 0000000..b909939 --- /dev/null +++ b/__tests__/lib/webhooks/fakeSupabase.ts @@ -0,0 +1,92 @@ +/** + * A tiny in-memory stand-in for the supabase-js query builder: just the calls + * supabaseStore makes (select/insert/update/delete, eq/neq/lt, order, limit, + * maybeSingle, in) plus the two claim RPCs. Enough to run the real hub end to end. + */ +type Row = Record; + +export function fakeSupabase() { + const tables = new Map(); + const rows = (t: string) => { + if (!tables.has(t)) tables.set(t, []); + return tables.get(t)!; + }; + const calls: string[] = []; + + function from(table: string) { + calls.push(table); + let op: "select" | "insert" | "update" | "delete" = "select"; + let payload: Row | Row[] | null = null; + const filters: Array<(r: Row) => boolean> = []; + let orderBy: { col: string; asc: boolean } | null = null; + let lim: number | null = null; + let single = false; + + const run = () => { + const all = rows(table); + if (op === "insert") { + const list = Array.isArray(payload) ? payload : [payload!]; + for (const r of list) { + if (all.some((x) => x.id === r.id)) return { data: null, error: { message: "duplicate key" } }; + } + all.push(...list.map((r) => ({ ...r }))); + return { data: null, error: null }; + } + const hit = all.filter((r) => filters.every((f) => f(r))); + if (op === "update") { + for (const r of hit) Object.assign(r, payload); + return { data: null, error: null }; + } + if (op === "delete") { + tables.set(table, all.filter((r) => !hit.includes(r))); + return { data: null, error: null }; + } + let out = hit.map((r) => ({ ...r })); + if (orderBy) { + const { col, asc } = orderBy; + out.sort((a, b) => (Number(a[col]) - Number(b[col])) * (asc ? 1 : -1)); + } + if (lim !== null) out = out.slice(0, lim); + if (single) return { data: out[0] ?? null, error: null }; + return { data: out, error: null }; + }; + + const q: any = { + select: () => q, + insert: (p: Row | Row[]) => ((op = "insert"), (payload = p), q), + update: (p: Row) => ((op = "update"), (payload = p), q), + delete: () => ((op = "delete"), q), + eq: (c: string, v: unknown) => (filters.push((r) => r[c] === v), q), + neq: (c: string, v: unknown) => (filters.push((r) => r[c] !== v), q), + in: (c: string, vs: unknown[]) => (filters.push((r) => vs.includes(r[c])), q), + lt: (c: string, v: number) => (filters.push((r) => Number(r[c]) < v), q), + order: (col: string, o: { ascending?: boolean } = {}) => ((orderBy = { col, asc: o.ascending !== false }), q), + limit: (n: number) => ((lim = n), q), + maybeSingle: () => ((single = true), q), + then: (res: (v: unknown) => unknown, rej?: (e: unknown) => unknown) => Promise.resolve(run()).then(res, rej), + }; + return q; + } + + async function rpc(fn: string, args: Record) { + if (fn === "claim_hub_webhook_events") { + const now = Date.now(); + const due = rows("hub_webhook_events") + .filter((r) => !r.claimed_until || Number(r.claimed_until) < now) + .sort((a, b) => Number(a.id) - Number(b.id)) + .slice(0, Number(args.p_limit)); + for (const r of due) r.claimed_until = now + 120_000; + return { data: due.map((r) => ({ ...r })), error: null }; + } + if (fn !== "claim_hub_webhook_deliveries") return { data: null, error: { message: `no rpc ${fn}` } }; + const { p_now, p_limit, p_lease_until } = args as { p_now: number; p_limit: number; p_lease_until: number }; + const due = rows("hub_webhook_deliveries") + .filter((r) => r.status === "pending" && Number(r.next_at) <= p_now) + .sort((a, b) => Number(a.next_at) - Number(b.next_at)) + .slice(0, p_limit); + for (const r of due) r.next_at = p_lease_until; + return { data: due.map((r) => ({ ...r })), error: null }; + } + + return { from, rpc, tables, rows, calls }; +} diff --git a/__tests__/lib/webhooks/hub.test.ts b/__tests__/lib/webhooks/hub.test.ts new file mode 100644 index 0000000..4573c67 --- /dev/null +++ b/__tests__/lib/webhooks/hub.test.ts @@ -0,0 +1,103 @@ +// @vitest-environment node +import { describe, it, expect } from 'vitest' +import { verifyAndParse } from '@profullstack/webhooks' +import { createHooks, pumpEvents, WEBHOOK_EVENT_TYPES, WEBHOOK_SOURCE, EVENTS_TABLE } from '@/lib/webhooks/hub' +import { supabaseStore } from '@/lib/webhooks/supabase-store' +import { fakeSupabase } from './fakeSupabase' + +// The whole path a trigger's outbox row takes: claim, emit, drain, deliver, +// and a receiver verifying it with the secret it was shown. + +function setup() { + const sb = fakeSupabase() + const sent: Array<{ url: string; headers: Record; body: string }> = [] + const fetchImpl = (async (url: string, init: RequestInit) => { + sent.push({ url: String(url), headers: init.headers as Record, body: String(init.body) }) + return new Response('ok', { status: 200 }) + }) as unknown as typeof fetch + const hooks = createHooks(supabaseStore(sb), { + fetchImpl, + guard: { allowHttp: true, allowPrivate: true }, + log: () => {}, + }) + return { sb, hooks, sent } +} + +const outbox = (sb: ReturnType, row: Record) => sb.rows(EVENTS_TABLE).push(row) + +describe('saasrow webhooks hub', () => { + it('emits SaaSRow events and validates endpoint patterns against them', async () => { + const { hooks } = setup() + expect(hooks.eventTypes).toEqual([...WEBHOOK_EVENT_TYPES]) + await expect(hooks.createEndpoint('u1', { url: 'https://hooks.example.com/x', events: ['gig.*'] })).rejects.toThrow() + const { endpoint, secret } = await hooks.createEndpoint('u1', { url: 'https://hooks.example.com/x', events: ['product.*'] }) + expect(endpoint.events).toEqual(['product.*']) + expect(secret).toMatch(/^whsec_/) + }) + + it('turns an outbox row into a signed CloudEvent delivery and removes the row', async () => { + const { sb, hooks, sent } = setup() + const { secret } = await hooks.createEndpoint('owner-1', { url: 'https://hooks.example.com/saasrow', events: ['product.upvoted'] }) + outbox(sb, { + id: 7, + owner: 'owner-1', + type: 'product.upvoted', + subject: 'product:abc', + data: { product: { id: 'abc', title: 'Thing', upvotes: 3 }, vote: { id: 'v1', member: true } }, + claimed_until: null, + }) + // Another owner's event and an event this endpoint did not subscribe to. + outbox(sb, { id: 8, owner: 'owner-2', type: 'product.upvoted', subject: null, data: {}, claimed_until: null }) + outbox(sb, { id: 9, owner: 'owner-1', type: 'karma.changed', subject: null, data: { points: 1 }, claimed_until: null }) + + expect(await pumpEvents(hooks, sb)).toBe(3) + expect(sb.rows(EVENTS_TABLE)).toHaveLength(0) + expect(sb.rows('hub_webhook_deliveries')).toHaveLength(1) + + await hooks.drain() + expect(sent).toHaveLength(1) + const [d] = sent + expect(d.url).toBe('https://hooks.example.com/saasrow') + + const parsed = verifyAndParse({ headers: d.headers, body: d.body, secret }) + expect(parsed.ok).toBe(true) + if (!parsed.ok) return + expect(parsed.id).toBe('evt_saasrow_7') + expect(parsed.event).toMatchObject({ + specversion: '1.0', + id: 'evt_saasrow_7', + type: 'product.upvoted', + source: WEBHOOK_SOURCE, + subject: 'product:abc', + data: { product: { id: 'abc', title: 'Thing', upvotes: 3 }, vote: { member: true } }, + }) + expect(typeof parsed.event.time).toBe('string') + expect(sb.rows('hub_webhook_deliveries')[0]).toMatchObject({ status: 'delivered', attempts: 1 }) + }) + + it('a wrong secret does not verify', async () => { + const { sb, hooks, sent } = setup() + await hooks.createEndpoint('owner-1', { url: 'https://hooks.example.com/a' }) + outbox(sb, { id: 1, owner: 'owner-1', type: 'product.approved', subject: null, data: {}, claimed_until: null }) + await pumpEvents(hooks, sb) + await hooks.drain() + const other = (await hooks.createEndpoint('owner-9', { url: 'https://hooks.example.com/b' })).secret + expect(verifyAndParse({ headers: sent[0].headers, body: sent[0].body, secret: other }).ok).toBe(false) + }) + + it('leaves a row it could not emit for the next pump', async () => { + const { sb, hooks } = setup() + const broken = { ...hooks, emit: async () => { throw new Error('db down') } } as unknown as typeof hooks + outbox(sb, { id: 1, owner: 'o', type: 'product.approved', subject: null, data: {}, claimed_until: null }) + expect(await pumpEvents(broken, sb)).toBe(1) + expect(sb.rows(EVENTS_TABLE)).toHaveLength(1) + // Claimed: the same pump does not take it again until the lease runs out. + expect(await pumpEvents(hooks, sb)).toBe(0) + }) + + it('throws when the claim RPC fails rather than dropping events', async () => { + const { hooks } = setup() + const sb = { from: () => ({}), rpc: async () => ({ data: null, error: { message: 'boom' } }) } + await expect(pumpEvents(hooks, sb as any)).rejects.toThrow(/claim: boom/) + }) +}) diff --git a/__tests__/lib/webhooks/supabaseStore.test.ts b/__tests__/lib/webhooks/supabaseStore.test.ts new file mode 100644 index 0000000..b8e68d1 --- /dev/null +++ b/__tests__/lib/webhooks/supabaseStore.test.ts @@ -0,0 +1,98 @@ +// @vitest-environment node +import { describe, it, expect } from "vitest"; +import type { Delivery, Endpoint } from "@profullstack/webhooks"; +import { supabaseStore, HUB_TABLES } from "@/lib/webhooks/supabase-store"; +import { fakeSupabase } from "./fakeSupabase"; + +const endpoint = (over: Partial = {}): Endpoint => ({ + id: "wh_1", + owner: "user-1", + url: "https://hooks.example.com/", + secret: "whsec_abc=", + events: ["product.*"], + description: "", + enabled: true, + failures: 0, + disabledReason: null, + createdAt: 1000, + updatedAt: 1000, + ...over, +}); + +const delivery = (over: Partial = {}): Delivery => ({ + id: "dlv_1", + endpointId: "wh_1", + owner: "user-1", + eventId: "ev_1", + eventType: "product.upvoted", + payload: "{}", + status: "pending", + attempts: 0, + nextAt: 5000, + lastStatus: null, + lastError: null, + createdAt: 1000, + updatedAt: 1000, + ...over, +}); + +describe("supabaseStore", () => { + it("round-trips endpoints through the hub_ table columns", async () => { + const sb = fakeSupabase(); + const store = supabaseStore(sb); + await store.insertEndpoint(endpoint()); + expect(sb.rows(HUB_TABLES.endpoints)[0]).toMatchObject({ events: '["product.*"]', enabled: 1, disabled_reason: null, created_at: 1000 }); + expect(await store.getEndpoint("wh_1")).toEqual(endpoint()); + expect(await store.getEndpoint("nope")).toBeNull(); + + await store.updateEndpoint("wh_1", { enabled: false, failures: 3, disabledReason: "off", events: ["*"], updatedAt: 2000 }); + expect(await store.getEndpoint("wh_1")).toMatchObject({ enabled: false, failures: 3, disabledReason: "off", events: ["*"], updatedAt: 2000 }); + + await store.insertEndpoint(endpoint({ id: "wh_2", createdAt: 500 })); + await store.insertEndpoint(endpoint({ id: "wh_3", owner: "user-2" })); + expect((await store.listEndpoints("user-1")).map((e) => e.id)).toEqual(["wh_2", "wh_1"]); + }); + + it("claims due deliveries once and leases them", async () => { + const sb = fakeSupabase(); + const store = supabaseStore(sb); + await store.insertDeliveries([delivery(), delivery({ id: "dlv_2", nextAt: 9000 }), delivery({ id: "dlv_3", status: "delivered", nextAt: 0 })]); + const first = await store.claimDue(6000, 10, 60_000); + expect(first.map((d) => d.id)).toEqual(["dlv_1"]); + expect(await store.claimDue(6000, 10, 60_000)).toEqual([]); + expect((await store.getDelivery("dlv_1"))!.nextAt).toBe(66_000); + }); + + it("lists, updates, prunes and deletes deliveries", async () => { + const sb = fakeSupabase(); + const store = supabaseStore(sb); + await store.insertEndpoint(endpoint()); + await store.insertDeliveries([ + delivery({ id: "a", createdAt: 1 }), + delivery({ id: "b", createdAt: 2, endpointId: "wh_other" }), + delivery({ id: "c", createdAt: 3, owner: "user-2" }), + ]); + expect((await store.listDeliveries({ owner: "user-1" })).map((d) => d.id)).toEqual(["b", "a"]); + expect((await store.listDeliveries({ owner: "user-1", endpointId: "wh_1" })).map((d) => d.id)).toEqual(["a"]); + expect((await store.listDeliveries({ owner: "user-1", limit: 1 })).map((d) => d.id)).toEqual(["b"]); + + await store.updateDelivery("a", { status: "delivered", attempts: 1, lastStatus: 200, updatedAt: 10 }); + expect(await store.getDelivery("a")).toMatchObject({ status: "delivered", attempts: 1, lastStatus: 200 }); + + await store.prune(100); + expect((await store.listDeliveries({ owner: "user-1" })).map((d) => d.id)).toEqual(["b"]); + + await store.insertDeliveries([delivery({ id: "d" })]); + await store.deleteEndpoint("wh_1"); + expect(await store.getEndpoint("wh_1")).toBeNull(); + expect(await store.getDelivery("d")).toBeNull(); + expect(await store.getDelivery("b")).not.toBeNull(); + }); + + it("throws the Supabase error message instead of losing a write", async () => { + const sb = fakeSupabase(); + const store = supabaseStore(sb); + await store.insertEndpoint(endpoint()); + await expect(store.insertEndpoint(endpoint())).rejects.toThrow(/insert endpoint: duplicate key/); + }); +}); diff --git a/app/api/autoblog-webhook/route.ts b/app/api/autoblog-webhook/route.ts index 9fbba72..bb3a3b9 100644 --- a/app/api/autoblog-webhook/route.ts +++ b/app/api/autoblog-webhook/route.ts @@ -1,5 +1,5 @@ import { createClient } from '@supabase/supabase-js' -import { verifyAndParse } from '@profullstack/autoblog' +import { verifyAutoblog } from '@/lib/autoblogReceiver' function slugify(text: string): string { return text @@ -37,13 +37,7 @@ export async function POST(req: Request) { } const body = await req.text() - const headers = Object.fromEntries(req.headers) - - const result = verifyAndParse({ - headers, - body, - opts: { secret: config.webhook_secret }, - }) + const result = verifyAutoblog({ headers: req.headers, body, secret: config.webhook_secret }) if (!result.ok) { return new Response(result.reason ?? 'Unauthorized', { status: result.status ?? 401 }) diff --git a/app/api/v1/route.ts b/app/api/v1/route.ts index 72947ce..200c9b3 100644 --- a/app/api/v1/route.ts +++ b/app/api/v1/route.ts @@ -2,6 +2,7 @@ import { jsonResponse, corsPreflight, API_VERSION } from '@/lib/publicApi' import { siteUrl } from '@/lib/structuredData' import { MAX_PAGE_SIZE, DEFAULT_PAGE_SIZE } from '@/lib/distribution' import { LISTING_LIMITS } from '@/lib/listingInput' +import { WEBHOOK_EVENT_TYPES } from '@/lib/webhooks/events' export const dynamic = 'force-static' @@ -109,6 +110,22 @@ export function GET() { authentication: 'required', description: 'PATCH {"name"} renames a key. DELETE revokes it immediately.', }, + webhooks: { + url: `${base}/api/v1/webhooks`, + methods: ['GET', 'POST'], + authentication: 'required', + description: + 'Your webhook endpoints. GET lists them and event_types; POST {"url","events"?,"description"?} adds one and returns its signing secret once. SaaSRow POSTs CloudEvents signed with Standard Webhooks headers (webhook-id, webhook-timestamp, webhook-signature) for: ' + + WEBHOOK_EVENT_TYPES.join(', ') + + '. Failed deliveries retry with backoff for about three days.', + }, + webhook: { + url: `${base}/api/v1/webhooks/{id}`, + methods: ['GET', 'PATCH', 'DELETE'], + authentication: 'required', + description: + 'One endpoint. PATCH {"url"?,"events"?,"description"?,"enabled"?}. Also GET {id}/secret, POST {id}/rotate, POST {id}/test, GET {id}/deliveries, GET /api/v1/webhooks/deliveries, POST /api/v1/webhooks/deliveries/{delivery id}/retry.', + }, cli_login: { url: `${base}/api/v1/auth/cli`, method: 'POST', diff --git a/app/api/v1/webhooks/[[...path]]/route.ts b/app/api/v1/webhooks/[[...path]]/route.ts new file mode 100644 index 0000000..c817904 --- /dev/null +++ b/app/api/v1/webhooks/[[...path]]/route.ts @@ -0,0 +1,32 @@ +import { corsPreflight, errorResponse, errorToResponse } from '@/lib/publicApi' +import { handleWebhooksRequest } from '@/lib/webhooks/api' + +export const runtime = 'nodejs' +export const dynamic = 'force-dynamic' + +/** + * /api/v1/webhooks: the @profullstack/webhooks API, scoped to the caller's + * account. Same credentials as the rest of the write API (src/lib/apiAuth.ts): + * an API key (Authorization: Bearer sr_…), a listing management token + * (X-Management-Token), or the signed-in website member. + * + * GET / endpoints + event_types + * POST / { url, events?, description? } -> { endpoint, secret } + * GET|PATCH|DELETE /:id + * GET /:id/secret POST /:id/rotate POST /:id/test + * GET /:id/deliveries GET /deliveries POST /deliveries/:id/retry + */ +async function handle(req: Request): Promise { + try { + return (await handleWebhooksRequest(req)) ?? errorResponse('Not found', 404) + } catch (error) { + return errorToResponse(error, 'api/v1/webhooks') + } +} + +export const OPTIONS = () => corsPreflight() +export const GET = handle +export const POST = handle +export const PATCH = handle +export const PUT = handle +export const DELETE = handle diff --git a/bun.lock b/bun.lock index 7cd67ab..aec0a49 100644 --- a/bun.lock +++ b/bun.lock @@ -7,13 +7,13 @@ "dependencies": { "@netlify/plugin-nextjs": "^5.16.4", "@profullstack/affiliates": "^0.3.0", - "@profullstack/autoblog": "^0.4.0", "@profullstack/footer": "latest", "@profullstack/form-guard": "^0.2.2", "@profullstack/redeem-credits": "^0.2.0", "@profullstack/referrals": "^0.1.0", "@profullstack/stack": "^0.5.0", "@profullstack/throttle": "^0.4.0", + "@profullstack/webhooks": "0.2.1", "@profullstack/x402-gateway": "^0.15.2", "@radix-ui/react-slot": "^1.5.0", "@supabase/ssr": "^0.12.7", @@ -33,6 +33,7 @@ "tailwindcss-animate": "^1.0.7", }, "devDependencies": { + "@profullstack/autoblog": "^0.4.0", "@testing-library/jest-dom": "^7.0.1", "@testing-library/react": "^16.3.3", "@testing-library/user-event": "^14.6.8", @@ -254,6 +255,8 @@ "@profullstack/throttle": ["@profullstack/throttle@0.4.0", "", { "dependencies": { "@profullstack/x402-gateway": "^0.6.0 || ^0.7.0 || ^0.8.0 || ^0.9.0 || ^0.10.0 || ^0.11.0 || ^0.12.0 || ^0.13.0 || ^0.14.0 || ^0.15.0" } }, "sha512-Lj+XHt51OxyaJ2FqosZDyr9Z5RFB8z+ARixpoUfrDu3YlvFlct5GbbZM/C0qLZkUgzeuFSDvtYt9f0NNfBkFqA=="], + "@profullstack/webhooks": ["@profullstack/webhooks@0.2.1", "", {}, "sha512-QHeSkpu4O44oJ620qhQqxqFlD3u/OhupmXRnBHw4MGKuDTa0R+Y/727nnF7psSHxxSfnbWfNuVl15z3IXuaZiw=="], + "@profullstack/x402-gateway": ["@profullstack/x402-gateway@0.15.2", "", { "dependencies": { "@profullstack/keys": "^0.1.0" } }, "sha512-f0el7rPd13StA04OxA0PGrrG0kxg/vvGPtvI0XYn8wlTGPut3G+2UKcMRzSxQdaF4LkYAOLnkg9S+LhmIKvDBA=="], "@puppeteer/browsers": ["@puppeteer/browsers@3.2.6", "", { "dependencies": { "modern-tar": "^0.8.6", "yargs": "^18.0.0" }, "peerDependencies": { "proxy-agent": ">=8.0.1", "yauzl": "^2.10.0 || ^3.4.0" }, "optionalPeers": ["proxy-agent", "yauzl"], "bin": { "browsers": "lib/main-cli.js" } }, "sha512-MJP29gVoHznmqmwhH1tlZqzfeQ3aylTpSSa6WKZXtkwYUH02i9imBmWGTyZrbnAou5paX78SE4jMAUqnEPf8ug=="], diff --git a/cli/README.md b/cli/README.md index b558cbe..5cdc2b3 100644 --- a/cli/README.md +++ b/cli/README.md @@ -81,6 +81,29 @@ claude mcp add --transport http saasrow https://www.saasrow.com/api/mcp \ --header "Authorization: Bearer sr_…" ``` +## Webhooks + +SaaSRow POSTs a signed event to your URL when your listings are approved, +rejected, featured, upvoted or commented on, or your karma changes. + +```sh +saasrow webhooks # endpoints + the event types +saasrow webhooks add https://you.example/hook product.upvoted product.commented + # prints the signing secret once +saasrow webhooks test wh_… # send webhook.test now +saasrow webhooks log [wh_…] # recent deliveries +saasrow webhooks update wh_… --events 'product.*' --disable +saasrow webhooks rotate wh_… # new secret; the old one stops verifying +saasrow webhooks rm wh_… +``` + +Events: `product.approved`, `product.rejected`, `product.featured`, +`product.upvoted`, `product.commented`, `karma.changed`. Bodies are +CloudEvents JSON (`type`, `time`, `subject`, `data`) with Standard Webhooks +headers (`webhook-id`, `webhook-timestamp`, `webhook-signature`); verify them +with `verifyAndParse` from `@profullstack/webhooks` and the secret. Failed +deliveries retry with backoff for about three days. + ## Credits and invites ```sh diff --git a/cli/package.json b/cli/package.json index b700c39..e092818 100644 --- a/cli/package.json +++ b/cli/package.json @@ -1,6 +1,6 @@ { "name": "@profullstack/saasrow", - "version": "0.3.0", + "version": "0.4.0", "type": "module", "description": "Command-line client for saasrow.com: search the software directory, and create and manage your free listings and API keys from the terminal.", "keywords": ["saasrow", "saas", "directory", "listing", "cli", "mcp", "agent"], diff --git a/cli/src/args.mjs b/cli/src/args.mjs index dacd19b..c219129 100644 --- a/cli/src/args.mjs +++ b/cli/src/args.mjs @@ -2,7 +2,7 @@ // positionals. No dependency, no magic; every command declares nothing and // reads what it needs from `flags`. -const BOOLEAN_FLAGS = new Set(['json', 'yes', 'help', 'version', 'all', 'revoke']) +const BOOLEAN_FLAGS = new Set(['json', 'yes', 'help', 'version', 'all', 'revoke', 'enable', 'disable']) const ALIASES = { y: 'yes', h: 'help', v: 'version' } diff --git a/cli/src/cli.mjs b/cli/src/cli.mjs index 67d1030..f96d1d2 100644 --- a/cli/src/cli.mjs +++ b/cli/src/cli.mjs @@ -3,7 +3,7 @@ // and a scripted prompt. import { hostname } from 'node:os' -import { parseArgs } from './args.mjs' +import { parseArgs, listFlag } from './args.mjs' import { createClient, ApiClientError, VERSION } from './client.mjs' import { loadConfig, @@ -46,6 +46,19 @@ Karma karma rules how karma is earned leaderboard [--period week|month|all] [--limit n] +Webhooks (require login) + webhooks your endpoints and the events SaaSRow sends + webhooks add [events…] [--description …] + POST signed events to your URL; prints the secret once + webhooks get + webhooks update [--url …] [--events a,b] [--description …] [--enable|--disable] + webhooks test send a webhook.test event now + webhooks secret print the signing secret + webhooks rotate [--yes] replace the secret (the old one stops verifying) + webhooks log [id] [--limit n] recent deliveries + webhooks retry send a delivery again + webhooks rm [--yes] + Credits and invites (require login) redeem redeem a credit code for days of Premium invites your invites, what is left, your affiliate link @@ -461,7 +474,138 @@ async function invites({ client, positionals, flags, io }) { return 0 } -const COMMANDS = { login, logout, whoami, keys, listings, karma, leaderboard, redeem, invites, search, categories, vocabulary } +// --------------------------------------------------------------------------- + +const enc = encodeURIComponent + +function deliveryTable(rows) { + return table(rows, [ + { key: (d) => date(d.created_at), label: 'DATE' }, + { key: 'event_type', label: 'EVENT' }, + { key: (d) => `${d.status}${d.last_status ? ` ${d.last_status}` : ''}`, label: 'STATUS' }, + { key: 'attempts', label: 'TRIES' }, + { key: (d) => d.last_error ?? '', label: 'ERROR', max: 40 }, + { key: 'id', label: 'ID' }, + ]) +} + +function endpointSummary(e) { + const lines = [ + `${e.url} [${e.enabled ? 'on' : `off${e.disabled_reason ? `: ${e.disabled_reason}` : ''}`}]`, + ` id: ${e.id}`, + ` events: ${e.events.join(', ')}`, + ] + if (e.description) lines.push(` note: ${e.description}`) + if (e.failures) lines.push(` failures: ${e.failures} in a row`) + lines.push(` created: ${date(e.created_at)}`) + return `${lines.join('\n')}\n` +} + +async function webhooks({ client, positionals, flags, io }) { + needKey(client) + const [sub, ...rest] = positionals + const need = (v, what) => { + if (!v) throw new UsageError(`Usage: saasrow webhooks ${sub} <${what}>`) + return v + } + const print = (result, human) => (flags.json ? io.out(json(result)) : io.out(human), 0) + switch (sub ?? 'list') { + case 'list': + case 'ls': { + const r = await client.get('/api/v1/webhooks') + if (flags.json) return io.out(json(r)), 0 + if (r.endpoints.length === 0) io.out('No webhooks yet. Add one: saasrow webhooks add [events…]\n') + else { + io.out( + table(r.endpoints, [ + { key: 'id', label: 'ID' }, + { key: 'url', label: 'URL', max: 50 }, + { key: 'events', label: 'EVENTS', max: 40 }, + { key: (e) => (e.enabled ? 'on' : 'off'), label: 'STATUS' }, + ]), + ) + } + io.out(`\nEvents: ${r.event_types.join(', ')}\n`) + return 0 + } + case 'add': + case 'create': { + const [url, ...events] = rest + need(url, 'https url') + const list = events.length ? events.flatMap((e) => e.split(',')).filter(Boolean) : listFlag(flags.events) + const body = { url } + if (list?.length) body.events = list + if (typeof flags.description === 'string') body.description = flags.description + const r = await client.post('/api/v1/webhooks', body) + if (flags.json) return io.out(json(r)), 0 + io.out(`${r.secret}\n`) + io.err(`Added ${r.endpoint.id} -> ${r.endpoint.url} (${r.endpoint.events.join(', ')}). That is its signing secret; verify the webhook-signature header with it.\n`) + return 0 + } + case 'get': + case 'show': { + const r = await client.get(`/api/v1/webhooks/${enc(need(rest[0], 'id'))}`) + return print(r, endpointSummary(r)) + } + case 'update': + case 'edit': { + const id = need(rest[0], 'id') + const patch = {} + if (typeof flags.url === 'string') patch.url = flags.url + const events = listFlag(flags.events) + if (events) patch.events = events + if (typeof flags.description === 'string') patch.description = flags.description + if (flags.enable) patch.enabled = true + if (flags.disable) patch.enabled = false + if (!Object.keys(patch).length) throw new UsageError('Nothing to change. Pass --url, --events, --description, --enable or --disable.') + const r = await client.patch(`/api/v1/webhooks/${enc(id)}`, patch) + return print(r, endpointSummary(r)) + } + case 'test': { + const r = await client.post(`/api/v1/webhooks/${enc(need(rest[0], 'id'))}/test`) + if (flags.json) return io.out(json(r)), 0 + io.out(r.ok ? `Delivered (HTTP ${r.status}).\n` : `Failed: ${r.error ?? `HTTP ${r.status}`}.\n`) + return r.ok ? 0 : 1 + } + case 'secret': { + const r = await client.get(`/api/v1/webhooks/${enc(need(rest[0], 'id'))}/secret`) + return print(r, `${r.secret}\n`) + } + case 'rotate': { + const id = need(rest[0], 'id') + if (!(await confirm(io, `Replace the signing secret of ${id}? The old one stops verifying immediately.`, flags))) { + return io.err('Cancelled.\n'), 1 + } + const r = await client.post(`/api/v1/webhooks/${enc(id)}/rotate`) + return print(r, `${r.secret}\n`) + } + case 'log': + case 'deliveries': { + const query = flags.limit ? { limit: flags.limit } : undefined + const path = rest[0] ? `/api/v1/webhooks/${enc(rest[0])}/deliveries` : '/api/v1/webhooks/deliveries' + const r = await client.get(path, { query }) + if (flags.json) return io.out(json(r)), 0 + io.out(r.deliveries.length ? deliveryTable(r.deliveries) : 'No deliveries yet.\n') + return 0 + } + case 'retry': { + const r = await client.post(`/api/v1/webhooks/deliveries/${enc(need(rest[0], 'delivery id'))}/retry`) + return print(r, `Queued ${r.id} (${r.event_type}) to send again.\n`) + } + case 'rm': + case 'remove': + case 'delete': { + const id = need(rest[0], 'id') + if (!(await confirm(io, `Remove webhook ${id} and its delivery log?`, flags))) return io.err('Cancelled.\n'), 1 + const r = await client.delete(`/api/v1/webhooks/${enc(id)}`) + return print(r, `Removed ${id}.\n`) + } + default: + throw new UsageError(`Unknown webhooks command "${sub}". Try: list, add, get, update, test, secret, rotate, log, retry, rm.`) + } +} + +const COMMANDS = { login, logout, whoami, keys, listings, karma, leaderboard, redeem, invites, webhooks, search, categories, vocabulary } /** * @param {string[]} argv diff --git a/cli/test/cli.test.mjs b/cli/test/cli.test.mjs index 2177015..cfcb2bd 100644 --- a/cli/test/cli.test.mjs +++ b/cli/test/cli.test.mjs @@ -300,3 +300,53 @@ test('invites create prints the link once; invites lists them', async () => { h.cleanup() } }) + +test('webhooks: list, add, test, log and rm talk to /api/v1/webhooks', async () => { + const endpoint = { id: 'wh_1', url: 'https://hooks.example.com/x', events: ['product.*'], description: '', enabled: true, failures: 0, disabled_reason: null, created_at: '2026-10-28T00:00:00Z' } + const h = harness( + { + 'GET /api/v1/webhooks': () => [200, { endpoints: [endpoint], event_types: ['product.approved', 'karma.changed'] }], + 'POST /api/v1/webhooks': () => [201, { endpoint, secret: 'whsec_abc' }], + 'POST /api/v1/webhooks/wh_1/test': () => [200, { ok: false, status: 500, error: 'HTTP 500', delivery_id: 'dlv_1' }], + 'GET /api/v1/webhooks/deliveries': () => [200, { deliveries: [{ id: 'dlv_1', event_type: 'webhook.test', status: 'failed', attempts: 1, last_status: 500, last_error: 'HTTP 500', created_at: '2026-10-28T00:00:00Z' }] }], + 'PATCH /api/v1/webhooks/wh_1': () => [200, { ...endpoint, enabled: false }], + 'DELETE /api/v1/webhooks/wh_1': () => [200, { deleted: true }], + }, + { env: { SAASROW_API_KEY: 'sr_test' } }, + ) + try { + assert.equal(await h.run(['webhooks']), 0) + assert.match(h.out, /wh_1/) + assert.match(h.out, /Events: product\.approved, karma\.changed/) + + assert.equal(await h.run(['webhooks', 'add', 'https://hooks.example.com/x', 'product.*', '--description', 'ci']), 0) + assert.match(h.out, /whsec_abc/) + const post = h.calls.find((c) => c.method === 'POST' && c.path === '/api/v1/webhooks') + assert.deepEqual(post.body, { url: 'https://hooks.example.com/x', events: ['product.*'], description: 'ci' }) + assert.equal(post.headers.Authorization, 'Bearer sr_test') + + assert.equal(await h.run(['webhooks', 'test', 'wh_1']), 1) + assert.match(h.out, /Failed: HTTP 500/) + + assert.equal(await h.run(['webhooks', 'log']), 0) + assert.match(h.out, /webhook\.test/) + + assert.equal(await h.run(['webhooks', 'update', 'wh_1', '--disable']), 0) + assert.deepEqual(h.calls.at(-1).body, { enabled: false }) + + assert.equal(await h.run(['webhooks', 'rm', 'wh_1', '--yes']), 0) + assert.match(h.out, /Removed wh_1/) + } finally { + h.cleanup() + } +}) + +test('webhooks needs a key', async () => { + const h = harness({}) + try { + assert.equal(await h.run(['webhooks']), 2) + assert.match(h.err, /Not logged in/) + } finally { + h.cleanup() + } +}) diff --git a/instrumentation.ts b/instrumentation.ts new file mode 100644 index 0000000..2866eac --- /dev/null +++ b/instrumentation.ts @@ -0,0 +1,12 @@ +/** + * Runs once when the Next server boots (not per request, not in tests). + * Starts customer webhooks: the outbox pump that turns database events into + * deliveries, and the drainer that sends them with retries and backoff + * (@profullstack/webhooks, src/lib/webhooks/hub.ts). + */ +export async function register() { + if (process.env.NEXT_RUNTIME !== 'nodejs') return + if (!process.env.NEXT_PUBLIC_SUPABASE_URL || !process.env.SUPABASE_SERVICE_ROLE_KEY) return + const { startWebhooks } = await import('@/lib/webhooks/hub') + startWebhooks() +} diff --git a/package.json b/package.json index c52f92f..64671d0 100644 --- a/package.json +++ b/package.json @@ -1,5 +1,5 @@ { - "version": "1.0.0", + "version": "1.1.0", "type": "module", "name": "anima-project", "description": "A Next.js project using Supabase", @@ -17,13 +17,13 @@ "dependencies": { "@netlify/plugin-nextjs": "^5.16.4", "@profullstack/affiliates": "^0.3.0", - "@profullstack/autoblog": "^0.4.0", "@profullstack/footer": "latest", "@profullstack/form-guard": "^0.2.2", "@profullstack/redeem-credits": "^0.2.0", "@profullstack/referrals": "^0.1.0", "@profullstack/stack": "^0.5.0", "@profullstack/throttle": "^0.4.0", + "@profullstack/webhooks": "0.2.1", "@profullstack/x402-gateway": "^0.15.2", "@radix-ui/react-slot": "^1.5.0", "@supabase/ssr": "^0.12.7", @@ -43,6 +43,7 @@ "tailwindcss-animate": "^1.0.7" }, "devDependencies": { + "@profullstack/autoblog": "^0.4.0", "@testing-library/jest-dom": "^7.0.1", "@testing-library/react": "^16.3.3", "@testing-library/user-event": "^14.6.8", diff --git a/src/components/WebhooksPanel.tsx b/src/components/WebhooksPanel.tsx new file mode 100644 index 0000000..192709b --- /dev/null +++ b/src/components/WebhooksPanel.tsx @@ -0,0 +1,387 @@ +'use client' + +import { useCallback, useEffect, useState } from 'react' +import { WEBHOOK_EVENT_HELP } from '@/lib/webhooks/events' + +// Webhook endpoints for the listing management page. Talks to /api/v1/webhooks +// with the management token from the URL, which the server resolves to the +// same account an API key would. A signing secret is shown once, when it is +// created or rotated, and on request after that. + +interface Endpoint { + id: string + url: string + events: string[] + description: string + enabled: boolean + failures: number + disabled_reason: string | null + created_at: string +} + +interface Delivery { + id: string + endpoint_id: string + event_type: string + status: 'pending' | 'delivered' | 'failed' + attempts: number + last_status: number | null + last_error: string | null + created_at: string +} + +async function hooksRequest(token: string, path: string, init: RequestInit = {}) { + const res = await fetch(`/api/v1/webhooks${path}`, { + ...init, + headers: { + 'Content-Type': 'application/json', + 'X-Management-Token': token, + ...(init.headers ?? {}), + }, + }) + const body = await res.json().catch(() => ({})) + if (!res.ok) { + throw new Error(body?.error?.message || body?.error || `Request failed (${res.status})`) + } + return body +} + +function when(iso: string | null): string { + if (!iso) return '' + return new Date(iso).toLocaleString() +} + +export function WebhooksPanel({ managementToken }: { managementToken: string }) { + const [endpoints, setEndpoints] = useState([]) + const [eventTypes, setEventTypes] = useState([]) + const [loading, setLoading] = useState(true) + const [error, setError] = useState(null) + const [notice, setNotice] = useState(null) + const [url, setUrl] = useState('') + const [description, setDescription] = useState('') + const [picked, setPicked] = useState([]) + const [creating, setCreating] = useState(false) + const [secret, setSecret] = useState<{ id: string; value: string; label: string } | null>(null) + const [copied, setCopied] = useState(false) + const [log, setLog] = useState<{ id: string; rows: Delivery[] } | null>(null) + const [busy, setBusy] = useState(null) + + const load = useCallback(async () => { + try { + const body = await hooksRequest(managementToken, '') + setEndpoints(body.endpoints ?? []) + setEventTypes(body.event_types ?? []) + setError(null) + } catch (err) { + setError(err instanceof Error ? err.message : 'Could not load webhooks') + } finally { + setLoading(false) + } + }, [managementToken]) + + useEffect(() => { + load() + }, [load]) + + const run = async (key: string, fn: () => Promise) => { + setBusy(key) + setError(null) + setNotice(null) + try { + await fn() + } catch (err) { + setError(err instanceof Error ? err.message : 'Something went wrong') + } finally { + setBusy(null) + } + } + + const showSecret = (id: string, value: string, label: string) => { + setSecret({ id, value, label }) + setCopied(false) + } + + const handleCreate = (e: React.FormEvent) => { + e.preventDefault() + setCreating(true) + run('create', async () => { + const body = await hooksRequest(managementToken, '', { + method: 'POST', + body: JSON.stringify({ + url: url.trim(), + events: picked.length ? picked : ['*'], + description: description.trim(), + }), + }) + showSecret(body.endpoint.id, body.secret, `Signing secret for ${body.endpoint.url}`) + setUrl('') + setDescription('') + setPicked([]) + await load() + }).finally(() => setCreating(false)) + } + + const test = (ep: Endpoint) => + run(`test:${ep.id}`, async () => { + const r = await hooksRequest(managementToken, `/${ep.id}/test`, { method: 'POST' }) + setNotice( + r.ok + ? `Test delivered to ${ep.url} (HTTP ${r.status}).` + : `Test to ${ep.url} failed: ${r.error ?? `HTTP ${r.status}`}.`, + ) + if (log?.id === ep.id) await openLog(ep.id) + }) + + const rotate = (ep: Endpoint) => { + if (!window.confirm(`Replace the signing secret for ${ep.url}? The old one stops verifying immediately.`)) return + run(`rotate:${ep.id}`, async () => { + const r = await hooksRequest(managementToken, `/${ep.id}/rotate`, { method: 'POST' }) + showSecret(ep.id, r.secret, `New signing secret for ${ep.url}`) + }) + } + + const reveal = (ep: Endpoint) => + run(`secret:${ep.id}`, async () => { + const r = await hooksRequest(managementToken, `/${ep.id}/secret`) + showSecret(ep.id, r.secret, `Signing secret for ${ep.url}`) + }) + + const toggle = (ep: Endpoint) => + run(`toggle:${ep.id}`, async () => { + await hooksRequest(managementToken, `/${ep.id}`, { + method: 'PATCH', + body: JSON.stringify({ enabled: !ep.enabled }), + }) + await load() + }) + + const remove = (ep: Endpoint) => { + if (!window.confirm(`Remove the webhook to ${ep.url}? Its delivery log goes with it.`)) return + run(`remove:${ep.id}`, async () => { + await hooksRequest(managementToken, `/${ep.id}`, { method: 'DELETE' }) + if (secret?.id === ep.id) setSecret(null) + if (log?.id === ep.id) setLog(null) + await load() + }) + } + + const openLog = async (id: string) => { + const r = await hooksRequest(managementToken, `/${id}/deliveries?limit=20`) + setLog({ id, rows: r.deliveries ?? [] }) + } + + const retry = (d: Delivery) => + run(`retry:${d.id}`, async () => { + await hooksRequest(managementToken, `/deliveries/${d.id}/retry`, { method: 'POST' }) + setNotice('Queued to send again.') + await openLog(d.endpoint_id) + }) + + const copySecret = async () => { + if (!secret) return + try { + await navigator.clipboard.writeText(secret.value) + setCopied(true) + } catch { + setCopied(false) + } + } + + const togglePick = (t: string) => + setPicked((p) => (p.includes(t) ? p.filter((x) => x !== t) : [...p, t])) + + const link = 'text-[#4FFFE3] hover:underline disabled:opacity-50' + + return ( +
+

Webhooks

+

+ SaaSRow POSTs a signed JSON event to your URL when your listings are approved, featured, + upvoted or commented on, or your karma changes. Bodies are CloudEvents; verify the{' '} + webhook-signature header with the + signing secret (Standard Webhooks, e.g.{' '} + @profullstack/webhooks). Failed + deliveries are retried for about three days. +

+

+ Also from the terminal (saasrow webhooks), + the API and the{' '} + MCP server. +

+ + {error && ( +
+ {error} +
+ )} + {notice && ( +
+ {notice} +
+ )} + + {secret && ( +
+

{secret.label}

+

+ Keep it on your server only. Anyone with it can forge events to your URL. +

+
+ + {secret.value} + + + +
+
+ )} + +
+
+ setUrl(e.target.value)} + placeholder="https://your-app.example.com/hooks/saasrow" + className="flex-1 min-w-0 bg-[#1a1a1a] text-white font-ubuntu rounded-lg px-4 py-2 border border-white/10 focus:border-[#4FFFE3] focus:outline-none" + /> + setDescription(e.target.value)} + placeholder="Note (optional)" + maxLength={200} + className="sm:w-48 bg-[#1a1a1a] text-white font-ubuntu rounded-lg px-4 py-2 border border-white/10 focus:border-[#4FFFE3] focus:outline-none" + /> + +
+ {eventTypes.length > 0 && ( +
+ Events (none ticked = all of them) + {eventTypes.map((t) => ( + + ))} +
+ )} +
+ + {loading ? ( +

Loading webhooks…

+ ) : endpoints.length === 0 ? ( +

No webhooks yet.

+ ) : ( +
    + {endpoints.map((ep) => ( +
  • +
    +
    +

    + {ep.url} + {!ep.enabled && ( + + off{ep.disabled_reason ? ` (${ep.disabled_reason})` : ''} + + )} +

    +

    + {ep.id} · {ep.events.join(', ')} + {ep.description ? ` · ${ep.description}` : ''} + {ep.failures > 0 ? ` · ${ep.failures} failures in a row` : ''} +

    +
    +
    + + + + + + +
    +
    + {log?.id === ep.id && ( +
    + {log.rows.length === 0 ? ( +

    No deliveries yet.

    + ) : ( + + + {log.rows.map((d) => ( + + + + + + + + + ))} + +
    {when(d.created_at)}{d.event_type} + {d.status} + {d.last_status ? ` ${d.last_status}` : ''} + ×{d.attempts}{d.last_error ?? ''} + {d.status !== 'pending' && ( + + )} +
    + )} +
    + )} +
  • + ))} +
+ )} +
+ ) +} diff --git a/src/lib/autoblogReceiver.ts b/src/lib/autoblogReceiver.ts new file mode 100644 index 0000000..6a4d1d5 --- /dev/null +++ b/src/lib/autoblogReceiver.ts @@ -0,0 +1,92 @@ +import { timingSafeEqual } from 'node:crypto' +import { getHeader, verifyAndParse, type HeaderBag } from '@profullstack/webhooks' + +// The crawlproof autoblog posts new blog posts to /api/autoblog-webhook. Its +// sender (@profullstack/autoblog sendWebhook) sends the shared secret as +// `Authorization: Bearer ` and signs Standard Webhooks style +// (`webhook-signature: v1,`) keyed with +// the secret string's UTF-8 bytes, whatever it looks like. +// +// @profullstack/webhooks verifies the same headers, but it would decode a +// `whsec_…` secret as base64 and so compute a different key. Passing the +// secret as bytes keeps the key exactly what the sender uses for any secret +// format (today's is 64 hex characters). The bearer check and the post +// envelope check are kept from the old receiver. + +export interface AutoblogPost { + id: string + title: string + slug: string + html: string + status: string + published_at: string + tags: string[] + categories: string[] + excerpt?: string + markdown?: string + url?: string + canonical_url?: string + updated_at?: string + author?: { name?: string; url?: string } + featured_image?: { url?: string; alt?: string } +} + +export type AutoblogResult = + | { ok: true; post: AutoblogPost; event: Record } + | { ok: false; status: 400 | 401; reason: string } + +function sameString(a: string, b: string): boolean { + const x = Buffer.from(a) + const y = Buffer.from(b) + return x.length === y.length && timingSafeEqual(x, y) +} + +function isPost(p: unknown): p is AutoblogPost { + if (!p || typeof p !== 'object') return false + const o = p as Record + return ( + typeof o.id === 'string' && + typeof o.title === 'string' && + typeof o.slug === 'string' && + typeof o.html === 'string' && + typeof o.status === 'string' && + typeof o.published_at === 'string' && + Array.isArray(o.tags) && + Array.isArray(o.categories) + ) +} + +export function verifyAutoblog(input: { + headers: HeaderBag + body: string + secret: string + now?: () => number +}): AutoblogResult { + const { headers, body, secret, now } = input + if (!secret) return { ok: false, status: 401, reason: 'no secret configured' } + const bearer = (getHeader(headers, 'authorization') ?? '').replace(/^Bearer\s+/i, '') + if (!bearer || !sameString(bearer, secret)) return { ok: false, status: 401, reason: 'invalid bearer' } + + const parsed = verifyAndParse>({ + headers, + body, + secret: Buffer.from(secret, 'utf8'), + ...(now ? { now } : {}), + }) + if (!parsed.ok) return { ok: false, status: parsed.status, reason: parsed.reason } + + const event = parsed.event + const data = event.data as Record | undefined + if ( + event.specversion !== '1.0' || + typeof event.id !== 'string' || + typeof event.type !== 'string' || + typeof event.source !== 'string' || + !data || + typeof data !== 'object' || + !isPost(data.post) + ) { + return { ok: false, status: 400, reason: 'invalid event envelope' } + } + return { ok: true, post: data.post, event } +} diff --git a/src/lib/mcpServer.ts b/src/lib/mcpServer.ts index 9eef529..ab33723 100644 --- a/src/lib/mcpServer.ts +++ b/src/lib/mcpServer.ts @@ -34,6 +34,8 @@ import { karmaRules, karmaSummary, leaderboard, redeemCredit } from './karma' import { KARMA_PER_CREDIT, parsePeriod } from './karmaRules' import { createInviteFor, inviteOverview, redeemCreditCode } from './credits' import { siteOriginFrom } from './inviteRules' +import { WebhookError } from '@profullstack/webhooks' +import { getHooks, WEBHOOK_EVENT_HELP, WEBHOOK_EVENT_TYPES } from './webhooks/hub' const SITE_ORIGIN = 'https://saasrow.com' const mcpOrigin = (ctx: McpContext) => (ctx.request ? siteOriginFrom(ctx.request.headers) : SITE_ORIGIN) @@ -259,6 +261,72 @@ export const WRITE_TOOLS = [ additionalProperties: false, }, }, + { + name: 'list_webhooks', + description: `Your webhook endpoints, and the events SaaSRow can send to them (${WEBHOOK_EVENT_TYPES.join(', ')}). ${AUTH_NOTE}`, + inputSchema: { type: 'object', properties: {}, additionalProperties: false }, + }, + { + name: 'create_webhook', + description: `Register an HTTPS URL that SaaSRow POSTs signed events to (Standard Webhooks, CloudEvents JSON) when your listings are approved, featured, upvoted or commented on, or your karma changes. Returns the signing secret once. ${AUTH_NOTE}`, + inputSchema: { + type: 'object', + properties: { + url: { type: 'string', description: 'Public https:// URL that receives the POSTs.' }, + events: { + type: 'array', + items: { type: 'string' }, + description: `Event types or patterns (product.*, *). Default all. One of: ${WEBHOOK_EVENT_TYPES.join(', ')}.`, + }, + description: { type: 'string', description: 'A note for yourself.' }, + }, + required: ['url'], + additionalProperties: false, + }, + }, + { + name: 'update_webhook', + description: `Change a webhook endpoint's URL, events or description, or switch it on or off. ${AUTH_NOTE}`, + inputSchema: { + type: 'object', + properties: { + id: { type: 'string', description: 'The webhook endpoint id (wh_…).' }, + url: { type: 'string' }, + events: { type: 'array', items: { type: 'string' } }, + description: { type: 'string' }, + enabled: { type: 'boolean' }, + }, + required: ['id'], + additionalProperties: false, + }, + }, + { + name: 'delete_webhook', + description: `Remove a webhook endpoint and its delivery log. ${AUTH_NOTE}`, + inputSchema: { type: 'object', properties: { id: { type: 'string', description: 'The webhook endpoint id (wh_…).' } }, required: ['id'], additionalProperties: false }, + }, + { + name: 'test_webhook', + description: `Send a webhook.test event to an endpoint now and report the HTTP status it answered. ${AUTH_NOTE}`, + inputSchema: { type: 'object', properties: { id: { type: 'string', description: 'The webhook endpoint id (wh_…).' } }, required: ['id'], additionalProperties: false }, + }, + { + name: 'rotate_webhook_secret', + description: `Replace an endpoint's signing secret; the old one stops verifying at once. Returns the new secret. ${AUTH_NOTE}`, + inputSchema: { type: 'object', properties: { id: { type: 'string', description: 'The webhook endpoint id (wh_…).' } }, required: ['id'], additionalProperties: false }, + }, + { + name: 'webhook_deliveries', + description: `Recent webhook deliveries (status, attempts, last HTTP status), for one endpoint or all of them. ${AUTH_NOTE}`, + inputSchema: { + type: 'object', + properties: { + id: { type: 'string', description: 'Only this endpoint (optional).' }, + limit: { type: 'integer', minimum: 1, maximum: 200 }, + }, + additionalProperties: false, + }, + }, ] as const export const TOOLS = [...READ_TOOLS, ...WRITE_TOOLS] as const @@ -369,6 +437,47 @@ async function callTool(name: string, args: Record, ctx: McpCon await deleteOwnedListing(principal, id) return toolResult({ deleted: true, id }) } + case 'list_webhooks': + return toolResult({ + endpoints: await getHooks().listEndpoints(principal.userId), + event_types: WEBHOOK_EVENT_TYPES.map((type) => ({ type, when: WEBHOOK_EVENT_HELP[type] })), + }) + case 'create_webhook': + return toolResult( + await getHooks().createEndpoint(principal.userId, { + url: String(args.url ?? ''), + events: args.events as string[] | string | undefined, + description: typeof args.description === 'string' ? args.description : undefined, + }), + ) + case 'update_webhook': { + const id = requireId(args) + if (!id) return toolError('id is required') + const { id: _id, ...patch } = args + return toolResult(await getHooks().updateEndpoint(principal.userId, id, patch)) + } + case 'delete_webhook': { + const id = requireId(args) + if (!id) return toolError('id is required') + return toolResult(await getHooks().deleteEndpoint(principal.userId, id)) + } + case 'test_webhook': { + const id = requireId(args) + if (!id) return toolError('id is required') + return toolResult(await getHooks().test(principal.userId, id)) + } + case 'rotate_webhook_secret': { + const id = requireId(args) + if (!id) return toolError('id is required') + return toolResult(await getHooks().rotateSecret(principal.userId, id)) + } + case 'webhook_deliveries': + return toolResult({ + deliveries: await getHooks().deliveries(principal.userId, { + endpointId: requireId(args) ?? undefined, + limit: typeof args.limit === 'number' ? args.limit : 50, + }), + }) default: return toolError(`Unknown tool: ${name}`) } @@ -400,7 +509,7 @@ export async function handleMcpMessage( return ok(id, { protocolVersion, capabilities: { tools: { listChanged: false } }, - serverInfo: { name: 'saasrow', version: '1.1.0', websiteUrl: siteUrl() }, + serverInfo: { name: 'saasrow', version: '1.2.0', websiteUrl: siteUrl() }, instructions: 'SaaSRow is a directory of software and SaaS products. Use search_products to find tools by use case, audience, platform, pricing model, or as an alternative to a named competitor. Call get_vocabulary first if you need the exact filter terms. ' + 'With an API key you can also create_listing for a product you make and manage it with list_my_listings, update_listing and delete_listing, and check my_karma. ' + @@ -422,7 +531,7 @@ export async function handleMcpMessage( return ok(id, await callTool(name, args, ctx)) } catch (error) { // Validation, ownership and conflict errors are the caller's to fix. - if (error instanceof ApiError) return ok(id, toolError(error.message)) + if (error instanceof ApiError || error instanceof WebhookError) return ok(id, toolError(error.message)) console.error('[mcp] tool call failed', name, error) return ok(id, toolError(`Tool "${name}" failed.`)) } diff --git a/src/lib/webhooks/api.ts b/src/lib/webhooks/api.ts new file mode 100644 index 0000000..4c3b8ed --- /dev/null +++ b/src/lib/webhooks/api.ts @@ -0,0 +1,26 @@ +import 'server-only' +import { webhooksApi } from '@profullstack/webhooks' +import { requirePrincipal } from '../apiAuth' +import { getHooks } from './hub' + +const CORS: Record = { + 'Access-Control-Allow-Origin': '*', + 'Access-Control-Allow-Methods': 'GET, POST, PATCH, PUT, DELETE, OPTIONS', + 'Access-Control-Allow-Headers': 'Content-Type, Accept, Authorization, X-Management-Token', +} + +/** + * Authenticate (401 via ApiError when there is no valid credential), then hand + * the request to the package's router with the account as the owner. Null for + * a path outside /api/v1/webhooks. + */ +export async function handleWebhooksRequest(req: Request): Promise { + const principal = await requirePrincipal(req) + const api = webhooksApi(getHooks(), { basePath: '/api/v1/webhooks', owner: () => principal.userId }) + const res = await api(req) + if (!res) return null + const headers = new Headers(res.headers) + for (const [k, v] of Object.entries(CORS)) headers.set(k, v) + headers.set('Cache-Control', 'no-store') + return new Response(res.body, { status: res.status, headers }) +} diff --git a/src/lib/webhooks/events.ts b/src/lib/webhooks/events.ts new file mode 100644 index 0000000..2718fbb --- /dev/null +++ b/src/lib/webhooks/events.ts @@ -0,0 +1,23 @@ +// The events SaaSRow sends to customer webhooks. Plain data, safe for client +// components. Keep in step with the triggers in the 20261028160000_webhooks_hub migration. + +export const WEBHOOK_EVENT_TYPES = [ + 'product.approved', + 'product.rejected', + 'product.featured', + 'product.upvoted', + 'product.commented', + 'karma.changed', +] as const + +export type WebhookEventType = (typeof WEBHOOK_EVENT_TYPES)[number] + +export const WEBHOOK_EVENT_HELP: Record = { + 'product.approved': 'your listing was approved and is live in the directory', + 'product.rejected': 'your listing was rejected in review', + 'product.featured': 'your listing moved to the Featured or Premium tier', + 'product.upvoted': 'someone upvoted your listing', + 'product.commented': 'a comment or review on your listing went public', + 'karma.changed': 'your karma went up or down', +} + diff --git a/src/lib/webhooks/hub.ts b/src/lib/webhooks/hub.ts new file mode 100644 index 0000000..b81a106 --- /dev/null +++ b/src/lib/webhooks/hub.ts @@ -0,0 +1,130 @@ +import { createClient } from '@supabase/supabase-js' +import { Webhooks, type WebhooksOptions, type WebhookStore } from '@profullstack/webhooks' +import { supabaseStore, type SupabaseLike } from './supabase-store' +import { WEBHOOK_EVENT_TYPES } from './events' + +/** + * Customer webhooks for SaaSRow, on @profullstack/webhooks. + * + * A listing owner registers HTTPS endpoints and SaaSRow POSTs signed + * CloudEvents (Standard Webhooks headers) to them. Endpoints and a durable + * delivery outbox live in Supabase (hub_webhook_* tables); the drainer started + * once at server boot (instrumentation.ts) sends with retries and backoff. + * + * The events themselves are written by Postgres triggers into + * hub_webhook_events (see the 20261028160000_webhooks_hub migration), because + * most of them happen outside this app: approvals and anonymous votes in edge + * functions, karma in SQL. pumpEvents() turns each row into hooks.emit(). + * + * No webhooks existed before this, so there is no legacy signature to keep. + */ + +export { WEBHOOK_EVENT_TYPES, WEBHOOK_EVENT_HELP, type WebhookEventType } from './events' + +export const WEBHOOK_SOURCE = 'https://saasrow.com' + +export function createHooks(store: WebhookStore, options: Partial = {}): Webhooks { + return new Webhooks({ + store, + source: WEBHOOK_SOURCE, + eventTypes: [...WEBHOOK_EVENT_TYPES], + maxEndpoints: 10, + guard: process.env.NODE_ENV === 'production' ? {} : { allowHttp: true }, + log: (m) => console.warn(`[webhooks] ${m}`), + ...options, + }) +} + +/** The supabase-js service client, built here rather than via supabaseAdmin so boot code can load it. */ +export function serviceClient(): SupabaseLike { + const url = process.env.NEXT_PUBLIC_SUPABASE_URL + const key = process.env.SUPABASE_SERVICE_ROLE_KEY + if (!url || !key) { + throw new Error('Supabase service env not configured (NEXT_PUBLIC_SUPABASE_URL / SUPABASE_SERVICE_ROLE_KEY)') + } + return createClient(url, key, { auth: { persistSession: false, autoRefreshToken: false } }) as unknown as SupabaseLike +} + +const KEY = Symbol.for('saasrow.webhooks.hub') +type HubState = { hooks: Webhooks; client: SupabaseLike; pump?: ReturnType } +type HubGlobal = typeof globalThis & { [KEY]?: HubState } + +/** + * The process-wide hub. Kept on globalThis because Next bundles route handlers + * and instrumentation separately: a module-level singleton would give each its + * own instance, and only the one that called start() would drain. + */ +function state(): HubState { + const g = globalThis as HubGlobal + if (!g[KEY]) { + const client = serviceClient() + g[KEY] = { client, hooks: createHooks(supabaseStore(client)) } + } + return g[KEY]! +} + +export function getHooks(): Webhooks { + return state().hooks +} + +export const EVENTS_TABLE = 'hub_webhook_events' +export const CLAIM_EVENTS_RPC = 'claim_hub_webhook_events' + +interface OutboxRow { + id: number | string + owner: string + type: string + subject: string | null + data: unknown +} + +/** + * Move up to `limit` outbox rows into the hub: emit each (which queues one + * delivery per subscribed endpoint) and delete it. The CloudEvent id comes + * from the row id, so a row emitted twice after a crash carries the same + * webhook-id and receivers can drop the repeat. Returns how many rows it took. + */ +export async function pumpEvents(hooks: Webhooks, client: SupabaseLike, limit = 100): Promise { + const { data, error } = await client.rpc(CLAIM_EVENTS_RPC, { p_limit: limit }) + if (error) throw new Error(`webhooks outbox: claim: ${error.message}`) + const rows = (data ?? []) as unknown as OutboxRow[] + const done: Array = [] + for (const row of rows) { + try { + await hooks.emit(row.owner, row.type, row.data ?? {}, { + ...(row.subject ? { subject: row.subject } : {}), + id: `evt_saasrow_${row.id}`, + }) + done.push(row.id) + } catch (err) { + // Left claimed; the lease runs out and the next pump tries again. + console.warn(`[webhooks] outbox event ${row.id} (${row.type}) not emitted: ${(err as Error).message}`) + } + } + if (done.length) { + const del = await client.from(EVENTS_TABLE).delete().in('id', done) + if (del?.error) throw new Error(`webhooks outbox: delete: ${del.error.message}`) + } + return rows.length +} + +/** Start the delivery drainer and the outbox pump, once per server process. */ +export function startWebhooks(intervalMs = 5000): void { + const s = state() + if (s.pump) return + s.hooks.start() + let running = false + s.pump = setInterval(async () => { + if (running) return + running = true + try { + // Keep going while full pages come back, so a burst drains in one tick. + while ((await pumpEvents(s.hooks, s.client)) === 100) {} + } catch (err) { + console.warn(`[webhooks] outbox pump: ${(err as Error).message}`) + } finally { + running = false + } + }, intervalMs) + s.pump.unref?.() +} diff --git a/src/lib/webhooks/supabase-store.ts b/src/lib/webhooks/supabase-store.ts new file mode 100644 index 0000000..65d947f --- /dev/null +++ b/src/lib/webhooks/supabase-store.ts @@ -0,0 +1,202 @@ +import type { Delivery, Endpoint, WebhookStore } from "@profullstack/webhooks"; + +/** + * A @profullstack/webhooks store on supabase-js. + * + * SaaSRow has no direct Postgres connection string (only the Supabase URL and the + * service role key), so the package's postgresStore cannot be used. This keeps + * the same two tables (schemaSql('pg', 'hub_'), see the + * 20261028160000_webhooks_hub migration) and the same claim semantics: due + * rows are taken with FOR UPDATE SKIP LOCKED inside the + * claim_hub_webhook_deliveries RPC, because PostgREST cannot lock rows. + * + * Times are unix milliseconds (BIGINT); `events` is JSON text; `enabled` is 0/1. + */ + +type Row = Record; +type Result = { data: T | null; error: { message: string } | null }; + +/** The slice of a supabase-js client this store needs (any client works; tests pass a fake). */ +export interface SupabaseLike { + from(table: string): any; + rpc(fn: string, args: Record): PromiseLike>; +} + +export const HUB_TABLES = { + endpoints: "hub_webhook_endpoints", + deliveries: "hub_webhook_deliveries", + claim: "claim_hub_webhook_deliveries", +} as const; + +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 check(r: Result, what: string): T | null { + if (r.error) throw new Error(`webhooks store: ${what}: ${r.error.message}`); + return r.data; +} + +export function toEndpoint(r: Row | null | undefined): Endpoint | null { + if (!r) return null; + let events: string[] = ["*"]; + try { + const parsed = typeof r.events === "string" ? JSON.parse(r.events) : r.events; + if (Array.isArray(parsed)) events = parsed.map(String); + } catch { + // keep ["*"] + } + return { + id: String(r.id), + owner: String(r.owner), + url: String(r.url), + secret: String(r.secret), + events, + description: (r.description as string) ?? "", + enabled: Number(r.enabled) === 1 || r.enabled === true, + failures: Number(r.failures ?? 0), + disabledReason: (r.disabled_reason as string | null) ?? null, + createdAt: Number(r.created_at), + updatedAt: Number(r.updated_at), + }; +} + +export function toDelivery(r: Row | null | undefined): 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: r.status as Delivery["status"], + attempts: Number(r.attempts ?? 0), + nextAt: Number(r.next_at), + lastStatus: num(r.last_status), + lastError: (r.last_error as string | null) ?? null, + createdAt: Number(r.created_at), + updatedAt: Number(r.updated_at), + }; +} + +function sqlValue(key: string, value: unknown) { + if (key === "events") return JSON.stringify(value); + if (key === "enabled") return value ? 1 : 0; + return value ?? null; +} + +function columns(fields: Record, patch: Record): Row { + const out: Row = {}; + for (const [k, v] of Object.entries(patch)) { + if (fields[k] && v !== undefined) out[fields[k]] = sqlValue(k, v); + } + return out; +} + +const endpointRow = (e: Endpoint): Row => ({ + id: e.id, + owner: e.owner, + url: e.url, + secret: e.secret, + events: JSON.stringify(e.events), + description: e.description, + enabled: e.enabled ? 1 : 0, + failures: e.failures, + disabled_reason: e.disabledReason, + created_at: e.createdAt, + updated_at: e.updatedAt, +}); + +const deliveryRow = (d: Delivery): Row => ({ + id: d.id, + endpoint_id: d.endpointId, + owner: d.owner, + event_id: d.eventId, + event_type: d.eventType, + payload: d.payload, + status: d.status, + attempts: d.attempts, + next_at: d.nextAt, + last_status: d.lastStatus, + last_error: d.lastError, + created_at: d.createdAt, + updated_at: d.updatedAt, +}); + +export function supabaseStore(client: SupabaseLike): WebhookStore { + const T = HUB_TABLES; + return { + async insertEndpoint(e) { + check(await client.from(T.endpoints).insert(endpointRow(e)), "insert endpoint"); + }, + async getEndpoint(id) { + return toEndpoint(check(await client.from(T.endpoints).select("*").eq("id", id).maybeSingle(), "get endpoint")); + }, + async listEndpoints(owner) { + const rows = check( + await client.from(T.endpoints).select("*").eq("owner", owner).order("created_at", { ascending: true }), + "list endpoints", + ); + return (rows ?? []).map((r) => toEndpoint(r) as Endpoint); + }, + async updateEndpoint(id, patch) { + const set = columns(ENDPOINT_FIELDS, patch as Record); + if (!Object.keys(set).length) return; + check(await client.from(T.endpoints).update(set).eq("id", id), "update endpoint"); + }, + async deleteEndpoint(id) { + check(await client.from(T.deliveries).delete().eq("endpoint_id", id), "delete deliveries"); + check(await client.from(T.endpoints).delete().eq("id", id), "delete endpoint"); + }, + async insertDeliveries(rows) { + if (!rows.length) return; + check(await client.from(T.deliveries).insert(rows.map(deliveryRow)), "insert deliveries"); + }, + async getDelivery(id) { + return toDelivery(check(await client.from(T.deliveries).select("*").eq("id", id).maybeSingle(), "get delivery")); + }, + async listDeliveries({ owner, endpointId, limit = 50 }) { + let q = client.from(T.deliveries).select("*").eq("owner", owner); + if (endpointId) q = q.eq("endpoint_id", endpointId); + const rows = check(await q.order("created_at", { ascending: false }).limit(limit), "list deliveries"); + return (rows ?? []).map((r) => toDelivery(r) as Delivery); + }, + async updateDelivery(id, patch) { + const set = columns(DELIVERY_FIELDS, patch as Record); + if (!Object.keys(set).length) return; + check(await client.from(T.deliveries).update(set).eq("id", id), "update delivery"); + }, + async claimDue(now, limit, leaseMs) { + const rows = check( + await client.rpc(T.claim, { p_now: now, p_limit: limit, p_lease_until: now + leaseMs }), + "claim due", + ); + return (rows ?? []).map((r) => toDelivery(r) as Delivery); + }, + async prune(before) { + check( + await client.from(T.deliveries).delete().neq("status", "pending").lt("updated_at", before), + "prune", + ); + }, + }; +} diff --git a/src/views/ManageListings.tsx b/src/views/ManageListings.tsx index a3d26b7..f23e784 100644 --- a/src/views/ManageListings.tsx +++ b/src/views/ManageListings.tsx @@ -10,6 +10,7 @@ import { BasicAnalytics } from '../components/BasicAnalytics' import { PremiumAnalytics } from '../components/PremiumAnalytics' import { ScreenshotGallery } from '../components/ScreenshotGallery' import { ApiKeysPanel } from '../components/ApiKeysPanel' +import { WebhooksPanel } from '../components/WebhooksPanel' import { RedeemCodePanel } from '../components/RedeemCodePanel' import { callFn } from '@/lib/clientApi' import { adoptManagementToken } from '@/lib/member' @@ -588,6 +589,8 @@ export default function ManageListings() { {token && } + {token && } + {showAddForm && (

Add New Listing

diff --git a/supabase/migrations/20261028160000_webhooks_hub.sql b/supabase/migrations/20261028160000_webhooks_hub.sql new file mode 100644 index 0000000..36d822f --- /dev/null +++ b/supabase/migrations/20261028160000_webhooks_hub.sql @@ -0,0 +1,404 @@ +-- Customer webhooks on @profullstack/webhooks 0.2.1. +-- +-- A listing owner registers HTTPS endpoints (/api/v1/webhooks, the manage +-- page, `saasrow webhooks`, MCP) and SaaSRow POSTs signed CloudEvents to them +-- when something happens to their products or their karma. +-- +-- Three parts: +-- +-- 1. The hub's own tables, exactly schemaSql('pg', 'hub_') of the package, so +-- a later switch to its postgresStore needs no data change. The Next app +-- has no Postgres connection string (only supabase-js), so it reads and +-- writes them through a small WebhookStore adapter +-- (src/lib/webhooks/supabase-store.ts); claim_hub_webhook_deliveries is +-- the FOR UPDATE SKIP LOCKED claim PostgREST cannot express. +-- +-- 2. An event outbox, hub_webhook_events. SaaSRow's domain events happen in +-- many places: Next routes, Deno edge functions (approval, CoinPay +-- upgrades, anonymous votes and comments), and Postgres functions (karma). +-- Triggers on the tables those all write to put one row here per event, +-- and the Next server turns each row into hooks.emit() (src/lib/webhooks/ +-- hub.ts, pumpEvents). Rows are only written for owners who have an enabled +-- endpoint, and a trigger failure never fails the write that caused it. +-- +-- 3. The triggers: +-- software_submissions status -> approved product.approved +-- status -> rejected product.rejected +-- tier -> featured|premium product.featured +-- votes a new upvote product.upvoted +-- comments a comment goes public product.commented +-- karma_events any karma row karma.changed +-- The owner of a product event is listing_owner(submission): the account +-- that owns the listing. karma.changed goes to the member whose karma moved. +-- +-- Only the server (service role) touches any of this: RLS on, no policies, +-- functions granted to service_role alone. +-- +-- Idempotent: safe to re-run. + +-- ------------------------------------------------------------ hub tables -- + +create table if not exists public.hub_webhook_endpoints ( + id text primary key, + owner text not null, + url text not null, + secret text not null, + events text not null default '["*"]', + description text not null default '', + enabled integer not null default 1, + failures integer not null default 0, + disabled_reason text, + created_at bigint not null, + updated_at bigint not null +); +create index if not exists hub_webhook_endpoints_owner on public.hub_webhook_endpoints(owner); + +create table if not exists public.hub_webhook_deliveries ( + id text primary key, + endpoint_id text not null, + owner text not null, + event_id text not null, + event_type text not null, + payload text not null, + status text not null default 'pending', + attempts integer not null default 0, + next_at bigint not null, + last_status integer, + last_error text, + created_at bigint not null, + updated_at bigint not null +); +create index if not exists hub_webhook_deliveries_due on public.hub_webhook_deliveries(status, next_at); +create index if not exists hub_webhook_deliveries_endpoint on public.hub_webhook_deliveries(endpoint_id, created_at); +create index if not exists hub_webhook_deliveries_owner on public.hub_webhook_deliveries(owner, created_at); + +create table if not exists public.hub_webhook_events ( + id bigserial primary key, + owner text not null, + type text not null, + subject text, + data jsonb not null default '{}'::jsonb, + created_at timestamptz not null default now(), + claimed_until timestamptz +); + +alter table public.hub_webhook_endpoints enable row level security; +alter table public.hub_webhook_deliveries enable row level security; +alter table public.hub_webhook_events enable row level security; +revoke all on public.hub_webhook_endpoints from anon, authenticated; +revoke all on public.hub_webhook_deliveries from anon, authenticated; +revoke all on public.hub_webhook_events from anon, authenticated; +revoke all on sequence public.hub_webhook_events_id_seq from anon, authenticated; +grant all on public.hub_webhook_endpoints to service_role; +grant all on public.hub_webhook_deliveries to service_role; +grant all on public.hub_webhook_events to service_role; +grant usage, select on sequence public.hub_webhook_events_id_seq to service_role; + +-- Take up to p_limit due deliveries and lease them until p_lease_until, so two +-- app processes never send the same row. +create or replace function public.claim_hub_webhook_deliveries(p_now bigint, p_limit integer, p_lease_until bigint) +returns setof public.hub_webhook_deliveries +language sql +volatile +set search_path to 'public' +as $function$ + update public.hub_webhook_deliveries + set next_at = p_lease_until + where id in ( + select id from public.hub_webhook_deliveries + where status = 'pending' and next_at <= p_now + order by next_at + limit p_limit + for update skip locked + ) + returning *; +$function$; +revoke execute on function public.claim_hub_webhook_deliveries(bigint, integer, bigint) from public, anon, authenticated; +grant execute on function public.claim_hub_webhook_deliveries(bigint, integer, bigint) to service_role; + +-- Take up to p_limit outbox events for two minutes. The app deletes each one +-- once hooks.emit() has queued it; a crashed pump's lease simply runs out. +create or replace function public.claim_hub_webhook_events(p_limit integer) +returns setof public.hub_webhook_events +language sql +volatile +set search_path to 'public' +as $function$ + update public.hub_webhook_events + set claimed_until = now() + interval '2 minutes' + where id in ( + select id from public.hub_webhook_events + where claimed_until is null or claimed_until < now() + order by id + limit p_limit + for update skip locked + ) + returning *; +$function$; +revoke execute on function public.claim_hub_webhook_events(integer) from public, anon, authenticated; +grant execute on function public.claim_hub_webhook_events(integer) to service_role; + +-- ----------------------------------------------------------------- outbox -- + +-- Does this account have anywhere to send events? Checked before a trigger +-- builds a payload, so accounts without webhooks cost one index probe. +create or replace function public.hub_webhook_wanted(p_owner uuid) +returns boolean +language sql +stable +security definer +set search_path to 'public' +as $function$ + select p_owner is not null + and exists (select 1 from public.hub_webhook_endpoints where owner = p_owner::text and enabled = 1) +$function$; +revoke execute on function public.hub_webhook_wanted(uuid) from public, anon, authenticated; +grant execute on function public.hub_webhook_wanted(uuid) to service_role; + +create or replace function public.hub_webhook_enqueue(p_owner uuid, p_type text, p_subject text, p_data jsonb) +returns void +language plpgsql +security definer +set search_path to 'public' +as $function$ +begin + if not public.hub_webhook_wanted(p_owner) then + return; + end if; + insert into public.hub_webhook_events (owner, type, subject, data) + values (p_owner::text, p_type, p_subject, coalesce(p_data, '{}'::jsonb)); +end; +$function$; +revoke execute on function public.hub_webhook_enqueue(uuid, text, text, jsonb) from public, anon, authenticated; +grant execute on function public.hub_webhook_enqueue(uuid, text, text, jsonb) to service_role; + +-- The product as a receiver sees it: what it is and where it lives. +create or replace function public.hub_webhook_product(p_submission uuid) +returns jsonb +language sql +stable +security definer +set search_path to 'public' +as $function$ + select jsonb_build_object( + 'id', s.id, + 'title', s.title, + 'url', s.url, + 'status', s.status, + 'tier', s.tier, + 'upvotes', coalesce(s.upvotes, 0), + 'downvotes', coalesce(s.downvotes, 0), + 'saasrow_url', 'https://saasrow.com/software/' || s.id + ) + from public.software_submissions s + where s.id = p_submission +$function$; +revoke execute on function public.hub_webhook_product(uuid) from public, anon, authenticated; +grant execute on function public.hub_webhook_product(uuid) to service_role; + +-- ---------------------------------------------------------------- triggers -- +-- Each one is AFTER, SECURITY DEFINER, and swallows its own errors: a webhook +-- must never be the reason a vote, comment, approval or payment fails. The +-- "zz_" names make them fire after the app's own AFTER triggers (Postgres +-- fires same-event triggers in name order), so vote counts and karma totals +-- already include the change. + +create or replace function public.hub_webhook_on_listing() +returns trigger +language plpgsql +security definer +set search_path to 'public' +as $function$ +declare + v_owner uuid; + v_product jsonb; + v_old_status text := case when tg_op = 'UPDATE' then old.status else null end; + v_old_tier text := case when tg_op = 'UPDATE' then old.tier else null end; + v_featured boolean; + v_was_featured boolean; +begin + begin + v_featured := coalesce(new.tier, 'free') in ('featured', 'premium'); + v_was_featured := tg_op = 'UPDATE' and coalesce(v_old_tier, 'free') in ('featured', 'premium'); + if new.status is not distinct from v_old_status + and (not v_featured or new.tier is not distinct from v_old_tier) then + return null; + end if; + + v_owner := public.listing_owner(new.id); + if not public.hub_webhook_wanted(v_owner) then + return null; + end if; + v_product := public.hub_webhook_product(new.id); + + if new.status is distinct from v_old_status and new.status in ('approved', 'rejected') then + perform public.hub_webhook_enqueue( + v_owner, + 'product.' || new.status, + 'product:' || new.id, + jsonb_build_object('product', v_product, 'previous_status', v_old_status) + ); + end if; + + if v_featured and new.tier is distinct from v_old_tier then + perform public.hub_webhook_enqueue( + v_owner, + 'product.featured', + 'product:' || new.id, + jsonb_build_object( + 'product', v_product, + 'tier', new.tier, + 'previous_tier', coalesce(v_old_tier, 'free'), + 'upgrade', v_was_featured, + 'featured_until', new.featured_until + ) + ); + end if; + exception when others then + raise warning 'hub_webhook_on_listing: %', sqlerrm; + end; + return null; +end; +$function$; + +create or replace function public.hub_webhook_on_vote() +returns trigger +language plpgsql +security definer +set search_path to 'public' +as $function$ +declare + v_owner uuid; +begin + begin + if new.vote_type <> 'upvote' then + return null; + end if; + if tg_op = 'UPDATE' and old.vote_type = 'upvote' then + return null; + end if; + v_owner := public.listing_owner(new.submission_id); + if not public.hub_webhook_wanted(v_owner) or v_owner is not distinct from new.user_id then + return null; + end if; + perform public.hub_webhook_enqueue( + v_owner, + 'product.upvoted', + 'product:' || new.submission_id, + jsonb_build_object( + 'product', public.hub_webhook_product(new.submission_id), + 'vote', jsonb_build_object( + 'id', new.id, + 'member', new.user_id is not null and not coalesce(new.is_anonymous, false), + 'at', new.created_at + ) + ) + ); + exception when others then + raise warning 'hub_webhook_on_vote: %', sqlerrm; + end; + return null; +end; +$function$; + +create or replace function public.hub_webhook_on_comment() +returns trigger +language plpgsql +security definer +set search_path to 'public' +as $function$ +declare + v_owner uuid; +begin + begin + -- Public means verified: on insert for members, on moderation for the rest. + if not coalesce(new.is_verified, false) then + return null; + end if; + if tg_op = 'UPDATE' and coalesce(old.is_verified, false) then + return null; + end if; + v_owner := public.listing_owner(new.submission_id); + if not public.hub_webhook_wanted(v_owner) or v_owner is not distinct from new.user_id then + return null; + end if; + -- Never the commenter's email or address: just what the page shows. + perform public.hub_webhook_enqueue( + v_owner, + 'product.commented', + 'product:' || new.submission_id, + jsonb_build_object( + 'product', public.hub_webhook_product(new.submission_id), + 'comment', jsonb_build_object( + 'id', new.id, + 'author_name', new.author_name, + 'content', new.content, + 'rating', new.rating, + 'member', new.user_id is not null, + 'created_at', new.created_at + ) + ) + ); + exception when others then + raise warning 'hub_webhook_on_comment: %', sqlerrm; + end; + return null; +end; +$function$; + +create or replace function public.hub_webhook_on_karma() +returns trigger +language plpgsql +security definer +set search_path to 'public' +as $function$ +begin + begin + if not public.hub_webhook_wanted(new.user_id) then + return null; + end if; + perform public.hub_webhook_enqueue( + new.user_id, + 'karma.changed', + 'karma:' || new.id, + jsonb_build_object( + 'action', new.action, + 'points', new.points, + 'reversal', new.bucket = 'reversed', + 'total', (select coalesce(sum(points), 0) from public.karma_events where user_id = new.user_id), + 'product', case when new.submission_id is null then null + else public.hub_webhook_product(new.submission_id) end, + 'at', new.created_at + ) + ); + exception when others then + raise warning 'hub_webhook_on_karma: %', sqlerrm; + end; + return null; +end; +$function$; + +revoke execute on function public.hub_webhook_on_listing() from public, anon, authenticated; +revoke execute on function public.hub_webhook_on_vote() from public, anon, authenticated; +revoke execute on function public.hub_webhook_on_comment() from public, anon, authenticated; +revoke execute on function public.hub_webhook_on_karma() from public, anon, authenticated; + +drop trigger if exists zz_hub_webhook_listing on public.software_submissions; +create trigger zz_hub_webhook_listing + after insert or update of status, tier on public.software_submissions + for each row execute function public.hub_webhook_on_listing(); + +drop trigger if exists zz_hub_webhook_vote on public.votes; +create trigger zz_hub_webhook_vote + after insert or update of vote_type on public.votes + for each row execute function public.hub_webhook_on_vote(); + +drop trigger if exists zz_hub_webhook_comment on public.comments; +create trigger zz_hub_webhook_comment + after insert or update of is_verified on public.comments + for each row execute function public.hub_webhook_on_comment(); + +drop trigger if exists zz_hub_webhook_karma on public.karma_events; +create trigger zz_hub_webhook_karma + after insert on public.karma_events + for each row execute function public.hub_webhook_on_karma();