diff --git a/backend/src/controllers/webhook.controller.ts b/backend/src/controllers/webhook.controller.ts index 0976db09..a5048880 100644 --- a/backend/src/controllers/webhook.controller.ts +++ b/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' }); } 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..cae51aaa --- /dev/null +++ b/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(); diff --git a/backend/src/routes/__tests__/webhook.routes.test.ts b/backend/src/routes/__tests__/webhook.routes.test.ts index 0cc8f491..7d18d127 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'; @@ -10,20 +10,39 @@ 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'; diff --git a/backend/src/routes/webhook.routes.ts b/backend/src/routes/webhook.routes.ts index c4eabf25..df0274f5 100644 --- a/backend/src/routes/webhook.routes.ts +++ b/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) => { diff --git a/backend/src/services/forecasting/liquidityAlertChecker.ts b/backend/src/services/forecasting/liquidityAlertChecker.ts index 600a84ef..2eda2dfd 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('liquidity.insufficient', payload, organizationId); } const alertEmails = liquiditySettings.alertEmails || []; diff --git a/backend/src/services/webhook.service.ts b/backend/src/services/webhook.service.ts index d8208f38..cfda87c0 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 { v4 as uuidv4 } from 'uuid'; +import pool from '../config/database.js'; export interface WebhookSubscription { id: string; @@ -9,8 +11,29 @@ 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; +} + +interface WebhookSubscriptionUpdates { + url?: string; + secret?: string; + events?: string[]; +} + +function toSubscription(row: WebhookSubscriptionRow): WebhookSubscription { + return { + id: row.id, + url: row.url, + secret: row.secret, + events: row.events, + organizationId: row.organization_id, + }; +} export class WebhookService { static async subscribe( @@ -19,36 +42,94 @@ 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 = uuidv4(); + 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] + ); - static listSubscriptions(organizationId: number): WebhookSubscription[] { - return subscriptions.filter((s) => s.organizationId === organizationId); + return toSubscription(result.rows[0]); } - static deleteSubscription(id: string, organizationId: number): boolean { - const index = subscriptions.findIndex( - (s) => s.id === id && 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] ); - if (index !== -1) { - subscriptions.splice(index, 1); - return true; + + return result.rows.map(toSubscription); + } + + static async updateSubscription( + id: string, + organizationId: number, + updates: WebhookSubscriptionUpdates + ): Promise { + const assignments: string[] = []; + const values: unknown[] = []; + + if (updates.url !== undefined) { + values.push(updates.url); + assignments.push(`url = $${values.length}`); + } + if (updates.secret !== undefined) { + values.push(updates.secret); + assignments.push(`secret = $${values.length}`); } - return false; + if (updates.events !== undefined) { + values.push(updates.events); + assignments.push(`events = $${values.length}::text[]`); + } + + if (assignments.length === 0) { + return null; + } + + values.push(id, organizationId); + const idParameter = values.length - 1; + const organizationParameter = values.length; + + const result = await pool.query( + `UPDATE webhook_subscriptions + SET ${assignments.join(', ')} + WHERE id = $${idParameter} AND organization_id = $${organizationParameter} + RETURNING id, url, secret, events, organization_id`, + values + ); + + return result.rows[0] ? toSubscription(result.rows[0]) : null; } - static async dispatch(eventType: string, payload: any): Promise { - const relevantSubscriptions = subscriptions.filter( - (s) => s.events.includes(eventType) || s.events.includes('*') + static async deleteSubscription(id: string, organizationId: number): Promise { + const result = await pool.query( + `DELETE FROM webhook_subscriptions + WHERE id = $1 AND organization_id = $2`, + [id, organizationId] + ); + + return (result.rowCount ?? 0) > 0; + } + + static async dispatch(eventType: string, payload: any, organizationId: number): Promise { + // A subscription belonging to another employer must never receive this + // organization's financial or payment events, including wildcard events. + if (!Number.isSafeInteger(organizationId) || organizationId <= 0) { + throw new Error('Verified organization scope required for webhook delivery'); + } + const result = await pool.query( + `SELECT id, url, secret, events, organization_id + FROM webhook_subscriptions + WHERE organization_id = $2 + AND (events @> ARRAY[$1]::text[] + OR events @> ARRAY['*']::text[])`, + [eventType, organizationId] ); + const relevantSubscriptions = result.rows.map(toSubscription); const dispatchPromises = relevantSubscriptions.map(async (sub) => { const timestamp = Date.now().toString();