From 42034f3ab43effa32584fb08d2f90ad4cbec35f2 Mon Sep 17 00:00:00 2001 From: MJigah <56768247+MJigah@users.noreply.github.com> Date: Thu, 22 Oct 2026 13:32:47 +0100 Subject: [PATCH] feat: add webhook circuit breaker After 5 consecutive delivery failures the circuit opens for the webhook endpoint for 24h: deliveries are skipped and no retries are scheduled. Opening the circuit raises an admin alert and emails the merchant. Admins can reset an open circuit with POST /admin/webhooks/:id/circuit/reset. Closes #457 --- ...792454600000-AddWebhookCircuitOpenUntil.ts | 13 +++ .../templates/webhook-circuit-open.html.hbs | 1 + .../templates/webhook-circuit-open.text.hbs | 6 ++ src/webhooks/entities/webhook.entity.ts | 4 + .../webhook-circuit-admin.controller.ts | 25 ++++++ src/webhooks/webhook-delivery.processor.ts | 79 +++++++++++++++++-- src/webhooks/webhooks.module.ts | 5 +- src/webhooks/webhooks.service.ts | 11 +++ 8 files changed, 137 insertions(+), 7 deletions(-) create mode 100644 src/database/migrations/1792454600000-AddWebhookCircuitOpenUntil.ts create mode 100644 src/email/templates/webhook-circuit-open.html.hbs create mode 100644 src/email/templates/webhook-circuit-open.text.hbs create mode 100644 src/webhooks/webhook-circuit-admin.controller.ts diff --git a/src/database/migrations/1792454600000-AddWebhookCircuitOpenUntil.ts b/src/database/migrations/1792454600000-AddWebhookCircuitOpenUntil.ts new file mode 100644 index 00000000..ed2df39f --- /dev/null +++ b/src/database/migrations/1792454600000-AddWebhookCircuitOpenUntil.ts @@ -0,0 +1,13 @@ +import { MigrationInterface, QueryRunner } from 'typeorm'; + +export class AddWebhookCircuitOpenUntil1792454600000 implements MigrationInterface { + name = 'AddWebhookCircuitOpenUntil1792454600000'; + + async up(queryRunner: QueryRunner): Promise { + await queryRunner.query(`ALTER TABLE "webhooks" ADD COLUMN IF NOT EXISTS "circuitOpenUntil" timestamptz`); + } + + async down(queryRunner: QueryRunner): Promise { + await queryRunner.query(`ALTER TABLE "webhooks" DROP COLUMN IF EXISTS "circuitOpenUntil"`); + } +} diff --git a/src/email/templates/webhook-circuit-open.html.hbs b/src/email/templates/webhook-circuit-open.html.hbs new file mode 100644 index 00000000..cf18ae7e --- /dev/null +++ b/src/email/templates/webhook-circuit-open.html.hbs @@ -0,0 +1 @@ +

Hello {{businessName}},

Your webhook endpoint {{url}} failed {{failureCount}} times in a row, so we have paused deliveries to it until {{reopensAt}}.

Last error: {{lastError}}

Please fix the endpoint. Deliveries resume automatically after the pause, or sooner if our support team resets it.

