feat: persist webhooks to SQLite, add delivery UI, and switch event i…

backend/src/dal/index.js

@@ -7,7 +7,7 @@ import {
 } from './sqliteCampaignRepository.js';
 import { assertAuditLogRepository } from './auditLogRepository.js';
 import { createSqliteAuditLogRepository } from './sqliteAuditLogRepository.js';
-import { WebhookRepository } from './webhookRepository.js';
+import { createSqliteWebhookRepository } from './sqliteWebhookRepository.js';
 import { createSqliteReferralRepository } from './sqliteReferralRepository.js';
 import { assertApiKeyRepository } from './apiKeyRepository.js';
 import { createSqliteApiKeyRepository } from './sqliteApiKeyRepository.js';
@@ -83,7 +83,7 @@ export async function createDal({
     auditLogs: assertAuditLogRepository(
       auditLogRepository ?? pgAuditLogs ?? createSqliteAuditLogRepository({ db }),
     ),
-    webhooks: webhookRepository ?? new WebhookRepository(db),
+    webhooks: webhookRepository ?? createSqliteWebhookRepository({ db }),
     referrals: createSqliteReferralRepository({ db }),
     variants: createSqliteVariantRepository({ db }),
     cohorts: createSqliteCohortRepository({ db }),

backend/src/dal/sqliteWebhookRepository.js

@@ -0,0 +1,208 @@
+import { randomUUID } from 'node:crypto';
+
+function rowToWebhook(row) {
+  if (!row) return null;
+  return {
+    id: row.id,
+    url: row.url,
+    secret: row.secret,
+    events: JSON.parse(row.events || '[]'),
+    active: row.active === 1,
+    description: row.description || '',
+    createdAt: row.created_at,
+    updatedAt: row.updated_at,
+  };
+}
+
+function rowToDelivery(row) {
+  if (!row) return null;
+  let payload = null;
+  try {
+    payload = row.payload ? JSON.parse(row.payload) : null;
+  } catch {
+    payload = row.payload;
+  }
+  return {
+    id: row.id,
+    webhookId: row.webhook_id,
+    event: row.event,
+    payload,
+    statusCode: row.status_code,
+    error: row.error || null,
+    attempts: row.attempts,
+    nextRetryAt: row.next_retry_at || null,
+    createdAt: row.created_at,
+    updatedAt: row.updated_at,
+  };
+}
+
+export function createSqliteWebhookRepository({ db }) {
+  function create({ url, events, secret, description = '' }) {
+    const id = randomUUID();
+    const now = new Date().toISOString();
+    const resolvedSecret = secret || randomUUID();
+    const eventsJson = JSON.stringify(Array.isArray(events) ? events : []);
+
+    db.prepare(`
+      INSERT INTO webhooks (id, url, secret, events, active, description, created_at, updated_at)
+      VALUES (?, ?, ?, ?, 1, ?, ?, ?)
+    `).run(id, url, resolvedSecret, eventsJson, description, now, now);
+
+    return {
+      id,
+      url,
+      secret: resolvedSecret,
+      events: Array.isArray(events) ? events : [],
+      active: true,
+      description,
+      createdAt: now,
+      updatedAt: now,
+    };
+  }
+
+  function getById(id) {
+    const row = db.prepare('SELECT * FROM webhooks WHERE id = ?').get(id);
+    return rowToWebhook(row);
+  }
+
+  function list(filters = {}) {
+    if (filters.active !== undefined) {
+      const rows = db
+        .prepare('SELECT * FROM webhooks WHERE active = ? ORDER BY created_at DESC')
+        .all(filters.active ? 1 : 0);
+      return rows.map(rowToWebhook);
+    }
+    return db.prepare('SELECT * FROM webhooks ORDER BY created_at DESC').all().map(rowToWebhook);
+  }
+
+  function update(id, updates) {
+    const existing = db.prepare('SELECT * FROM webhooks WHERE id = ?').get(id);
+    if (!existing) return null;
+
+    const now = new Date().toISOString();
+    const merged = rowToWebhook(existing);
+
+    if (updates.url !== undefined) merged.url = updates.url;
+    if (updates.events !== undefined) merged.events = updates.events;
+    if (updates.active !== undefined) merged.active = updates.active;
+    if (updates.description !== undefined) merged.description = updates.description;
+    if (updates.secret !== undefined) merged.secret = updates.secret;
+    merged.updatedAt = now;
+
+    db.prepare(`
+      UPDATE webhooks
+      SET url = ?, events = ?, active = ?, description = ?, secret = ?, updated_at = ?
+      WHERE id = ?
+    `).run(
+      merged.url,
+      JSON.stringify(merged.events),
+      merged.active ? 1 : 0,
+      merged.description,
+      merged.secret,
+      merged.updatedAt,
+      id,
+    );
+
+    return merged;
+  }
+
+  function del(id) {
+    const result = db.prepare('DELETE FROM webhooks WHERE id = ?').run(id);
+    return result.changes > 0;
+  }
+
+  function recordDelivery({ webhookId, event, payload, statusCode, error }) {
+    const id = randomUUID();
+    const now = new Date().toISOString();
+    const failed = statusCode >= 400 || statusCode === 0;
+    const nextRetryAt = failed ? new Date(Date.now() + 60_000).toISOString() : null;
+    const payloadJson = payload !== undefined ? JSON.stringify(payload) : null;
+
+    db.prepare(`
+      INSERT INTO webhook_deliveries
+        (id, webhook_id, event, payload, status_code, error, attempts, next_retry_at, created_at, updated_at)
+      VALUES (?, ?, ?, ?, ?, ?, 1, ?, ?, ?)
+    `).run(id, webhookId, event, payloadJson, statusCode, error || null, nextRetryAt, now, now);
+
+    return {
+      id,
+      webhookId,
+      event,
+      payload,
+      statusCode,
+      error: error || null,
+      attempts: 1,
+      nextRetryAt,
+      createdAt: now,
+      updatedAt: now,
+    };
+  }
+
+  function getDeliveryById(id) {
+    const row = db.prepare('SELECT * FROM webhook_deliveries WHERE id = ?').get(id);
+    return rowToDelivery(row);
+  }
+
+  function listDeliveries(webhookId, filters = {}) {
+    const limit = filters.limit || 100;
+    const rows = db
+      .prepare(
+        'SELECT * FROM webhook_deliveries WHERE webhook_id = ? ORDER BY created_at DESC LIMIT ?',
+      )
+      .all(webhookId, limit);
+    return rows.map(rowToDelivery);
+  }
+
+  function getPendingRetries() {
+    const now = new Date().toISOString();
+    const rows = db
+      .prepare(
+        `SELECT * FROM webhook_deliveries
+         WHERE next_retry_at IS NOT NULL AND next_retry_at <= ? AND attempts < 5`,
+      )
+      .all(now);
+    return rows.map(rowToDelivery);
+  }
+
+  function updateDelivery(id, updates) {
+    const existing = db.prepare('SELECT * FROM webhook_deliveries WHERE id = ?').get(id);
+    if (!existing) return null;
+
+    const now = new Date().toISOString();
+    const merged = rowToDelivery(existing);
+
+    if (updates.statusCode !== undefined) merged.statusCode = updates.statusCode;
+    if (updates.error !== undefined) merged.error = updates.error;
+    if (updates.attempts !== undefined) merged.attempts = updates.attempts;
+    if ('nextRetryAt' in updates) merged.nextRetryAt = updates.nextRetryAt;
+    merged.updatedAt = now;
+
+    db.prepare(`
+      UPDATE webhook_deliveries
+      SET status_code = ?, error = ?, attempts = ?, next_retry_at = ?, updated_at = ?
+      WHERE id = ?
+    `).run(
+      merged.statusCode,
+      merged.error,
+      merged.attempts,
+      merged.nextRetryAt,
+      merged.updatedAt,
+      id,
+    );
+
+    return merged;
+  }
+
+  return {
+    create,
+    getById,
+    list,
+    update,
+    delete: del,
+    recordDelivery,
+    getDeliveryById,
+    listDeliveries,
+    getPendingRetries,
+    updateDelivery,
+  };
+}

backend/src/db/migrations/028_webhooks.js

@@ -0,0 +1,43 @@
+export const version = 28;
+export const description = 'Add webhooks and webhook_deliveries tables for persistent webhook storage';
+
+export function up(db) {
+  db.exec(`
+    CREATE TABLE IF NOT EXISTS webhooks (
+      id          TEXT    PRIMARY KEY,
+      url         TEXT    NOT NULL,
+      secret      TEXT    NOT NULL,
+      events      TEXT    NOT NULL DEFAULT '[]',
+      active      INTEGER NOT NULL DEFAULT 1,
+      description TEXT    NOT NULL DEFAULT '',
+      created_at  TEXT    NOT NULL,
+      updated_at  TEXT    NOT NULL
+    );
+
+    CREATE TABLE IF NOT EXISTS webhook_deliveries (
+      id            TEXT    PRIMARY KEY,
+      webhook_id    TEXT    NOT NULL REFERENCES webhooks(id) ON DELETE CASCADE,
+      event         TEXT    NOT NULL,
+      payload       TEXT,
+      status_code   INTEGER NOT NULL DEFAULT 0,
+      error         TEXT,
+      attempts      INTEGER NOT NULL DEFAULT 1,
+      next_retry_at TEXT,
+      created_at    TEXT    NOT NULL,
+      updated_at    TEXT    NOT NULL
+    );
+
+    CREATE INDEX IF NOT EXISTS idx_webhook_deliveries_webhook_id
+      ON webhook_deliveries(webhook_id);
+    CREATE INDEX IF NOT EXISTS idx_webhook_deliveries_retry
+      ON webhook_deliveries(next_retry_at)
+      WHERE next_retry_at IS NOT NULL;
+  `);
+}
+
+export function down(db) {
+  db.exec(`
+    DROP TABLE IF EXISTS webhook_deliveries;
+    DROP TABLE IF EXISTS webhooks;
+  `);
+}

backend/src/index.js

@@ -92,7 +92,6 @@
 import { createModerationService } from './moderation/moderationService.js';
 import { createContentModerationMiddleware } from './middleware/contentModeration.js';
 import createFaucetRoutes from './routes/faucet.js';
-import createWebhookRoutes from './routes/webhooks.js';
 import createStatusRoutes from './routes/status.js';
 
 const DEFAULT_PORT = 3001;
@@ -732,6 +731,26 @@
     durableJobQueue.start();
   }
 
+  // Event indexer: use Horizon SSE for near-instant indexing; fall back to 30s polling
+  if (!options.disableJobs && (rewardsContractId || campaignContractId)) {
+    const contractIds = [rewardsContractId, campaignContractId].filter(Boolean);
+    if (stellarConfig.horizonUrl) {
+      eventIndexer.startSse({
+        contractIds,
+        horizonUrl: stellarConfig.horizonUrl,
+        allowHttp: !stellarConfig.horizonUrl.startsWith('https'),
+      });
+    } else {
+      const indexerPollMs = 30_000;
+      for (const contractId of contractIds) {
+        setInterval(async () => {
+          const cursor = eventIndexer.getCursor(contractId);
+          await eventIndexer.poll(contractId, cursor);
+        }, indexerPollMs).unref?.();
+      }
+    }
+  }
+
   // #552 — Operator balance monitoring job
   const operatorBalanceJob = createOperatorBalanceJob({
     db: dal.db,
@@ -2251,22 +2270,22 @@
       return res.json(webhook);
     });
 
     app.put(`${prefix}/webhooks/:id`, rateLimiter, idempotencyMiddleware, ...guard, (req, res) => {
       const { url, events, active } = req.body;
       const before = webhookRepository.getById(req.params.id);
       if (!before) {
         return res.status(404).json({ error: 'Webhook not found', code: 'WEBHOOK_NOT_FOUND' });
       }
       const updates = {};
       if (url !== undefined) updates.url = url;
       if (events !== undefined) updates.events = events;
       if (active !== undefined) updates.active = active;
       const webhook = webhookRepository.update(req.params.id, updates);
       recordAuditEntry(req, {
         action: 'update',
         entity: 'webhook',
         entityId: webhook.id,
         diff: { before, after: webhook },
       });
       return res.json(webhook);
     });
@@ -2297,6 +2316,85 @@
       return res.json(paginateItems(deliveries, req.query));
     });
 
