fix(webhooks): persist tenant-scoped subscriptions across replicas (#447)

backend/src/controllers/webhook.controller.ts

@@ -29,26 +29,36 @@ export class WebhookController {
     }
   }
 
-  static listSubscriptions(req: Request, res: Response) {
-    const organizationId = req.tenantId!;
-    const subscriptions = WebhookService.listSubscriptions(organizationId);
-    res.json(subscriptions);
+  static async listSubscriptions(req: Request, res: Response) {
+    try {
+      const organizationId = req.tenantId!;
+      const subscriptions = await WebhookService.listSubscriptions(organizationId);
+      res.json(subscriptions);
+    } catch {
+      res.status(500).json({ error: 'Internal Server Error' });
+    }
   }
 
-  static deleteSubscription(req: Request, res: Response) {
-    const { id } = req.params;
-    const organizationId = req.tenantId!;
-    const success = WebhookService.deleteSubscription(id as string, organizationId);
-    if (success) {
-      res.status(204).send();
-      return;
+  static async deleteSubscription(req: Request, res: Response) {
+    try {
+      const { id } = req.params;
+      const organizationId = req.tenantId!;
+      const success = await WebhookService.deleteSubscription(id as string, organizationId);
+      if (success) {
+        res.status(204).send();
+        return;
+      }
+      res.status(404).json({ error: 'Subscription not found' });
+    } catch {
+      res.status(500).json({ error: 'Internal Server Error' });
     }
-    res.status(404).json({ error: 'Subscription not found' });
   }
 
   static async triggerMockEvent(req: Request, res: Response) {
     const { event, payload } = req.body;
+    const organizationId = req.tenantId!;
     await WebhookService.dispatch(
+      organizationId,
       (event as string) || 'payment.completed',
       payload || { id: 'test_tx_123', amount: 100 }
     );

backend/src/db/migrations/032_create_webhook_subscriptions.sql

@@ -0,0 +1,15 @@
+-- Persist webhook subscriptions so all backend replicas share the same tenant-scoped registry.
+CREATE TABLE IF NOT EXISTS webhook_subscriptions (
+  id UUID PRIMARY KEY,
+  organization_id INTEGER NOT NULL REFERENCES organizations(id) ON DELETE CASCADE,
+  url TEXT NOT NULL,
+  secret TEXT NOT NULL,
+  events TEXT[] NOT NULL DEFAULT ARRAY['*']::TEXT[],
+  created_at TIMESTAMPTZ NOT NULL DEFAULT NOW()
+);
+
+CREATE INDEX IF NOT EXISTS idx_webhook_subscriptions_org_id
+  ON webhook_subscriptions (organization_id);
+
+COMMENT ON TABLE webhook_subscriptions IS
+  'Tenant-scoped outbound webhook subscriptions shared by all backend replicas.';

backend/src/routes/__tests__/webhook.routes.test.ts

@@ -2,7 +2,7 @@ import request from 'supertest';
 import express, { Request, Response, NextFunction } from 'express';
 import jwt from 'jsonwebtoken';
 import webhookRoutes from '../webhook.routes.js';
-import { WebhookService } from '../../services/webhook.service.js';
+import { pool } from '../../config/database.js';
 
 const JWT_SECRET = 'test-secret';
 
@@ -46,13 +46,67 @@ const app = express();
 app.use(express.json());
 app.use('/webhooks', webhookRoutes);
 
+interface StoredWebhookSubscription {
+  id: string;
+  organization_id: number;
+  url: string;
+  secret: string;
+  events: string[];
+  created_at: Date;
+}
+
+let webhookRows: StoredWebhookSubscription[] = [];
+
 describe('Webhook Routes - Auth and Tenant Isolation', () => {
-  beforeEach(async () => {
-    // Clear in-memory subscriptions by listing and deleting for both tenants
-    const subsA = WebhookService.listSubscriptions(10);
-    for (const s of subsA) WebhookService.deleteSubscription(s.id, 10);
-    const subsB = WebhookService.listSubscriptions(20);
-    for (const s of subsB) WebhookService.deleteSubscription(s.id, 20);
+  beforeEach(() => {
+    webhookRows = [];
+
+    (pool.query as jest.Mock).mockImplementation(async (sql: unknown, params: unknown[] = []) => {
+      if (typeof sql !== 'string') return { rows: [] };
+      const normalized = sql.replace(/\s+/g, ' ').trim();
+
+      if (normalized.startsWith('INSERT INTO webhook_subscriptions')) {
+        const row: StoredWebhookSubscription = {
+          id: String(params[0]),
+          organization_id: Number(params[1]),
+          url: String(params[2]),
+          secret: String(params[3]),
+          events: params[4] as string[],
+          created_at: new Date(),
+        };
+        webhookRows.push(row);
+        return { rows: [row] };
+      }
+
+      if (
+        normalized.startsWith('SELECT id, url, secret, events, organization_id') &&
+        normalized.includes('FROM webhook_subscriptions')
+      ) {
+        const organizationId = Number(params[0]);
+        return {
+          rows: webhookRows
+            .filter((row) => row.organization_id === organizationId)
+            .sort(
+              (left, right) =>
+                left.created_at.getTime() - right.created_at.getTime() ||
+                left.id.localeCompare(right.id)
+            ),
+        };
+      }
+
+      if (normalized.startsWith('DELETE FROM webhook_subscriptions')) {
+        const [id, organizationId] = [String(params[0]), Number(params[1])];
+        const deleted = webhookRows.find(
+          (row) => row.id === id && row.organization_id === organizationId
+        );
+        webhookRows = webhookRows.filter(
+          (row) => !(row.id === id && row.organization_id === organizationId)
+        );
+        return { rows: deleted ? [{ id: deleted.id }] : [] };
+      }
+
+      return { rows: [] };
+    });
   });
 
   describe('Authentication required', () => {
@@ -100,6 +154,9 @@ describe('Webhook Routes - Auth and Tenant Isolation', () => {
 
       expect(resA.status).toBe(201);
       expect(resA.body.organizationId).toBe(10);
+      expect(resA.body.id).toMatch(
+        /^[0-9a-f]{8}-[0-9a-f]{4}-4[0-9a-f]{3}-[89ab][0-9a-f]{3}-[0-9a-f]{12}$/
+      );
 
       const listA = await request(app)
         .get('/webhooks/subscriptions')
@@ -110,6 +167,25 @@ describe('Webhook Routes - Auth and Tenant Isolation', () => {
       expect(listA.body[0].organizationId).toBe(10);
     });
 
+    it('reads subscriptions written by another backend instance from shared storage', async () => {
+      webhookRows.push({
+        id: '00000000-0000-4000-8000-000000000001',
+        organization_id: 10,
+        url: 'https://other-pod.example.com/hook',
+        secret: 's'.repeat(16),
+        events: ['payment.completed'],
+        created_at: new Date('2026-10-08T00:00:00Z'),
+      });
+
+      const listA = await request(app)
+        .get('/webhooks/subscriptions')
+        .set('Authorization', `Bearer ${tenantAToken}`);
+
+      expect(listA.status).toBe(200);
+      expect(listA.body).toHaveLength(1);
+      expect(listA.body[0].url).toBe('https://other-pod.example.com/hook');
+    });
+
     it('tenant B cannot see tenant A subscriptions', async () => {
       await request(app)
         .post('/webhooks/subscribe')

backend/src/services/forecasting/liquidityAlertChecker.ts

@@ -125,7 +125,7 @@ export class LiquidityAlertChecker {
     };
 
     if (webhookUrl) {
-      await WebhookService.dispatch('liquidity.insufficient', payload);
+      await WebhookService.dispatch(organizationId, 'liquidity.insufficient', payload);
     }
 
     const alertEmails = liquiditySettings.alertEmails || [];

backend/src/services/webhook.service.ts

@@ -1,5 +1,7 @@
 import axios from 'axios';
 import CryptoJS from 'crypto-js';
+import { randomUUID } from 'node:crypto';
+import { pool } from '../config/database.js';
 
 export interface WebhookSubscription {
   id: string;
@@ -9,8 +11,23 @@ export interface WebhookSubscription {
   organizationId: number;
 }
 
-// In-memory storage for demonstration (in a real app, this would be a database)
-const subscriptions: WebhookSubscription[] = [];
+interface WebhookSubscriptionRow {
+  id: string;
+  url: string;
+  secret: string;
+  events: string[];
+  organization_id: number;
+}
+
+function mapSubscription(row: WebhookSubscriptionRow): WebhookSubscription {
+  return {
+    id: row.id,
+    url: row.url,
+    secret: row.secret,
+    events: row.events,
+    organizationId: Number(row.organization_id),
+  };
+}
 
 export class WebhookService {
   static async subscribe(
@@ -19,34 +36,46 @@ export class WebhookService {
     secret: string,
     events: string[]
   ): Promise<WebhookSubscription> {
-    const subscription: WebhookSubscription = {
-      id: Math.random().toString(36).substring(2, 11),
-      url,
-      secret,
-      events,
-      organizationId,
-    };
-    subscriptions.push(subscription);
-    return subscription;
+    const id = randomUUID();
+    const result = await pool.query<WebhookSubscriptionRow>(
+      `INSERT INTO webhook_subscriptions (id, organization_id, url, secret, events)
+       VALUES ($1, $2, $3, $4, $5::text[])
+       RETURNING id, url, secret, events, organization_id`,
+      [id, organizationId, url, secret, events]
+    );
+
+    return mapSubscription(result.rows[0]);
   }
 
-  static listSubscriptions(organizationId: number): WebhookSubscription[] {
-    return subscriptions.filter((s) => s.organizationId === organizationId);
+  static async listSubscriptions(organizationId: number): Promise<WebhookSubscription[]> {
+    const result = await pool.query<WebhookSubscriptionRow>(
+      `SELECT id, url, secret, events, organization_id
+       FROM webhook_subscriptions
+       WHERE organization_id = $1
+       ORDER BY created_at ASC, id ASC`,
+      [organizationId]
+    );
+
+    return result.rows.map(mapSubscription);
   }
 
-  static deleteSubscription(id: string, organizationId: number): boolean {
-    const index = subscriptions.findIndex(
-      (s) => s.id === id && s.organizationId === organizationId
+  static async deleteSubscription(id: string, organizationId: number): Promise<boolean> {
+    const result = await pool.query(
+      `DELETE FROM webhook_subscriptions
+       WHERE id = $1 AND organization_id = $2
+       RETURNING id`,
+      [id, organizationId]
     );
-    if (index !== -1) {
-      subscriptions.splice(index, 1);
-      return true;
-    }
-    return false;
+
+    return result.rows.length > 0;
   }
 
-  static async dispatch(eventType: string, payload: any): Promise<void> {
-    const relevantSubscriptions = subscriptions.filter(
+  static async dispatch(
+    organizationId: number,
+    eventType: string,
+    payload: any
+  ): Promise<void> {
+    const relevantSubscriptions = (await this.listSubscriptions(organizationId)).filter(
       (s) => s.events.includes(eventType) || s.events.includes('*')
     );