diff --git a/.env.example b/.env.example index 88d2851..fe71790 100644 --- a/.env.example +++ b/.env.example @@ -28,6 +28,10 @@ # DB_DRIVER=postgres # DATABASE_URL=postgres://user:pass@host:5432/dbname +# --- webhooks (outbound; deliveries are persisted + retried) --- +# WEBHOOK_MAX_ATTEMPTS=5 # attempts before a delivery is marked failed +# WEBHOOK_TIMEOUT_MS=10000 # per-delivery HTTP timeout + # --- realtime: in-memory (default) | redis --- # Unset = single-process in-memory pub/sub + presence (the zero-config default). # Set REDIS_URL to scale horizontally across multiple server processes: websocket diff --git a/README.md b/README.md index 04696f4..a83d665 100644 --- a/README.md +++ b/README.md @@ -33,6 +33,7 @@ That's it — no clone, no config, no database to provision. Open | **Files** | Upload / download / list / delete — local disk by default, **S3** (or any S3-compatible store) optional | | **AI** | Chat (JSON or SSE streaming) over a **pluggable provider** — Anthropic, any OpenAI-compatible endpoint (OpenAI, **LiteLLM**, Ollama…), or **Amazon Bedrock** | | **Realtime** | Websocket pub/sub with presence (join/leave/typing + roster), history, and an optional **Redis** fan-out for multi-process scale | +| **Webhooks** | Outbound HTTP on data/file events — HMAC-signed, with a **persisted, retried** delivery queue and delivery history | | **Multi-tenant** | Optional isolated projects — one VM hosts many sites (set `ADMIN_KEY`) | | **Backup** | Whole-instance export / restore as one JSON file | @@ -354,6 +355,43 @@ In multi-tenant mode, the live feed of a `read:public` collection project id** (`{ project }`) — a read-only subscription, so a published frontend gets live updates without a secret. Other channels still require a project key. +### Webhooks — outbound events + +Register a URL to receive `data.*` / `file.*` events as HTTP POSTs. Deliveries are +**persisted and retried** with exponential backoff (so a brief outage on your side +doesn't drop events), and you can inspect delivery history. + +```bash +# Subscribe to created/updated docs in the "orders" collection +curl -X POST localhost:3737/v1/webhooks -H 'content-type: application/json' -d '{ + "url": "https://example.com/hook", + "events": ["data.created", "data.updated"], + "collections": ["orders"] +}' # -> { id, url, events, secret: "whsec_…", ... } (secret shown once) + +curl localhost:3737/v1/webhooks # list (without secrets) +curl localhost:3737/v1/webhooks//deliveries # recent attempts + status +curl -X POST localhost:3737/v1/webhooks//rotate-secret +curl -X DELETE localhost:3737/v1/webhooks/ +``` + +Event types use a `resource.action` shape (`data.created`, `data.updated`, +`data.deleted`, `file.created`, `file.deleted`); subscribe to exact types, a prefix +(`data.*`), or all (`*`), and optionally filter by `collections`. Each delivery is a +JSON POST `{ id, event, createdAt, project, data }` with headers: + +| Header | Meaning | +| --- | --- | +| `X-Zero-Event` | the event type | +| `X-Zero-Delivery` | unique delivery id | +| `X-Zero-Timestamp` | send time (ms) | +| `X-Zero-Signature` | `sha256=` — HMAC-SHA256 of `"."` keyed by the hook's `secret` | + +Verify by recomputing the HMAC over `` `${timestamp}.${rawBody}` `` with your stored +secret and comparing to `X-Zero-Signature`. Management is key-gated (and per-project +in multi-tenant); since webhooks make outbound requests, run with a key and restrict +egress if the instance is internet-exposed. + ## Multi-tenant projects By default the server runs in **open mode**: one implicit project, no auth — exactly diff --git a/public/docs.html b/public/docs.html index f224465..71de777 100644 --- a/public/docs.html +++ b/public/docs.html @@ -68,6 +68,7 @@

Docs

REST · Files REST · AI REST · Realtime + REST · Webhooks REST · Admin / backup REST · Meta & health Queries & filters @@ -276,6 +277,22 @@

REST API — Realtime

# Zero(base, { project: 'prj_…' }).data('posts').subscribe(fn) // read-only # Scale out: set REDIS_URL and run many processes — frames + presence fan out via Redis. +

REST API — Webhooks

+ + + + + + + + +
POST/v1/webhooksRegister { url, events:[…], collections?:[…], secret?, active? } → returns the hook + signing secret (shown here only).
GET/v1/webhooksList hooks (secrets omitted).
GET·PATCH·DELETE/v1/webhooks/:idRead (with secret) / update url·events·collections·active / delete.
POST/v1/webhooks/:id/rotate-secretIssue a new signing secret.
GET/v1/webhooks/:id/deliveriesRecent delivery attempts: status, attempts, last error/status code.
+

Events: data.created · data.updated · data.deleted · file.created · file.deleted. Subscribe to exact types, a prefix (data.*), or all (*). Deliveries are persisted and retried with backoff.

+ 
# Each delivery POSTs { id, event, createdAt, project, data } with headers:
+#   X-Zero-Event, X-Zero-Delivery, X-Zero-Timestamp, X-Zero-Signature: sha256=<hex>
+# Verify: hmacSHA256(secret, `${X-Zero-Timestamp}.${rawBody}`) === signature
+const sig = crypto.createHmac('sha256', secret).update(ts + '.' + rawBody).digest('hex');
+

