diff --git a/backend/src/controllers/webhook.controller.ts b/backend/src/controllers/webhook.controller.ts index 0976db09..7eb67818 100644 --- a/backend/src/controllers/webhook.controller.ts +++ b/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 } ); diff --git a/backend/src/db/migrations/032_create_webhook_subscriptions.sql b/backend/src/db/migrations/032_create_webhook_subscriptions.sql new file mode 100644 index 00000000..f8ca7721 --- /dev/null +++ b/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.'; diff --git a/backend/src/routes/__tests__/webhook.routes.test.ts b/backend/src/routes/__tests__/webhook.routes.test.ts index 0cc8f491..8ee8f78c 100644 --- a/backend/src/routes/__tests__/webhook.routes.test.ts +++ b/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') diff --git a/backend/src/services/forecasting/liquidityAlertChecker.ts b/backend/src/services/forecasting/liquidityAlertChecker.ts index 600a84ef..88a119a3 100644 --- a/backend/src/services/forecasting/liquidityAlertChecker.ts +++ b/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 || []; diff --git a/backend/src/services/webhook.service.ts b/backend/src/services/webhook.service.ts index d8208f38..75c30881 100644 --- a/backend/src/services/webhook.service.ts +++ b/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 { - 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( + `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 { + const result = await pool.query( + `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 { + 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 { - const relevantSubscriptions = subscriptions.filter( + static async dispatch( + organizationId: number, + eventType: string, + payload: any + ): Promise { + const relevantSubscriptions = (await this.listSubscriptions(organizationId)).filter( (s) => s.events.includes(eventType) || s.events.includes('*') );