Add public API keys, outbound webhooks, and bulk SMS campaigns with settings UI and org modules.
This commit is contained in:
@@ -0,0 +1,296 @@
|
||||
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<WebhookEndpointDto[]> {
|
||||
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<WebhookEndpointDto & { secret: string }> {
|
||||
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<WebhookEndpointDto> {
|
||||
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<void> {
|
||||
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<WebhookDeliveryDto[]> {
|
||||
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<string, unknown>,
|
||||
): 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<string, unknown>,
|
||||
): Promise<void> {
|
||||
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<number> {
|
||||
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<void> {
|
||||
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-Mento-Event': row.event,
|
||||
'X-Mento-Signature': `sha256=${signature}`,
|
||||
'User-Agent': 'MentoCRM-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<string>(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(),
|
||||
};
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user