feat: end-to-end webhook monitoring with Horizon SSE (#94/#61)

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: <without secret> }
+  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;

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;

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<string, () => 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,
+};

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<void>}
+ */
+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 };