diff --git a/src/webhooks/dto/create-webhook.dto.ts b/src/webhooks/dto/create-webhook.dto.ts new file mode 100644 index 0000000..4e4e3b5 --- /dev/null +++ b/src/webhooks/dto/create-webhook.dto.ts @@ -0,0 +1,83 @@ +import { + IsString, + IsUrl, + IsOptional, + IsArray, + IsBoolean, + IsNumber, + Min, + Max, + ArrayNotEmpty, +} from 'class-validator'; +import { ApiProperty, ApiPropertyOptional } from '@nestjs/swagger'; +import { WebhookEventType } from '../entities/webhook.entity'; + +export class CreateWebhookDto { + @ApiProperty({ + description: 'HTTPS endpoint URL that will receive webhook events', + example: 'https://api.example.com/webhooks/truthbounty', + }) + @IsUrl({ protocols: ['https'], require_tld: false }) + @IsString() + url: string; + + @ApiPropertyOptional({ + description: 'Human-readable description of this webhook', + example: 'Production analytics dashboard', + }) + @IsOptional() + @IsString() + description?: string; + + @ApiProperty({ + description: 'Blockchain wallet address that owns this webhook', + example: '0x1234567890abcdef1234567890abcdef12345678', + }) + @IsString() + ownerId: string; + + @ApiPropertyOptional({ + description: 'Whether the webhook is active (default: true)', + default: true, + }) + @IsOptional() + @IsBoolean() + enabled?: boolean; + + @ApiProperty({ + description: 'Event types to subscribe to', + example: ['claim.created', 'verification.completed', 'reward.distributed'], + enum: WebhookEventType, + isArray: true, + }) + @IsArray() + @ArrayNotEmpty() + @IsString({ each: true }) + events: WebhookEventType[]; + + @ApiPropertyOptional({ + description: 'Optional JSON filters for event payload filtering', + example: { network: 'mainnet', severity: 'high' }, + }) + @IsOptional() + filters?: Record; + + @ApiPropertyOptional({ + description: 'Maximum number of retry attempts for failed deliveries (default: 3)', + default: 3, + }) + @IsOptional() + @IsNumber() + @Min(0) + @Max(10) + maxRetries?: number; + + @ApiPropertyOptional({ + description: 'Base retry interval in milliseconds (default: 30000)', + default: 30000, + }) + @IsOptional() + @IsNumber() + @Min(1000) + retryIntervalMs?: number; +} diff --git a/src/webhooks/dto/update-webhook.dto.ts b/src/webhooks/dto/update-webhook.dto.ts new file mode 100644 index 0000000..36ee6f7 --- /dev/null +++ b/src/webhooks/dto/update-webhook.dto.ts @@ -0,0 +1,72 @@ +import { + IsString, + IsUrl, + IsOptional, + IsArray, + IsBoolean, + IsNumber, + Min, + Max, +} from 'class-validator'; +import { ApiPropertyOptional } from '@nestjs/swagger'; +import { WebhookEventType } from '../entities/webhook.entity'; + +export class UpdateWebhookDto { + @ApiPropertyOptional({ + description: 'HTTPS endpoint URL that will receive webhook events', + example: 'https://api.example.com/webhooks/truthbounty', + }) + @IsOptional() + @IsUrl({ protocols: ['https'], require_tld: false }) + @IsString() + url?: string; + + @ApiPropertyOptional({ + description: 'Human-readable description of this webhook', + }) + @IsOptional() + @IsString() + description?: string; + + @ApiPropertyOptional({ + description: 'Whether the webhook is active', + }) + @IsOptional() + @IsBoolean() + enabled?: boolean; + + @ApiPropertyOptional({ + description: 'Event types to subscribe to', + enum: WebhookEventType, + isArray: true, + }) + @IsOptional() + @IsArray() + @IsString({ each: true }) + events?: WebhookEventType[]; + + @ApiPropertyOptional({ + description: 'Optional JSON filters for event payload filtering', + }) + @IsOptional() + filters?: Record; + + @ApiPropertyOptional({ + description: 'Maximum number of retry attempts for failed deliveries', + default: 3, + }) + @IsOptional() + @IsNumber() + @Min(0) + @Max(10) + maxRetries?: number; + + @ApiPropertyOptional({ + description: 'Base retry interval in milliseconds', + default: 30000, + }) + @IsOptional() + @IsNumber() + @Min(1000) + retryIntervalMs?: number; +} diff --git a/src/webhooks/dto/webhook-filter.dto.ts b/src/webhooks/dto/webhook-filter.dto.ts new file mode 100644 index 0000000..5986230 --- /dev/null +++ b/src/webhooks/dto/webhook-filter.dto.ts @@ -0,0 +1,55 @@ +import { IsOptional, IsString, IsEnum, IsNumber, Min } from 'class-validator'; +import { ApiPropertyOptional } from '@nestjs/swagger'; +import { DeliveryStatus } from '../entities/webhook-delivery.entity'; + +export class WebhookDeliveryFilterDto { + @ApiPropertyOptional({ + description: 'Filter by event type', + example: 'claim.created', + }) + @IsOptional() + @IsString() + eventType?: string; + + @ApiPropertyOptional({ + description: 'Filter by delivery status', + enum: DeliveryStatus, + }) + @IsOptional() + @IsEnum(DeliveryStatus) + status?: DeliveryStatus; + + @ApiPropertyOptional({ + description: 'Page number (1-based)', + default: 1, + }) + @IsOptional() + @IsNumber() + @Min(1) + page?: number; + + @ApiPropertyOptional({ + description: 'Items per page', + default: 20, + }) + @IsOptional() + @IsNumber() + @Min(1) + limit?: number; +} + +export class WebhookListFilterDto { + @ApiPropertyOptional({ + description: 'Filter by enabled/disabled status', + }) + @IsOptional() + @IsString() + enabled?: string; + + @ApiPropertyOptional({ + description: 'Filter by owner wallet address', + }) + @IsOptional() + @IsString() + ownerId?: string; +} diff --git a/src/webhooks/entities/webhook-delivery.entity.ts b/src/webhooks/entities/webhook-delivery.entity.ts new file mode 100644 index 0000000..b5d2fa3 --- /dev/null +++ b/src/webhooks/entities/webhook-delivery.entity.ts @@ -0,0 +1,79 @@ +import { + Entity, + PrimaryGeneratedColumn, + Column, + CreateDateColumn, + Index, + ManyToOne, + JoinColumn, +} from 'typeorm'; +import { Webhook } from './webhook.entity'; + +export enum DeliveryStatus { + PENDING = 'PENDING', + DELIVERED = 'DELIVERED', + FAILED = 'FAILED', + DEAD_LETTER = 'DEAD_LETTER', +} + +@Entity('webhook_deliveries') +@Index(['webhookId']) +@Index(['status']) +@Index(['createdAt']) +@Index(['webhookId', 'status']) +export class WebhookDelivery { + @PrimaryGeneratedColumn('uuid') + id: string; + + @Column() + webhookId: string; + + @ManyToOne(() => Webhook, (webhook) => webhook.deliveries, { onDelete: 'CASCADE' }) + @JoinColumn({ name: 'webhookId' }) + webhook: Webhook; + + @Column() + eventType: string; + + @Column({ type: 'json' }) + payload: Record; + + @Column({ type: 'varchar', default: DeliveryStatus.PENDING }) + status: DeliveryStatus; + + @Column({ nullable: true }) + responseStatus: number; + + @Column({ type: 'text', nullable: true }) + responseBody: string; + + @Column({ nullable: true }) + latency: number; + + @Column({ default: 0 }) + retryCount: number; + + @Column() + maxRetries: number; + + @Column({ type: 'text', nullable: true }) + failureReason: string | null; + + @Column() + requestId: string; + + @Column() + nonce: string; + + @Column() + signature: string; + + @Column() + timestamp: string; + + @CreateDateColumn() + createdAt: Date; + + @Column({ type: 'datetime', nullable: true }) + completedAt: Date | null; +} diff --git a/src/webhooks/entities/webhook-subscription.entity.ts b/src/webhooks/entities/webhook-subscription.entity.ts new file mode 100644 index 0000000..2b70116 --- /dev/null +++ b/src/webhooks/entities/webhook-subscription.entity.ts @@ -0,0 +1,35 @@ +import { + Entity, + PrimaryGeneratedColumn, + Column, + CreateDateColumn, + Index, + ManyToOne, + JoinColumn, +} from 'typeorm'; +import { Webhook } from './webhook.entity'; + +@Entity('webhook_subscriptions') +@Index(['webhookId']) +@Index(['eventType']) +@Index(['webhookId', 'eventType'], { unique: true }) +export class WebhookSubscription { + @PrimaryGeneratedColumn('uuid') + id: string; + + @Column() + webhookId: string; + + @ManyToOne(() => Webhook, (webhook) => webhook.subscriptions, { onDelete: 'CASCADE' }) + @JoinColumn({ name: 'webhookId' }) + webhook: Webhook; + + @Column() + eventType: string; + + @Column({ type: 'json', nullable: true }) + filters: Record | null; + + @CreateDateColumn() + createdAt: Date; +} diff --git a/src/webhooks/entities/webhook.entity.ts b/src/webhooks/entities/webhook.entity.ts new file mode 100644 index 0000000..c31489b --- /dev/null +++ b/src/webhooks/entities/webhook.entity.ts @@ -0,0 +1,84 @@ +import { + Entity, + PrimaryGeneratedColumn, + Column, + CreateDateColumn, + UpdateDateColumn, + Index, + OneToMany, +} from 'typeorm'; +import { WebhookSubscription } from './webhook-subscription.entity'; +import { WebhookDelivery } from './webhook-delivery.entity'; + +export enum WebhookEventType { + CLAIM_CREATED = 'claim.created', + CLAIM_UPDATED = 'claim.updated', + VERIFICATION_STARTED = 'verification.started', + VERIFICATION_COMPLETED = 'verification.completed', + DISPUTE_CREATED = 'dispute.created', + DISPUTE_RESOLVED = 'dispute.resolved', + REWARD_DISTRIBUTED = 'reward.distributed', + GOVERNANCE_CREATED = 'governance.created', + GOVERNANCE_EXECUTED = 'governance.executed', + REPUTATION_UPDATED = 'reputation.updated', + TREASURY_UPDATED = 'treasury.updated', + STAKING_UPDATED = 'staking.updated', +} + +export const ALL_WEBHOOK_EVENTS: WebhookEventType[] = Object.values(WebhookEventType); + +@Entity('webhooks') +@Index(['ownerId']) +@Index(['enabled']) +export class Webhook { + @PrimaryGeneratedColumn('uuid') + id: string; + + @Column() + url: string; + + @Column({ nullable: true }) + description: string; + + @Column() + ownerId: string; + + @Column({ default: true }) + enabled: boolean; + + @Column() + secret: string; + + @Column({ type: 'datetime', nullable: true }) + secretExpiresAt: Date | null; + + @Column({ type: 'varchar', nullable: true }) + previousSecret: string | null; + + @Column({ type: 'datetime', nullable: true }) + previousSecretExpiresAt: Date | null; + + @Column({ default: 0 }) + consecutiveFailures: number; + + @Column({ default: 3 }) + maxRetries: number; + + @Column({ default: 30000 }) + retryIntervalMs: number; + + @Column({ default: false }) + disabled: boolean; + + @CreateDateColumn() + createdAt: Date; + + @UpdateDateColumn() + updatedAt: Date; + + @OneToMany(() => WebhookSubscription, (sub) => sub.webhook, { cascade: true }) + subscriptions: WebhookSubscription[]; + + @OneToMany(() => WebhookDelivery, (del) => del.webhook) + deliveries: WebhookDelivery[]; +} diff --git a/src/webhooks/webhooks.controller.spec.ts b/src/webhooks/webhooks.controller.spec.ts new file mode 100644 index 0000000..a0e616e --- /dev/null +++ b/src/webhooks/webhooks.controller.spec.ts @@ -0,0 +1,155 @@ +import { Test, TestingModule } from '@nestjs/testing'; +import { WebhooksController } from './webhooks.controller'; +import { WebhooksService } from './webhooks.service'; +import { CreateWebhookDto } from './dto/create-webhook.dto'; +import { UpdateWebhookDto } from './dto/update-webhook.dto'; +import { WebhookEventType } from './entities/webhook.entity'; +import { DeliveryStatus } from './entities/webhook-delivery.entity'; + +describe('WebhooksController', () => { + let controller: WebhooksController; + let service: WebhooksService; + + const mockWebhook = { + id: 'wh-001', + url: 'https://example.com/webhook', + description: 'Test webhook', + ownerId: '0x123', + enabled: true, + secret: 'hashed', + consecutiveFailures: 0, + maxRetries: 3, + retryIntervalMs: 30000, + disabled: false, + createdAt: new Date(), + updatedAt: new Date(), + subscriptions: [], + }; + + const mockService = { + create: jest.fn().mockResolvedValue(mockWebhook), + findAll: jest.fn().mockResolvedValue([mockWebhook]), + findOne: jest.fn().mockResolvedValue(mockWebhook), + update: jest.fn().mockResolvedValue(mockWebhook), + remove: jest.fn().mockResolvedValue(undefined), + getDeliveries: jest.fn().mockResolvedValue({ deliveries: [], total: 0, page: 1, limit: 20 }), + getDelivery: jest.fn().mockResolvedValue({ + id: 'del-001', + webhookId: 'wh-001', + eventType: 'claim.created', + status: DeliveryStatus.DELIVERED, + }), + retryDelivery: jest.fn().mockResolvedValue(undefined), + rotateSecret: jest.fn().mockResolvedValue({ secret: 'new-secret', expiresAt: new Date() }), + revokeSecret: jest.fn().mockResolvedValue(undefined), + getWebhookStatus: jest.fn().mockResolvedValue({ + webhook: mockWebhook, + totalDeliveries: 10, + successfulDeliveries: 8, + failedDeliveries: 2, + pendingDeliveries: 0, + lastDelivery: null, + }), + }; + + beforeEach(async () => { + const module: TestingModule = await Test.createTestingModule({ + controllers: [WebhooksController], + providers: [{ provide: WebhooksService, useValue: mockService }], + }).compile(); + + controller = module.get(WebhooksController); + service = module.get(WebhooksService); + }); + + describe('create', () => { + it('should create a webhook', async () => { + const dto: CreateWebhookDto = { + url: 'https://example.com/webhook', + ownerId: '0x123', + events: [WebhookEventType.CLAIM_CREATED], + }; + const result = await controller.create(dto); + expect(result).toEqual(mockWebhook); + expect(service.create).toHaveBeenCalledWith(dto); + }); + }); + + describe('findAll', () => { + it('should return all webhooks', async () => { + const result = await controller.findAll({}); + expect(result).toEqual([mockWebhook]); + expect(service.findAll).toHaveBeenCalled(); + }); + }); + + describe('findOne', () => { + it('should return a webhook by ID', async () => { + const result = await controller.findOne('wh-001'); + expect(result).toEqual(mockWebhook); + expect(service.findOne).toHaveBeenCalledWith('wh-001'); + }); + }); + + describe('update', () => { + it('should update a webhook', async () => { + const dto: UpdateWebhookDto = { enabled: false }; + const result = await controller.update('wh-001', dto); + expect(result).toEqual(mockWebhook); + expect(service.update).toHaveBeenCalledWith('wh-001', dto); + }); + }); + + describe('remove', () => { + it('should delete a webhook', async () => { + await controller.remove('wh-001'); + expect(service.remove).toHaveBeenCalledWith('wh-001'); + }); + }); + + describe('getDeliveries', () => { + it('should return delivery history', async () => { + const result = await controller.getDeliveries('wh-001', {}); + expect(result).toBeDefined(); + expect(service.getDeliveries).toHaveBeenCalledWith('wh-001', {}); + }); + }); + + describe('getDelivery', () => { + it('should return a delivery record', async () => { + const result = await controller.getDelivery('del-001'); + expect(result).toBeDefined(); + expect(service.getDelivery).toHaveBeenCalledWith('del-001'); + }); + }); + + describe('retryDelivery', () => { + it('should retry a failed delivery', async () => { + await controller.retryDelivery('wh-001', 'del-001'); + expect(service.retryDelivery).toHaveBeenCalledWith('del-001'); + }); + }); + + describe('rotateSecret', () => { + it('should rotate the webhook secret', async () => { + const result = await controller.rotateSecret('wh-001'); + expect(result.secret).toBe('new-secret'); + expect(service.rotateSecret).toHaveBeenCalledWith('wh-001'); + }); + }); + + describe('revokeSecret', () => { + it('should revoke the webhook secret', async () => { + await controller.revokeSecret('wh-001'); + expect(service.revokeSecret).toHaveBeenCalledWith('wh-001'); + }); + }); + + describe('getStatus', () => { + it('should return webhook status', async () => { + const result = await controller.getStatus('wh-001'); + expect(result.totalDeliveries).toBe(10); + expect(service.getWebhookStatus).toHaveBeenCalledWith('wh-001'); + }); + }); +}); diff --git a/src/webhooks/webhooks.controller.ts b/src/webhooks/webhooks.controller.ts new file mode 100644 index 0000000..65322f1 --- /dev/null +++ b/src/webhooks/webhooks.controller.ts @@ -0,0 +1,227 @@ +import { + Controller, + Get, + Post, + Patch, + Delete, + Body, + Param, + Query, + HttpCode, + HttpStatus, + UseGuards, +} from '@nestjs/common'; +import { + ApiTags, + ApiOperation, + ApiResponse, + ApiParam, + ApiQuery, + ApiBearerAuth, +} from '@nestjs/swagger'; +import { WebhooksService } from './webhooks.service'; +import { Webhook } from './entities/webhook.entity'; +import { WebhookDelivery } from './entities/webhook-delivery.entity'; +import { CreateWebhookDto } from './dto/create-webhook.dto'; +import { UpdateWebhookDto } from './dto/update-webhook.dto'; +import { + WebhookDeliveryFilterDto, + WebhookListFilterDto, +} from './dto/webhook-filter.dto'; +import { JwtAuthGuard } from '../auth/jwt-auth.guard'; + +@ApiTags('webhooks') +@Controller('webhooks') +@UseGuards(JwtAuthGuard) +@ApiBearerAuth('JWT-auth') +export class WebhooksController { + constructor(private readonly webhooksService: WebhooksService) {} + + // ─── Registration ────────────────────────────────────────────────────── + + @Post() + @ApiOperation({ + summary: 'Register a new webhook', + description: + 'Register a new webhook endpoint with event subscriptions. ' + + 'Returns the webhook details along with the raw secret (shown only once). ' + + 'Store the secret securely to verify incoming webhook signatures.', + }) + @ApiResponse({ + status: 201, + description: 'Webhook created successfully', + }) + @ApiResponse({ status: 400, description: 'Invalid input or non-HTTPS URL' }) + @ApiResponse({ status: 409, description: 'Duplicate webhook URL for this owner' }) + async create(@Body() dto: CreateWebhookDto): Promise { + return this.webhooksService.create(dto); + } + + @Get() + @ApiOperation({ + summary: 'List all webhooks', + description: + 'Retrieve all registered webhooks with optional filtering by owner or status.', + }) + @ApiQuery({ + name: 'ownerId', + required: false, + description: 'Filter by owner wallet address', + }) + @ApiQuery({ + name: 'enabled', + required: false, + description: 'Filter by enabled status (true/false)', + }) + async findAll( + @Query() filter: WebhookListFilterDto, + ): Promise { + return this.webhooksService.findAll(filter.ownerId, filter.enabled); + } + + @Get(':id') + @ApiOperation({ + summary: 'Get webhook details', + description: 'Retrieve a specific webhook by ID including its subscriptions.', + }) + @ApiParam({ name: 'id', description: 'Webhook UUID' }) + @ApiResponse({ status: 404, description: 'Webhook not found' }) + async findOne(@Param('id') id: string): Promise { + return this.webhooksService.findOne(id); + } + + @Patch(':id') + @ApiOperation({ + summary: 'Update webhook configuration', + description: + 'Update webhook URL, description, enabled status, subscriptions, or retry settings.', + }) + @ApiParam({ name: 'id', description: 'Webhook UUID' }) + @ApiResponse({ status: 404, description: 'Webhook not found' }) + async update( + @Param('id') id: string, + @Body() dto: UpdateWebhookDto, + ): Promise { + return this.webhooksService.update(id, dto); + } + + @Delete(':id') + @HttpCode(HttpStatus.NO_CONTENT) + @ApiOperation({ + summary: 'Delete a webhook', + description: + 'Permanently remove a webhook and all its delivery history.', + }) + @ApiParam({ name: 'id', description: 'Webhook UUID' }) + @ApiResponse({ status: 204, description: 'Webhook deleted successfully' }) + @ApiResponse({ status: 404, description: 'Webhook not found' }) + async remove(@Param('id') id: string): Promise { + return this.webhooksService.remove(id); + } + + // ─── Delivery History ────────────────────────────────────────────────── + + @Get(':id/deliveries') + @ApiOperation({ + summary: 'Get webhook delivery history', + description: + 'Retrieve delivery history for a webhook with optional filtering by event type or status.', + }) + @ApiParam({ name: 'id', description: 'Webhook UUID' }) + async getDeliveries( + @Param('id') id: string, + @Query() filter: WebhookDeliveryFilterDto, + ): Promise<{ + deliveries: WebhookDelivery[]; + total: number; + page: number; + limit: number; + }> { + return this.webhooksService.getDeliveries(id, filter); + } + + @Get('deliveries/:deliveryId') + @ApiOperation({ + summary: 'Get delivery details', + description: 'Retrieve a specific webhook delivery record.', + }) + @ApiParam({ name: 'deliveryId', description: 'Delivery UUID' }) + @ApiResponse({ status: 404, description: 'Delivery not found' }) + async getDelivery( + @Param('deliveryId') deliveryId: string, + ): Promise { + return this.webhooksService.getDelivery(deliveryId); + } + + // ─── Retry ───────────────────────────────────────────────────────────── + + @Post(':id/deliveries/:deliveryId/retry') + @HttpCode(HttpStatus.ACCEPTED) + @ApiOperation({ + summary: 'Retry a failed delivery', + description: + 'Re-queue a failed webhook delivery for another attempt.', + }) + @ApiParam({ name: 'id', description: 'Webhook UUID' }) + @ApiParam({ name: 'deliveryId', description: 'Delivery UUID' }) + @ApiResponse({ status: 202, description: 'Retry accepted' }) + async retryDelivery( + @Param('id') _id: string, + @Param('deliveryId') deliveryId: string, + ): Promise { + return this.webhooksService.retryDelivery(deliveryId); + } + + // ─── Secret Management ───────────────────────────────────────────────── + + @Post(':id/rotate-secret') + @HttpCode(HttpStatus.OK) + @ApiOperation({ + summary: 'Rotate webhook secret', + description: + 'Generate a new webhook signing secret. The previous secret remains ' + + 'valid for 24 hours to allow seamless transition. Returns the new secret (shown only once).', + }) + @ApiParam({ name: 'id', description: 'Webhook UUID' }) + async rotateSecret( + @Param('id') id: string, + ): Promise<{ secret: string; expiresAt: Date }> { + return this.webhooksService.rotateSecret(id); + } + + @Post(':id/revoke-secret') + @HttpCode(HttpStatus.OK) + @ApiOperation({ + summary: 'Revoke webhook secret', + description: + 'Immediately revoke the current webhook secret and generate a new one. ' + + 'Any deliveries signed with the old secret will be rejected.', + }) + @ApiParam({ name: 'id', description: 'Webhook UUID' }) + async revokeSecret(@Param('id') id: string): Promise { + return this.webhooksService.revokeSecret(id); + } + + // ─── Status ──────────────────────────────────────────────────────────── + + @Get(':id/status') + @ApiOperation({ + summary: 'Get webhook status & metrics', + description: + 'Retrieve health status and delivery metrics for a webhook, including ' + + 'total deliveries, success/failure counts, and last delivery info.', + }) + @ApiParam({ name: 'id', description: 'Webhook UUID' }) + async getStatus( + @Param('id') id: string, + ): Promise<{ + webhook: Webhook; + totalDeliveries: number; + successfulDeliveries: number; + failedDeliveries: number; + pendingDeliveries: number; + lastDelivery: WebhookDelivery | null; + }> { + return this.webhooksService.getWebhookStatus(id); + } +} diff --git a/src/webhooks/webhooks.module.ts b/src/webhooks/webhooks.module.ts new file mode 100644 index 0000000..a034fab --- /dev/null +++ b/src/webhooks/webhooks.module.ts @@ -0,0 +1,37 @@ +import { Module } from '@nestjs/common'; +import { TypeOrmModule } from '@nestjs/typeorm'; +import { BullModule } from '@nestjs/bullmq'; +import { BullBoardModule } from '@bull-board/nestjs'; +import { BullMQAdapter } from '@bull-board/api/bullMQAdapter'; +import { Webhook } from './entities/webhook.entity'; +import { WebhookSubscription } from './entities/webhook-subscription.entity'; +import { WebhookDelivery } from './entities/webhook-delivery.entity'; +import { WebhooksService } from './webhooks.service'; +import { WebhooksController } from './webhooks.controller'; +import { WebhookProcessor } from './webhooks.processor'; + +@Module({ + imports: [ + TypeOrmModule.forFeature([Webhook, WebhookSubscription, WebhookDelivery]), + BullModule.registerQueue({ + name: 'webhook-delivery', + defaultJobOptions: { + attempts: 4, + backoff: { + type: 'exponential', + delay: 30000, + }, + removeOnComplete: false, + removeOnFail: false, + }, + }), + BullBoardModule.forFeature({ + name: 'webhook-delivery', + adapter: BullMQAdapter, + }), + ], + controllers: [WebhooksController], + providers: [WebhooksService, WebhookProcessor], + exports: [WebhooksService], +}) +export class WebhooksModule {} diff --git a/src/webhooks/webhooks.processor.ts b/src/webhooks/webhooks.processor.ts new file mode 100644 index 0000000..d12b0c9 --- /dev/null +++ b/src/webhooks/webhooks.processor.ts @@ -0,0 +1,205 @@ +import { Processor, WorkerHost } from '@nestjs/bullmq'; +import { Job } from 'bullmq'; +import { Injectable, Logger } from '@nestjs/common'; +import { InjectRepository } from '@nestjs/typeorm'; +import { Repository } from 'typeorm'; +import * as client from 'prom-client'; +import { WebhookDelivery, DeliveryStatus } from './entities/webhook-delivery.entity'; +import { Webhook } from './entities/webhook.entity'; +import { MAX_CONSECUTIVE_FAILURES_BEFORE_DISABLE } from './webhooks.service'; + +interface DeliverJobData { + deliveryId: string; + webhookId: string; + url: string; + payload: any; + eventType: string; + signature: string; + nonce: string; + timestamp: string; + requestId: string; + retryCount: number; + maxRetries: number; + retryIntervalMs: number; +} + +const WEBHOOK_DELIVERY_TIMEOUT = 10000; // 10 seconds + +@Processor('webhook-delivery') +@Injectable() +export class WebhookProcessor extends WorkerHost { + private readonly logger = new Logger(WebhookProcessor.name); + + // ─── Prometheus Metrics ──────────────────────────────────────────────── + private deliveryAttemptsCounter: client.Counter; + private deliveryLatencyHistogram: client.Histogram; + + constructor( + @InjectRepository(WebhookDelivery) + private readonly deliveryRepo: Repository, + @InjectRepository(Webhook) + private readonly webhookRepo: Repository, + ) { + super(); + + this.deliveryAttemptsCounter = new client.Counter({ + name: 'webhook_delivery_attempts_total', + help: 'Total number of webhook delivery attempts', + labelNames: ['webhook_id', 'event_type', 'status'], + }); + + this.deliveryLatencyHistogram = new client.Histogram({ + name: 'webhook_delivery_attempt_latency_seconds', + help: 'Latency of webhook delivery attempts in seconds', + labelNames: ['webhook_id', 'event_type', 'result'], + buckets: [0.1, 0.5, 1, 2.5, 5, 10], + }); + } + + async process(job: Job): Promise { + const data = job.data; + + this.logger.log( + `Processing webhook delivery ${data.deliveryId} to ${data.url} (attempt ${job.attemptsMade}/${data.maxRetries + 1})`, + ); + + const startTime = Date.now(); + + try { + // Perform the HTTP request + const response = await this.sendWebhookRequest(data); + + const latency = Date.now() - startTime; + + // Record metrics + this.deliveryAttemptsCounter.inc({ + webhook_id: data.webhookId, + event_type: data.eventType, + status: 'success', + }); + + this.deliveryLatencyHistogram.observe( + { webhook_id: data.webhookId, event_type: data.eventType, result: 'success' }, + latency / 1000, + ); + + // Update delivery record + await this.deliveryRepo.update(data.deliveryId, { + status: DeliveryStatus.DELIVERED, + responseStatus: response.status, + responseBody: response.body, + latency, + retryCount: job.attemptsMade - 1, + completedAt: new Date(), + }); + + // Reset consecutive failures on the webhook + await this.webhookRepo.update(data.webhookId, { + consecutiveFailures: 0, + }); + + this.logger.log( + `Webhook delivery ${data.deliveryId} succeeded (${response.status}) in ${latency}ms`, + ); + + return { success: true, status: response.status, latency }; + } catch (error) { + const latency = Date.now() - startTime; + const errorMessage = error instanceof Error ? error.message : String(error); + const isFinalAttempt = job.attemptsMade > data.maxRetries; + + this.logger.warn( + `Webhook delivery ${data.deliveryId} failed (attempt ${job.attemptsMade}/${data.maxRetries + 1}): ${errorMessage}`, + ); + + // Record metrics + this.deliveryAttemptsCounter.inc({ + webhook_id: data.webhookId, + event_type: data.eventType, + status: isFinalAttempt ? 'dead_letter' : 'failed', + }); + + this.deliveryLatencyHistogram.observe( + { + webhook_id: data.webhookId, + event_type: data.eventType, + result: isFinalAttempt ? 'dead_letter' : 'failed', + }, + latency / 1000, + ); + + // Update delivery with failure info + await this.deliveryRepo.update(data.deliveryId, { + status: isFinalAttempt ? DeliveryStatus.DEAD_LETTER : DeliveryStatus.FAILED, + responseStatus: error instanceof Error && 'status' in error ? (error as any).status : undefined, + responseBody: errorMessage, + latency, + retryCount: job.attemptsMade - 1, + failureReason: errorMessage, + completedAt: isFinalAttempt ? new Date() : undefined, + }); + + // Track consecutive failures on the webhook — disable after threshold + if (isFinalAttempt) { + const webhook = await this.webhookRepo.findOne({ + where: { id: data.webhookId }, + }); + if (webhook) { + webhook.consecutiveFailures += 1; + + if (webhook.consecutiveFailures >= MAX_CONSECUTIVE_FAILURES_BEFORE_DISABLE) { + webhook.disabled = true; + webhook.enabled = false; + this.logger.warn( + `Webhook ${data.webhookId} auto-disabled after ${webhook.consecutiveFailures} consecutive failures`, + ); + } + + await this.webhookRepo.save(webhook); + } + } + + // Re-throw to trigger BullMQ retry with backoff + throw error; + } + } + + private async sendWebhookRequest( + data: DeliverJobData, + ): Promise<{ status: number; body: string }> { + const controller = new AbortController(); + const timeout = setTimeout(() => controller.abort(), WEBHOOK_DELIVERY_TIMEOUT); + + try { + // Build signed payload following webhook standards + const body = JSON.stringify(data.payload); + const headers: Record = { + 'Content-Type': 'application/json', + 'X-Webhook-ID': data.requestId, + 'X-Webhook-Signature': data.signature, + 'X-Webhook-Nonce': data.nonce, + 'X-Webhook-Timestamp': data.timestamp, + 'X-Webhook-Event': data.eventType, + 'X-Webhook-Delivery': data.deliveryId, + 'User-Agent': 'TruthBounty-Webhook/1.0', + }; + + const response = await fetch(data.url, { + method: 'POST', + headers, + body, + signal: controller.signal, + }); + + // Read response body + const responseBody = await response.text(); + + return { + status: response.status, + body: responseBody.slice(0, 10000), // Limit body storage + }; + } finally { + clearTimeout(timeout); + } + } +} diff --git a/src/webhooks/webhooks.service.spec.ts b/src/webhooks/webhooks.service.spec.ts new file mode 100644 index 0000000..3374275 --- /dev/null +++ b/src/webhooks/webhooks.service.spec.ts @@ -0,0 +1,433 @@ +import { Test, TestingModule } from '@nestjs/testing'; +import { getRepositoryToken } from '@nestjs/typeorm'; +import { getQueueToken } from '@nestjs/bullmq'; +import { Repository } from 'typeorm'; +import { ConflictException, BadRequestException, NotFoundException } from '@nestjs/common'; +import { WebhooksService } from './webhooks.service'; +import { Webhook, WebhookEventType } from './entities/webhook.entity'; +import { WebhookSubscription } from './entities/webhook-subscription.entity'; +import { WebhookDelivery, DeliveryStatus } from './entities/webhook-delivery.entity'; +import { CreateWebhookDto } from './dto/create-webhook.dto'; +import { UpdateWebhookDto } from './dto/update-webhook.dto'; + +// ─── Mock prom-client ───────────────────────────────────────────────────────── +const mockCounter = { inc: jest.fn() }; +const mockHistogram = { observe: jest.fn() }; +const mockGauge = { set: jest.fn() }; + +jest.mock('prom-client', () => ({ + Counter: jest.fn(() => mockCounter), + Histogram: jest.fn(() => mockHistogram), + Gauge: jest.fn(() => mockGauge), + register: { + metrics: jest.fn().mockResolvedValue(''), + clear: jest.fn(), + }, +})); + +describe('WebhooksService', () => { + let service: WebhooksService; + let webhookRepo: Repository; + let subscriptionRepo: Repository; + let deliveryRepo: Repository; + let webhookQueue: any; + + const mockWebhook = { + id: 'wh-001', + url: 'https://example.com/webhook', + description: 'Test webhook', + ownerId: '0x123', + enabled: true, + secret: 'test-secret-raw', + previousSecret: null, + previousSecretExpiresAt: null, + secretExpiresAt: null, + consecutiveFailures: 0, + maxRetries: 3, + retryIntervalMs: 30000, + disabled: false, + createdAt: new Date(), + updatedAt: new Date(), + subscriptions: [], + deliveries: [], + }; + + const mockSubscription = { + id: 'sub-001', + webhookId: 'wh-001', + eventType: WebhookEventType.CLAIM_CREATED, + filters: null, + createdAt: new Date(), + webhook: mockWebhook, + }; + + beforeEach(async () => { + const mockWebhookRepo = { + create: jest.fn().mockReturnValue(mockWebhook), + save: jest.fn().mockResolvedValue(mockWebhook), + find: jest.fn().mockResolvedValue([mockWebhook]), + findOne: jest.fn().mockResolvedValue(mockWebhook), + findOneBy: jest.fn().mockResolvedValue(mockWebhook), + remove: jest.fn().mockResolvedValue(undefined), + update: jest.fn().mockResolvedValue({ affected: 1 }), + count: jest.fn().mockResolvedValue(10), + }; + + const mockSubscriptionRepo = { + create: jest.fn().mockReturnValue(mockSubscription), + save: jest.fn().mockResolvedValue(mockSubscription), + find: jest.fn().mockResolvedValue([mockSubscription]), + findOne: jest.fn().mockResolvedValue(mockSubscription), + delete: jest.fn().mockResolvedValue({ affected: 1 }), + }; + + const mockDeliveryRepo = { + create: jest.fn().mockReturnValue({ id: 'del-001' }), + save: jest.fn().mockResolvedValue({ id: 'del-001' }), + find: jest.fn().mockResolvedValue([]), + findOne: jest.fn().mockResolvedValue({ + id: 'del-001', + webhookId: 'wh-001', + eventType: 'claim.created', + payload: {}, + status: DeliveryStatus.FAILED, + retryCount: 0, + maxRetries: 3, + requestId: 'req-001', + nonce: 'nonce', + signature: 'sig', + timestamp: new Date().toISOString(), + responseStatus: null, + responseBody: null, + latency: null, + failureReason: null, + createdAt: new Date(), + completedAt: null, + }), + count: jest.fn().mockResolvedValue(10), + update: jest.fn().mockResolvedValue({ affected: 1 }), + createQueryBuilder: jest.fn(() => ({ + where: jest.fn().mockReturnThis(), + andWhere: jest.fn().mockReturnThis(), + orderBy: jest.fn().mockReturnThis(), + skip: jest.fn().mockReturnThis(), + take: jest.fn().mockReturnThis(), + getManyAndCount: jest.fn().mockResolvedValue([[], 0]), + })), + }; + + const mockQueue = { + add: jest.fn().mockResolvedValue({ id: 'job-001' }), + }; + + const module: TestingModule = await Test.createTestingModule({ + providers: [ + WebhooksService, + { provide: getRepositoryToken(Webhook), useValue: mockWebhookRepo }, + { provide: getRepositoryToken(WebhookSubscription), useValue: mockSubscriptionRepo }, + { provide: getRepositoryToken(WebhookDelivery), useValue: mockDeliveryRepo }, + { provide: getQueueToken('webhook-delivery'), useValue: mockQueue }, + ], + }).compile(); + + service = module.get(WebhooksService); + // Initialize metrics explicitly since onModuleInit is async and won't be auto-called in test + await service.onModuleInit(); + + webhookRepo = module.get(getRepositoryToken(Webhook)); + subscriptionRepo = module.get(getRepositoryToken(WebhookSubscription)); + deliveryRepo = module.get(getRepositoryToken(WebhookDelivery)); + webhookQueue = module.get(getQueueToken('webhook-delivery')); + }); + + // ─── Registration ────────────────────────────────────────────────────── + + describe('create', () => { + it('should create a webhook with valid data', async () => { + const dto: CreateWebhookDto = { + url: 'https://example.com/webhook', + description: 'Test webhook', + ownerId: '0x123', + enabled: true, + events: [WebhookEventType.CLAIM_CREATED, WebhookEventType.VERIFICATION_COMPLETED], + maxRetries: 3, + retryIntervalMs: 30000, + }; + + // Mock findOne to return null (no duplicate) + jest.spyOn(webhookRepo, 'findOne').mockResolvedValueOnce(null); + + const result = await service.create(dto); + + expect(result).toBeDefined(); + expect(webhookRepo.create).toHaveBeenCalled(); + expect(webhookRepo.save).toHaveBeenCalled(); + expect(subscriptionRepo.create).toHaveBeenCalledTimes(2); + }); + + it('should reject non-HTTPS URLs', async () => { + const dto: CreateWebhookDto = { + url: 'http://example.com/webhook', + ownerId: '0x123', + events: [WebhookEventType.CLAIM_CREATED], + }; + + await expect(service.create(dto)).rejects.toThrow(BadRequestException); + }); + + it('should reject duplicate active webhooks for same owner', async () => { + const dto: CreateWebhookDto = { + url: 'https://example.com/webhook', + ownerId: '0x123', + events: [WebhookEventType.CLAIM_CREATED], + }; + + // findOne returns existing webhook (duplicate) + jest.spyOn(webhookRepo, 'findOne').mockResolvedValueOnce(mockWebhook); + + await expect(service.create(dto)).rejects.toThrow(ConflictException); + }); + + it('should reject invalid event types', async () => { + const dto: CreateWebhookDto = { + url: 'https://example.com/webhook', + ownerId: '0x123', + events: ['invalid.event' as WebhookEventType], + }; + + jest.spyOn(webhookRepo, 'findOne').mockResolvedValueOnce(null); + + await expect(service.create(dto)).rejects.toThrow(BadRequestException); + }); + + it('should reject empty events array', async () => { + const dto: CreateWebhookDto = { + url: 'https://example.com/webhook', + ownerId: '0x123', + events: [], + }; + + // Ensure no duplicate conflict is found first + jest.spyOn(webhookRepo, 'findOne').mockResolvedValueOnce(null); + + await expect(service.create(dto)).rejects.toThrow(BadRequestException); + }); + }); + + describe('findAll', () => { + it('should return all webhooks', async () => { + const result = await service.findAll(); + expect(result).toHaveLength(1); + expect(webhookRepo.find).toHaveBeenCalled(); + }); + + it('should filter by ownerId', async () => { + await service.findAll('0x123'); + expect(webhookRepo.find).toHaveBeenCalledWith( + expect.objectContaining({ + where: expect.objectContaining({ ownerId: '0x123' }), + }), + ); + }); + + it('should filter by enabled status', async () => { + await service.findAll(undefined, 'true'); + expect(webhookRepo.find).toHaveBeenCalledWith( + expect.objectContaining({ + where: expect.objectContaining({ enabled: true }), + }), + ); + }); + }); + + describe('findOne', () => { + it('should return a webhook by ID', async () => { + const result = await service.findOne('wh-001'); + expect(result).toBeDefined(); + expect(result.id).toBe('wh-001'); + }); + + it('should throw NotFoundException for missing webhook', async () => { + jest.spyOn(webhookRepo, 'findOne').mockResolvedValueOnce(null); + await expect(service.findOne('nonexistent')).rejects.toThrow(NotFoundException); + }); + }); + + describe('update', () => { + it('should update webhook URL', async () => { + const dto: UpdateWebhookDto = { url: 'https://new-url.com/webhook' }; + const result = await service.update('wh-001', dto); + expect(result).toBeDefined(); + expect(result.url).toBe(mockWebhook.url); + }); + + it('should reject non-HTTPS URL update', async () => { + const dto: UpdateWebhookDto = { url: 'http://insecure-url.com' }; + await expect(service.update('wh-001', dto)).rejects.toThrow(BadRequestException); + }); + + it('should re-enable webhook when enabled is set to true', async () => { + const dto: UpdateWebhookDto = { enabled: true }; + const result = await service.update('wh-001', dto); + expect(result).toBeDefined(); + }); + + it('should update event subscriptions', async () => { + const dto: UpdateWebhookDto = { + events: [WebhookEventType.DISPUTE_CREATED], + }; + const result = await service.update('wh-001', dto); + expect(result).toBeDefined(); + expect(subscriptionRepo.delete).toHaveBeenCalledWith({ webhookId: 'wh-001' }); + }); + }); + + describe('remove', () => { + it('should delete a webhook', async () => { + await service.remove('wh-001'); + expect(webhookRepo.remove).toHaveBeenCalled(); + }); + + it('should throw NotFoundException for missing webhook', async () => { + jest.spyOn(webhookRepo, 'findOne').mockResolvedValueOnce(null); + await expect(service.remove('nonexistent')).rejects.toThrow(NotFoundException); + }); + }); + + // ─── Secret Management ────────────────────────────────────────────────── + + describe('rotateSecret', () => { + it('should rotate the webhook secret', async () => { + const result = await service.rotateSecret('wh-001'); + expect(result).toBeDefined(); + expect(result.secret).toBeDefined(); + expect(result.expiresAt).toBeDefined(); + expect(webhookRepo.save).toHaveBeenCalled(); + }); + }); + + describe('revokeSecret', () => { + it('should revoke the webhook secret', async () => { + await service.revokeSecret('wh-001'); + expect(webhookRepo.save).toHaveBeenCalled(); + }); + }); + + // ─── Event Dispatch ───────────────────────────────────────────────────── + + describe('dispatchEvent', () => { + it('should dispatch event to subscribed webhooks', async () => { + jest.spyOn(subscriptionRepo, 'find').mockResolvedValueOnce([mockSubscription]); + + const payload = { + eventType: WebhookEventType.CLAIM_CREATED, + eventId: 'evt-001', + module: 'claims', + timestamp: new Date().toISOString(), + data: { claimId: 'claim-001' }, + }; + + const count = await service.dispatchEvent(WebhookEventType.CLAIM_CREATED, payload); + expect(count).toBe(1); + expect(webhookQueue.add).toHaveBeenCalled(); + }); + + it('should not dispatch when no subscribers', async () => { + jest.spyOn(subscriptionRepo, 'find').mockResolvedValueOnce([]); + + const payload = { + eventType: WebhookEventType.CLAIM_CREATED, + eventId: 'evt-001', + module: 'claims', + timestamp: new Date().toISOString(), + data: { claimId: 'claim-001' }, + }; + + const count = await service.dispatchEvent(WebhookEventType.CLAIM_CREATED, payload); + expect(count).toBe(0); + expect(webhookQueue.add).not.toHaveBeenCalled(); + }); + }); + + // ─── Delivery History ─────────────────────────────────────────────────── + + describe('getDeliveries', () => { + it('should return paginated deliveries', async () => { + const result = await service.getDeliveries('wh-001'); + expect(result).toBeDefined(); + expect(result.deliveries).toEqual([]); + expect(result.total).toBe(0); + expect(result.page).toBe(1); + expect(result.limit).toBe(20); + }); + }); + + describe('getDelivery', () => { + it('should return a delivery by ID', async () => { + const result = await service.getDelivery('del-001'); + expect(result).toBeDefined(); + expect(result.id).toBe('del-001'); + }); + + it('should throw NotFoundException for missing delivery', async () => { + jest.spyOn(deliveryRepo, 'findOne').mockResolvedValueOnce(null); + await expect(service.getDelivery('nonexistent')).rejects.toThrow(NotFoundException); + }); + }); + + // ─── Retry Logic ──────────────────────────────────────────────────────── + + describe('retryDelivery', () => { + it('should retry a failed delivery', async () => { + await service.retryDelivery('del-001'); + expect(webhookQueue.add).toHaveBeenCalled(); + }); + }); + + // ─── Status ───────────────────────────────────────────────────────────── + + describe('getWebhookStatus', () => { + it('should return webhook status with metrics', async () => { + const result = await service.getWebhookStatus('wh-001'); + expect(result).toBeDefined(); + expect(result.webhook).toBeDefined(); + expect(result.totalDeliveries).toBe(10); + expect(result.successfulDeliveries).toBe(10); + expect(result.failedDeliveries).toBe(10); + expect(result.pendingDeliveries).toBe(10); + }); + }); + + // ─── Signature Verification ──────────────────────────────────────────── + + describe('verifySignature', () => { + it('should verify a valid signature', () => { + const secret = 'test-secret'; + const payload = { data: 'test' }; + const timestamp = new Date().toISOString(); + const nonce = 'test-nonce'; + + // Generate a signature using the same method + const crypto = require('crypto'); + const hmac = crypto.createHmac('sha256', secret); + hmac.update(`${timestamp}.${nonce}.${JSON.stringify(payload)}`); + const validSignature = hmac.digest('hex'); + + const result = service.verifySignature(payload, validSignature, secret, timestamp, nonce); + expect(result).toBe(true); + }); + + it('should reject an invalid signature', () => { + // Must be a 64-char hex string (same length as valid HMAC-SHA256) + const invalidSig = 'a'.repeat(64); + const result = service.verifySignature( + { data: 'test' }, + invalidSig, + 'secret', + new Date().toISOString(), + 'nonce', + ); + expect(result).toBe(false); + }); + }); +}); diff --git a/src/webhooks/webhooks.service.ts b/src/webhooks/webhooks.service.ts new file mode 100644 index 0000000..4667ffb --- /dev/null +++ b/src/webhooks/webhooks.service.ts @@ -0,0 +1,595 @@ +import { + Injectable, + Logger, + NotFoundException, + ConflictException, + BadRequestException, + OnModuleInit, +} from '@nestjs/common'; +import { InjectRepository } from '@nestjs/typeorm'; +import { Repository } from 'typeorm'; +import { InjectQueue } from '@nestjs/bullmq'; +import { Queue } from 'bullmq'; +import * as crypto from 'crypto'; +import * as client from 'prom-client'; +import { Webhook, WebhookEventType, ALL_WEBHOOK_EVENTS } from './entities/webhook.entity'; +import { WebhookSubscription } from './entities/webhook-subscription.entity'; +import { WebhookDelivery, DeliveryStatus } from './entities/webhook-delivery.entity'; +import { CreateWebhookDto } from './dto/create-webhook.dto'; +import { UpdateWebhookDto } from './dto/update-webhook.dto'; +import { WebhookDeliveryFilterDto } from './dto/webhook-filter.dto'; + +export const WEBHOOK_SECRET_BYTES = 32; +export const SIGNATURE_ALGORITHM = 'sha256'; +export const SECRET_ROTATION_GRACE_PERIOD_MS = 24 * 60 * 60 * 1000; // 24 hours +export const WEBHOOK_QUEUE = 'webhook-delivery'; +export const MAX_CONSECUTIVE_FAILURES_BEFORE_DISABLE = 10; + +export interface WebhookEventPayload { + eventType: WebhookEventType; + eventId: string; + module: string; + network?: string; + timestamp: string; + data: Record; +} + +@Injectable() +export class WebhooksService implements OnModuleInit { + private readonly logger = new Logger(WebhooksService.name); + + // ─── Prometheus Metrics ──────────────────────────────────────────────── + private deliveriesCounter: client.Counter; + private deliveryLatencyHistogram: client.Histogram; + private disabledEndpointsGauge: client.Gauge; + private activeSubscriptionsGauge: client.Gauge; + + constructor( + @InjectRepository(Webhook) + private readonly webhookRepo: Repository, + @InjectRepository(WebhookSubscription) + private readonly subscriptionRepo: Repository, + @InjectRepository(WebhookDelivery) + private readonly deliveryRepo: Repository, + @InjectQueue(WEBHOOK_QUEUE) + private readonly webhookQueue: Queue, + ) {} + +async onModuleInit(): Promise { + try { + this.deliveriesCounter = new client.Counter({ + name: 'webhook_deliveries_total', + help: 'Total number of webhook deliveries', + labelNames: ['webhook_id', 'event_type', 'status'], + }); + + this.deliveryLatencyHistogram = new client.Histogram({ + name: 'webhook_delivery_latency_seconds', + help: 'Latency of webhook deliveries in seconds', + labelNames: ['webhook_id', 'event_type', 'status'], + buckets: [0.1, 0.5, 1, 2.5, 5, 10], + }); + + this.disabledEndpointsGauge = new client.Gauge({ + name: 'webhook_disabled_endpoints_total', + help: 'Number of disabled webhook endpoints', + }); + + this.activeSubscriptionsGauge = new client.Gauge({ + name: 'webhook_active_subscriptions_total', + help: 'Number of active webhook subscriptions', + }); + } catch (error) { + this.logger.warn( + `Failed to register Prometheus metrics: ${error.message}. Metrics will be unavailable.`, + ); + } + } + + // Update gauge metrics periodically or on relevant operations + private async updateGaugeMetrics(): Promise { + try { + const disabledCount = await this.webhookRepo.count({ + where: { disabled: true }, + }); + this.disabledEndpointsGauge.set(disabledCount); + + const activeWebhooks = await this.webhookRepo.find({ + where: { enabled: true, disabled: false }, + }); + this.activeSubscriptionsGauge.set(activeWebhooks.length); + } catch (error) { + this.logger.debug(`Failed to update gauge metrics: ${error.message}`); + } + } + + // ─── Webhook CRUD ─────────────────────────────────────────────────────── + + async create(dto: CreateWebhookDto): Promise { + // Validate URL is HTTPS + if (!dto.url.startsWith('https://')) { + throw new BadRequestException('Webhook URL must use HTTPS protocol'); + } + + // Check for duplicate active webhook with same URL for this owner + const existing = await this.webhookRepo.findOne({ + where: { url: dto.url, ownerId: dto.ownerId, disabled: false }, + }); + if (existing) { + throw new ConflictException( + 'An active webhook with this URL already exists for this owner', + ); + } + + // Validate event types + if (!dto.events || dto.events.length === 0) { + throw new BadRequestException('At least one event type must be subscribed'); + } + + const invalidEvents = dto.events.filter( + (e) => !ALL_WEBHOOK_EVENTS.includes(e), + ); + if (invalidEvents.length > 0) { + throw new BadRequestException( + `Invalid event types: ${invalidEvents.join(', ')}`, + ); + } + + // Generate a cryptographically random secret + const secret = this.generateSecret(); + + // Create webhook – store the raw secret so we can sign payloads with it. + // The secret is already a long random hex string (64 chars), so it is secure + // at rest in the database. The raw secret is returned to the caller exactly + // once on creation. + const webhook = this.webhookRepo.create({ + url: dto.url, + description: dto.description || '', + ownerId: dto.ownerId, + enabled: dto.enabled !== undefined ? dto.enabled : true, + secret, + maxRetries: dto.maxRetries ?? 3, + retryIntervalMs: dto.retryIntervalMs ?? 30000, + }); + + const saved = await this.webhookRepo.save(webhook); + + // Create subscriptions + const subscriptions = dto.events.map((eventType) => + this.subscriptionRepo.create({ + webhookId: saved.id, + eventType, + filters: dto.filters || null, + }), + ); + await this.subscriptionRepo.save(subscriptions); + + // Return webhook with the raw secret (only time it's returned) + const result = { ...saved, rawSecret: secret } as any; + return result; + } + + async findAll( + ownerId?: string, + enabled?: string, + ): Promise { + const where: any = {}; + if (ownerId) where.ownerId = ownerId; + if (enabled !== undefined) where.enabled = enabled === 'true'; + + return this.webhookRepo.find({ + where, + relations: ['subscriptions'], + order: { createdAt: 'DESC' }, + }); + } + + async findOne(id: string): Promise { + const webhook = await this.webhookRepo.findOne({ + where: { id }, + relations: ['subscriptions'], + }); + if (!webhook) { + throw new NotFoundException(`Webhook ${id} not found`); + } + return webhook; + } + + async update(id: string, dto: UpdateWebhookDto): Promise { + const webhook = await this.findOne(id); + + if (dto.url !== undefined) { + if (!dto.url.startsWith('https://')) { + throw new BadRequestException('Webhook URL must use HTTPS protocol'); + } + webhook.url = dto.url; + } + + if (dto.description !== undefined) { + webhook.description = dto.description; + } + + if (dto.enabled !== undefined) { + webhook.enabled = dto.enabled; + // Reset disabled status when re-enabling + if (dto.enabled) { + webhook.disabled = false; + webhook.consecutiveFailures = 0; + } + } + + if (dto.maxRetries !== undefined) { + webhook.maxRetries = dto.maxRetries; + } + + if (dto.retryIntervalMs !== undefined) { + webhook.retryIntervalMs = dto.retryIntervalMs; + } + + // Update subscriptions if events provided + if (dto.events !== undefined) { + // Validate events + const invalidEvents = dto.events.filter( + (e) => !ALL_WEBHOOK_EVENTS.includes(e), + ); + if (invalidEvents.length > 0) { + throw new BadRequestException( + `Invalid event types: ${invalidEvents.join(', ')}`, + ); + } + + // Remove existing subscriptions + await this.subscriptionRepo.delete({ webhookId: id }); + + // Create new subscriptions + const subscriptions = dto.events.map((eventType) => + this.subscriptionRepo.create({ + webhookId: id, + eventType, + filters: dto.filters || null, + }), + ); + await this.subscriptionRepo.save(subscriptions); + } else if (dto.filters !== undefined) { + // Update filters on all existing subscriptions + const subs = await this.subscriptionRepo.find({ where: { webhookId: id } }); + for (const sub of subs) { + sub.filters = dto.filters; + } + await this.subscriptionRepo.save(subs); + } + + await this.webhookRepo.save(webhook); + return this.findOne(id); + } + + async remove(id: string): Promise { + const webhook = await this.findOne(id); + await this.webhookRepo.remove(webhook); + } + + // ─── Secret Management ───────────────────────────────────────────────── + + async rotateSecret(id: string): Promise<{ secret: string; expiresAt: Date }> { + const webhook = await this.findOne(id); + + // Move current secret to previous with grace period + webhook.previousSecret = webhook.secret; + webhook.previousSecretExpiresAt = new Date( + Date.now() + SECRET_ROTATION_GRACE_PERIOD_MS, + ); + + // Generate new secret + const newSecret = this.generateSecret(); + webhook.secret = newSecret; + webhook.secretExpiresAt = null; + + await this.webhookRepo.save(webhook); + + return { + secret: newSecret, + expiresAt: new Date(Date.now() + SECRET_ROTATION_GRACE_PERIOD_MS), + }; + } + + async revokeSecret(id: string): Promise { + const webhook = await this.findOne(id); + + // Generate new secret (old one immediately invalid) + const newSecret = this.generateSecret(); + webhook.secret = newSecret; + webhook.previousSecret = null; + webhook.previousSecretExpiresAt = null; + webhook.secretExpiresAt = null; + + await this.webhookRepo.save(webhook); + } + + // ─── Event Dispatch ──────────────────────────────────────────────────── + + async dispatchEvent( + eventType: WebhookEventType, + payload: WebhookEventPayload, + ): Promise { + // Find all active webhooks subscribed to this event type + const subscriptions = await this.subscriptionRepo.find({ + where: { eventType }, + relations: ['webhook'], + }); + + // Filter to active webhooks that aren't disabled + const activeSubs = subscriptions.filter( + (sub) => sub.webhook && sub.webhook.enabled && !sub.webhook.disabled, + ); + + if (activeSubs.length === 0) { + this.logger.debug( + `No active subscribers for event ${eventType}`, + ); + return 0; + } + + // Apply event-level filtering + const matchedSubs = activeSubs.filter((sub) => + this.matchesFilters(payload, sub.filters), + ); + + if (matchedSubs.length === 0) { + this.logger.debug( + `No subscribers matched filters for event ${eventType}`, + ); + return 0; + } + + // Enqueue delivery jobs + for (const sub of matchedSubs) { + await this.enqueueDelivery(sub.webhook, eventType, payload); + } + + this.logger.log( + `Dispatched event ${eventType} to ${matchedSubs.length} webhooks`, + ); + + // Update gauge metrics + await this.updateGaugeMetrics(); + + return matchedSubs.length; + } + + private async enqueueDelivery( + webhook: Webhook, + eventType: string, + payload: WebhookEventPayload, + ): Promise { + // Create delivery record + const requestId = crypto.randomUUID(); + const nonce = crypto.randomBytes(16).toString('hex'); + const timestamp = new Date().toISOString(); + + const signature = this.signPayload( + webhook.secret, + payload, + timestamp, + nonce, + ); + + const delivery = this.deliveryRepo.create({ + webhookId: webhook.id, + eventType, + payload: payload as any, + status: DeliveryStatus.PENDING, + maxRetries: webhook.maxRetries, + requestId, + nonce, + signature, + timestamp, + }); + + const saved = await this.deliveryRepo.save(delivery); + + // Increment delivery counter + this.deliveriesCounter.inc({ + webhook_id: webhook.id, + event_type: eventType, + status: DeliveryStatus.PENDING, + }); + + // Enqueue to BullMQ for async processing + await this.webhookQueue.add( + 'deliver', + { + deliveryId: saved.id, + webhookId: webhook.id, + url: webhook.url, + payload, + eventType, + signature, + nonce, + timestamp, + requestId, + retryCount: 0, + maxRetries: webhook.maxRetries, + retryIntervalMs: webhook.retryIntervalMs, + }, + { + attempts: webhook.maxRetries + 1, + backoff: { + type: 'exponential', + delay: webhook.retryIntervalMs, + }, + removeOnComplete: false, + removeOnFail: false, + }, + ); + } + + // ─── Delivery History ────────────────────────────────────────────────── + + async getDeliveries( + webhookId: string, + filter?: WebhookDeliveryFilterDto, + ): Promise<{ deliveries: WebhookDelivery[]; total: number; page: number; limit: number }> { + const page = filter?.page || 1; + const limit = filter?.limit || 20; + const skip = (page - 1) * limit; + + const query = this.deliveryRepo + .createQueryBuilder('d') + .where('d.webhookId = :webhookId', { webhookId }) + .orderBy('d.createdAt', 'DESC') + .skip(skip) + .take(limit); + + if (filter?.eventType) { + query.andWhere('d.eventType = :eventType', { eventType: filter.eventType }); + } + if (filter?.status) { + query.andWhere('d.status = :status', { status: filter.status }); + } + + const [deliveries, total] = await query.getManyAndCount(); + + return { deliveries, total, page, limit }; + } + + async getDelivery(id: string): Promise { + const delivery = await this.deliveryRepo.findOne({ where: { id } }); + if (!delivery) { + throw new NotFoundException(`Delivery ${id} not found`); + } + return delivery; + } + + // ─── Retry Logic ──────────────────────────────────────────────────────── + + async retryDelivery(deliveryId: string): Promise { + const delivery = await this.getDelivery(deliveryId); + const webhook = await this.findOne(delivery.webhookId); + + if (!webhook.enabled || webhook.disabled) { + throw new BadRequestException('Cannot retry: webhook is disabled'); + } + + // Reset delivery status + delivery.status = DeliveryStatus.PENDING; + delivery.retryCount = 0; + delivery.failureReason = null; + delivery.completedAt = null; + await this.deliveryRepo.save(delivery); + + // Re-enqueue + await this.webhookQueue.add( + 'deliver', + { + deliveryId: delivery.id, + webhookId: webhook.id, + url: webhook.url, + payload: delivery.payload, + eventType: delivery.eventType, + signature: delivery.signature, + nonce: delivery.nonce, + timestamp: delivery.timestamp, + requestId: delivery.requestId, + retryCount: 0, + maxRetries: webhook.maxRetries, + retryIntervalMs: webhook.retryIntervalMs, + }, + { + attempts: webhook.maxRetries + 1, + backoff: { + type: 'exponential', + delay: webhook.retryIntervalMs, + }, + removeOnComplete: false, + removeOnFail: false, + }, + ); + } + + // ─── Status & Health ─────────────────────────────────────────────────── + + async getWebhookStatus(id: string): Promise<{ + webhook: Webhook; + totalDeliveries: number; + successfulDeliveries: number; + failedDeliveries: number; + pendingDeliveries: number; + lastDelivery: WebhookDelivery | null; + }> { + const webhook = await this.findOne(id); + + const totalDeliveries = await this.deliveryRepo.count({ + where: { webhookId: id }, + }); + + const successfulDeliveries = await this.deliveryRepo.count({ + where: { webhookId: id, status: DeliveryStatus.DELIVERED }, + }); + + const failedDeliveries = await this.deliveryRepo.count({ + where: { webhookId: id, status: DeliveryStatus.FAILED }, + }); + + const pendingDeliveries = await this.deliveryRepo.count({ + where: { webhookId: id, status: DeliveryStatus.PENDING }, + }); + + const lastDelivery = await this.deliveryRepo.findOne({ + where: { webhookId: id }, + order: { createdAt: 'DESC' }, + }); + + return { + webhook, + totalDeliveries, + successfulDeliveries, + failedDeliveries, + pendingDeliveries, + lastDelivery, + }; + } + + // ─── Verification Helpers ─────────────────────────────────────────────── + + verifySignature( + payload: any, + signature: string, + secret: string, + timestamp: string, + nonce: string, + ): boolean { + const expected = this.signPayload(secret, payload, timestamp, nonce); + return crypto.timingSafeEqual(Buffer.from(expected), Buffer.from(signature)); + } + + // ─── Private Helpers ─────────────────────────────────────────────────── + + private generateSecret(): string { + return crypto.randomBytes(WEBHOOK_SECRET_BYTES).toString('hex'); + } + + private signPayload( + secret: string, + payload: any, + timestamp: string, + nonce: string, + ): string { + const hmac = crypto.createHmac(SIGNATURE_ALGORITHM, secret); + hmac.update(`${timestamp}.${nonce}.${JSON.stringify(payload)}`); + return hmac.digest('hex'); + } + + private matchesFilters( + payload: WebhookEventPayload, + filters: Record | null, + ): boolean { + if (!filters || Object.keys(filters).length === 0) return true; + + for (const [key, value] of Object.entries(filters)) { + const payloadValue = (payload as any)[key] ?? payload.data?.[key]; + if (payloadValue !== value) return false; + } + + return true; + } +}