Add webhook event filtering support

stellar-payment-platform/prisma/schema.prisma

@@ -44,6 +44,7 @@ model Webhook {
   user         User      @relation(fields: [username], references: [username], onDelete: Cascade)
   url          String
   secret       String // HMAC-SHA256 signing secret for payload verification
+  events       String[]  @default(["*"]) // Optional event filter; "*" preserves the legacy all-events behavior.
   createdAt    DateTime  @default(now()) @map("created_at")
   lastSentAt   DateTime? @map("last_sent_at")
   failingSince DateTime? @map("failing_since")

stellar-payment-platform/src/routes/v1/webhookRoutes.js

@@ -146,6 +146,20 @@ const isValidWebhookUrl = (url) => {
   }
 };
 
+const normalizeWebhookEvents = (input) => {
+  if (input === undefined || input === null) return ['*'];
+  const raw = Array.isArray(input) ? input : [input];
+  const events = raw
+    .filter((value) => typeof value === 'string' && value.trim())
+    .map((value) => value.trim())
+    .filter((value, index, arr) => arr.indexOf(value) === index);
+
+  if (events.length === 0) return ['*'];
+  if (events.includes('*')) return ['*'];
+
+  return events;
+};
+
 router.post('/webhooks', asyncHandler(async (req, res, next) => {
   try {
     if (!req.is('application/json')) {
@@ -154,6 +168,7 @@ router.post('/webhooks', asyncHandler(async (req, res, next) => {
 
     const user = await authenticateWebhookCall(req);
     const rawUrl = typeof req.body?.url === 'string' ? req.body.url.trim() : '';
+    const events = normalizeWebhookEvents(req.body?.events);
 
     if (!isValidWebhookUrl(rawUrl)) {
       return res.status(400).json({ error: 'Invalid webhook URL. Must be http or https.' });
@@ -171,6 +186,7 @@ router.post('/webhooks', asyncHandler(async (req, res, next) => {
           username: user.username,
           url: rawUrl,
           secret,
+          events,
           createdAt: now,
         },
       });
@@ -188,11 +204,11 @@ router.post('/webhooks', asyncHandler(async (req, res, next) => {
       if (!shouldFallbackToLocalRegistry(error)) throw error;
 
       await poolRun(
-        `INSERT INTO webhooks (id, username, url, secret, created_at, last_sent_at, failing_since)
-         VALUES (?, ?, ?, ?, ?, NULL, NULL)`,
-        [id, user.username, rawUrl, secret, now.toISOString()],
+        `INSERT INTO webhooks (id, username, url, secret, events, created_at, last_sent_at, failing_since)
+         VALUES (?, ?, ?, ?, ?, ?, NULL, NULL)`,
+        [id, user.username, rawUrl, secret, JSON.stringify(events), now.toISOString()],
       );
-      webhook = { id, username: user.username, url: rawUrl, createdAt: now.toISOString() };
+      webhook = { id, username: user.username, url: rawUrl, events, createdAt: now.toISOString() };
     }
 
     return res.status(201).json({
@@ -201,6 +217,7 @@ router.post('/webhooks', asyncHandler(async (req, res, next) => {
         id: webhook.id,
         username: webhook.username,
         url: webhook.url,
+        events: Array.isArray(webhook.events) ? webhook.events : normalizeWebhookEvents(webhook.events),
         secret,
         created_at: (webhook.createdAt instanceof Date
           ? webhook.createdAt
@@ -235,14 +252,15 @@ router.get('/webhooks', asyncHandler(async (req, res, next) => {
     } catch (error) {
       if (!shouldFallbackToLocalRegistry(error)) throw error;
       const rows = await poolAll(
-        `SELECT id, username, url, created_at, last_sent_at, failing_since
+        `SELECT id, username, url, events, created_at, last_sent_at, failing_since
          FROM webhooks WHERE username = ? ORDER BY created_at DESC`,
         [user.username],
       );
       webhooks = rows.map((r) => ({
         id: r.id,
         username: r.username,
         url: r.url,
+        events: Array.isArray(r.events) ? r.events : (typeof r.events === 'string' ? JSON.parse(r.events || '[]') : ['*']),
         createdAt: r.created_at,
         lastSentAt: r.last_sent_at,
         failingSince: r.failing_since,
@@ -254,6 +272,7 @@ router.get('/webhooks', asyncHandler(async (req, res, next) => {
       webhooks: webhooks.map((w) => ({
         id: w.id,
         url: w.url,
+        events: Array.isArray(w.events) ? w.events : normalizeWebhookEvents(w.events),
         created_at: (w.createdAt instanceof Date ? w.createdAt : new Date(w.createdAt)).toISOString(),
         last_sent_at: w.lastSentAt
           ? (w.lastSentAt instanceof Date ? w.lastSentAt : new Date(w.lastSentAt)).toISOString()

stellar-payment-platform/src/webhookWorker.js

@@ -20,6 +20,17 @@ const computeSignature = (secret, rawBody) => {
   return crypto.createHmac('sha256', secret).update(rawBody).digest('hex');
 };
 
+const webhookEventMatches = (webhook, eventName) => {
+  const subscriptions = Array.isArray(webhook?.events) ? webhook.events : ['*'];
+  const normalized = subscriptions
+    .filter((value) => typeof value === 'string')
+    .map((value) => value.trim())
+    .filter(Boolean);
+
+  if (normalized.length === 0 || normalized.includes('*')) return true;
+  return normalized.includes(eventName);
+};
+
 const fetchWebhooksForAddress = async (prisma, poolGetFn, stellarAddress) => {
   try {
     return await prisma.webhook.findMany({
@@ -31,14 +42,15 @@ const fetchWebhooksForAddress = async (prisma, poolGetFn, stellarAddress) => {
         username: true,
         url: true,
         secret: true,
-        failingSince: true,
+       events: true,
+       failingSince: true,
       },
     });
   } catch (error) {
     if (!shouldFallbackToLocalRegistry(error)) throw error;
 
     const rows = await poolGetFn(
-      `SELECT w.id, w.username, w.url, w.secret, w.failing_since
+      `SELECT w.id, w.username, w.url, w.secret, w.events, w.failing_since
        FROM webhooks w
        INNER JOIN username_registry u ON u.username = w.username
        WHERE u.address = ?`,
@@ -49,6 +61,7 @@ const fetchWebhooksForAddress = async (prisma, poolGetFn, stellarAddress) => {
       username: r.username,
       url: r.url,
       secret: r.secret,
+      events: Array.isArray(r.events) ? r.events : (typeof r.events === 'string' ? JSON.parse(r.events || '[]') : ['*']),
       failingSince: r.failing_since ? new Date(r.failing_since) : null,
     }));
   }
@@ -174,6 +187,11 @@ const dispatchPaymentWebhooks = async ({
   };
 
   for (const wh of webhooks) {
+    if (!webhookEventMatches(wh, payload.event)) {
+      logger.info(`[webhook-worker] Skipping webhook id=${wh.id} url=${wh.url} for event=${payload.event} due to subscription filter`);
+      continue;
+    }
+
     const now = new Date();
     try {
       await sendWebhook(wh.url, payload, wh.secret);
@@ -207,12 +225,13 @@ const listStaleFailingWebhooks = async (prisma, poolAllFn) => {
         username: true,
         url: true,
         secret: true,
+       events: true,
       },
     });
   } catch (error) {
     if (!shouldFallbackToLocalRegistry(error)) throw error;
     const rows = await poolAllFn(
-      `SELECT id, username, url, secret FROM webhooks
+      `SELECT id, username, url, secret, events FROM webhooks
        WHERE failing_since IS NOT NULL AND failing_since >= ?`,
       [cutoff.toISOString()],
     );
@@ -221,11 +240,16 @@ const listStaleFailingWebhooks = async (prisma, poolAllFn) => {
       username: r.username,
       url: r.url,
       secret: r.secret,
+      events: Array.isArray(r.events) ? r.events : (typeof r.events === 'string' ? JSON.parse(r.events || '[]') : ['*']),
     }));
   }
 };
 
 const sendLivenessPing = async (prisma, poolRunFn, webhook) => {
+  if (!webhookEventMatches(webhook, 'webhook.ping')) {
+    return false;
+  }
+
   const payload = {
     event: 'webhook.ping',
     event_id: `ping-${crypto.randomBytes(16).toString('hex')}`,

stellar-payment-platform/tests/payment-metadata.test.js

@@ -130,4 +130,70 @@ describe('payment intent metadata', () => {
     const requestBody = JSON.parse(fetchSpy.mock.calls[0][1].body);
     expect(requestBody.data.metadata).toEqual(metadata);
   });
+
+  test('delivers payment events only when the merchant subscribed to payment.received', async () => {
+    const prisma = {
+      webhook: {
+        findMany: jest.fn().mockResolvedValue([{
+          id: 'webhook-1',
+          url: 'https://merchant.example/webhooks',
+          secret: 'secret',
+          events: ['payment.received'],
+        }]),
+        update: jest.fn().mockResolvedValue({}),
+      },
+    };
+    const fetchSpy = jest.spyOn(global, 'fetch').mockResolvedValue({ ok: true, status: 200 });
+
+    await dispatchPaymentWebhooks({
+      prisma,
+      poolGetFn: jest.fn(),
+      poolRunFn: jest.fn(),
+      payment: {
+        id: 'payment-2',
+        type: 'payment',
+        transaction_hash: 'transaction-2',
+        from: 'GSOURCE',
+        to: 'GDESTINATION',
+        amount: '25.00',
+        asset_type: 'native',
+      },
+    });
+
+    expect(fetchSpy).toHaveBeenCalledTimes(1);
+    const requestBody = JSON.parse(fetchSpy.mock.calls[0][1].body);
+    expect(requestBody.event).toBe('payment.received');
+  });
+
+  test('skips delivery for unsubscribed webhook event types', async () => {
+    const prisma = {
+      webhook: {
+        findMany: jest.fn().mockResolvedValue([{
+          id: 'webhook-1',
+          url: 'https://merchant.example/webhooks',
+          secret: 'secret',
+          events: ['registration.updated'],
+        }]),
+        update: jest.fn().mockResolvedValue({}),
+      },
+    };
+    const fetchSpy = jest.spyOn(global, 'fetch').mockResolvedValue({ ok: true, status: 200 });
+
+    await dispatchPaymentWebhooks({
+      prisma,
+      poolGetFn: jest.fn(),
+      poolRunFn: jest.fn(),
+      payment: {
+        id: 'payment-3',
+        type: 'payment',
+        transaction_hash: 'transaction-3',
+        from: 'GSOURCE',
+        to: 'GDESTINATION',
+        amount: '25.00',
+        asset_type: 'native',
+      },
+    });
+
+    expect(fetchSpy).not.toHaveBeenCalled();
+  });
 });