import { BadRequestException, Injectable, Logger, NotFoundException, } from '@nestjs/common'; import { Cron } from '@nestjs/schedule'; import { createHmac, randomBytes } from 'crypto'; import type { CreateWebhookEndpointRequest, UpdateWebhookEndpointRequest, WebhookDeliveryDto, WebhookEndpointDto, WebhookEvent, } from '@mento/shared'; import { WEBHOOK_EVENTS } from '@mento/shared'; import { Prisma } from '@prisma/client'; import { PrismaService } from '../prisma/prisma.service'; @Injectable() export class WebhooksService { private readonly logger = new Logger(WebhooksService.name); constructor(private readonly prisma: PrismaService) {} async listEndpoints(organizationId: string): Promise { const rows = await this.prisma.webhookEndpoint.findMany({ where: { organizationId }, orderBy: { createdAt: 'desc' }, }); return rows.map((row) => this.toEndpointDto(row)); } async createEndpoint( organizationId: string, dto: CreateWebhookEndpointRequest, ): Promise { const name = dto.name?.trim(); const url = dto.url?.trim(); if (!name) throw new BadRequestException('نام وب‌هوک الزامی است'); if (!url || !/^https?:\/\//i.test(url)) { throw new BadRequestException('آدرس وب‌هوک باید با http یا https شروع شود'); } const events = this.normalizeEvents(dto.events); if (!events.length) throw new BadRequestException('حداقل یک رویداد انتخاب کنید'); const secret = dto.secret?.trim() || randomBytes(24).toString('hex'); const row = await this.prisma.webhookEndpoint.create({ data: { organizationId, name, url, secret, events, isActive: dto.isActive ?? true, }, }); return { ...this.toEndpointDto(row), secret }; } async updateEndpoint( organizationId: string, id: string, dto: UpdateWebhookEndpointRequest, ): Promise { const existing = await this.prisma.webhookEndpoint.findFirst({ where: { id, organizationId }, }); if (!existing) throw new NotFoundException('وب‌هوک پیدا نشد'); const data: Prisma.WebhookEndpointUpdateInput = {}; if (dto.name !== undefined) { const name = dto.name.trim(); if (!name) throw new BadRequestException('نام وب‌هوک الزامی است'); data.name = name; } if (dto.url !== undefined) { const url = dto.url.trim(); if (!url || !/^https?:\/\//i.test(url)) { throw new BadRequestException('آدرس وب‌هوک نامعتبر است'); } data.url = url; } if (dto.events !== undefined) { const events = this.normalizeEvents(dto.events); if (!events.length) throw new BadRequestException('حداقل یک رویداد انتخاب کنید'); data.events = events; } if (dto.secret !== undefined && dto.secret.trim()) { data.secret = dto.secret.trim(); } if (dto.isActive !== undefined) data.isActive = Boolean(dto.isActive); const row = await this.prisma.webhookEndpoint.update({ where: { id }, data, }); return this.toEndpointDto(row); } async removeEndpoint(organizationId: string, id: string): Promise { const existing = await this.prisma.webhookEndpoint.findFirst({ where: { id, organizationId }, }); if (!existing) throw new NotFoundException('وب‌هوک پیدا نشد'); await this.prisma.webhookEndpoint.delete({ where: { id } }); } async listDeliveries( organizationId: string, endpointId?: string, ): Promise { const rows = await this.prisma.webhookDelivery.findMany({ where: { organizationId, ...(endpointId ? { endpointId } : {}), }, orderBy: { createdAt: 'desc' }, take: 50, }); return rows.map((row) => ({ id: row.id, event: row.event, status: row.status, statusCode: row.statusCode, attempts: row.attempts, lastError: row.lastError, createdAt: row.createdAt.toISOString(), deliveredAt: row.deliveredAt?.toISOString() ?? null, })); } /** Fire-and-forget enqueue for domain events */ emit( organizationId: string, event: WebhookEvent, data: Record, ): void { void this.enqueue(organizationId, event, data).catch((err) => { this.logger.warn(`webhook enqueue failed: ${String(err)}`); }); } async enqueue( organizationId: string, event: WebhookEvent, data: Record, ): Promise { const endpoints = await this.prisma.webhookEndpoint.findMany({ where: { organizationId, isActive: true, events: { has: event }, }, select: { id: true }, }); if (!endpoints.length) return; const payload = { id: randomBytes(8).toString('hex'), event, createdAt: new Date().toISOString(), data, }; await this.prisma.webhookDelivery.createMany({ data: endpoints.map((ep) => ({ organizationId, endpointId: ep.id, event, payload: payload as Prisma.InputJsonValue, status: 'pending', nextAttemptAt: new Date(), })), }); } @Cron('*/1 * * * *') async processDue(): Promise { const now = new Date(); const pending = await this.prisma.webhookDelivery.findMany({ where: { status: { in: ['pending', 'failed'] }, nextAttemptAt: { lte: now }, attempts: { lt: 5 }, }, include: { endpoint: true }, orderBy: { nextAttemptAt: 'asc' }, take: 40, }); let done = 0; for (const row of pending) { if (!row.endpoint.isActive) { await this.prisma.webhookDelivery.update({ where: { id: row.id }, data: { status: 'cancelled', lastError: 'endpoint inactive' }, }); continue; } await this.deliver(row.id); done += 1; } return done; } private async deliver(id: string): Promise { const row = await this.prisma.webhookDelivery.findUnique({ where: { id }, include: { endpoint: true }, }); if (!row) return; const body = JSON.stringify(row.payload); const signature = createHmac('sha256', row.endpoint.secret) .update(body) .digest('hex'); try { const res = await fetch(row.endpoint.url, { method: 'POST', headers: { 'Content-Type': 'application/json', 'X-Montact-Event': row.event, 'X-Montact-Signature': `sha256=${signature}`, 'User-Agent': 'MontactCRM-Webhook/1.0', }, body, signal: AbortSignal.timeout(15000), }); if (res.ok) { await this.prisma.webhookDelivery.update({ where: { id }, data: { status: 'delivered', statusCode: res.status, attempts: row.attempts + 1, deliveredAt: new Date(), lastError: null, }, }); return; } const errText = await res.text().catch(() => ''); await this.fail(row.id, row.attempts + 1, res.status, errText.slice(0, 400)); } catch (err) { await this.fail(row.id, row.attempts + 1, null, String(err).slice(0, 400)); } } private async fail( id: string, attempts: number, statusCode: number | null, lastError: string, ) { const delayMin = Math.min(60, 2 ** Math.min(attempts, 5)); await this.prisma.webhookDelivery.update({ where: { id }, data: { status: attempts >= 5 ? 'dead' : 'failed', statusCode, attempts, lastError, nextAttemptAt: new Date(Date.now() + delayMin * 60_000), }, }); } private normalizeEvents(events: string[]): string[] { const allowed = new Set(WEBHOOK_EVENTS); return [...new Set((events ?? []).map((e) => e.trim()).filter((e) => allowed.has(e)))]; } private toEndpointDto(row: { id: string; name: string; url: string; events: string[]; isActive: boolean; createdAt: Date; updatedAt: Date; }): WebhookEndpointDto { return { id: row.id, name: row.name, url: row.url, events: row.events, isActive: row.isActive, createdAt: row.createdAt.toISOString(), updatedAt: row.updatedAt.toISOString(), }; } }