\ No newline at end of file diff --git a/src/email/templates/webhook-circuit-open.text.hbs b/src/email/templates/webhook-circuit-open.text.hbs new file mode 100644 index 00000000..1597b877 --- /dev/null +++ b/src/email/templates/webhook-circuit-open.text.hbs @@ -0,0 +1,6 @@ +Hello {{businessName}}, + +Your webhook endpoint {{url}} failed {{failureCount}} times in a row, so we have paused deliveries to it until {{reopensAt}}. +Last error: {{lastError}} + +Please fix the endpoint. Deliveries resume automatically after the pause, or sooner if our support team resets it. diff --git a/src/webhooks/entities/webhook.entity.ts b/src/webhooks/entities/webhook.entity.ts index e7d6d462..5ce8ec33 100644 --- a/src/webhooks/entities/webhook.entity.ts +++ b/src/webhooks/entities/webhook.entity.ts @@ -48,6 +48,10 @@ export class Webhook { @Column({ nullable: true }) lastDeliveredAt: Date; + /** While in the future the circuit is open and deliveries are skipped (#457). */ + @Column({ type: 'timestamptz', nullable: true }) + circuitOpenUntil: Date | null; + @CreateDateColumn() createdAt: Date; diff --git a/src/webhooks/webhook-circuit-admin.controller.ts b/src/webhooks/webhook-circuit-admin.controller.ts new file mode 100644 index 00000000..40660503 --- /dev/null +++ b/src/webhooks/webhook-circuit-admin.controller.ts @@ -0,0 +1,25 @@ +import { Controller, Param, ParseUUIDPipe, Post, UseGuards } from '@nestjs/common'; +import { ApiBearerAuth, ApiOperation, ApiTags } from '@nestjs/swagger'; +import { ApiErrors } from '../common/api-errors.decorator'; +import { Roles } from '../auth/decorators/roles.decorator'; +import { MerchantRole } from '../merchants/entities/merchant.entity'; +import { RolesGuard } from '../auth/guards/roles.guard'; +import { JwtAuthGuard } from '../auth/guards/jwt.guard'; +import { AdminTwoFactorGuard } from '../auth/guards/admin-two-factor.guard'; +import { WebhooksService } from './webhooks.service'; + +@ApiTags('admin / webhooks') +@ApiBearerAuth() +@UseGuards(JwtAuthGuard, AdminTwoFactorGuard, RolesGuard) +@Roles(MerchantRole.SUPERADMIN) +@ApiErrors() +@Controller('admin/webhooks') +export class WebhookCircuitAdminController { + constructor(private readonly webhooksService: WebhooksService) {} + + @Post(':id/circuit/reset') + @ApiOperation({ summary: 'Manually reset an open webhook circuit' }) + async resetCircuit(@Param('id', ParseUUIDPipe) id: string) { + return this.webhooksService.resetCircuit(id); + } +} diff --git a/src/webhooks/webhook-delivery.processor.ts b/src/webhooks/webhook-delivery.processor.ts index 3bd027c3..b89ca215 100644 --- a/src/webhooks/webhook-delivery.processor.ts +++ b/src/webhooks/webhook-delivery.processor.ts @@ -16,6 +16,12 @@ import { WebhookDeliveryLog } from "./entities/webhook-delivery-log.entity"; import { Merchant } from "../merchants/entities/merchant.entity"; import { WebhookFailureAlertService } from "./webhook-failure-alert.service"; import { MetricsService } from '../prometheus/metrics.service'; +import { AdminAlertService } from '../alerts/admin-alert.service'; +import { AdminAlertType } from '../alerts/admin-alert.entity'; +import { EmailService } from '../email/email.service'; + +export const WEBHOOK_CIRCUIT_FAILURE_THRESHOLD = 5; +export const WEBHOOK_CIRCUIT_OPEN_MS = 24 * 60 * 60 * 1000; interface WebhookDeliveryJobData { webhookId: string; @@ -46,12 +52,24 @@ export class WebhookDeliveryProcessor { private readonly queueConfig: QueueConfigService, private readonly config: ConfigService, private readonly failureAlertService: WebhookFailureAlertService, + private readonly adminAlerts: AdminAlertService, + private readonly emailService: EmailService, @Optional() private readonly metrics?: MetricsService, ) {} @Process(WEBHOOK_DELIVERY_JOB) async process(job: Job): Promise { const data = job.data; + + // Circuit breaker (#457): skip delivery while the endpoint's circuit is open. + const current = await this.webhookRepo.findOne({ where: { id: data.webhookId } }); + if (current?.circuitOpenUntil && new Date(current.circuitOpenUntil).getTime() > Date.now()) { + this.logger.warn( + `Webhook circuit open for ${data.url} until ${new Date(current.circuitOpenUntil).toISOString()}; skipping delivery.`, + ); + return; + } + const startedAt = Date.now(); const requestBody = data.body; const signature = this.sign(requestBody, data.secret); @@ -108,9 +126,11 @@ export class WebhookDeliveryProcessor { errorMessage, retryDelay, ); - await this.markWebhookFailure(data.webhookId, errorMessage); + const circuitOpened = await this.markWebhookFailure(data.webhookId, errorMessage); - if (attemptNumber < maxAttempts) { + if (circuitOpened) { + this.logger.warn(`Webhook circuit opened for ${data.url}; no further retries scheduled.`); + } else if (attemptNumber < maxAttempts) { const retryPayload = { ...data, attemptNumber: attemptNumber + 1 }; const retryHash = crypto.createHash('sha256').update(`${data.webhookId}:${data.event}:${retryPayload.attemptNumber}:${data.body}`).digest('hex').slice(0, 32); @@ -164,11 +184,11 @@ export class WebhookDeliveryProcessor { // Atomic write — avoids a read-modify-write race with concurrent deliveries. await this.webhookRepo.update( { id: webhookId }, - { failureCount: 0, lastDeliveredAt: new Date() }, + { failureCount: 0, lastDeliveredAt: new Date(), circuitOpenUntil: null }, ); } - private async markWebhookFailure(webhookId: string, lastError: string) { + private async markWebhookFailure(webhookId: string, lastError: string): Promise { // Atomic increment (UPDATE ... SET failureCount = failureCount + 1) so that // concurrent deliveries for the same webhook can't lose an increment to a // last-write-wins race and under-count towards the auto-deactivation threshold. @@ -177,12 +197,12 @@ export class WebhookDeliveryProcessor { 'failureCount', 1, ); - if (!affected) return; + if (!affected) return false; const webhook = await this.webhookRepo.findOne({ where: { id: webhookId }, }); - if (!webhook) return; + if (!webhook) return false; if (webhook.failureCount >= 10 && webhook.isActive) { await this.webhookRepo.update({ id: webhookId }, { isActive: false }); @@ -197,5 +217,52 @@ export class WebhookDeliveryProcessor { const settingsUrl = `${this.config.get('FRONTEND_URL', '')}/dashboard/settings/webhooks`; await this.failureAlertService.notifyIfNeeded(webhook, merchant.email, lastError, settingsUrl); } + + return this.openCircuitIfNeeded(webhook, lastError); + } + + /** Opens the circuit after N consecutive failures; alerts admins and emails the merchant. */ + private async openCircuitIfNeeded(webhook: Webhook, lastError: string): Promise { + const alreadyOpen = + webhook.circuitOpenUntil && new Date(webhook.circuitOpenUntil).getTime() > Date.now(); + if (webhook.failureCount < WEBHOOK_CIRCUIT_FAILURE_THRESHOLD || alreadyOpen) { + return false; + } + + const reopensAt = new Date(Date.now() + WEBHOOK_CIRCUIT_OPEN_MS); + await this.webhookRepo.update({ id: webhook.id }, { circuitOpenUntil: reopensAt }); + + await this.adminAlerts.raise({ + type: AdminAlertType.WEBHOOK_FAILURE, + dedupeKey: `webhook-circuit:${webhook.id}`, + message: `Webhook circuit opened for merchant ${webhook.merchantId} endpoint ${webhook.url} after ${webhook.failureCount} consecutive failures.`, + metadata: { + webhookId: webhook.id, + merchantId: webhook.merchantId, + url: webhook.url, + reopensAt: reopensAt.toISOString(), + lastError, + }, + thresholdValue: webhook.failureCount, + }); + + const merchant = await this.merchantRepo.findOne({ where: { id: webhook.merchantId } }); + if (merchant?.email) { + await this.emailService.queue( + merchant.email, + 'webhook-circuit-open', + { + subject: 'Webhook deliveries paused', + businessName: merchant.businessName ?? '', + url: webhook.url, + failureCount: webhook.failureCount, + reopensAt: reopensAt.toISOString(), + lastError, + }, + merchant.id, + ); + } + + return true; } } diff --git a/src/webhooks/webhooks.module.ts b/src/webhooks/webhooks.module.ts index c1ff443c..936456e5 100644 --- a/src/webhooks/webhooks.module.ts +++ b/src/webhooks/webhooks.module.ts @@ -16,6 +16,8 @@ import { Merchant } from '../merchants/entities/merchant.entity'; import { NotificationsModule } from '../notifications/notifications.module'; import { CacheModule } from '../cache/cache.module'; import { PrometheusModule } from '../prometheus/prometheus.module'; +import { EmailModule } from '../email/email.module'; +import { WebhookCircuitAdminController } from './webhook-circuit-admin.controller'; @Module({ imports: [ @@ -26,8 +28,9 @@ import { PrometheusModule } from '../prometheus/prometheus.module'; NotificationsModule, CacheModule, PrometheusModule, + EmailModule, ], - controllers: [WebhooksController], + controllers: [WebhooksController, WebhookCircuitAdminController], providers: [WebhooksService, WebhookDeliveryProcessor, WebhookDeliveryService, WebhookFailureAlertService, QueueConfigService], exports: [WebhooksService, WebhookDeliveryService], }) diff --git a/src/webhooks/webhooks.service.ts b/src/webhooks/webhooks.service.ts index b0fdde6a..b106173e 100644 --- a/src/webhooks/webhooks.service.ts +++ b/src/webhooks/webhooks.service.ts @@ -30,6 +30,17 @@ export class WebhooksService { private merchantsRepo?: Repository, ) {} + /** Admin: close an open webhook circuit and clear the failure streak (#457). */ + async resetCircuit(webhookId: string): Promise { + const webhook = await this.webhooksRepo.findOne({ where: { id: webhookId } }); + if (!webhook) { + throw new NotFoundException('Webhook not found'); + } + webhook.circuitOpenUntil = null; + webhook.failureCount = 0; + return this.webhooksRepo.save(webhook); + } + async updateRetryConfig(merchantId: string, config: UpdateWebhookRetryConfigDto) { if (!this.merchantsRepo) throw new Error('Merchant repository is unavailable'); const merchant = await this.merchantsRepo.findOne({ where: { id: merchantId } });