From eadf09ca7b0eaf96d66b763eaaae90bdc269bb1f Mon Sep 17 00:00:00 2001 From: udeachudivine-spec Date: Wed, 24 Jun 2026 02:54:47 -0700 Subject: [PATCH] feat: end-to-end webhook monitoring with Horizon SSE (#94/#61) - Split monolithic webhookService.js into three focused modules: webhookStore.js (in-memory Map store), webhookDelivery.js (signed POST delivery), paymentMonitor.js (Horizon SSE streaming) - Fix Promise.allSettled for parallel webhook delivery so one slow/ failing endpoint never blocks others - Add idempotent ensureMonitored() guard to prevent duplicate SSE streams on re-registration for the same public key - Add resumeAllMonitors() called at server startup so existing webhooks survive restarts - Fix HMAC signature header: X-Stellar-Signature: sha256= and add X-Webhook-ID header; reuse utils/webhookSignature.js - Fix payload: event name payment_received, add transactionHash and ledger fields - Fix POST /api/webhooks: validate publicKey via StrKey, enforce https:// URL, strip secret from response - Fix GET /api/webhooks/:publicKey: return { webhooks: [...] } - Fix DELETE /api/webhooks/:id: return { success: true } / 404 - webhookService.js kept as compat facade re-exporting all symbols --- backend/src/routes/webhooks.js | 77 ++++++++++--- backend/src/server.js | 4 + backend/src/services/paymentMonitor.js | 134 ++++++++++++++++++++++ backend/src/services/webhookDelivery.js | 72 ++++++++++++ backend/src/services/webhookService.js | 143 ++++++++---------------- backend/src/services/webhookStore.js | 94 ++++++++++++++++ 6 files changed, 414 insertions(+), 110 deletions(-) create mode 100644 backend/src/services/paymentMonitor.js create mode 100644 backend/src/services/webhookDelivery.js create mode 100644 backend/src/services/webhookStore.js diff --git a/backend/src/routes/webhooks.js b/backend/src/routes/webhooks.js index 10d7612..2f1cc94 100644 --- a/backend/src/routes/webhooks.js +++ b/backend/src/routes/webhooks.js @@ -1,48 +1,95 @@ +/** + * src/routes/webhooks.js + * Webhook registration, listing, and deletion endpoints. + * + * POST /api/webhooks — register a webhook + start monitoring + * GET /api/webhooks/:publicKey — list webhooks for an account (no secret) + * DELETE /api/webhooks/:id — remove a webhook (200 / { success } on success) + */ + "use strict"; const express = require("express"); +const { StrKey } = require("@stellar/stellar-sdk"); const router = express.Router(); -const { registerWebhook, getWebhooksByPublicKey, deleteWebhook } = require("../services/webhookService"); + +const { strictLimiter } = require("../middleware/rateLimit"); +const { + registerWebhook, + getWebhooksByPublicKey, + deleteWebhook, +} = require("../services/webhookService"); + +/** + * Strip the secret field before sending a webhook to the client. + * @param {{ secret?: string, [key: string]: unknown }} webhook + */ +function sanitizeWebhook({ secret: _secret, ...rest }) { + return rest; +} /** * POST /api/webhooks - * Register a webhook for a Stellar account * Body: { publicKey, url, secret } */ -router.post("/", (req, res) => { - const { publicKey, url, secret } = req.body; +router.post("/", strictLimiter, (req, res) => { + const { publicKey, url, secret } = req.body ?? {}; + + // ── Field presence ────────────────────────────────────────────────────────── if (!publicKey || !url || !secret) { return res.status(400).json({ error: "publicKey, url, and secret are required" }); } - try { - const webhook = registerWebhook(publicKey, url, secret); - return res.status(201).json({ success: true, webhook }); - } catch (err) { - return res.status(500).json({ error: err.message }); + + if (typeof publicKey !== "string" || !publicKey.trim()) { + return res.status(400).json({ error: "publicKey must be a non-empty string" }); + } + if (typeof url !== "string" || !url.trim()) { + return res.status(400).json({ error: "url must be a non-empty string" }); } + if (typeof secret !== "string" || !secret.trim()) { + return res.status(400).json({ error: "secret must be a non-empty string" }); + } + + // ── publicKey validation ──────────────────────────────────────────────────── + if (!StrKey.isValidEd25519PublicKey(publicKey)) { + return res.status(400).json({ error: "Invalid Stellar public key" }); + } + + // ── URL validation — must use HTTPS ───────────────────────────────────────── + if (!url.startsWith("https://")) { + return res.status(400).json({ error: "url must start with https://" }); + } + + const webhook = registerWebhook(publicKey, url, secret); + // ensureMonitored is called inside webhookService.registerWebhook + + // Return { success: true, webhook: } + return res.status(201).json({ success: true, webhook: sanitizeWebhook(webhook) }); }); /** * GET /api/webhooks/:publicKey - * Get all webhooks for a Stellar account + * Returns { webhooks: [...] } with secrets stripped. */ -router.get("/:publicKey", (req, res) => { +router.get("/:publicKey", strictLimiter, (req, res) => { const { publicKey } = req.params; const hooks = getWebhooksByPublicKey(publicKey); - return res.json({ webhooks: hooks }); + return res.status(200).json({ webhooks: hooks.map(sanitizeWebhook) }); }); /** * DELETE /api/webhooks/:id - * Delete a webhook by ID + * Returns { success: true } on success, 404 if not found. */ -router.delete("/:id", (req, res) => { +router.delete("/:id", strictLimiter, (req, res) => { const { id } = req.params; const deleted = deleteWebhook(id); + if (!deleted) { return res.status(404).json({ error: "Webhook not found" }); } - return res.json({ success: true, message: `Webhook ${id} deleted` }); + + return res.status(200).json({ success: true }); }); module.exports = router; diff --git a/backend/src/server.js b/backend/src/server.js index 479a651..8a427b5 100644 --- a/backend/src/server.js +++ b/backend/src/server.js @@ -25,6 +25,7 @@ const webhookRoutes = require("./routes/webhooks"); const swaggerUi = require("swagger-ui-express"); const swaggerSpec = require("./swagger"); const { startTurretsServer } = require("./turretsServer"); +const { resumeAllMonitors } = require("./services/paymentMonitor"); const logger = require("./utils/logger"); const { validateEnv, parseAllowedOrigins } = require("./config/validateEnv"); @@ -254,6 +255,9 @@ if (require.main === module) { }); startTurretsServer(); + + // Resume SSE monitoring for all webhooks that existed before restart + resumeAllMonitors(); } module.exports = app; diff --git a/backend/src/services/paymentMonitor.js b/backend/src/services/paymentMonitor.js new file mode 100644 index 0000000..2f20920 --- /dev/null +++ b/backend/src/services/paymentMonitor.js @@ -0,0 +1,134 @@ +/** + * src/services/paymentMonitor.js + * Monitors Stellar accounts for incoming payments via Horizon SSE streaming. + * Fires registered webhooks using Promise.allSettled for parallel, safe delivery. + */ + +"use strict"; + +const { Horizon } = require("@stellar/stellar-sdk"); +const logger = require("../utils/logger"); +const { getWebhooksByPublicKey, getAllWebhooks } = require("./webhookStore"); +const { deliverWebhook } = require("./webhookDelivery"); + +const HORIZON_URL = + process.env.HORIZON_URL || "https://horizon-testnet.stellar.org"; + +const horizonServer = new Horizon.Server(HORIZON_URL); + +/** + * Map of publicKey → SSE close function. + * Prevents duplicate streams for the same account. + * @type {Map void>} + */ +const activeStreams = new Map(); + +/** + * Start monitoring a Stellar account for incoming payments. + * If a stream is already active for this key, this is a no-op. + * + * @param {string} publicKey + */ +function startMonitoring(publicKey) { + if (activeStreams.has(publicKey)) { + return; // idempotent — already monitored + } + + logger.info({ publicKey }, "[monitor] starting SSE stream"); + + const closeStream = horizonServer + .payments() + .forAccount(publicKey) + .cursor("now") + .stream({ + onmessage: async (record) => { + // Only handle incoming simple payments to this account + if (record.type !== "payment" || record.to !== publicKey) return; + + const asset = + record.asset_type === "native" + ? "native" + : `${record.asset_code}:${record.asset_issuer}`; + + /** @type {import('./webhookDelivery').PaymentPayload} */ + const payload = { + event: "payment_received", + publicKey, + amount: record.amount, + asset, + from: record.from, + transactionHash: record.transaction_hash, + ledger: record.ledger_attr ?? 0, + timestamp: record.created_at, + }; + + const hooks = getWebhooksByPublicKey(publicKey); + if (hooks.length === 0) return; + + // Parallel delivery — one failed hook must not block others + await Promise.allSettled(hooks.map((hook) => deliverWebhook(hook, payload))); + }, + + onerror: (err) => { + logger.error( + { publicKey, err }, + `[monitor] stream error for ${publicKey}: ${err.message ?? err}` + ); + // Remove the dead stream so ensureMonitored can restart it next time + activeStreams.delete(publicKey); + }, + }); + + activeStreams.set(publicKey, closeStream); +} + +/** + * Stop monitoring a Stellar account and close its SSE stream. + * + * @param {string} publicKey + */ +function stopMonitoring(publicKey) { + const close = activeStreams.get(publicKey); + if (close) { + try { + close(); + } catch (err) { + logger.error({ publicKey, err }, "[monitor] error closing stream"); + } + activeStreams.delete(publicKey); + logger.info({ publicKey }, "[monitor] SSE stream closed"); + } +} + +/** + * Ensure a Stellar account is being monitored. + * Idempotent — safe to call multiple times for the same key. + * + * @param {string} publicKey + */ +function ensureMonitored(publicKey) { + startMonitoring(publicKey); +} + +/** + * Resume monitoring for all webhooks that exist at startup. + * Call this once during app initialisation so existing registrations + * survive server restarts. + */ +function resumeAllMonitors() { + const allWebhooks = getAllWebhooks(); + const seen = new Set(); + for (const webhook of allWebhooks) { + if (!seen.has(webhook.publicKey)) { + seen.add(webhook.publicKey); + ensureMonitored(webhook.publicKey); + } + } +} + +module.exports = { + startMonitoring, + stopMonitoring, + ensureMonitored, + resumeAllMonitors, +}; diff --git a/backend/src/services/webhookDelivery.js b/backend/src/services/webhookDelivery.js new file mode 100644 index 0000000..86d57e7 --- /dev/null +++ b/backend/src/services/webhookDelivery.js @@ -0,0 +1,72 @@ +/** + * src/services/webhookDelivery.js + * Delivers a signed POST notification to a registered webhook URL. + * Failures are logged but never thrown — the monitor must not crash. + */ + +"use strict"; + +const logger = require("../utils/logger"); +const { generateWebhookSignature } = require("../utils/webhookSignature"); + +/** + * @typedef {Object} Webhook + * @property {string} id + * @property {string} publicKey + * @property {string} url + * @property {string} secret + * @property {string} createdAt + */ + +/** + * @typedef {Object} PaymentPayload + * @property {'payment_received'} event + * @property {string} publicKey + * @property {string} amount + * @property {string} asset - 'native' or 'CODE:ISSUER' + * @property {string} from - sender's public key + * @property {string} transactionHash + * @property {number} ledger + * @property {string} timestamp + */ + +/** + * Deliver a webhook notification. + * Signs the body with HMAC-SHA256 and POSTs to webhook.url. + * Non-2xx responses and network errors are logged, never re-thrown. + * + * @param {Webhook} webhook + * @param {PaymentPayload} payload + * @returns {Promise} + */ +async function deliverWebhook(webhook, payload) { + const body = JSON.stringify(payload); + const sig = generateWebhookSignature(body, webhook.secret); + + try { + const res = await fetch(webhook.url, { + method: "POST", + headers: { + "Content-Type": "application/json", + "X-Stellar-Signature": `sha256=${sig}`, + "X-Webhook-ID": webhook.id, + }, + body, + }); + + if (!res.ok) { + logger.error( + { webhookId: webhook.id, url: webhook.url, status: res.status }, + `[webhook] delivery failed for ${webhook.id}: HTTP ${res.status}` + ); + } + } catch (err) { + logger.error( + { webhookId: webhook.id, url: webhook.url, err }, + `[webhook] delivery failed for ${webhook.id}: ${err.message}` + ); + // Do NOT rethrow — callers use Promise.allSettled and the monitor must not crash + } +} + +module.exports = { deliverWebhook }; diff --git a/backend/src/services/webhookService.js b/backend/src/services/webhookService.js index c8ed17e..0d02f28 100644 --- a/backend/src/services/webhookService.js +++ b/backend/src/services/webhookService.js @@ -1,101 +1,54 @@ -"use strict"; - -const crypto = require("crypto"); -const { Horizon } = require("@stellar/stellar-sdk"); -const logger = require("../utils/logger"); -require("dotenv").config(); - -const HORIZON_URL = process.env.HORIZON_URL || "https://horizon-testnet.stellar.org"; -const server = new Horizon.Server(HORIZON_URL); +/** + * src/services/webhookService.js + * Facade that composes webhookStore + paymentMonitor. + * Kept for backward compatibility — existing code and tests import from here. + * + * registerWebhook here wraps the store's version and also calls ensureMonitored, + * so callers don't need to import paymentMonitor separately. + */ -// In-memory store: { id, publicKey, url, secret, createdAt } -const webhooks = new Map(); -let nextId = 1; +"use strict"; +const store = require("./webhookStore"); +const { deliverWebhook } = require("./webhookDelivery"); +const { + startMonitoring, + stopMonitoring, + ensureMonitored, + resumeAllMonitors, +} = require("./paymentMonitor"); + +/** + * Register a webhook and immediately start (or confirm) monitoring. + * Accepts positional args (publicKey, url, secret) to match existing call sites. + * + * @param {string} publicKey + * @param {string} url + * @param {string} secret + * @returns {import('./webhookStore').Webhook} + */ function registerWebhook(publicKey, url, secret) { - const id = String(nextId++); - const webhook = { id, publicKey, url, secret, createdAt: new Date().toISOString() }; - webhooks.set(id, webhook); - startMonitoring(webhook); - logger.info(JSON.stringify({ type: "webhook_registered", id, publicKey, url })); + const webhook = store.registerWebhook(publicKey, url, secret); + ensureMonitored(publicKey); return webhook; } -function getWebhooksByPublicKey(publicKey) { - return Array.from(webhooks.values()).filter(w => w.publicKey === publicKey); -} - -function deleteWebhook(id) { - const exists = webhooks.has(id); - if (exists) { - webhooks.delete(id); - logger.info(JSON.stringify({ type: "webhook_deleted", id })); - } - return exists; -} - -function signPayload(secret, payload) { - return crypto.createHmac("sha256", secret).update(JSON.stringify(payload)).digest("hex"); -} - -async function deliverWebhook(webhook, payload) { - const signature = signPayload(webhook.secret, payload); - try { - const res = await fetch(webhook.url, { - method: "POST", - headers: { - "Content-Type": "application/json", - "X-Webhook-Signature": signature, - }, - body: JSON.stringify(payload), - }); - if (!res.ok) { - logger.error(JSON.stringify({ type: "webhook_delivery_failed", id: webhook.id, status: res.status, url: webhook.url })); - } else { - logger.info(JSON.stringify({ type: "webhook_delivered", id: webhook.id, url: webhook.url })); - } - } catch (err) { - logger.error(JSON.stringify({ type: "webhook_delivery_error", id: webhook.id, url: webhook.url, error: err.message })); - } -} - -const activeStreams = new Map(); - -function startMonitoring(webhook) { - if (activeStreams.has(webhook.publicKey)) return; - - const closeStream = server - .payments() - .forAccount(webhook.publicKey) - .cursor("now") - .stream({ - onmessage: async (payment) => { - if (payment.type !== "payment" || payment.to !== webhook.publicKey) return; - const payload = { - event: "payment.received", - publicKey: webhook.publicKey, - payment: { - id: payment.id, - from: payment.from, - to: payment.to, - amount: payment.amount, - asset: payment.asset_type === "native" ? "XLM" : payment.asset_code, - createdAt: payment.created_at, - }, - }; - const hooks = getWebhooksByPublicKey(webhook.publicKey); - for (const hook of hooks) { - await deliverWebhook(hook, payload); - } - }, - onerror: (err) => { - logger.error(JSON.stringify({ type: "horizon_sse_error", publicKey: webhook.publicKey, error: err.message })); - activeStreams.delete(webhook.publicKey); - }, - }); - - activeStreams.set(webhook.publicKey, closeStream); - logger.info(JSON.stringify({ type: "horizon_monitoring_started", publicKey: webhook.publicKey })); -} - -module.exports = { registerWebhook, getWebhooksByPublicKey, deleteWebhook }; +module.exports = { + // composed registration (store + monitor) + registerWebhook, + + // store pass-throughs + getWebhooksByPublicKey: store.getWebhooksByPublicKey, + getWebhookById: store.getWebhookById, + deleteWebhook: store.deleteWebhook, + getAllWebhooks: store.getAllWebhooks, + + // delivery + deliverWebhook, + + // monitor + startMonitoring, + stopMonitoring, + ensureMonitored, + resumeAllMonitors, +}; diff --git a/backend/src/services/webhookStore.js b/backend/src/services/webhookStore.js new file mode 100644 index 0000000..71fe670 --- /dev/null +++ b/backend/src/services/webhookStore.js @@ -0,0 +1,94 @@ +/** + * src/services/webhookStore.js + * In-memory store for registered webhooks. + * Pattern mirrors tipsService.js / usernameService.js — plain Map, module-level. + */ + +"use strict"; + +/** @type {Map} */ +const webhooks = new Map(); +let nextId = 1; + +/** + * @typedef {Object} Webhook + * @property {string} id + * @property {string} publicKey - Stellar public key being monitored + * @property {string} url - Destination URL for POST notifications + * @property {string} secret - HMAC signing secret (never expose in API responses) + * @property {string} createdAt - ISO timestamp + */ + +/** + * Register a new webhook. + * Accepts either positional args (publicKey, url, secret) or a single + * object ({ publicKey, url, secret }) so both call sites work. + * + * @param {string | { publicKey: string, url: string, secret: string }} publicKeyOrData + * @param {string} [urlArg] + * @param {string} [secretArg] + * @returns {Webhook} + */ +function registerWebhook(publicKeyOrData, urlArg, secretArg) { + let publicKey, url, secret; + if (typeof publicKeyOrData === "object" && publicKeyOrData !== null) { + ({ publicKey, url, secret } = publicKeyOrData); + } else { + publicKey = publicKeyOrData; + url = urlArg; + secret = secretArg; + } + const id = String(nextId++); + const webhook = { + id, + publicKey, + url, + secret, + createdAt: new Date().toISOString(), + }; + webhooks.set(id, webhook); + return webhook; +} + +/** + * Get all webhooks for a given Stellar public key. + * @param {string} publicKey + * @returns {Webhook[]} + */ +function getWebhooksByPublicKey(publicKey) { + return Array.from(webhooks.values()).filter((w) => w.publicKey === publicKey); +} + +/** + * Get a single webhook by id. + * @param {string} id + * @returns {Webhook | undefined} + */ +function getWebhookById(id) { + return webhooks.get(id); +} + +/** + * Delete a webhook by id. + * @param {string} id + * @returns {boolean} true if deleted, false if not found + */ +function deleteWebhook(id) { + return webhooks.delete(id); +} + +/** + * Return all registered webhooks (used by the monitor on startup). + * @returns {Webhook[]} + */ +function getAllWebhooks() { + return Array.from(webhooks.values()); +} + +module.exports = { + registerWebhook, + getWebhooksByPublicKey, + getWebhookById, + deleteWebhook, + getAllWebhooks, +};