1
25import { Inject, Injectable } from "@nestjs/common";
26import { Queue } from "bullmq";
27import { jobTraceMetadata } from "../../common/trace-metadata.js";
28import { PrismaService } from "../../common/prisma.service.js";
29import { AuditService } from "../../common/audit.service.js";
30import { RedisService } from "../../common/redis.service.js";
31import { ApiError } from "../../common/errors.js";
32import {
33 aes256gcmEncrypt,
34 sha256hex
35} from "../../lib/crypto.js";
36import { assertRole } from "../../lib/authorize.js";
37import { loadEnv } from "../../config/env.js";
38import { parsePagination } from "../../lib/pagination.js";
39import {
40 assertSafeDeliveryUrl,
41 buildDeliveryEnvelope,
42 classifyDeliveryResult,
43 generateId,
44 generateWebhookSecret,
45 newDeliveryId,
46 postWebhookDelivery,
47 resolveSigningSecret,
48 validateWebhookEvents,
49 WEBHOOK_EVENT_NAMES
50} from "@factory/shared";
51
52
53const MAX_WEBHOOKS_PER_WORKSPACE = 20;
54
55const TEST_SEND_MESSAGE = "This is a test delivery from the Factory developer hub.";
56
57function secretPrefix(secretHash: string): string {
58
59
60
61 return `whsec_${secretHash.slice(0, 4)}`;
62}
63
64@Injectable()
65export class WebhooksService {
66 constructor(
67 @Inject(PrismaService) private readonly prisma: PrismaService,
68 @Inject(AuditService) private readonly audit: AuditService,
69 @Inject(RedisService) private readonly redis: RedisService
70 ) {}
71
72 private masterKey(): string {
73 const key = loadEnv().MASTER_ENCRYPTION_KEY;
74 if (!key) {
75 throw new ApiError(
76 "INTERNAL_ERROR",
77 500,
78 "MASTER_ENCRYPTION_KEY is not configured — webhook secrets cannot be stored."
79 );
80 }
81 return key;
82 }
83
84 private allowPrivate(): boolean {
85 return loadEnv().WEBHOOK_ALLOW_PRIVATE === "true";
86 }
87
88 private async requireAdmin(workspaceId: string, userId: string) {
89 await assertRole(this.prisma, { workspaceId, userId }, ["OWNER", "ADMIN"]);
90 }
91
92
93 private async validateUrl(url: string): Promise<string> {
94 let u: URL;
95 try {
96 u = new URL(url);
97 } catch {
98 throw ApiError.validation("Webhook URL must be a valid absolute URL.", [
99 { field: "url", issue: "invalid_url" }
100 ]);
101 }
102 if (u.protocol !== "https:" && u.protocol !== "http:") {
103 throw ApiError.validation("Webhook URL must be http(s).", [
104 { field: "url", issue: "invalid_scheme", received: u.protocol }
105 ]);
106 }
107 if (u.search) {
108 throw ApiError.validation("Webhook URL must not contain a query string.", [
109 { field: "url", issue: "query_string_not_allowed" }
110 ]);
111 }
112 try {
113 await assertSafeDeliveryUrl(u.toString(), this.allowPrivate());
114 } catch (err) {
115
116
117 if (err instanceof Error && err.name === "SsrfError") {
118 throw ApiError.validation("Webhook URL targets a private address.", [
119 { field: "url", issue: "private_address_not_allowed" }
120 ]);
121 }
122 throw err;
123 }
124 return u.toString();
125 }
126
127
128 private validateEvents(events: string[]): string[] {
129 const { valid, unknown } = validateWebhookEvents(events);
130 if (!valid) {
131 throw ApiError.validation("Unknown webhook event name.", [
132 {
133 field: "events",
134 issue: "unknown_event",
135 received: unknown,
136 validEvents: WEBHOOK_EVENT_NAMES
137 }
138 ]);
139 }
140 return events;
141 }
142
143 async create(
144 workspaceId: string,
145 userId: string,
146 input: { url: string; events?: string[]; secret?: string; active?: boolean },
147 req: any
148 ) {
149 await this.requireAdmin(workspaceId, userId);
150 const url = await this.validateUrl(input.url);
151 const events = this.validateEvents(input.events ?? ["*"]);
152
153 const count = await this.prisma.webhook.count({ where: { workspaceId } });
154 if (count >= MAX_WEBHOOKS_PER_WORKSPACE) {
155 throw ApiError.conflict("Workspace webhook limit reached (20).", [
156 { field: "url", issue: "webhook_limit", max: MAX_WEBHOOKS_PER_WORKSPACE }
157 ]);
158 }
159
160 const secret = input.secret ?? generateWebhookSecret();
161 const webhook = await this.prisma.webhook.create({
162 data: {
163 workspaceId,
164 url,
165 secretHash: sha256hex(secret),
166 secretEncrypted: Buffer.from(aes256gcmEncrypt(secret, this.masterKey()), "base64"),
167 events,
168 active: input.active ?? true,
169 createdById: userId
170 }
171 });
172 await this.audit.audit(req, req?.auth, {
173 action: "webhook.created",
174 resourceType: "webhook",
175 resourceId: webhook.id,
176 workspaceId,
177 metadata: { url, events }
178 });
179 return {
180 ...this.shape(webhook),
181 secret,
182 warning: "Store this secret now — it cannot be retrieved again."
183 };
184 }
185
186 async list(workspaceId: string, userId: string, query: { cursor?: string; limit?: string }) {
187 await this.requireAdmin(workspaceId, userId);
188 const { limit } = parsePagination(query);
189 const webhooks = await this.prisma.webhook.findMany({
190 where: { workspaceId },
191 orderBy: { createdAt: "asc" },
192 take: limit
193 });
194 return { data: webhooks.map((w) => this.shape(w)) };
195 }
196
197 async get(workspaceId: string, webhookId: string, userId: string) {
198 await this.requireAdmin(workspaceId, userId);
199 const webhook = await this.prisma.webhook.findFirst({ where: { id: webhookId, workspaceId } });
200 if (!webhook) throw ApiError.notFound("Webhook not found.");
201 const recent = await this.prisma.webhookDelivery.groupBy({
202 by: ["status"],
203 where: { webhookId },
204 _count: { _all: true }
205 });
206 const last = await this.prisma.webhookDelivery.findFirst({
207 where: { webhookId, status: "DELIVERED" },
208 orderBy: { deliveredAt: "desc" },
209 select: { deliveredAt: true }
210 });
211 const total = recent.reduce((n, r) => n + (r._count._all ?? 0), 0);
212 const failed =
213 (recent.find((r) => r.status === "FAILED")?._count._all ?? 0) +
214 (recent.find((r) => r.status === "DEAD_LETTER")?._count._all ?? 0) +
215 (recent.find((r) => r.status === "REJECTED")?._count._all ?? 0);
216 return {
217 ...this.shape(webhook),
218 recentDeliveries: {
219 total,
220 failed,
221 lastDeliveredAt: last?.deliveredAt ?? null
222 }
223 };
224 }
225
226
227 async update(
228 workspaceId: string,
229 webhookId: string,
230 userId: string,
231 patch: { url?: string; events?: string[]; active?: boolean },
232 req: any
233 ) {
234 await this.requireAdmin(workspaceId, userId);
235 const webhook = await this.prisma.webhook.findFirst({ where: { id: webhookId, workspaceId } });
236 if (!webhook) throw ApiError.notFound("Webhook not found.");
237 const data: Record<string, unknown> = {};
238 if (patch.url) data.url = await this.validateUrl(patch.url);
239 if (patch.events) data.events = this.validateEvents(patch.events);
240 if (patch.active !== undefined) data.active = patch.active;
241 const updated = await this.prisma.webhook.update({ where: { id: webhookId }, data });
242 await this.audit.audit(req, req?.auth, {
243 action: "webhook.updated",
244 resourceType: "webhook",
245 resourceId: webhookId,
246 workspaceId,
247 metadata: { fields: Object.keys(patch) }
248 });
249 return this.shape(updated);
250 }
251
252
253 async rotateSecret(workspaceId: string, webhookId: string, userId: string, req: any) {
254 await this.requireAdmin(workspaceId, userId);
255 const webhook = await this.prisma.webhook.findFirst({ where: { id: webhookId, workspaceId } });
256 if (!webhook) throw ApiError.notFound("Webhook not found.");
257 const secret = generateWebhookSecret();
258 const updated = await this.prisma.webhook.update({
259 where: { id: webhookId },
260 data: {
261 secretHash: sha256hex(secret),
262 secretEncrypted: Buffer.from(aes256gcmEncrypt(secret, this.masterKey()), "base64")
263 }
264 });
265 await this.audit.audit(req, req?.auth, {
266 action: "webhook.secret_rotated",
267 resourceType: "webhook",
268 resourceId: webhookId,
269 workspaceId
270
271 });
272 return {
273 ...this.shape(updated),
274 secret,
275 warning: "Store this secret now — it cannot be retrieved again. In-flight deliveries signed with the old secret will fail verification."
276 };
277 }
278
279
280 async remove(workspaceId: string, webhookId: string, userId: string, req: any) {
281 await this.requireAdmin(workspaceId, userId);
282 const webhook = await this.prisma.webhook.findFirst({ where: { id: webhookId, workspaceId } });
283 if (!webhook) throw ApiError.notFound("Webhook not found.");
284 await this.prisma.webhook.update({ where: { id: webhookId }, data: { active: false } });
285
286
287
288 await this.audit.audit(req, req?.auth, {
289 action: "webhook.revoked",
290 resourceType: "webhook",
291 resourceId: webhookId,
292 workspaceId,
293 metadata: { url: webhook.url }
294 });
295 return { deleted: true };
296 }
297
298
304 async test(workspaceId: string, webhookId: string, userId: string, req: any) {
305 await this.requireAdmin(workspaceId, userId);
306 const webhook = await this.prisma.webhook.findFirst({ where: { id: webhookId, workspaceId } });
307 if (!webhook) throw ApiError.notFound("Webhook not found.");
308 const ws = await this.prisma.workspace.findUnique({
309 where: { id: webhook.workspaceId },
310 select: { id: true, name: true, slug: true }
311 });
312 const eventId = generateId("evt");
313 const envelope = buildDeliveryEnvelope({
314 id: eventId,
315 type: "webhook.test",
316 workspaceId: webhook.workspaceId,
317 workspace: ws ?? undefined,
318 data: { message: TEST_SEND_MESSAGE, sentAt: new Date().toISOString() },
319 test: true
320 });
321 const deliveryId = newDeliveryId();
322 await this.prisma.webhookDelivery.create({
323 data: {
324 id: deliveryId,
325 webhookId,
326 event: "webhook.test",
327 payload: envelope as object,
328 status: "PENDING",
329 attempts: 1
330 }
331 });
332 const resolved = this.resolveSecretOrThrow(webhook);
333 const result = await postWebhookDelivery(
334 {
335 url: webhook.url,
336 secret: resolved,
337 event: "webhook.test",
338 envelope,
339 deliveryId,
340 attempt: 1
341 },
342 { allowPrivate: this.allowPrivate() }
343 );
344 const outcome = classifyDeliveryResult(result);
345 await this.prisma.webhookDelivery.update({
346 where: { id: deliveryId },
347 data: {
348 status: outcome,
349 statusCode: result.status || null,
350 error:
351 outcome === "DELIVERED"
352 ? null
353 : result.transportError ?? `non-2xx: HTTP ${result.status}`,
354 deliveredAt: outcome === "DELIVERED" ? new Date() : null,
355 latencyMs: result.latencyMs
356 }
357 });
358 await this.audit.audit(req, req?.auth, {
359 action: "webhook.test_sent",
360 resourceType: "webhook",
361 resourceId: webhookId,
362 workspaceId,
363 metadata: { deliveryId, outcome, statusCode: result.status || null }
364 });
365 return {
366 deliveryId,
367 status: outcome,
368 statusCode: result.status || null,
369 attempts: 1,
370 latencyMs: result.latencyMs
371 };
372 }
373
374
375 async deliveries(
376 workspaceId: string,
377 webhookId: string,
378 userId: string,
379 query: {
380 cursor?: string;
381 limit?: string;
382 status?: string;
383 event?: string;
384 since?: string;
385 until?: string;
386 }
387 ) {
388 await this.requireAdmin(workspaceId, userId);
389 const webhook = await this.prisma.webhook.findFirst({ where: { id: webhookId, workspaceId } });
390 if (!webhook) throw ApiError.notFound("Webhook not found.");
391 const { cursor, limit } = parsePagination(query);
392 const since = query.since ? new Date(query.since) : undefined;
393 const until = query.until ? new Date(query.until) : undefined;
394 const rows = await this.prisma.webhookDelivery.findMany({
395 where: {
396 webhookId,
397 ...(query.status ? { status: query.status as never } : {}),
398 ...(query.event ? { event: query.event } : {}),
399 ...(since || until
400 ? {
401 createdAt: {
402 ...(since ? { gte: since } : {}),
403 ...(until ? { lte: until } : {})
404 }
405 }
406 : {})
407 },
408 ...(cursor ? { cursor: { id: cursor.id }, skip: 1 } : {}),
409 orderBy: { createdAt: "desc" },
410 take: limit + 1
411 });
412 const page = rows.slice(0, limit);
413 const next = rows.length > limit ? page[page.length - 1] : undefined;
414 return {
415 data: page.map((d) => this.shapeDelivery(d)),
416 nextCursor: next ? Buffer.from(JSON.stringify({ id: next.id })).toString("base64url") : null
417 };
418 }
419
420
421 async delivery(workspaceId: string, webhookId: string, deliveryId: string, userId: string) {
422 await this.requireAdmin(workspaceId, userId);
423 const webhook = await this.prisma.webhook.findFirst({ where: { id: webhookId, workspaceId } });
424 if (!webhook) throw ApiError.notFound("Webhook not found.");
425 const d = await this.prisma.webhookDelivery.findFirst({
426 where: { id: deliveryId, webhookId }
427 });
428 if (!d) throw ApiError.notFound("Delivery not found.");
429 return {
430 ...this.shapeDelivery(d),
431 payload: d.payload
432 };
433 }
434
435
440 async retryDelivery(
441 workspaceId: string,
442 webhookId: string,
443 deliveryId: string,
444 userId: string,
445 req: any
446 ) {
447 await this.requireAdmin(workspaceId, userId);
448 const webhook = await this.prisma.webhook.findFirst({ where: { id: webhookId, workspaceId } });
449 if (!webhook) throw ApiError.notFound("Webhook not found.");
450 const d = await this.prisma.webhookDelivery.findFirst({ where: { id: deliveryId, webhookId } });
451 if (!d) throw ApiError.notFound("Delivery not found.");
452 if (d.status === "DELIVERED") {
453 throw ApiError.conflict("Delivery already succeeded — nothing to retry.", [
454 { field: "deliveryId", issue: "already_delivered" }
455 ]);
456 }
457 const envelope = d.payload as Record<string, unknown>;
458 await this.prisma.webhookDelivery.update({
459 where: { id: deliveryId },
460 data: { status: "PENDING", attempts: 0, error: null, nextRetryAt: null }
461 });
462 await this.enqueueDeliveryJob(webhook, d, envelope, 1);
463 await this.audit.audit(req, req?.auth, {
464 action: "webhook.delivery_retried",
465 resourceType: "webhook",
466 resourceId: webhookId,
467 workspaceId,
468 metadata: { deliveryId }
469 });
470 return { deliveryId, status: "PENDING", attempts: 0 };
471 }
472
473
479
486 async resubmit(
487 workspaceId: string,
488 webhookId: string,
489 userId: string,
490 input: {
491 deliveryId?: string;
492 all_dead?: boolean;
493 window?: { since?: string; until?: string };
494 },
495 req: any
496 ): Promise<{ resubmitted: number; deliveryIds: string[]; originDeliveryIds: string[] }> {
497 await this.requireAdmin(workspaceId, userId);
498 const webhook = await this.prisma.webhook.findFirst({ where: { id: webhookId, workspaceId } });
499 if (!webhook) throw ApiError.notFound("Webhook not found.");
500 const where: Record<string, unknown> = { webhookId };
501 if (input.deliveryId) {
502 const d = await this.prisma.webhookDelivery.findFirst({
503 where: { id: input.deliveryId, webhookId }
504 });
505 if (!d) throw ApiError.notFound("Delivery not found.");
506 if (d.status === "DELIVERED" || d.status === "PENDING") {
507 throw ApiError.conflict("Only FAILED, REJECTED or DEAD_LETTER deliveries can be resubmitted.", [
508 { field: "deliveryId", issue: "not_retriable", status: d.status }
509 ]);
510 }
511 where.id = input.deliveryId;
512 } else if (input.window?.since || input.window?.until) {
513 where.status = { in: ["DEAD_LETTER", "FAILED", "REJECTED"] as const };
514 where.createdAt = {
515 ...(input.window.since ? { gte: new Date(input.window.since) } : {}),
516 ...(input.window.until ? { lte: new Date(input.window.until) } : {})
517 };
518 } else {
519
520 where.status = "DEAD_LETTER";
521 }
522 const rows = await this.prisma.webhookDelivery.findMany({
523 where,
524 orderBy: { createdAt: "desc" },
525 take: 200
526 });
527 const deliveryIds: string[] = [];
528 const originDeliveryIds: string[] = [];
529 for (const d of rows) {
530 const envelope = d.payload as Record<string, unknown>;
531 const newId = newDeliveryId();
532 await this.prisma.webhookDelivery.create({
533 data: {
534 id: newId,
535 webhookId,
536 event: d.event,
537 payload: d.payload as object,
538 status: "PENDING",
539 attempts: 0,
540 originDeliveryId: d.id
541 }
542 });
543 await this.enqueueDeliveryJob(webhook, d, envelope, 1, newId);
544 deliveryIds.push(newId);
545 originDeliveryIds.push(d.id);
546 }
547 await this.audit.audit(req, req?.auth, {
548 action: "webhook.deliveries_resubmitted",
549 resourceType: "webhook",
550 resourceId: webhookId,
551 workspaceId,
552 metadata: { mode: input.deliveryId ? "single" : input.window ? "window" : "all_dead", count: deliveryIds.length }
553 });
554 return { resubmitted: deliveryIds.length, deliveryIds, originDeliveryIds };
555 }
556
557
558
559 private resolveSecretOrThrow(webhook: {
560 secretEncrypted: Uint8Array | null;
561 secretHash: string;
562 }): string {
563 const resolved = resolveSigningSecret(
564 this.masterKey(),
565 webhook.secretEncrypted,
566 webhook.secretHash
567 );
568 if (!resolved) {
569 throw new ApiError(
570 "INTERNAL_ERROR",
571 500,
572 "Webhook signing secret is unrecoverable — recreate the webhook."
573 );
574 }
575 return resolved.secret;
576 }
577
578
579 private async enqueueDeliveryJob(
580 webhook: { id: string; workspaceId: string },
581 delivery: { id: string; event: string },
582 envelope: Record<string, unknown>,
583 attempt: number,
584 deliveryId?: string
585 ): Promise<void> {
586 const targetId = deliveryId ?? delivery.id;
587 const eventId = typeof envelope.id === "string" ? envelope.id : generateId("evt");
588 const occurredAt = typeof envelope.occurredAt === "string" ? envelope.occurredAt : undefined;
589 try {
590 const q = new Queue("webhooks", { connection: this.redis.getRedis().options as never });
591 try {
592 await q.add("webhooks", {
593 kind: "row",
594 webhookId: webhook.id,
595 event: delivery.event,
596 payload: envelope.data ?? {},
597 deliveryId: targetId,
598 eventId,
599 ...(occurredAt ? { occurredAt } : {}),
600 attempt
601 },
602
603 { ...jobTraceMetadata() }
604 );
605 } finally {
606 await q.close();
607 }
608 } catch (err) {
609
610
611 console.warn(`webhook delivery ${targetId} enqueue failed: ${(err as Error).message}`);
612 }
613 }
614
615 private shape(webhook: {
616 id: string;
617 url: string;
618 events: string[];
619 active: boolean;
620 createdById: string;
621 workspaceId: string;
622 createdAt: Date;
623 updatedAt: Date;
624 secretHash: string;
625 }) {
626 return {
627 id: webhook.id,
628 workspaceId: webhook.workspaceId,
629 url: webhook.url,
630 events: webhook.events,
631 active: webhook.active,
632 secretPrefix: secretPrefix(webhook.secretHash),
633 createdById: webhook.createdById,
634 createdAt: webhook.createdAt,
635 updatedAt: webhook.updatedAt
636 };
637 }
638
639 private shapeDelivery(d: {
640 id: string;
641 webhookId: string;
642 event: string;
643 status: string;
644 attempts: number;
645 statusCode: number | null;
646 error: string | null;
647 deliveredAt: Date | null;
648 nextRetryAt: Date | null;
649 createdAt: Date;
650 originDeliveryId: string | null;
651 latencyMs?: number | null;
652 payload?: unknown;
653 }) {
654 const env = (d.payload ?? {}) as Record<string, unknown>;
655 return {
656 id: d.id,
657 webhookId: d.webhookId,
658 event: d.event,
659 eventId: typeof env.id === "string" ? env.id : undefined,
660 status: d.status,
661 attempts: d.attempts,
662 statusCode: d.statusCode,
663 error: d.error,
664 deliveredAt: d.deliveredAt,
665 nextRetryAt: d.nextRetryAt,
666 createdAt: d.createdAt,
667 originDeliveryId: d.originDeliveryId,
668 latencyMs: d.latencyMs ?? null
669 };
670 }
671}
672