REST API — Admin & backup

diff --git a/public/sdk.d.ts b/public/sdk.d.ts index 4a99b8a..05dcc26 100644 --- a/public/sdk.d.ts +++ b/public/sdk.d.ts @@ -238,6 +238,40 @@ declare namespace Zero { ai: { provider: string; ready: boolean; model: string | null }; } + interface Webhook { + id: string; + url: string; + events: string[]; + collections: string[]; + active: boolean; + createdAt: number; + /** Returned on create/get/rotate-secret only — used to verify signatures. */ + secret?: string; + } + + interface WebhookDelivery { + id: string; + event: string; + status: 'pending' | 'delivering' | 'delivered' | 'failed'; + attempts: number; + maxAttempts: number; + lastError: string | null; + lastStatusCode: number | null; + nextAttemptAt: number; + createdAt: number; + updatedAt: number; + } + + interface Webhooks { + create(config: { url: string; events: string[]; collections?: string[]; secret?: string; active?: boolean }): Promise; + list(): Promise; + get(id: string): Promise; + update(id: string, patch: { url?: string; events?: string[]; collections?: string[]; active?: boolean }): Promise; + rotateSecret(id: string): Promise; + remove(id: string): Promise<{ deleted: boolean }>; + deliveries(id: string): Promise; + } + interface Client { baseUrl: string; health(): Promise<{ ok: boolean; uptime: number }>; @@ -248,5 +282,6 @@ declare namespace Zero { files: Files; ai: AI; realtime(channel: string, opts?: RoomOptions): Room; + webhooks: Webhooks; } } diff --git a/public/sdk.js b/public/sdk.js index 359d79b..1beebaf 100644 --- a/public/sdk.js +++ b/public/sdk.js @@ -363,6 +363,20 @@ }; } + // Outbound webhooks (operator/tenant scope). create() returns the signing + // secret; deliveries() shows recent attempts + status. + var webhooks = { + create: function (config) { return request('POST', '/v1/webhooks', config || {}); }, + list: function () { return request('GET', '/v1/webhooks').then(function (r) { return r.webhooks; }); }, + get: function (id) { return request('GET', '/v1/webhooks/' + encodeURIComponent(id)); }, + update: function (id, patch) { return request('PATCH', '/v1/webhooks/' + encodeURIComponent(id), patch || {}); }, + rotateSecret: function (id) { return request('POST', '/v1/webhooks/' + encodeURIComponent(id) + '/rotate-secret'); }, + remove: function (id) { return request('DELETE', '/v1/webhooks/' + encodeURIComponent(id)); }, + deliveries: function (id) { + return request('GET', '/v1/webhooks/' + encodeURIComponent(id) + '/deliveries').then(function (r) { return r.deliveries; }); + }, + }; + return { baseUrl: base, health: function () { return request('GET', '/health'); }, @@ -377,6 +391,7 @@ files: files, ai: ai, realtime: realtime, + webhooks: webhooks, }; } diff --git a/src/config.ts b/src/config.ts index 1685cd5..c2b551c 100644 --- a/src/config.ts +++ b/src/config.ts @@ -94,6 +94,12 @@ export const config = { redisUrl: process.env.REDIS_URL ?? null, }, + // Outbound webhooks: deliveries are persisted and retried with backoff. + webhooks: { + maxAttempts: Number(process.env.WEBHOOK_MAX_ATTEMPTS ?? 5), + timeoutMs: Number(process.env.WEBHOOK_TIMEOUT_MS ?? 10_000), + }, + ai: { // Which backend serves /v1/ai/chat: anthropic (default) | openai | bedrock. // - anthropic: Anthropic Messages API (ANTHROPIC_API_KEY). diff --git a/src/db/postgres.ts b/src/db/postgres.ts index 40b3fac..fa79531 100644 --- a/src/db/postgres.ts +++ b/src/db/postgres.ts @@ -177,5 +177,37 @@ async function bootstrap(pool: pg.Pool): Promise { schema text NOT NULL, PRIMARY KEY (project, collection) ); + + CREATE TABLE IF NOT EXISTS webhooks ( + id text PRIMARY KEY, + project text NOT NULL DEFAULT 'default', + url text NOT NULL, + events text NOT NULL, + collections text NOT NULL DEFAULT '[]', + secret text NOT NULL, + active boolean NOT NULL DEFAULT true, + created_at bigint NOT NULL + ); + CREATE INDEX IF NOT EXISTS idx_webhooks_project ON webhooks (project); + + CREATE TABLE IF NOT EXISTS webhook_deliveries ( + id text PRIMARY KEY, + webhook_id text NOT NULL, + project text NOT NULL, + event text NOT NULL, + url text NOT NULL, + secret text NOT NULL, + payload text NOT NULL, + status text NOT NULL DEFAULT 'pending', + attempts integer NOT NULL DEFAULT 0, + max_attempts integer NOT NULL, + next_attempt_at bigint NOT NULL, + last_error text, + last_status_code integer, + created_at bigint NOT NULL, + updated_at bigint NOT NULL + ); + CREATE INDEX IF NOT EXISTS idx_wh_deliveries_due ON webhook_deliveries (status, next_attempt_at); + CREATE INDEX IF NOT EXISTS idx_wh_deliveries_hook ON webhook_deliveries (webhook_id, created_at DESC); `); } diff --git a/src/db/sqlite.ts b/src/db/sqlite.ts index 6037578..22c413c 100644 --- a/src/db/sqlite.ts +++ b/src/db/sqlite.ts @@ -199,6 +199,41 @@ export function bootstrap(db: Database.Database): void { ); `); + // --- outbound webhooks + their persisted, retried delivery queue --- + db.exec(` + CREATE TABLE IF NOT EXISTS webhooks ( + id TEXT PRIMARY KEY, + project TEXT NOT NULL DEFAULT 'default', + url TEXT NOT NULL, + events TEXT NOT NULL, -- JSON array of event patterns + collections TEXT NOT NULL DEFAULT '[]', -- JSON array; empty = all + secret TEXT NOT NULL, + active INTEGER NOT NULL DEFAULT 1, + created_at INTEGER NOT NULL + ); + CREATE INDEX IF NOT EXISTS idx_webhooks_project ON webhooks (project); + + CREATE TABLE IF NOT EXISTS webhook_deliveries ( + id TEXT PRIMARY KEY, + webhook_id TEXT NOT NULL, + project TEXT NOT NULL, + event TEXT NOT NULL, + url TEXT NOT NULL, -- snapshot, so a deleted/edited hook can't orphan it + secret TEXT NOT NULL, + payload TEXT NOT NULL, + status TEXT NOT NULL DEFAULT 'pending', -- pending|delivering|delivered|failed + attempts INTEGER NOT NULL DEFAULT 0, + max_attempts INTEGER NOT NULL, + next_attempt_at INTEGER NOT NULL, + last_error TEXT, + last_status_code INTEGER, + created_at INTEGER NOT NULL, + updated_at INTEGER NOT NULL + ); + CREATE INDEX IF NOT EXISTS idx_wh_deliveries_due ON webhook_deliveries (status, next_attempt_at); + CREATE INDEX IF NOT EXISTS idx_wh_deliveries_hook ON webhook_deliveries (webhook_id, created_at DESC); + `); + // --- full-text index (rebuilt if it predates the project column) --- if (tableExists(db, 'documents_fts') && !hasColumn(db, 'documents_fts', 'project')) { db.exec(`DROP TABLE documents_fts;`); diff --git a/src/modules/data/routes.ts b/src/modules/data/routes.ts index 181be20..3b9bceb 100644 --- a/src/modules/data/routes.ts +++ b/src/modules/data/routes.ts @@ -97,8 +97,25 @@ export async function dataRoutes(app: FastifyInstance): Promise { // Broadcast a change so clients subscribed to `data:` get live // updates. Channels are namespaced per project by the realtime routes. - const emit = (project: string, collection: string, event: Record) => + // Broadcast a change over realtime AND durably enqueue it for matching webhooks + // (best-effort: a webhook failure never blocks the write). data.created/updated + // carry the document; data.deleted carries just the id. + const emit = async ( + project: string, + collection: string, + event: { type: 'created' | 'updated' | 'deleted'; collection: string; id: string; document?: unknown }, + ): Promise => { app.realtime.publish(`${project}:data:${collection}`, event, { history: false }); + try { + await app.webhooks.enqueue(project, `data.${event.type}`, { + collection, + id: event.id, + ...(event.document !== undefined ? { document: event.document } : {}), + }); + } catch (err) { + app.log.error({ err }, 'webhook enqueue failed'); + } + }; // For an anonymous (public-ACL) request, the collection's hidden fields to // strip from read results; empty for authenticated callers (they see all). @@ -138,7 +155,7 @@ export async function dataRoutes(app: FastifyInstance): Promise { const { collection } = req.params as { collection: string }; const body = (req.body ?? {}) as Record; const doc = await store.create(req.projectId, collection, body); - emit(req.projectId, collection, { type: 'created', collection, id: doc.id, document: doc }); + await emit(req.projectId, collection, { type: 'created', collection, id: doc.id, document: doc }); reply.code(201); return doc; }); @@ -173,7 +190,7 @@ export async function dataRoutes(app: FastifyInstance): Promise { const created = await store.createMany(req.projectId, collection, items as Record[]); for (const doc of created) { - emit(req.projectId, collection, { type: 'created', collection, id: doc.id, document: doc }); + await emit(req.projectId, collection, { type: 'created', collection, id: doc.id, document: doc }); } reply.code(201); return { created }; @@ -341,7 +358,7 @@ export async function dataRoutes(app: FastifyInstance): Promise { const doc = await write((req.body ?? {}) as Record, ifMatchVersion(req)); if (!doc) return reply.code(404).send({ error: 'not_found' }); reply.header('ETag', String(doc.updatedAt)); - emit(req.projectId, collection, { type: 'updated', collection, id: doc.id, document: doc }); + await emit(req.projectId, collection, { type: 'updated', collection, id: doc.id, document: doc }); return doc; } catch (err) { if (err instanceof VersionConflictError) { @@ -386,14 +403,14 @@ export async function dataRoutes(app: FastifyInstance): Promise { }); } const ids = await store.deleteMany(req.projectId, collection, filters, orFilters); - for (const id of ids) emit(req.projectId, collection, { type: 'deleted', collection, id }); + for (const id of ids) await emit(req.projectId, collection, { type: 'deleted', collection, id }); return { deleted: ids.length }; }); app.delete('/v1/data/:collection/:id', async (req, reply) => { const { collection, id } = req.params as { collection: string; id: string }; if (!(await store.delete(req.projectId, collection, id))) return reply.code(404).send({ error: 'not_found' }); - emit(req.projectId, collection, { type: 'deleted', collection, id }); + await emit(req.projectId, collection, { type: 'deleted', collection, id }); return { deleted: true }; }); } diff --git a/src/modules/files/routes.ts b/src/modules/files/routes.ts index de85ef9..13a6a4c 100644 --- a/src/modules/files/routes.ts +++ b/src/modules/files/routes.ts @@ -8,6 +8,15 @@ import { FileStore } from './storage.js'; export async function fileRoutes(app: FastifyInstance): Promise { const store = new FileStore(app.db, app.blobs); + // Durably enqueue a file event for matching webhooks (best-effort). + const fire = async (project: string, event: string, data: Record): Promise => { + try { + await app.webhooks.enqueue(project, event, data); + } catch (err) { + app.log.error({ err }, 'webhook enqueue failed'); + } + }; + app.post('/v1/files', async (req, reply) => { const data = await req.file(); if (!data) { @@ -25,6 +34,12 @@ export async function fileRoutes(app: FastifyInstance): Promise { return reply.code(413).send({ error: 'file_too_large' }); } + await fire(req.projectId, 'file.created', { + id: meta.id, + name: meta.name, + contentType: meta.contentType, + size: meta.size, + }); reply.code(201); return meta; }); @@ -53,6 +68,7 @@ export async function fileRoutes(app: FastifyInstance): Promise { app.delete('/v1/files/:id', async (req, reply) => { const { id } = req.params as { id: string }; if (!(await store.delete(req.projectId, id))) return reply.code(404).send({ error: 'not_found' }); + await fire(req.projectId, 'file.deleted', { id }); return { deleted: true }; }); } diff --git a/src/modules/meta/routes.ts b/src/modules/meta/routes.ts index 2258695..8c20a97 100644 --- a/src/modules/meta/routes.ts +++ b/src/modules/meta/routes.ts @@ -92,6 +92,12 @@ export async function metaRoutes(app: FastifyInstance): Promise { provider: config.ai.provider, // anthropic | openai | bedrock (AI_PROVIDER) chat: 'POST /v1/ai/chat { messages, system?, model?, max_tokens?, stream? }', }, + webhooks: { + manage: 'POST/GET /v1/webhooks · GET/PATCH/DELETE /v1/webhooks/:id · POST /v1/webhooks/:id/rotate-secret (key-gated)', + deliveries: 'GET /v1/webhooks/:id/deliveries — recent delivery attempts + status', + events: ['data.created', 'data.updated', 'data.deleted', 'file.created', 'file.deleted'], + signing: 'HMAC-SHA256 over "." in X-Zero-Signature: sha256=', + }, realtime: { subscribe: 'WS /v1/realtime/:channel?as= (frames: type=message|presence; presence events: sync/join/leave/typing)', publish: 'POST /v1/realtime/:channel', diff --git a/src/modules/webhooks/dispatcher.ts b/src/modules/webhooks/dispatcher.ts new file mode 100644 index 0000000..0175725 --- /dev/null +++ b/src/modules/webhooks/dispatcher.ts @@ -0,0 +1,78 @@ +import { createHmac } from 'node:crypto'; +import type { FastifyBaseLogger } from 'fastify'; +import type { WebhookStore, DeliveryJob } from './store.js'; + +const POLL_MS = 1000; // how often to look for due deliveries +const BATCH = 20; // max deliveries claimed per tick + +/** `sha256=` HMAC over `"{timestamp}.{body}"`, the string subscribers re-sign. */ +export function sign(secret: string, timestamp: number, body: string): string { + return `sha256=${createHmac('sha256', secret).update(`${timestamp}.${body}`).digest('hex')}`; +} + +/** + * Background delivery worker. Each tick claims due deliveries (atomically, so it's + * safe to run one per process) and POSTs them with a signed body; non-2xx or + * network/timeout failures are retried with backoff by the store, up to the cap. + */ +export class WebhookDispatcher { + private timer: NodeJS.Timeout | null = null; + private running = false; + + constructor( + private readonly store: WebhookStore, + private readonly log: FastifyBaseLogger, + private readonly timeoutMs: number, + ) {} + + start(): void { + if (this.timer) return; + this.timer = setInterval(() => void this.tick(), POLL_MS); + this.timer.unref?.(); + } + + stop(): void { + if (this.timer) clearInterval(this.timer); + this.timer = null; + } + + /** Exposed for tests: run one delivery pass synchronously. */ + async tick(): Promise { + if (this.running) return; // don't overlap ticks + this.running = true; + try { + const due = await this.store.claimDue(Date.now(), BATCH); + await Promise.all(due.map((job) => this.deliver(job))); + } catch (err) { + this.log.error({ err }, 'webhook dispatch tick failed'); + } finally { + this.running = false; + } + } + + private async deliver(job: DeliveryJob): Promise { + const ts = Date.now(); + try { + const res = await fetch(job.url, { + method: 'POST', + signal: AbortSignal.timeout(this.timeoutMs), + headers: { + 'content-type': 'application/json', + 'user-agent': 'zero-config-data-api', + 'x-zero-event': job.event, + 'x-zero-delivery': job.id, + 'x-zero-timestamp': String(ts), + 'x-zero-signature': sign(job.secret, ts, job.payload), + }, + body: job.payload, + }); + if (res.ok) { + await this.store.markDelivered(job.id, job.attempts + 1, res.status); + } else { + await this.store.recordFailure(job, `HTTP ${res.status}`, res.status); + } + } catch (err) { + await this.store.recordFailure(job, (err as Error).message || 'request failed', null); + } + } +} diff --git a/src/modules/webhooks/routes.ts b/src/modules/webhooks/routes.ts new file mode 100644 index 0000000..154292e --- /dev/null +++ b/src/modules/webhooks/routes.ts @@ -0,0 +1,105 @@ +import type { FastifyInstance } from 'fastify'; + +const EVENT = /^[a-z]+\.(\*|[a-z]+)$|^\*$/; // e.g. data.created, file.*, * + +/** Validate the webhook URL: must be an absolute http(s) URL. */ +function validUrl(url: unknown): url is string { + if (typeof url !== 'string') return false; + try { + const u = new URL(url); + return u.protocol === 'http:' || u.protocol === 'https:'; + } catch { + return false; + } +} + +function validEvents(events: unknown): events is string[] { + return Array.isArray(events) && events.length > 0 && events.every((e) => typeof e === 'string' && EVENT.test(e)); +} + +/** + * Webhook management — scoped to the request's project (key-gated; never + * anonymous). Create returns the signing secret (also via get/rotate-secret); + * list omits it. Deliveries are persisted and retried by the dispatcher. + */ +export async function webhookRoutes(app: FastifyInstance): Promise { + const store = app.webhooks; + + app.post('/v1/webhooks', async (req, reply) => { + const body = (req.body ?? {}) as { + url?: unknown; + events?: unknown; + collections?: unknown; + secret?: unknown; + active?: unknown; + }; + if (!validUrl(body.url)) { + return reply.code(400).send({ error: 'bad_request', message: 'url must be an http(s) URL.' }); + } + if (!validEvents(body.events)) { + return reply.code(400).send({ + error: 'bad_request', + message: 'events must be a non-empty array of patterns like "data.created", "file.*", or "*".', + }); + } + if (body.collections !== undefined && !(Array.isArray(body.collections) && body.collections.every((c) => typeof c === 'string'))) { + return reply.code(400).send({ error: 'bad_request', message: 'collections must be an array of collection names.' }); + } + const created = await store.create(req.projectId, { + url: body.url, + events: body.events, + collections: body.collections as string[] | undefined, + secret: typeof body.secret === 'string' ? body.secret : undefined, + active: body.active === undefined ? undefined : !!body.active, + }); + reply.code(201); + return created; + }); + + app.get('/v1/webhooks', async (req) => ({ webhooks: await store.list(req.projectId) })); + + app.get('/v1/webhooks/:id', async (req, reply) => { + const { id } = req.params as { id: string }; + const wh = await store.get(req.projectId, id); + if (!wh) return reply.code(404).send({ error: 'not_found' }); + return wh; + }); + + app.patch('/v1/webhooks/:id', async (req, reply) => { + const { id } = req.params as { id: string }; + const body = (req.body ?? {}) as { url?: unknown; events?: unknown; collections?: unknown; active?: unknown }; + if (body.url !== undefined && !validUrl(body.url)) { + return reply.code(400).send({ error: 'bad_request', message: 'url must be an http(s) URL.' }); + } + if (body.events !== undefined && !validEvents(body.events)) { + return reply.code(400).send({ error: 'bad_request', message: 'events must be a non-empty array of valid patterns.' }); + } + const updated = await store.update(req.projectId, id, { + url: body.url as string | undefined, + events: body.events as string[] | undefined, + collections: body.collections as string[] | undefined, + active: body.active === undefined ? undefined : !!body.active, + }); + if (!updated) return reply.code(404).send({ error: 'not_found' }); + return updated; + }); + + app.post('/v1/webhooks/:id/rotate-secret', async (req, reply) => { + const { id } = req.params as { id: string }; + const rotated = await store.rotateSecret(req.projectId, id); + if (!rotated) return reply.code(404).send({ error: 'not_found' }); + return rotated; + }); + + app.delete('/v1/webhooks/:id', async (req, reply) => { + const { id } = req.params as { id: string }; + if (!(await store.delete(req.projectId, id))) return reply.code(404).send({ error: 'not_found' }); + return { deleted: true }; + }); + + app.get('/v1/webhooks/:id/deliveries', async (req, reply) => { + const { id } = req.params as { id: string }; + if (!(await store.get(req.projectId, id))) return reply.code(404).send({ error: 'not_found' }); + return { deliveries: await store.deliveries(req.projectId, id) }; + }); +} diff --git a/src/modules/webhooks/store.ts b/src/modules/webhooks/store.ts new file mode 100644 index 0000000..5ba49d3 --- /dev/null +++ b/src/modules/webhooks/store.ts @@ -0,0 +1,292 @@ +import { randomBytes } from 'node:crypto'; +import type { Db } from '../../db.js'; +import { newId } from '../../lib/id.js'; + +export interface Webhook { + id: string; + url: string; + events: string[]; + collections: string[]; + active: boolean; + createdAt: number; + /** Only returned on create / get / rotate-secret — used to verify signatures. */ + secret?: string; +} + +export interface Delivery { + id: string; + event: string; + status: 'pending' | 'delivering' | 'delivered' | 'failed'; + attempts: number; + maxAttempts: number; + lastError: string | null; + lastStatusCode: number | null; + nextAttemptAt: number; + createdAt: number; + updatedAt: number; +} + +/** A claimed delivery row the dispatcher needs in order to send it. */ +export interface DeliveryJob { + id: string; + event: string; + url: string; + secret: string; + payload: string; + attempts: number; + maxAttempts: number; +} + +interface WebhookRow { + id: string; + url: string; + events: string; + collections: string; + secret: string; + active: number | boolean; + created_at: number; +} + +const toBool = (v: number | boolean): boolean => v === true || v === 1; +const parseArr = (s: string): string[] => { + try { + const v = JSON.parse(s); + return Array.isArray(v) ? v.filter((x): x is string => typeof x === 'string') : []; + } catch { + return []; + } +}; + +/** Does an event type (e.g. `data.created`) match one of a hook's patterns? */ +export function matchEvent(patterns: string[], event: string): boolean { + return patterns.some( + (p) => p === '*' || p === event || (p.endsWith('.*') && event.startsWith(p.slice(0, -1))), + ); +} + +const CACHE_TTL_MS = 2000; // bounds cross-process staleness after a hook change + +/** + * Outbound webhooks: registration + a persisted, retried delivery queue. Events + * (data.* / file.*) are enqueued as durable delivery rows; a background + * dispatcher (see dispatcher.ts) sends and retries them. The list of active + * hooks per project is cached briefly so the write path doesn't hit the DB when + * a project has no webhooks (the common case). + */ +export class WebhookStore { + private readonly cache = new Map(); + constructor( + private readonly db: Db, + private readonly maxAttempts: number, + ) {} + + private toWebhook(r: WebhookRow, withSecret: boolean): Webhook { + return { + id: r.id, + url: r.url, + events: parseArr(r.events), + collections: parseArr(r.collections), + active: toBool(r.active), + createdAt: r.created_at, + ...(withSecret ? { secret: r.secret } : {}), + }; + } + + async create( + project: string, + input: { url: string; events: string[]; collections?: string[]; secret?: string; active?: boolean }, + ): Promise { + const row: WebhookRow = { + id: `wh_${newId()}`, + url: input.url, + events: JSON.stringify(input.events), + collections: JSON.stringify(input.collections ?? []), + secret: input.secret && input.secret.length > 0 ? input.secret : `whsec_${randomBytes(24).toString('base64url')}`, + active: input.active === false ? 0 : 1, + created_at: Date.now(), + }; + await this.db.run( + `INSERT INTO webhooks (id, project, url, events, collections, secret, active, created_at) + VALUES (?, ?, ?, ?, ?, ?, ?, ?)`, + [row.id, project, row.url, row.events, row.collections, row.secret, row.active, row.created_at], + ); + this.cache.delete(project); + return this.toWebhook(row, true); + } + + async list(project: string): Promise { + const rows = await this.db.all( + `SELECT * FROM webhooks WHERE project = ? ORDER BY created_at`, + [project], + ); + return rows.map((r) => this.toWebhook(r, false)); + } + + async get(project: string, id: string, withSecret = true): Promise { + const row = await this.db.get(`SELECT * FROM webhooks WHERE project = ? AND id = ?`, [project, id]); + return row ? this.toWebhook(row, withSecret) : null; + } + + async update( + project: string, + id: string, + patch: { url?: string; events?: string[]; collections?: string[]; active?: boolean }, + ): Promise { + const existing = await this.get(project, id); + if (!existing) return null; + const next = { + url: patch.url ?? existing.url, + events: JSON.stringify(patch.events ?? existing.events), + collections: JSON.stringify(patch.collections ?? existing.collections), + active: (patch.active ?? existing.active) ? 1 : 0, + }; + await this.db.run(`UPDATE webhooks SET url = ?, events = ?, collections = ?, active = ? WHERE project = ? AND id = ?`, [ + next.url, + next.events, + next.collections, + next.active, + project, + id, + ]); + this.cache.delete(project); + return this.get(project, id); + } + + async rotateSecret(project: string, id: string): Promise { + const existing = await this.get(project, id); + if (!existing) return null; + const secret = `whsec_${randomBytes(24).toString('base64url')}`; + await this.db.run(`UPDATE webhooks SET secret = ? WHERE project = ? AND id = ?`, [secret, project, id]); + this.cache.delete(project); + return { ...existing, secret }; + } + + async delete(project: string, id: string): Promise { + const res = await this.db.run(`DELETE FROM webhooks WHERE project = ? AND id = ?`, [project, id]); + this.cache.delete(project); + return res.changes > 0; + } + + /** Recent delivery attempts for a webhook, newest first (for debugging). */ + async deliveries(project: string, webhookId: string, limit = 50): Promise { + const rows = await this.db.all<{ + id: string; + event: string; + status: Delivery['status']; + attempts: number; + max_attempts: number; + last_error: string | null; + last_status_code: number | null; + next_attempt_at: number; + created_at: number; + updated_at: number; + }>( + `SELECT id, event, status, attempts, max_attempts, last_error, last_status_code, next_attempt_at, created_at, updated_at + FROM webhook_deliveries WHERE project = ? AND webhook_id = ? ORDER BY created_at DESC LIMIT ?`, + [project, webhookId, limit], + ); + return rows.map((r) => ({ + id: r.id, + event: r.event, + status: r.status, + attempts: r.attempts, + maxAttempts: r.max_attempts, + lastError: r.last_error, + lastStatusCode: r.last_status_code, + nextAttemptAt: r.next_attempt_at, + createdAt: r.created_at, + updatedAt: r.updated_at, + })); + } + + /** Active hooks for a project, cached briefly to keep the write path off the DB. */ + private async activeHooks(project: string): Promise { + const cached = this.cache.get(project); + if (cached && Date.now() - cached.at < CACHE_TTL_MS) return cached.hooks; + // `WHERE active` reads truthy on both backends (SQLite 1/0, Postgres boolean). + const rows = await this.db.all(`SELECT * FROM webhooks WHERE project = ? AND active`, [project]); + this.cache.set(project, { hooks: rows, at: Date.now() }); + return rows; + } + + /** + * Enqueue an event for every matching active hook in the project. Best-effort: + * a failure here is logged by the caller and never blocks the originating write. + * `data.collection` (when present) is matched against a hook's collection filter. + */ + async enqueue(project: string, event: string, data: Record): Promise { + const hooks = await this.activeHooks(project); + if (hooks.length === 0) return; + const collection = typeof data.collection === 'string' ? data.collection : null; + const now = Date.now(); + for (const h of hooks) { + if (!matchEvent(parseArr(h.events), event)) continue; + const collections = parseArr(h.collections); + if (collection && collections.length > 0 && !collections.includes(collection)) continue; + const id = `whd_${newId()}`; + const payload = JSON.stringify({ id, event, createdAt: now, project, data }); + await this.db.run( + `INSERT INTO webhook_deliveries + (id, webhook_id, project, event, url, secret, payload, status, attempts, max_attempts, next_attempt_at, created_at, updated_at) + VALUES (?, ?, ?, ?, ?, ?, ?, 'pending', 0, ?, ?, ?, ?)`, + [id, h.id, project, event, h.url, h.secret, payload, this.maxAttempts, now, now, now], + ); + } + } + + // --- dispatcher support -------------------------------------------------- + + /** Claim up to `limit` due deliveries (flipping pending → delivering atomically). */ + async claimDue(now: number, limit: number): Promise { + const rows = await this.db.all<{ + id: string; + event: string; + url: string; + secret: string; + payload: string; + attempts: number; + max_attempts: number; + }>( + `SELECT id, event, url, secret, payload, attempts, max_attempts FROM webhook_deliveries + WHERE status = 'pending' AND next_attempt_at <= ? ORDER BY next_attempt_at LIMIT ?`, + [now, limit], + ); + const claimed: DeliveryJob[] = []; + for (const r of rows) { + const res = await this.db.run( + `UPDATE webhook_deliveries SET status = 'delivering', updated_at = ? WHERE id = ? AND status = 'pending'`, + [Date.now(), r.id], + ); + if (res.changes === 1) { + claimed.push({ id: r.id, event: r.event, url: r.url, secret: r.secret, payload: r.payload, attempts: r.attempts, maxAttempts: r.max_attempts }); + } + } + return claimed; + } + + async markDelivered(id: string, attempts: number, statusCode: number): Promise { + await this.db.run( + `UPDATE webhook_deliveries SET status = 'delivered', attempts = ?, last_status_code = ?, last_error = NULL, updated_at = ? WHERE id = ?`, + [attempts, statusCode, Date.now(), id], + ); + } + + /** Record a failed attempt: reschedule with backoff, or mark failed at the cap. */ + async recordFailure(job: DeliveryJob, error: string, statusCode: number | null): Promise { + const attempts = job.attempts + 1; + const now = Date.now(); + if (attempts >= job.maxAttempts) { + await this.db.run( + `UPDATE webhook_deliveries SET status = 'failed', attempts = ?, last_error = ?, last_status_code = ?, updated_at = ? WHERE id = ?`, + [attempts, error.slice(0, 500), statusCode, now, job.id], + ); + return; + } + // Exponential backoff, capped at 1h: 5s, 10s, 20s, 40s, … + const delay = Math.min(60 * 60 * 1000, 5000 * 2 ** (attempts - 1)); + await this.db.run( + `UPDATE webhook_deliveries SET status = 'pending', attempts = ?, last_error = ?, last_status_code = ?, next_attempt_at = ?, updated_at = ? WHERE id = ?`, + [attempts, error.slice(0, 500), statusCode, now + delay, now, job.id], + ); + } +} diff --git a/src/server.ts b/src/server.ts index a7696d7..76a406b 100644 --- a/src/server.ts +++ b/src/server.ts @@ -19,6 +19,9 @@ import { metaRoutes } from './modules/meta/routes.js'; import { backupRoutes } from './modules/backup/routes.js'; import { registerRateLimit, type RateLimitOptions } from './rate-limit.js'; import { registerObservability } from './metrics.js'; +import { WebhookStore } from './modules/webhooks/store.js'; +import { WebhookDispatcher } from './modules/webhooks/dispatcher.js'; +import { webhookRoutes } from './modules/webhooks/routes.js'; import { AclStore } from './modules/acl/store.js'; import { SchemaStore, SchemaValidationError } from './modules/data/schema.js'; import { getBlobStore } from './modules/files/blob.js'; @@ -80,6 +83,8 @@ export interface BuildOptions { adminKey?: string | null; /** Override the CORS allowlist (comma-separated origins). null/undefined = use env. */ corsOrigin?: string | null; + /** Override the webhook max delivery attempts (used by the test suite). */ + webhookMaxAttempts?: number; } /** Resolve the @fastify/cors `origin` option: an allowlist array, or `true` (open). */ @@ -127,6 +132,12 @@ export async function buildServer(opts: BuildOptions = {}): Promise webhookDispatcher.stop()); app.decorateRequest('projectId', DEFAULT_PROJECT); // True unless the request reached a collection anonymously via a public ACL; // read routes use this to redact `hidden` fields from anonymous callers. @@ -266,6 +277,7 @@ export async function buildServer(opts: BuildOptions = {}): Promise; body: string }[]; close: () => Promise }> { + const requests: { headers: Record; body: string }[] = []; + const server = http.createServer((req, res) => { + let body = ''; + req.on('data', (c) => (body += c)); + req.on('end', () => { + requests.push({ headers: req.headers as Record, body }); + res.writeHead(status); + res.end('ok'); + }); + }); + await new Promise((r) => server.listen(0, '127.0.0.1', () => r())); + const port = (server.address() as AddressInfo).port; + return { + url: `http://127.0.0.1:${port}/hook`, + requests, + close: () => new Promise((r) => server.close(() => r())), + }; +} + let app: FastifyInstance; let base: string; // The browser SDK is a plain side-effecting script that assigns globalThis.Zero. @@ -675,6 +698,107 @@ test('files: upload, download, list, delete', async () => { assert.equal(res.status, 404); }); +test('webhooks: a data.created event is signed, delivered, and recorded', async () => { + const wait = (ms = 100) => new Promise((r) => setTimeout(r, ms)); + const rec = await startReceiver(200); + try { + const wh = await json( + await fetch(`${base}/v1/webhooks`, { + method: 'POST', + headers: { 'content-type': 'application/json' }, + body: JSON.stringify({ url: rec.url, events: ['data.created'], collections: ['hooked'] }), + }), + ); + assert.match(wh.id, /^wh_/); + assert.match(wh.secret, /^whsec_/); + + const doc = await json( + await fetch(`${base}/v1/data/hooked`, { + method: 'POST', + headers: { 'content-type': 'application/json' }, + body: JSON.stringify({ hello: 'world' }), + }), + ); + + await app.webhookDispatcher.tick(); + await wait(); + + assert.equal(rec.requests.length, 1, 'exactly one delivery'); + const got = rec.requests[0]!; + assert.equal(got.headers['x-zero-event'], 'data.created'); + assert.equal(got.headers['x-zero-delivery']?.startsWith('whd_'), true); + // The signature verifies over "." with the hook's secret. + const ts = got.headers['x-zero-timestamp']; + const expected = `sha256=${createHmac('sha256', wh.secret).update(`${ts}.${got.body}`).digest('hex')}`; + assert.equal(got.headers['x-zero-signature'], expected); + // Payload envelope is stable. + const payload = JSON.parse(got.body); + assert.equal(payload.event, 'data.created'); + assert.equal(payload.data.collection, 'hooked'); + assert.equal(payload.data.id, doc.id); + assert.equal(payload.data.document.hello, 'world'); + + const deliveries = (await json(await fetch(`${base}/v1/webhooks/${wh.id}/deliveries`))).deliveries; + assert.equal(deliveries.length, 1); + assert.equal(deliveries[0].status, 'delivered'); + + // list omits the secret; get includes it. + assert.equal((await json(await fetch(`${base}/v1/webhooks`))).webhooks.some((w: any) => w.secret), false); + + await fetch(`${base}/v1/webhooks/${wh.id}`, { method: 'DELETE' }); // don't fire for later tests + } finally { + await rec.close(); + } +}); + +test('webhooks: a failing endpoint is retried and recorded as failed at the cap', async () => { + const wait = (ms = 100) => new Promise((r) => setTimeout(r, ms)); + const rec = await startReceiver(500); // always errors + const server = await buildServer({ logger: false, webhookMaxAttempts: 1 }); + await server.listen({ host: '127.0.0.1', port: 0 }); + const b = `http://127.0.0.1:${(server.server.address() as AddressInfo).port}`; + try { + const wh = await json( + await fetch(`${b}/v1/webhooks`, { + method: 'POST', + headers: { 'content-type': 'application/json' }, + body: JSON.stringify({ url: rec.url, events: ['data.created'], collections: ['failers'] }), + }), + ); + await fetch(`${b}/v1/data/failers`, { + method: 'POST', + headers: { 'content-type': 'application/json' }, + body: JSON.stringify({ x: 1 }), + }); + await server.webhookDispatcher.tick(); + await wait(); + + assert.ok(rec.requests.length >= 1, 'the endpoint was attempted'); + const d = (await json(await fetch(`${b}/v1/webhooks/${wh.id}/deliveries`))).deliveries[0]; + assert.equal(d.status, 'failed'); // maxAttempts=1 → terminal after the first 500 + assert.equal(d.attempts, 1); + assert.equal(d.lastStatusCode, 500); + } finally { + await server.close(); + await rec.close(); + } +}); + +test('webhooks: bad url and bad events are rejected with 400', async () => { + const bad1 = await fetch(`${base}/v1/webhooks`, { + method: 'POST', + headers: { 'content-type': 'application/json' }, + body: JSON.stringify({ url: 'ftp://nope', events: ['data.created'] }), + }); + assert.equal(bad1.status, 400); + const bad2 = await fetch(`${base}/v1/webhooks`, { + method: 'POST', + headers: { 'content-type': 'application/json' }, + body: JSON.stringify({ url: 'https://example.com/h', events: ['not a valid event'] }), + }); + assert.equal(bad2.status, 400); +}); + test('realtime: broadcast reaches other subscribers', async () => { const wsUrl = `${base.replace('http', 'ws')}/v1/realtime/test-room`; const a = new WebSocket(wsUrl);