feat: webhook circuit breaker

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<void> {
+    await queryRunner.query(`ALTER TABLE "webhooks" ADD COLUMN IF NOT EXISTS "circuitOpenUntil" timestamptz`);
+  }
+
+  async down(queryRunner: QueryRunner): Promise<void> {
+    await queryRunner.query(`ALTER TABLE "webhooks" DROP COLUMN IF EXISTS "circuitOpenUntil"`);
+  }
+}

src/email/templates/webhook-circuit-open.html.hbs

@@ -0,0 +1 @@
+<p>Hello {{businessName}},</p><p>Your webhook endpoint {{url}} failed {{failureCount}} times in a row, so we have paused deliveries to it until {{reopensAt}}.</p><p>Last error: {{lastError}}</p><p>Please fix the endpoint. Deliveries resume automatically after the pause, or sooner if our support team resets it.</p>

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.

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;
 

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);
+  }
+}

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<WebhookDeliveryJobData>): Promise<void> {
     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<boolean> {
     // 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<boolean> {
+    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;
   }
 }

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';
 import { MerchantsModule } from '../merchants/merchants.module';
 import { NotificationWebhookService } from './notification-webhook.service';
 import { NotificationWebhookController } from './notification-webhook.controller';
@@ -30,9 +32,10 @@ import { PaymentNotificationSubscriber } from './payment-notification.subscriber
     NotificationsModule,
     CacheModule,
     PrometheusModule,
+    EmailModule,
     MerchantsModule,
   ],
-  controllers: [WebhooksController, NotificationWebhookController],
+  controllers: [WebhooksController, WebhookCircuitAdminController, NotificationWebhookController],
   providers: [WebhooksService, WebhookDeliveryProcessor, WebhookDeliveryService, WebhookFailureAlertService, QueueConfigService, NotificationWebhookService, PaymentNotificationSubscriber],
   exports: [WebhooksService, WebhookDeliveryService, NotificationWebhookService],
 })

src/webhooks/webhooks.service.ts

@@ -30,6 +30,17 @@ export class WebhooksService {
     private merchantsRepo?: Repository<Merchant>,
   ) {}
 
+  /** Admin: close an open webhook circuit and clear the failure streak (#457). */
+  async resetCircuit(webhookId: string): Promise<Webhook> {
+    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 } });