feat(webhooks): persist subscriptions in PostgreSQL

backend/src/controllers/webhook.controller.ts

@@ -5,9 +5,19 @@ import { z } from 'zod';
 const subscribeSchema = z.object({
   url: z.string().url(),
   secret: z.string().min(16),
-  events: z.array(z.string()).default(['*']),
+  events: z.array(z.string()).min(1).default(['*']),
 });
 
+const updateSubscriptionSchema = z
+  .object({
+    url: z.string().url().optional(),
+    secret: z.string().min(16).optional(),
+    events: z.array(z.string()).min(1).optional(),
+  })
+  .refine((data) => Object.keys(data).length > 0, {
+    message: 'At least one subscription field must be provided',
+  });
+
 export class WebhookController {
   static async subscribe(req: Request, res: Response) {
     try {
@@ -29,28 +39,62 @@ 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 updateSubscription(req: Request, res: Response) {
+    try {
+      const validatedData = updateSubscriptionSchema.parse(req.body);
+      const organizationId = req.tenantId!;
+      const subscription = await WebhookService.updateSubscription(
+        req.params.id as string,
+        organizationId,
+        validatedData
+      );
+
+      if (!subscription) {
+        res.status(404).json({ error: 'Subscription not found' });
+        return;
+      }
+
+      res.json(subscription);
+    } catch (error) {
+      if (error instanceof z.ZodError) {
+        res.status(400).json({ error: error.issues });
+        return;
+      }
+      res.status(500).json({ error: 'Internal Server Error' });
+    }
+  }
+
+  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;
     await WebhookService.dispatch(
       (event as string) || 'payment.completed',
-      payload || { id: 'test_tx_123', amount: 100 }
+      payload || { id: 'test_tx_123', amount: 100 },
+      req.tenantId!
     );
     res.json({ message: 'Mock event dispatched' });
   }

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

@@ -0,0 +1,24 @@
+-- Persist webhook subscriptions so they survive restarts and are visible to every API replica.
+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 CHECK (char_length(url) > 0),
+  secret TEXT NOT NULL CHECK (char_length(secret) >= 16),
+  events TEXT[] NOT NULL DEFAULT ARRAY['*']::TEXT[],
+  created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
+  updated_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
+  CONSTRAINT webhook_subscriptions_events_not_empty CHECK (cardinality(events) > 0)
+);
+
+CREATE INDEX IF NOT EXISTS idx_webhook_subscriptions_organization_id
+  ON webhook_subscriptions (organization_id);
+
+-- Dispatch matches an exact event or wildcard subscription with array containment.
+CREATE INDEX IF NOT EXISTS idx_webhook_subscriptions_events
+  ON webhook_subscriptions USING GIN (events);
+
+DROP TRIGGER IF EXISTS update_webhook_subscriptions_updated_at ON webhook_subscriptions;
+CREATE TRIGGER update_webhook_subscriptions_updated_at
+  BEFORE UPDATE ON webhook_subscriptions
+  FOR EACH ROW
+  EXECUTE FUNCTION update_updated_at_column();

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

@@ -2,28 +2,47 @@ 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';
 
 jest.mock('../../config/env.js', () => ({
   config: { JWT_SECRET: 'test-secret' },
 }));
 
-jest.mock('../../config/database.js', () => ({
-  pool: {
+jest.mock('../../config/database.js', () => {
+  const query = jest.fn();
+  const client = {
     query: jest.fn().mockResolvedValue({ rows: [] }),
-    connect: jest.fn().mockResolvedValue({
-      query: jest.fn().mockResolvedValue({}),
-      release: jest.fn(),
-    }),
-  },
-}));
+    release: jest.fn(),
+  };
+  const pool = {
+    query,
+    connect: jest.fn().mockResolvedValue(client),
+  };
+
+  return {
+    __esModule: true,
+    default: pool,
+    pool,
+  };
+});
 
 jest.mock('../../utils/logger.js', () => ({
   default: { error: jest.fn(), warn: jest.fn(), info: jest.fn(), debug: jest.fn() },
 }));
 
+interface StoredSubscription {
+  id: string;
+  organization_id: number;
+  url: string;
+  secret: string;
+  events: string[];
+  created_at: Date;
+}
+
+let storedSubscriptions: StoredSubscription[] = [];
+
 function makeToken(payload: object): string {
   return jwt.sign(payload, JWT_SECRET, { expiresIn: '1h' });
 }
@@ -47,12 +66,78 @@ app.use(express.json());
 app.use('/webhooks', webhookRoutes);
 
 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(() => {
+    storedSubscriptions = [];
+    jest.clearAllMocks();
+
+    (pool.query as jest.Mock).mockImplementation(async (sql: string, params: any[] = []) => {
+      const normalized = sql.replace(/\s+/g, ' ').trim();
+
+      if (normalized.startsWith('INSERT INTO webhook_subscriptions')) {
+        const row: StoredSubscription = {
+          id: params[0],
+          organization_id: params[1],
+          url: params[2],
+          secret: params[3],
+          events: params[4],
+          created_at: new Date(),
+        };
+        storedSubscriptions.push(row);
+        return { rows: [row], rowCount: 1 };
+      }
+
+      if (
+        normalized.startsWith('SELECT id, url, secret, events, organization_id') &&
+        normalized.includes('WHERE organization_id = $1')
+      ) {
+        return {
+          rows: storedSubscriptions.filter((row) => row.organization_id === params[0]),
+          rowCount: 0,
+        };
+      }
+
+      if (normalized.startsWith('UPDATE webhook_subscriptions')) {
+        const organizationId = params[params.length - 1];
+        const id = params[params.length - 2];
+        const row = storedSubscriptions.find(
+          (candidate) => candidate.id === id && candidate.organization_id === organizationId
+        );
+
+        if (!row) return { rows: [], rowCount: 0 };
+
+        let valueIndex = 0;
+        if (normalized.includes('url = $')) row.url = params[valueIndex++];
+        if (normalized.includes('secret = $')) row.secret = params[valueIndex++];
+        if (normalized.includes('events = $')) row.events = params[valueIndex++];
+
+        return { rows: [row], rowCount: 1 };
+      }
+
+      if (normalized.startsWith('DELETE FROM webhook_subscriptions')) {
+        const [id, organizationId] = params;
+        const before = storedSubscriptions.length;
+        storedSubscriptions = storedSubscriptions.filter(
+          (row) => !(row.id === id && row.organization_id === organizationId)
+        );
+        return { rows: [], rowCount: before - storedSubscriptions.length };
+      }
+
+      if (
+        normalized.startsWith('SELECT id, url, secret, events, organization_id') &&
+        normalized.includes('WHERE organization_id = $2')
+      ) {
+        return {
+          rows: storedSubscriptions.filter(
+            (row) => row.organization_id === params[1] &&
+              (row.events.includes(params[0]) || row.events.includes('*'))
+          ),
+          rowCount: 0,
+        };
+      }
+
+      // Tenant access logging and unrelated middleware queries are not under test here.
+      return { rows: [], rowCount: 0 };
+    });
   });
 
   describe('Authentication required', () => {
@@ -91,7 +176,7 @@ describe('Webhook Routes - Auth and Tenant Isolation', () => {
     });
   });
 
-  describe('Tenant isolation', () => {
+  describe('Tenant isolation and CRUD', () => {
     it('scopes subscriptions to the creating tenant', async () => {
       const resA = await request(app)
         .post('/webhooks/subscribe')
@@ -100,6 +185,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 +198,26 @@ 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 () => {
+      storedSubscriptions.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].id).toBe('00000000-0000-4000-8000-000000000001');
+      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')
@@ -124,6 +232,44 @@ describe('Webhook Routes - Auth and Tenant Isolation', () => {
       expect(listB.body).toHaveLength(0);
     });
 
+    it('tenant can update their own subscription', async () => {
+      const createRes = await request(app)
+        .post('/webhooks/subscribe')
+        .set('Authorization', `Bearer ${tenantAToken}`)
+        .send({
+          url: 'https://orgA.example.com/hook',
+          secret: 'a'.repeat(16),
+          events: ['payment.completed'],
+        });
+
+      const updateRes = await request(app)
+        .patch(`/webhooks/subscriptions/${createRes.body.id}`)
+        .set('Authorization', `Bearer ${tenantAToken}`)
+        .send({
+          url: 'https://orgA.example.com/replacement',
+          events: ['payment.failed'],
+        });
+
+      expect(updateRes.status).toBe(200);
+      expect(updateRes.body.url).toBe('https://orgA.example.com/replacement');
+      expect(updateRes.body.events).toEqual(['payment.failed']);
+      expect(updateRes.body.organizationId).toBe(10);
+    });
+
+    it('tenant B cannot update tenant A subscription by ID', async () => {
+      const createRes = await request(app)
+        .post('/webhooks/subscribe')
+        .set('Authorization', `Bearer ${tenantAToken}`)
+        .send({ url: 'https://orgA.example.com/hook', secret: 'a'.repeat(16), events: ['*'] });
+
+      const updateRes = await request(app)
+        .patch(`/webhooks/subscriptions/${createRes.body.id}`)
+        .set('Authorization', `Bearer ${tenantBToken}`)
+        .send({ url: 'https://orgB.example.com/steal' });
+
+      expect(updateRes.status).toBe(404);
+    });
+
     it('tenant B cannot delete tenant A subscription by ID', async () => {
       const createRes = await request(app)
         .post('/webhooks/subscribe')
@@ -186,6 +332,26 @@ describe('Webhook Routes - Auth and Tenant Isolation', () => {
       expect(res.body.message).toBe('Mock event dispatched');
     });
 
+    it('never selects another organization wildcard subscription for a tenant event', async () => {
+      process.env.NODE_ENV = 'development';
+
+      const created = await request(app)
+        .post('/webhooks/subscribe')
+        .set('Authorization', `Bearer ${tenantBToken}`)
+        .send({ url: 'https://orgB.example.com/hook', secret: 'b'.repeat(16), events: ['*'] });
+      expect(created.status).toBe(201);
+
+      const triggered = await request(app)
+        .post('/webhooks/test-trigger')
+        .set('Authorization', `Bearer ${tenantAToken}`)
+        .send({ event: 'payment.completed', payload: { id: 'private-tenant-A' } });
+      expect(triggered.status).toBe(200);
+      expect(pool.query).toHaveBeenCalledWith(
+        expect.stringContaining('WHERE organization_id = $2'),
+        ['payment.completed', 10]
+      );
+    });
+
     it('blocks test-trigger in production', async () => {
       process.env.NODE_ENV = 'production';
 

backend/src/routes/webhook.routes.ts

@@ -13,6 +13,7 @@ router.use(logTenantAccess);
 
 router.post('/subscribe', WebhookController.subscribe);
 router.get('/subscriptions', WebhookController.listSubscriptions);
+router.patch('/subscriptions/:id', WebhookController.updateSubscription);
 router.delete('/subscriptions/:id', WebhookController.deleteSubscription);
 
 const requireNonProduction = (req: Request, res: Response, next: NextFunction) => {

backend/src/services/forecasting/liquidityAlertChecker.ts

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