+    app.get(`${prefix}/webhooks/:id/deliveries/:deliveryId`, rateLimiter, ...guard, (req, res) => {
+      const webhook = webhookRepository.getById(req.params.id);
+      if (!webhook) {
+        return res.status(404).json({ error: 'Webhook not found', code: 'WEBHOOK_NOT_FOUND' });
+      }
+      const delivery = webhookRepository.getDeliveryById(req.params.deliveryId);
+      if (!delivery || delivery.webhookId !== req.params.id) {
+        return res.status(404).json({ error: 'Delivery not found', code: 'DELIVERY_NOT_FOUND' });
+      }
+      return res.json(delivery);
+    });
+
+    app.post(
+      `${prefix}/webhooks/:id/deliveries/:deliveryId/replay`,
+      rateLimiter,
+      idempotencyMiddleware,
+      ...guard,
+      async (req, res) => {
+        const webhook = webhookRepository.getById(req.params.id);
+        if (!webhook) {
+          return res.status(404).json({ error: 'Webhook not found', code: 'WEBHOOK_NOT_FOUND' });
+        }
+        const delivery = webhookRepository.getDeliveryById(req.params.deliveryId);
+        if (!delivery || delivery.webhookId !== req.params.id) {
+          return res
+            .status(404)
+            .json({ error: 'Delivery not found', code: 'DELIVERY_NOT_FOUND' });
+        }
+        try {
+          await webhookService.deliverWebhook(webhook, {
+            type: delivery.event,
+            data: delivery.payload,
+            timestamp: new Date().toISOString(),
+          });
+          return res.json({ replayed: true, webhookId: req.params.id, event: delivery.event });
+        } catch (err) {
+          log.warn({ err, webhookId: req.params.id }, 'Webhook replay error');
+          return res.status(502).json({ error: 'Replay delivery failed', code: 'REPLAY_FAILED' });
+        }
+      },
+    );
+
+    app.post(`${prefix}/webhooks/:id/test`, rateLimiter, ...guard, async (req, res) => {
+      const webhook = webhookRepository.getById(req.params.id);
+      if (!webhook) {
+        return res.status(404).json({ error: 'Webhook not found', code: 'WEBHOOK_NOT_FOUND' });
+      }
+      const eventType = req.body?.eventType || 'campaign.created';
+      try {
+        await webhookService.deliverWebhook(webhook, {
+          type: eventType,
+          data: { test: true, timestamp: new Date().toISOString() },
+          timestamp: new Date().toISOString(),
+        });
+        return res.json({ sent: true, webhookId: req.params.id, eventType });
+      } catch (err) {
+        log.warn({ err, webhookId: req.params.id }, 'Webhook test error');
+        return res.status(502).json({ error: 'Test delivery failed', code: 'TEST_FAILED' });
+      }
+    });
+
+    // POST /webhooks/verify — signature verification helper for consumers (no auth required)
+    app.post(`${prefix}/webhooks/verify`, rateLimiter, (req, res) => {
+      const { signature, secret, payload } = req.body ?? {};
+      if (!signature || !secret || payload === undefined) {
+        return res.status(400).json({
+          error: 'signature, secret, and payload are required',
+          code: 'VALIDATION_ERROR',
+        });
+      }
+      try {
+        const payloadStr = typeof payload === 'string' ? payload : JSON.stringify(payload);
+        const valid = webhookService.verifySignature(signature, secret, payloadStr);
+        return res.json({ valid });
+      } catch {
+        return res.json({ valid: false });
+      }
+    });
+
     // Referral routes (Issue #350)
     app.post(`${prefix}/campaigns/:id/referrals`, rateLimiter, (req, res) => {
       const campaign = campaignRepository.getById(req.params.id);
@@ -2580,7 +2678,7 @@
   app.use(`${API_V1_PREFIX}/faucet`, createFaucetRoutes);
 
   // #811 — Partner webhook subscription management
   app.use(`${API_V1_PREFIX}/webhooks`, createWebhookRoutes);
 
   // #818 — Public status page + incident communication
   app.use(`${API_V1_PREFIX}/status`, createStatusRoutes);