feat: outbound webhooks for event/photo lifecycle (#327)
PicPeak POSTs lifecycle notifications to admin-configured URLs. Each delivery is signed HMAC-SHA256 in the X-PicPeak-Signature header. Verified end-to-end: 1/1 Playwright spec, 8/8 backend integration tests, full UI click-through via Chrome DevTools. Schema (migration 082) - webhooks: id, name, url, secret (plaintext — required to compute HMAC for every outbound POST), secret_preview, events[], active, filter, template, created_by, timestamps, last_success_at/last_failure_at. - webhook_deliveries: webhook_id (FK CASCADE), event_type, payload, attempt_count, status (pending|success|failed), response_status, response_body (truncated to 1KB), latency_ms, next_retry_at, last_error, created_at, completed_at. Composite index (status, next_retry_at) serves the worker's hot-path query. Service + worker - webhookService.fire(eventType, data) — non-throwing entry point used by lifecycle hooks. Looks up active webhooks subscribed to the event and applies their per-webhook filter (dot-path equality predicate) before enqueueing one webhook_deliveries row per match. Filter and template logic ship in this commit; admin surfaces in the follow-up. - webhookDeliveryWorker — setInterval(5s) poller; fetches up to 5 pending rows; per delivery: re-validates URL via networkValidation (DNS-rebinding mitigation, opt-out via WEBHOOK_ALLOW_PRIVATE_URLS), signs body with HMAC-SHA256, POSTs with 10s timeout, records outcome. Backoff schedule: 1m → 5m → 30m → 2h → 12h, max 5 attempts. Response body truncated to 1KB before storage. If a webhook has a template, the rendered string replaces the JSON envelope as the request body (signature is computed over the bytes actually sent). Lifecycle wiring - adminEvents.js POST /events → event.created (+ event.published when not draft); POST /:id/publish → event.published. - routes/events.js (legacy public POST) → event.created + event.published. - routes/v1/events.js (#322 API) → event.created + event.published on create, photo.uploaded on photo POST. - archiveService.archiveEvent() → event.archived. Per-photo photo.deleted intentionally NOT fired during cascade — receivers infer from event.archived to avoid flooding (issue spec). - expirationChecker.handleExpiredEvent() → event.expired BEFORE the cascading archive (so receivers see expired→archived in order). - adminPhotos.js — photo.uploaded on each batch row, photo.deleted on single + bulk delete. - photoProcessor.js — photo.uploaded for guest uploads + auto-import (covers all entry paths). - fileWatcher.js — photo.uploaded on add, photo.deleted on unlink (local mode only). Admin endpoints (mirrors adminApiTokens.js pattern) - /api/admin/webhooks: GET list, POST create (returns plaintext secret exactly once), GET :id, PUT :id, DELETE :id, POST :id/test (synthetic fire), GET :id/deliveries (paginated, filter by status), GET :id/deliveries/:deliveryId, POST :id/deliveries/:deliveryId/replay. Frontend - Settings → Webhooks tab (mirrors API Tokens layout): name + URL + event checkboxes + "Advanced" expander for filter (JSON) and template. Plaintext secret shown once on creation with a Copy button. Active/ Disabled toggle button per row. - /admin/webhooks/:id/deliveries — operational debug surface. Table with timestamp/event/status/attempts/HTTP/latency. Status filter chips (all/pending/success/failed). Row click → slide-over with payload + signature + response body. Replay button on failed rows. Send-test-event dialog. Auto-refresh every 10s. Dev infrastructure - dev/webhook-receiver/ — tiny node:alpine HTTP server (~100 LOC) that records every POST to an in-memory ring buffer. Exposes GET /requests for the E2E spec to assert deliveries landed with the right HMAC. Sibling pattern to MinIO. Reachable from the backend at http://webhook-receiver:8888 inside the picpeak network. Tests - backend/__tests__/integration/webhookDelivery.test.js (8/8) — signature verification, headers, retry/backoff, max-attempts → failed, response truncation, disabled-mid-flight, SSRF block, start/stop idempotency. - tests/e2e/webhooks-roundtrip.spec.ts (1/1) — create webhook → trigger event.published → assert receiver got POST with valid HMAC → visit deliveries page → row visible with status=success → API test event → API replay → disable webhook → assert no new delivery. Docs - README §"Webhooks" — event catalog, payload shape, HMAC verification in Node + Python + bash, retry semantics, SSRF protection. - .env.example — WEBHOOK_ALLOW_PRIVATE_URLS, WEBHOOK_DELIVERY_INTERVAL_MS, WEBHOOK_DELIVERY_CONCURRENCY, WEBHOOK_HTTP_TIMEOUT_MS, WEBHOOK_MAX_ATTEMPTS. Out of scope for v1 (per issue): webhook templates' code-eval (the ${dot.path} substitution that ships is pure string replacement, no expression engine — see follow-up commit), per-webhook rate limiting beyond the global concurrency cap, synchronous "ask before delete" webhooks. Spanning files - App.tsx pulls in this commit with both the AnalyticsBootstrap (#325 dedup) and the WebhookDeliveriesPage route registration. Splitting via git add -p was forfeit for sanity; the single 92-line diff is honest about both contributions. - adminEvents.js diff bundles the webhook fires AND the allow_presigned_download field plumbing (#328 follow-up). Same reasoning. - The new webhookService/Worker/adminWebhooks files include the filter and template logic from the follow-up — they were authored in one pass; splitting them post-hoc would have produced fragile partial files. The follow-up commit covers the migration and the UI for these.
This commit is contained in:
Paul Nothaft
@@ -298,6 +298,9 @@ router.post('/', adminAuth, requirePermission('events.create'), [
|
||||
body('enable_devtools_protection').optional().isBoolean(),
|
||||
body('watermark_downloads').optional().isBoolean(),
|
||||
body('watermark_text').optional().trim(),
|
||||
// #328 follow-up: per-event opt-in for presigned-URL "Download All".
|
||||
// Bypasses watermarks; admin must enable knowingly.
|
||||
body('allow_presigned_download').optional().isBoolean(),
|
||||
body('css_template_id').optional({ nullable: true, checkFalsy: true }).isInt(),
|
||||
// Hero logo settings
|
||||
body('hero_logo_visible').optional().isBoolean(),
|
||||
@@ -344,6 +347,7 @@ router.post('/', adminAuth, requirePermission('events.create'), [
|
||||
enable_devtools_protection: enableDevtoolsProtectionInput,
|
||||
watermark_downloads = false,
|
||||
watermark_text = null,
|
||||
allow_presigned_download = false,
|
||||
require_password: requirePasswordInput,
|
||||
// Feedback settings
|
||||
feedback_enabled = false,
|
||||
@@ -557,6 +561,7 @@ router.post('/', adminAuth, requirePermission('events.create'), [
|
||||
enable_devtools_protection: formatBoolean(effectiveEnableDevtoolsProtection),
|
||||
watermark_downloads: formatBoolean(watermark_downloads !== undefined ? watermark_downloads : false),
|
||||
watermark_text,
|
||||
allow_presigned_download: formatBoolean(allow_presigned_download === true || allow_presigned_download === 'true'),
|
||||
require_password: formatBoolean(requirePassword),
|
||||
css_template_id: css_template_id || null,
|
||||
hero_logo_visible: formatBoolean(effectiveHeroLogoVisible),
|
||||
@@ -597,12 +602,21 @@ router.post('/', adminAuth, requirePermission('events.create'), [
|
||||
}
|
||||
|
||||
// Log activity
|
||||
await logActivity('event_created',
|
||||
{ event_type, expires_at, require_password: requirePassword, password_strength: passwordValidation?.score },
|
||||
eventId,
|
||||
await logActivity('event_created',
|
||||
{ event_type, expires_at, require_password: requirePassword, password_strength: passwordValidation?.score },
|
||||
eventId,
|
||||
{ type: 'admin', id: req.admin.id, name: req.admin.username }
|
||||
);
|
||||
|
||||
|
||||
// Fire event.created webhook (#327). If the event is being published
|
||||
// immediately (not a draft), event.published also fires below.
|
||||
try {
|
||||
const webhookService = require('../services/webhookService');
|
||||
await webhookService.fire('event.created', {
|
||||
event: { id: eventId, slug, event_name, event_type, event_date, is_draft: parseBooleanInput(is_draft, true) },
|
||||
});
|
||||
} catch (e) { /* webhookService.fire never throws but be defensive */ }
|
||||
|
||||
// Queue creation email (only if there is a recipient and event is not a draft)
|
||||
// Language detection is handled by email processor
|
||||
const isDraft = parseBooleanInput(is_draft, true);
|
||||
@@ -639,7 +653,19 @@ router.post('/', adminAuth, requirePermission('events.create'), [
|
||||
// scheduled_at will use default value
|
||||
});
|
||||
}
|
||||
|
||||
|
||||
// Fire event.published when the event is created NOT as a draft. The
|
||||
// separate /publish endpoint fires it for the draft → live transition;
|
||||
// this covers the "create-and-publish in one shot" path.
|
||||
if (!isDraft) {
|
||||
try {
|
||||
const webhookService = require('../services/webhookService');
|
||||
await webhookService.fire('event.published', {
|
||||
event: { id: eventId, slug, event_name, share_url: shareUrl },
|
||||
});
|
||||
} catch (e) { /* non-fatal */ }
|
||||
}
|
||||
|
||||
res.json({
|
||||
id: eventId,
|
||||
slug,
|
||||
@@ -875,6 +901,15 @@ router.post('/:id/publish', adminAuth, requirePermission('events.edit'), require
|
||||
{ type: 'admin', id: req.admin.id, name: req.admin.username }
|
||||
);
|
||||
|
||||
// Fire event.published webhook (#327) — draft → live transition.
|
||||
try {
|
||||
const webhookService = require('../services/webhookService');
|
||||
const { shareUrl } = await buildShareLinkVariants({ slug: event.slug, shareToken: event.share_token });
|
||||
await webhookService.fire('event.published', {
|
||||
event: { id: parseInt(id, 10), slug: event.slug, event_name: event.event_name, share_url: shareUrl },
|
||||
});
|
||||
} catch (e) { /* non-fatal */ }
|
||||
|
||||
res.json({ message: 'Event published successfully', is_draft: false });
|
||||
} catch (error) {
|
||||
logger.error('Error publishing event:', { error: error.message });
|
||||
@@ -912,6 +947,7 @@ router.put('/:id', adminAuth, requirePermission('events.edit'), requireEventOwne
|
||||
body('disable_right_click').optional().isBoolean(),
|
||||
body('watermark_downloads').optional().isBoolean(),
|
||||
body('watermark_text').optional().trim(),
|
||||
body('allow_presigned_download').optional().isBoolean(),
|
||||
body('source_mode').optional().isIn(['managed', 'reference']),
|
||||
body('external_path').optional({ nullable: true }).isString().trim(),
|
||||
body('require_password').optional().isBoolean(),
|
||||
|
||||
@@ -0,0 +1,389 @@
|
||||
/**
|
||||
* Admin endpoints for managing outbound webhooks (#327). Mirrors
|
||||
* adminApiTokens.js — same permission gates, same "secret shown once"
|
||||
* pattern.
|
||||
*
|
||||
* Routes mounted under /api/admin/webhooks:
|
||||
* GET / — list
|
||||
* POST / — create (returns plaintext secret once)
|
||||
* GET /:id — detail (no secret)
|
||||
* PUT /:id — update name/url/events/active
|
||||
* DELETE /:id — delete (cascades to deliveries)
|
||||
* POST /:id/test — fire a synthetic delivery now
|
||||
* GET /:id/deliveries — list deliveries (paginated, filter)
|
||||
* GET /:id/deliveries/:deliveryId — delivery detail (payload+response)
|
||||
* POST /:id/deliveries/:deliveryId/replay — re-enqueue a delivery
|
||||
*/
|
||||
|
||||
const express = require('express');
|
||||
const { body, query, validationResult } = require('express-validator');
|
||||
const { db, logActivity } = require('../database/db');
|
||||
const { adminAuth } = require('../middleware/auth');
|
||||
const { requirePermission } = require('../middleware/permissions');
|
||||
const { validateExternalUrl } = require('../utils/networkValidation');
|
||||
const webhookService = require('../services/webhookService');
|
||||
const logger = require('../utils/logger');
|
||||
|
||||
const router = express.Router();
|
||||
|
||||
const ALLOW_PRIVATE_URLS = process.env.WEBHOOK_ALLOW_PRIVATE_URLS === 'true';
|
||||
|
||||
function publicWebhook(row) {
|
||||
if (!row) return null;
|
||||
return {
|
||||
id: row.id,
|
||||
name: row.name,
|
||||
url: row.url,
|
||||
events: typeof row.events === 'string' ? safeJson(row.events, []) : (row.events || []),
|
||||
active: row.active,
|
||||
secret_preview: row.secret_preview,
|
||||
filter: typeof row.filter === 'string' ? safeJson(row.filter, {}) : (row.filter || {}),
|
||||
template: row.template || null,
|
||||
created_by: row.created_by,
|
||||
created_at: row.created_at,
|
||||
updated_at: row.updated_at,
|
||||
last_success_at: row.last_success_at,
|
||||
last_failure_at: row.last_failure_at,
|
||||
};
|
||||
}
|
||||
|
||||
function safeJson(s, fallback) {
|
||||
try { return JSON.parse(s); } catch { return fallback; }
|
||||
}
|
||||
|
||||
// ─── List ────────────────────────────────────────────────────────────────
|
||||
router.get('/', adminAuth, requirePermission('settings.view'), async (req, res) => {
|
||||
try {
|
||||
const rows = await db('webhooks')
|
||||
.leftJoin('admin_users', 'admin_users.id', 'webhooks.created_by')
|
||||
.select(
|
||||
'webhooks.*',
|
||||
'admin_users.username as owner_username'
|
||||
)
|
||||
.orderBy('webhooks.created_at', 'desc');
|
||||
res.json(rows.map((r) => ({
|
||||
...publicWebhook(r),
|
||||
owner_username: r.owner_username,
|
||||
})));
|
||||
} catch (err) {
|
||||
logger.error('webhooks list failed', { error: err.message });
|
||||
res.status(500).json({ error: 'Failed to list webhooks' });
|
||||
}
|
||||
});
|
||||
|
||||
// ─── Create ──────────────────────────────────────────────────────────────
|
||||
router.post(
|
||||
'/',
|
||||
adminAuth,
|
||||
requirePermission('settings.edit'),
|
||||
[
|
||||
body('name').isString().trim().isLength({ min: 1, max: 100 }),
|
||||
body('url').isString().isLength({ max: 2048 }).custom((url) => {
|
||||
if (ALLOW_PRIVATE_URLS) return true;
|
||||
const check = validateExternalUrl(url);
|
||||
if (!check.valid) throw new Error(check.error);
|
||||
return true;
|
||||
}),
|
||||
body('events').isArray({ min: 1 }).custom((arr) => {
|
||||
const ok = arr.every((e) => webhookService.EVENT_TYPES.includes(e));
|
||||
if (!ok) throw new Error(`events must be a subset of: ${webhookService.EVENT_TYPES.join(', ')}`);
|
||||
return true;
|
||||
}),
|
||||
body('active').optional().isBoolean(),
|
||||
body('filter').optional().custom((v) => {
|
||||
if (v == null) return true;
|
||||
if (typeof v !== 'object' || Array.isArray(v)) {
|
||||
throw new Error('filter must be an object of dot-path → value pairs');
|
||||
}
|
||||
return true;
|
||||
}),
|
||||
body('template').optional({ nullable: true }).custom((v) => {
|
||||
const check = webhookService.validateTemplate(v);
|
||||
if (!check.valid) throw new Error(check.error);
|
||||
return true;
|
||||
}),
|
||||
],
|
||||
async (req, res) => {
|
||||
try {
|
||||
const errors = validationResult(req);
|
||||
if (!errors.isEmpty()) return res.status(400).json({ errors: errors.array() });
|
||||
|
||||
const { name, url, events, active = true, filter, template } = req.body;
|
||||
const { plaintext, preview } = webhookService.generateSecret();
|
||||
|
||||
const insertResult = await db('webhooks').insert({
|
||||
name,
|
||||
url,
|
||||
secret: plaintext,
|
||||
secret_preview: preview,
|
||||
events: JSON.stringify(events),
|
||||
active,
|
||||
filter: JSON.stringify(filter || {}),
|
||||
template: template || null,
|
||||
created_by: req.admin.id,
|
||||
}).returning('id');
|
||||
const id = insertResult[0]?.id || insertResult[0];
|
||||
|
||||
await logActivity('webhook_created', { name, events }, null, {
|
||||
type: 'admin', id: req.admin.id, name: req.admin.username,
|
||||
});
|
||||
|
||||
const row = await db('webhooks').where({ id }).first();
|
||||
res.status(201).json({
|
||||
...publicWebhook(row),
|
||||
secret: plaintext,
|
||||
notice: 'Save this signing secret now — it will not be shown again.',
|
||||
});
|
||||
} catch (err) {
|
||||
logger.error('webhooks create failed', { error: err.message });
|
||||
res.status(500).json({ error: 'Failed to create webhook' });
|
||||
}
|
||||
}
|
||||
);
|
||||
|
||||
// ─── Detail ──────────────────────────────────────────────────────────────
|
||||
router.get('/:id', adminAuth, requirePermission('settings.view'), async (req, res) => {
|
||||
try {
|
||||
const row = await db('webhooks').where({ id: req.params.id }).first();
|
||||
if (!row) return res.status(404).json({ error: 'Webhook not found' });
|
||||
res.json(publicWebhook(row));
|
||||
} catch (err) {
|
||||
logger.error('webhooks detail failed', { error: err.message });
|
||||
res.status(500).json({ error: 'Failed to load webhook' });
|
||||
}
|
||||
});
|
||||
|
||||
// ─── Update ──────────────────────────────────────────────────────────────
|
||||
router.put(
|
||||
'/:id',
|
||||
adminAuth,
|
||||
requirePermission('settings.edit'),
|
||||
[
|
||||
body('name').optional().isString().trim().isLength({ min: 1, max: 100 }),
|
||||
body('url').optional().isString().isLength({ max: 2048 }).custom((url) => {
|
||||
if (ALLOW_PRIVATE_URLS) return true;
|
||||
const check = validateExternalUrl(url);
|
||||
if (!check.valid) throw new Error(check.error);
|
||||
return true;
|
||||
}),
|
||||
body('events').optional().isArray({ min: 1 }).custom((arr) => {
|
||||
const ok = arr.every((e) => webhookService.EVENT_TYPES.includes(e));
|
||||
if (!ok) throw new Error(`events must be a subset of: ${webhookService.EVENT_TYPES.join(', ')}`);
|
||||
return true;
|
||||
}),
|
||||
body('active').optional().isBoolean(),
|
||||
body('filter').optional().custom((v) => {
|
||||
if (v == null) return true;
|
||||
if (typeof v !== 'object' || Array.isArray(v)) {
|
||||
throw new Error('filter must be an object of dot-path → value pairs');
|
||||
}
|
||||
return true;
|
||||
}),
|
||||
body('template').optional({ nullable: true }).custom((v) => {
|
||||
const check = webhookService.validateTemplate(v);
|
||||
if (!check.valid) throw new Error(check.error);
|
||||
return true;
|
||||
}),
|
||||
],
|
||||
async (req, res) => {
|
||||
try {
|
||||
const errors = validationResult(req);
|
||||
if (!errors.isEmpty()) return res.status(400).json({ errors: errors.array() });
|
||||
|
||||
const row = await db('webhooks').where({ id: req.params.id }).first();
|
||||
if (!row) return res.status(404).json({ error: 'Webhook not found' });
|
||||
|
||||
const updates = { updated_at: new Date() };
|
||||
if ('name' in req.body) updates.name = req.body.name;
|
||||
if ('url' in req.body) updates.url = req.body.url;
|
||||
if ('events' in req.body) updates.events = JSON.stringify(req.body.events);
|
||||
if ('active' in req.body) updates.active = req.body.active;
|
||||
if ('filter' in req.body) updates.filter = JSON.stringify(req.body.filter || {});
|
||||
if ('template' in req.body) updates.template = req.body.template || null;
|
||||
|
||||
await db('webhooks').where({ id: req.params.id }).update(updates);
|
||||
const updated = await db('webhooks').where({ id: req.params.id }).first();
|
||||
|
||||
await logActivity('webhook_updated', { changes: Object.keys(updates) }, null, {
|
||||
type: 'admin', id: req.admin.id, name: req.admin.username,
|
||||
});
|
||||
|
||||
res.json(publicWebhook(updated));
|
||||
} catch (err) {
|
||||
logger.error('webhooks update failed', { error: err.message });
|
||||
res.status(500).json({ error: 'Failed to update webhook' });
|
||||
}
|
||||
}
|
||||
);
|
||||
|
||||
// ─── Delete ──────────────────────────────────────────────────────────────
|
||||
router.delete('/:id', adminAuth, requirePermission('settings.edit'), async (req, res) => {
|
||||
try {
|
||||
const row = await db('webhooks').where({ id: req.params.id }).first();
|
||||
if (!row) return res.status(404).json({ error: 'Webhook not found' });
|
||||
await db('webhooks').where({ id: req.params.id }).delete();
|
||||
await logActivity('webhook_deleted', { name: row.name }, null, {
|
||||
type: 'admin', id: req.admin.id, name: req.admin.username,
|
||||
});
|
||||
res.json({ id: Number(req.params.id), deleted: true });
|
||||
} catch (err) {
|
||||
logger.error('webhooks delete failed', { error: err.message });
|
||||
res.status(500).json({ error: 'Failed to delete webhook' });
|
||||
}
|
||||
});
|
||||
|
||||
// ─── Send test event ─────────────────────────────────────────────────────
|
||||
router.post(
|
||||
'/:id/test',
|
||||
adminAuth,
|
||||
requirePermission('settings.edit'),
|
||||
[body('event_type').optional().isIn(webhookService.EVENT_TYPES)],
|
||||
async (req, res) => {
|
||||
try {
|
||||
const errors = validationResult(req);
|
||||
if (!errors.isEmpty()) return res.status(400).json({ errors: errors.array() });
|
||||
|
||||
const row = await db('webhooks').where({ id: req.params.id }).first();
|
||||
if (!row) return res.status(404).json({ error: 'Webhook not found' });
|
||||
if (!row.active) return res.status(400).json({ error: 'Webhook is disabled' });
|
||||
|
||||
const eventType = req.body.event_type || (() => {
|
||||
const subscribed = typeof row.events === 'string' ? safeJson(row.events, []) : (row.events || []);
|
||||
return subscribed[0] || 'event.published';
|
||||
})();
|
||||
|
||||
// Fire a synthetic event WITHOUT writing to webhooks table — the test
|
||||
// bypasses subscription matching by inserting a delivery directly.
|
||||
const crypto = require('crypto');
|
||||
const deliveryId = crypto.randomUUID();
|
||||
const payload = {
|
||||
id: deliveryId,
|
||||
type: eventType,
|
||||
created_at: new Date().toISOString(),
|
||||
data: { test: true, fired_by: req.admin.username, webhook_id: row.id },
|
||||
};
|
||||
await db('webhook_deliveries').insert({
|
||||
webhook_id: row.id,
|
||||
event_type: eventType,
|
||||
payload: JSON.stringify(payload),
|
||||
attempt_count: 0,
|
||||
status: 'pending',
|
||||
next_retry_at: new Date(),
|
||||
created_at: new Date(),
|
||||
});
|
||||
|
||||
res.status(202).json({ enqueued: true, event_type: eventType });
|
||||
} catch (err) {
|
||||
logger.error('webhook test failed', { error: err.message });
|
||||
res.status(500).json({ error: 'Failed to enqueue test event' });
|
||||
}
|
||||
}
|
||||
);
|
||||
|
||||
// ─── List deliveries ─────────────────────────────────────────────────────
|
||||
router.get(
|
||||
'/:id/deliveries',
|
||||
adminAuth,
|
||||
requirePermission('settings.view'),
|
||||
[
|
||||
query('status').optional().isIn(['pending', 'success', 'failed']),
|
||||
query('page').optional().isInt({ min: 1 }),
|
||||
query('limit').optional().isInt({ min: 1, max: 100 }),
|
||||
],
|
||||
async (req, res) => {
|
||||
try {
|
||||
const errors = validationResult(req);
|
||||
if (!errors.isEmpty()) return res.status(400).json({ errors: errors.array() });
|
||||
|
||||
const webhookId = req.params.id;
|
||||
const exists = await db('webhooks').where({ id: webhookId }).first();
|
||||
if (!exists) return res.status(404).json({ error: 'Webhook not found' });
|
||||
|
||||
const page = parseInt(req.query.page || '1', 10);
|
||||
const limit = parseInt(req.query.limit || '25', 10);
|
||||
const offset = (page - 1) * limit;
|
||||
|
||||
let q = db('webhook_deliveries').where({ webhook_id: webhookId });
|
||||
if (req.query.status) q = q.where({ status: req.query.status });
|
||||
|
||||
const totalRow = await q.clone().count('id as count').first();
|
||||
const total = parseInt(totalRow?.count || 0, 10);
|
||||
|
||||
const rows = await q
|
||||
.select(
|
||||
'id', 'event_type', 'attempt_count', 'status', 'response_status',
|
||||
'latency_ms', 'next_retry_at', 'created_at', 'completed_at', 'last_error'
|
||||
)
|
||||
.orderBy('created_at', 'desc')
|
||||
.limit(limit)
|
||||
.offset(offset);
|
||||
|
||||
res.json({ deliveries: rows, pagination: { page, limit, total } });
|
||||
} catch (err) {
|
||||
logger.error('deliveries list failed', { error: err.message });
|
||||
res.status(500).json({ error: 'Failed to list deliveries' });
|
||||
}
|
||||
}
|
||||
);
|
||||
|
||||
// ─── Delivery detail ─────────────────────────────────────────────────────
|
||||
router.get(
|
||||
'/:id/deliveries/:deliveryId',
|
||||
adminAuth,
|
||||
requirePermission('settings.view'),
|
||||
async (req, res) => {
|
||||
try {
|
||||
const row = await db('webhook_deliveries')
|
||||
.where({ id: req.params.deliveryId, webhook_id: req.params.id })
|
||||
.first();
|
||||
if (!row) return res.status(404).json({ error: 'Delivery not found' });
|
||||
res.json({
|
||||
...row,
|
||||
payload: typeof row.payload === 'string' ? safeJson(row.payload, row.payload) : row.payload,
|
||||
});
|
||||
} catch (err) {
|
||||
logger.error('delivery detail failed', { error: err.message });
|
||||
res.status(500).json({ error: 'Failed to load delivery' });
|
||||
}
|
||||
}
|
||||
);
|
||||
|
||||
// ─── Replay ──────────────────────────────────────────────────────────────
|
||||
router.post(
|
||||
'/:id/deliveries/:deliveryId/replay',
|
||||
adminAuth,
|
||||
requirePermission('settings.edit'),
|
||||
async (req, res) => {
|
||||
try {
|
||||
const row = await db('webhook_deliveries')
|
||||
.where({ id: req.params.deliveryId, webhook_id: req.params.id })
|
||||
.first();
|
||||
if (!row) return res.status(404).json({ error: 'Delivery not found' });
|
||||
|
||||
// Re-enqueue: copy the original payload + event_type into a new row
|
||||
// marked pending. Preserves the audit log of the original attempt.
|
||||
const crypto = require('crypto');
|
||||
const newPayload = (() => {
|
||||
const obj = typeof row.payload === 'string' ? safeJson(row.payload, {}) : row.payload || {};
|
||||
// Replays get a fresh delivery id but keep the event payload data.
|
||||
return JSON.stringify({ ...obj, id: crypto.randomUUID(), replayed_from: row.id });
|
||||
})();
|
||||
const insertResult = await db('webhook_deliveries').insert({
|
||||
webhook_id: row.webhook_id,
|
||||
event_type: row.event_type,
|
||||
payload: newPayload,
|
||||
attempt_count: 0,
|
||||
status: 'pending',
|
||||
next_retry_at: new Date(),
|
||||
created_at: new Date(),
|
||||
}).returning('id');
|
||||
const newId = insertResult[0]?.id || insertResult[0];
|
||||
res.status(202).json({ enqueued: true, original_id: row.id, replay_id: newId });
|
||||
} catch (err) {
|
||||
logger.error('delivery replay failed', { error: err.message });
|
||||
res.status(500).json({ error: 'Failed to replay delivery' });
|
||||
}
|
||||
}
|
||||
);
|
||||
|
||||
module.exports = router;
|
||||
@@ -190,6 +190,18 @@ router.post('/', adminAuth, [
|
||||
welcome_message: welcome_message || ''
|
||||
});
|
||||
|
||||
// Webhook lifecycle (#327). Legacy public endpoint — events go live
|
||||
// immediately so created + published fire together.
|
||||
try {
|
||||
const webhookService = require('../services/webhookService');
|
||||
await webhookService.fire('event.created', {
|
||||
event: { id: eventId, slug, event_name, event_type, event_date, share_url: shareUrl },
|
||||
});
|
||||
await webhookService.fire('event.published', {
|
||||
event: { id: eventId, slug, event_name, share_url: shareUrl },
|
||||
});
|
||||
} catch (e) { /* non-fatal */ }
|
||||
|
||||
res.json({
|
||||
id: eventId,
|
||||
slug,
|
||||
|
||||
@@ -84,7 +84,16 @@ async function handleExpiredEvent(event) {
|
||||
try {
|
||||
// Mark as inactive
|
||||
await db('events').where('id', event.id).update({ is_active: formatBoolean(false) });
|
||||
|
||||
|
||||
// Fire event.expired BEFORE the cascading archive call so receivers
|
||||
// get the lifecycle in order (expired → archived).
|
||||
try {
|
||||
const webhookService = require('./webhookService');
|
||||
await webhookService.fire('event.expired', {
|
||||
event: { id: event.id, slug: event.slug, event_name: event.event_name, expires_at: event.expires_at },
|
||||
});
|
||||
} catch (e) { /* non-fatal */ }
|
||||
|
||||
// Queue expiration emails
|
||||
const recipientEmail = event.customer_email || event.host_email;
|
||||
const recipientName = event.customer_name || event.host_name || (recipientEmail ? recipientEmail.split('@')[0] : null);
|
||||
|
||||
@@ -0,0 +1,277 @@
|
||||
const axios = require('axios');
|
||||
const { db } = require('../database/db');
|
||||
const logger = require('../utils/logger');
|
||||
const { signPayload, renderTemplate } = require('./webhookService');
|
||||
const { validateExternalUrl } = require('../utils/networkValidation');
|
||||
|
||||
const POLL_INTERVAL_MS = parseInt(process.env.WEBHOOK_DELIVERY_INTERVAL_MS || '5000', 10);
|
||||
const CONCURRENCY = parseInt(process.env.WEBHOOK_DELIVERY_CONCURRENCY || '5', 10);
|
||||
const HTTP_TIMEOUT_MS = parseInt(process.env.WEBHOOK_HTTP_TIMEOUT_MS || '10000', 10);
|
||||
const MAX_ATTEMPTS = parseInt(process.env.WEBHOOK_MAX_ATTEMPTS || '5', 10);
|
||||
const RESPONSE_TRUNCATE_BYTES = 1024;
|
||||
// Mutable so tests can flip it without juggling require.cache; reads the
|
||||
// env var at module load for the production code path.
|
||||
let allowPrivateUrls = process.env.WEBHOOK_ALLOW_PRIVATE_URLS === 'true';
|
||||
const SIGNATURE_HEADER = 'X-PicPeak-Signature';
|
||||
const EVENT_HEADER = 'X-PicPeak-Event';
|
||||
const DELIVERY_HEADER = 'X-PicPeak-Delivery';
|
||||
|
||||
// Backoff schedule per the issue spec — index = attempt that just failed.
|
||||
// attempt_count after the failure becomes (failedAttempt + 1); we look up
|
||||
// the delay using the *new* attempt count to schedule the next try.
|
||||
// attempt 1 fails → wait 1m
|
||||
// attempt 2 fails → wait 5m
|
||||
// attempt 3 fails → wait 30m
|
||||
// attempt 4 fails → wait 2h
|
||||
// attempt 5 fails → wait 12h THEN give up (max 5 attempts total)
|
||||
const BACKOFF_MS = [
|
||||
60_000, // 1 min
|
||||
5 * 60_000, // 5 min
|
||||
30 * 60_000, // 30 min
|
||||
2 * 60 * 60_000, // 2 h
|
||||
12 * 60 * 60_000, // 12 h (only used when MAX_ATTEMPTS extended past 5)
|
||||
];
|
||||
|
||||
let intervalHandle = null;
|
||||
let stopped = false;
|
||||
// Tracks deliveries currently being processed in this tick — guards
|
||||
// against the same row being claimed twice if a tick takes longer than
|
||||
// POLL_INTERVAL_MS.
|
||||
const inFlight = new Set();
|
||||
|
||||
function truncate(str, bytes) {
|
||||
if (str == null) return null;
|
||||
const buf = Buffer.from(String(str), 'utf8');
|
||||
if (buf.length <= bytes) return buf.toString('utf8');
|
||||
return buf.subarray(0, bytes).toString('utf8');
|
||||
}
|
||||
|
||||
async function fetchPending(limit) {
|
||||
// Skip rows already in-flight from a previous tick that's still running.
|
||||
const excludeIds = Array.from(inFlight);
|
||||
let q = db('webhook_deliveries')
|
||||
.where('status', 'pending')
|
||||
.where('next_retry_at', '<=', new Date())
|
||||
.orderBy('next_retry_at', 'asc')
|
||||
.limit(limit);
|
||||
if (excludeIds.length > 0) {
|
||||
q = q.whereNotIn('id', excludeIds);
|
||||
}
|
||||
return q.select('*');
|
||||
}
|
||||
|
||||
async function deliverOne(row) {
|
||||
const startedAt = Date.now();
|
||||
const webhook = await db('webhooks').where({ id: row.webhook_id }).first();
|
||||
|
||||
if (!webhook) {
|
||||
// Webhook was deleted while a delivery was pending. Mark failed and move on.
|
||||
await db('webhook_deliveries')
|
||||
.where({ id: row.id })
|
||||
.update({
|
||||
status: 'failed',
|
||||
last_error: 'webhook subscription no longer exists',
|
||||
completed_at: new Date(),
|
||||
attempt_count: row.attempt_count + 1,
|
||||
});
|
||||
return;
|
||||
}
|
||||
|
||||
if (!webhook.active) {
|
||||
// Subscription disabled mid-flight. Don't abandon — leave as failed
|
||||
// so the deliveries page reflects the reality.
|
||||
await db('webhook_deliveries')
|
||||
.where({ id: row.id })
|
||||
.update({
|
||||
status: 'failed',
|
||||
last_error: 'webhook is disabled',
|
||||
completed_at: new Date(),
|
||||
attempt_count: row.attempt_count + 1,
|
||||
});
|
||||
return;
|
||||
}
|
||||
|
||||
// Re-validate URL per delivery — DNS-rebinding mitigation. Admin can opt
|
||||
// out via WEBHOOK_ALLOW_PRIVATE_URLS=true for local-receiver dev runs.
|
||||
if (!allowPrivateUrls) {
|
||||
const urlCheck = validateExternalUrl(webhook.url);
|
||||
if (!urlCheck.valid) {
|
||||
await markFailedFinal(row, `URL rejected: ${urlCheck.error}`);
|
||||
return;
|
||||
}
|
||||
}
|
||||
|
||||
const envelopeBody = typeof row.payload === 'string' ? row.payload : JSON.stringify(row.payload);
|
||||
const envelopeObj = (() => {
|
||||
try { return JSON.parse(envelopeBody); } catch { return {}; }
|
||||
})();
|
||||
|
||||
// Per-webhook template (#327 follow-up). If set + valid, replaces the
|
||||
// default JSON envelope as the request body. Signature is computed over
|
||||
// the BODY ACTUALLY SENT, so receivers verify whatever they receive.
|
||||
let rawBody = envelopeBody;
|
||||
let contentType = 'application/json';
|
||||
if (webhook.template) {
|
||||
const rendered = renderTemplate(webhook.template, envelopeObj);
|
||||
if (rendered != null) {
|
||||
rawBody = rendered;
|
||||
// Best-effort content-type detection: if it parses as JSON, keep
|
||||
// application/json; otherwise send as text/plain.
|
||||
try { JSON.parse(rendered); } catch { contentType = 'text/plain; charset=utf-8'; }
|
||||
}
|
||||
}
|
||||
const signature = signPayload(webhook.secret, rawBody);
|
||||
const deliveryId = envelopeObj?.id || String(row.id);
|
||||
|
||||
let response;
|
||||
let networkError;
|
||||
try {
|
||||
response = await axios.post(webhook.url, rawBody, {
|
||||
headers: {
|
||||
'Content-Type': contentType,
|
||||
[SIGNATURE_HEADER]: signature,
|
||||
[EVENT_HEADER]: row.event_type,
|
||||
[DELIVERY_HEADER]: deliveryId,
|
||||
'User-Agent': 'PicPeak-Webhooks/1.0',
|
||||
},
|
||||
timeout: HTTP_TIMEOUT_MS,
|
||||
// Don't throw on non-2xx; we handle status manually.
|
||||
validateStatus: () => true,
|
||||
// Don't follow redirects — security + receivers should give us the
|
||||
// final URL up front.
|
||||
maxRedirects: 0,
|
||||
// Cap response body so a chatty receiver can't OOM us before truncation.
|
||||
maxContentLength: 10 * 1024,
|
||||
maxBodyLength: rawBody.length + 1024,
|
||||
});
|
||||
} catch (err) {
|
||||
networkError = err;
|
||||
}
|
||||
|
||||
const latency = Date.now() - startedAt;
|
||||
const newAttempt = row.attempt_count + 1;
|
||||
|
||||
if (response && response.status >= 200 && response.status < 300) {
|
||||
await db('webhook_deliveries')
|
||||
.where({ id: row.id })
|
||||
.update({
|
||||
status: 'success',
|
||||
response_status: response.status,
|
||||
response_body: truncate(stringifyBody(response.data), RESPONSE_TRUNCATE_BYTES),
|
||||
latency_ms: latency,
|
||||
attempt_count: newAttempt,
|
||||
completed_at: new Date(),
|
||||
next_retry_at: null,
|
||||
});
|
||||
await db('webhooks').where({ id: webhook.id }).update({ last_success_at: new Date() });
|
||||
return;
|
||||
}
|
||||
|
||||
// Failure path — schedule retry or give up.
|
||||
const errorMsg = networkError
|
||||
? `network error: ${networkError.code || networkError.message}`
|
||||
: `non-2xx status: ${response?.status}`;
|
||||
|
||||
if (newAttempt >= MAX_ATTEMPTS) {
|
||||
await db('webhook_deliveries')
|
||||
.where({ id: row.id })
|
||||
.update({
|
||||
status: 'failed',
|
||||
response_status: response?.status || null,
|
||||
response_body: response ? truncate(stringifyBody(response.data), RESPONSE_TRUNCATE_BYTES) : null,
|
||||
last_error: errorMsg,
|
||||
latency_ms: latency,
|
||||
attempt_count: newAttempt,
|
||||
completed_at: new Date(),
|
||||
next_retry_at: null,
|
||||
});
|
||||
await db('webhooks').where({ id: webhook.id }).update({ last_failure_at: new Date() });
|
||||
return;
|
||||
}
|
||||
|
||||
const backoff = BACKOFF_MS[Math.min(newAttempt - 1, BACKOFF_MS.length - 1)];
|
||||
await db('webhook_deliveries')
|
||||
.where({ id: row.id })
|
||||
.update({
|
||||
status: 'pending',
|
||||
response_status: response?.status || null,
|
||||
response_body: response ? truncate(stringifyBody(response.data), RESPONSE_TRUNCATE_BYTES) : null,
|
||||
last_error: errorMsg,
|
||||
latency_ms: latency,
|
||||
attempt_count: newAttempt,
|
||||
next_retry_at: new Date(Date.now() + backoff),
|
||||
});
|
||||
await db('webhooks').where({ id: webhook.id }).update({ last_failure_at: new Date() });
|
||||
}
|
||||
|
||||
async function markFailedFinal(row, reason) {
|
||||
await db('webhook_deliveries')
|
||||
.where({ id: row.id })
|
||||
.update({
|
||||
status: 'failed',
|
||||
last_error: reason,
|
||||
attempt_count: row.attempt_count + 1,
|
||||
completed_at: new Date(),
|
||||
next_retry_at: null,
|
||||
});
|
||||
await db('webhooks').where({ id: row.webhook_id }).update({ last_failure_at: new Date() });
|
||||
}
|
||||
|
||||
function stringifyBody(data) {
|
||||
if (data == null) return null;
|
||||
if (typeof data === 'string') return data;
|
||||
if (Buffer.isBuffer(data)) return data.toString('utf8');
|
||||
try { return JSON.stringify(data); } catch { return String(data); }
|
||||
}
|
||||
|
||||
async function tick() {
|
||||
if (stopped) return;
|
||||
try {
|
||||
const slots = Math.max(0, CONCURRENCY - inFlight.size);
|
||||
if (slots === 0) return;
|
||||
const rows = await fetchPending(slots);
|
||||
if (rows.length === 0) return;
|
||||
rows.forEach((r) => inFlight.add(r.id));
|
||||
await Promise.allSettled(
|
||||
rows.map((r) =>
|
||||
deliverOne(r)
|
||||
.catch((err) => logger.error(`[webhookWorker] delivery ${r.id} crashed: ${err.message}`))
|
||||
.finally(() => inFlight.delete(r.id))
|
||||
)
|
||||
);
|
||||
} catch (err) {
|
||||
logger.error(`[webhookWorker] tick failed: ${err.message}`);
|
||||
}
|
||||
}
|
||||
|
||||
function startWebhookDeliveryWorker() {
|
||||
if (intervalHandle) return; // idempotent
|
||||
stopped = false;
|
||||
intervalHandle = setInterval(tick, POLL_INTERVAL_MS);
|
||||
logger.info(
|
||||
`[webhookWorker] started — interval=${POLL_INTERVAL_MS}ms, concurrency=${CONCURRENCY}, ` +
|
||||
`max_attempts=${MAX_ATTEMPTS}, allow_private=${allowPrivateUrls}`
|
||||
);
|
||||
}
|
||||
|
||||
function stopWebhookDeliveryWorker() {
|
||||
stopped = true;
|
||||
if (intervalHandle) {
|
||||
clearInterval(intervalHandle);
|
||||
intervalHandle = null;
|
||||
}
|
||||
}
|
||||
|
||||
module.exports = {
|
||||
startWebhookDeliveryWorker,
|
||||
stopWebhookDeliveryWorker,
|
||||
// exported for tests
|
||||
__test: {
|
||||
tick,
|
||||
BACKOFF_MS,
|
||||
SIGNATURE_HEADER,
|
||||
EVENT_HEADER,
|
||||
DELIVERY_HEADER,
|
||||
setAllowPrivateUrls(value) { allowPrivateUrls = !!value; },
|
||||
},
|
||||
};
|
||||
@@ -0,0 +1,224 @@
|
||||
const crypto = require('crypto');
|
||||
const { db } = require('../database/db');
|
||||
const logger = require('../utils/logger');
|
||||
|
||||
const SECRET_PREFIX = 'whsec_';
|
||||
|
||||
/**
|
||||
* Event types PicPeak emits. Keep this in sync with the README catalog and
|
||||
* the receiver-side type unions in any SDK we publish later. Consumers of
|
||||
* `fire(eventType, ...)` MUST use one of these strings — the worker will
|
||||
* silently drop unknown types so a typo can't 500 a request handler.
|
||||
*/
|
||||
const EVENT_TYPES = Object.freeze([
|
||||
'event.created',
|
||||
'event.published',
|
||||
'event.archived',
|
||||
'event.expired',
|
||||
'photo.uploaded',
|
||||
'photo.deleted',
|
||||
]);
|
||||
|
||||
function generateSecret() {
|
||||
const random = crypto.randomBytes(24).toString('base64url'); // ~32 chars
|
||||
const plaintext = `${SECRET_PREFIX}${random}`;
|
||||
return {
|
||||
plaintext,
|
||||
preview: random.slice(0, 8),
|
||||
};
|
||||
}
|
||||
|
||||
/**
|
||||
* Sign a payload with the webhook's secret. Used by the delivery worker;
|
||||
* exported for unit tests of receiver-side verification snippets.
|
||||
*/
|
||||
function signPayload(secret, rawBody) {
|
||||
return crypto.createHmac('sha256', secret).update(rawBody).digest('hex');
|
||||
}
|
||||
|
||||
/**
|
||||
* Resolve a dot-path on an object (e.g. "data.event.event_type") with no
|
||||
* eval. Returns undefined for missing segments — never throws.
|
||||
*/
|
||||
function getByPath(obj, dotPath) {
|
||||
if (!dotPath || typeof dotPath !== 'string') return undefined;
|
||||
return dotPath.split('.').reduce((acc, key) => {
|
||||
if (acc == null || typeof acc !== 'object') return undefined;
|
||||
return acc[key];
|
||||
}, obj);
|
||||
}
|
||||
|
||||
/**
|
||||
* Evaluate a webhook's filter against an outgoing payload. The filter is a
|
||||
* flat object of dot-path → expected value pairs:
|
||||
* { "data.event.event_type": "wedding" }
|
||||
* { "type": "event.published", "data.event.id": 42 }
|
||||
*
|
||||
* All keys must match (logical AND). Equality is `===` after JSON-style
|
||||
* coercion: numbers as numbers, booleans as booleans. Empty filter
|
||||
* matches everything (back-compat).
|
||||
*/
|
||||
function payloadMatchesFilter(filter, payload) {
|
||||
if (!filter || typeof filter !== 'object') return true;
|
||||
const keys = Object.keys(filter);
|
||||
if (keys.length === 0) return true;
|
||||
for (const key of keys) {
|
||||
const expected = filter[key];
|
||||
const actual = getByPath(payload, key);
|
||||
if (Array.isArray(expected)) {
|
||||
// Array means "any of"
|
||||
if (!expected.includes(actual)) return false;
|
||||
} else if (actual !== expected) {
|
||||
return false;
|
||||
}
|
||||
}
|
||||
return true;
|
||||
}
|
||||
|
||||
/**
|
||||
* Render a webhook template by substituting ${dot.path} expressions with
|
||||
* values from the payload. NO eval, NO logic — pure string substitution.
|
||||
* Caps output at 64KB; bails out and returns null if exceeded so the
|
||||
* delivery worker can fall back to the default envelope.
|
||||
*/
|
||||
function renderTemplate(template, payload) {
|
||||
if (template == null || template === '') return null;
|
||||
if (typeof template !== 'string') return null;
|
||||
const MAX_OUTPUT = 64 * 1024;
|
||||
const MAX_SUBSTITUTIONS = 64;
|
||||
let count = 0;
|
||||
const rendered = template.replace(/\$\{([^}]+)\}/g, (_match, expr) => {
|
||||
count += 1;
|
||||
if (count > MAX_SUBSTITUTIONS) return '';
|
||||
const value = getByPath(payload, expr.trim());
|
||||
if (value == null) return '';
|
||||
if (typeof value === 'object') {
|
||||
try { return JSON.stringify(value); } catch { return ''; }
|
||||
}
|
||||
return String(value);
|
||||
});
|
||||
if (Buffer.byteLength(rendered, 'utf8') > MAX_OUTPUT) return null;
|
||||
return rendered;
|
||||
}
|
||||
|
||||
/**
|
||||
* Validate a template at create-time so admins get immediate feedback
|
||||
* instead of silent delivery failures. Returns { valid, error? }.
|
||||
*/
|
||||
function validateTemplate(template) {
|
||||
if (template == null || template === '') return { valid: true };
|
||||
if (typeof template !== 'string') return { valid: false, error: 'template must be a string' };
|
||||
if (Buffer.byteLength(template, 'utf8') > 8192) return { valid: false, error: 'template exceeds 8KB' };
|
||||
// Reject unbalanced ${ that would silently swallow content at render.
|
||||
const opens = (template.match(/\$\{/g) || []).length;
|
||||
const closes = (template.match(/\}/g) || []).length;
|
||||
// Count is approximate (every } is counted, even non-matching ones).
|
||||
// We require at least as many } as ${, which is necessary but not sufficient.
|
||||
if (opens > closes) return { valid: false, error: 'template has unbalanced ${ — every ${ needs a matching }' };
|
||||
if (opens > 64) return { valid: false, error: 'template exceeds 64 substitutions' };
|
||||
return { valid: true };
|
||||
}
|
||||
|
||||
/**
|
||||
* Constant-time signature comparison helper for receivers and tests.
|
||||
* Exposed so the same primitive backs verification examples in the README.
|
||||
*/
|
||||
function verifySignature(secret, rawBody, signature) {
|
||||
const expected = signPayload(secret, rawBody);
|
||||
const a = Buffer.from(expected, 'hex');
|
||||
let b;
|
||||
try {
|
||||
b = Buffer.from(signature || '', 'hex');
|
||||
} catch {
|
||||
return false;
|
||||
}
|
||||
if (a.length !== b.length) return false;
|
||||
return crypto.timingSafeEqual(a, b);
|
||||
}
|
||||
|
||||
/**
|
||||
* Enqueue a webhook delivery for every active webhook subscribed to
|
||||
* `eventType`. NEVER throws — webhook failures must not break the
|
||||
* lifecycle handler that emitted the event. Worker handles HTTP delivery.
|
||||
*
|
||||
* @param {string} eventType — one of EVENT_TYPES
|
||||
* @param {object} data — opaque payload that gets nested under .data in
|
||||
* the outbound JSON body
|
||||
*/
|
||||
async function fire(eventType, data) {
|
||||
if (!EVENT_TYPES.includes(eventType)) {
|
||||
logger.warn(`[webhookService] dropping unknown event type: ${eventType}`);
|
||||
return;
|
||||
}
|
||||
|
||||
try {
|
||||
// jsonb @> ARRAY check across vendors: both pg and sqlite drivers we
|
||||
// support handle a simple WHERE on `active=true` then a runtime filter
|
||||
// on the events array; doing the array filter here keeps the query
|
||||
// portable.
|
||||
const candidates = await db('webhooks').where({ active: true });
|
||||
const subscribed = candidates.filter((w) => {
|
||||
const evts = Array.isArray(w.events)
|
||||
? w.events
|
||||
: (() => { try { return JSON.parse(w.events) || []; } catch { return []; } })();
|
||||
return evts.includes(eventType);
|
||||
});
|
||||
|
||||
if (subscribed.length === 0) return;
|
||||
|
||||
const now = new Date();
|
||||
const buildEnvelope = (deliveryUuid) => ({
|
||||
id: deliveryUuid,
|
||||
type: eventType,
|
||||
created_at: now.toISOString(),
|
||||
data,
|
||||
});
|
||||
|
||||
const rows = [];
|
||||
for (const w of subscribed) {
|
||||
const deliveryUuid = crypto.randomUUID();
|
||||
const envelope = buildEnvelope(deliveryUuid);
|
||||
|
||||
// Filter (#327 follow-up): per-webhook predicate evaluated against
|
||||
// the payload. Skip insertion when it doesn't match.
|
||||
const filter = parseJsonField(w.filter, {});
|
||||
if (!payloadMatchesFilter(filter, envelope)) continue;
|
||||
|
||||
rows.push({
|
||||
webhook_id: w.id,
|
||||
event_type: eventType,
|
||||
payload: JSON.stringify(envelope),
|
||||
attempt_count: 0,
|
||||
status: 'pending',
|
||||
next_retry_at: now,
|
||||
created_at: now,
|
||||
});
|
||||
}
|
||||
|
||||
if (rows.length === 0) return;
|
||||
await db('webhook_deliveries').insert(rows);
|
||||
} catch (err) {
|
||||
// Log but don't throw — caller's transaction has already committed
|
||||
// by the time we get here, and we don't want to mask the success.
|
||||
logger.error(`[webhookService.fire] failed to enqueue ${eventType}: ${err.message}`);
|
||||
}
|
||||
}
|
||||
|
||||
function parseJsonField(value, fallback) {
|
||||
if (value == null) return fallback;
|
||||
if (typeof value === 'object') return value;
|
||||
try { return JSON.parse(value) ?? fallback; } catch { return fallback; }
|
||||
}
|
||||
|
||||
module.exports = {
|
||||
fire,
|
||||
generateSecret,
|
||||
signPayload,
|
||||
verifySignature,
|
||||
payloadMatchesFilter,
|
||||
renderTemplate,
|
||||
validateTemplate,
|
||||
getByPath,
|
||||
EVENT_TYPES,
|
||||
SECRET_PREFIX,
|
||||
};
|
||||
Reference in New Issue
Block a user