import { Controller, Post, Body, Headers, Param, Req, HttpCode, HttpStatus, Logger, UnauthorizedException, } from "@nestjs/common"; import { ConfigService } from "@nestjs/config"; import { PrismaService } from "../prisma/prisma.service"; import type { IntegrationProvider } from "@prisma/client"; import * as crypto from "crypto"; import type { Request } from "express"; /** Extended Request with raw body (added by NestJS rawBody middleware) */ interface RawBodyRequest extends Request { rawBody?: Buffer; } import { WebhookIdempotencyService } from "./webhook-idempotency.service"; import { SelfHostedClientsService } from "./clients/self-hosted-clients.service"; import { ChatwootSyncService } from "../support/chatwoot-sync.service"; import { EventsGateway } from "../websocket/events.gateway"; function timingSafeCompare(provided?: string, expected?: string): boolean { if (!provided || !expected) return false; const bufProvided = Buffer.from(provided); const bufExpected = Buffer.from(expected); if (bufProvided.length !== bufExpected.length) return false; return crypto.timingSafeEqual(bufProvided, bufExpected); } /** * Verify a raw-body HMAC webhook signature (Chatwoot). * * Chatwoot signs the RAW request body with the shared webhook secret and * ships the digest in `X-Chatwoot-Signature` as `sha256=` (older * installs may use `sha1=`). Some versions include the * `X-Chatwoot-Timestamp` in the signed content (`.`) — we * accept both forms since both still require a valid HMAC under the shared * secret. We recompute over the raw body (captured via Nest's * `rawBody: true` — see main.ts) and compare in constant time. */ function verifyRawBodyHmac( rawBody: Buffer | undefined, signature: string | undefined, secret: string, timestamp?: string, allowSha1 = false, ): boolean { if (!rawBody || !signature || !secret) return false; const provided = signature.replace(/^sha256=/i, "").replace(/^sha1=/i, ""); const messages = [rawBody]; if (timestamp) { messages.push(Buffer.from(`${timestamp}.${rawBody.toString("utf8")}`, "utf8")); } const candidates: string[] = []; for (const msg of messages) { candidates.push(crypto.createHmac("sha256", secret).update(msg).digest("hex")); if (allowSha1) { candidates.push(crypto.createHmac("sha1", secret).update(msg).digest("hex")); } } return candidates.some((expected) => timingSafeCompare(provided, expected)); } // ═══ Webhook Payload Types ═════════════════════════════════ interface ChatwootWebhookPayload { event: string; message_type?: string; content?: string; conversation?: { id: number; status: string; contact_inbox?: { contact_id: number }; assignee?: { id: number }; }; contact?: { id: number; name: string; email: string; phone_number?: string; }; message?: { id: number; content: string; message_type: "incoming" | "outgoing" | "activity"; }; } interface DocumensoWebhookPayload { event: string; documentId?: number; document?: { id: number; title: string; status: string; }; data?: { id?: number; }; } interface FormbricksWebhookPayload { event?: string; type?: string; surveyId?: string; responseId?: string; data?: { id?: string; surveyId?: string; userId?: string; finished?: boolean; data?: Record; }; } interface ListmonkWebhookPayload { event: string; subscriber_id?: number; subscriber?: { id: number; email: string; name: string; status: string; lists?: Array<{ id: number; name: string }>; }; data?: { subscriber?: { id: number; email: string; name: string; status: string; lists?: Array<{ id: number; name: string }>; }; }; } interface LiveKitWebhookPayload { [key: string]: unknown; event: string; room?: { sid: string; name: string; empty_timeout: number; }; egressInfo?: { roomName?: string; status?: string; }; recording?: { sid: string; room_sid: string; name: string; status: string; }; participant?: { id: string; identity: string; name: string; }; track?: { sid: string; name: string; kind: string; }; } /** Generic webhook body for Cal.com (already well-structured in handler) */ type CalComWebhookPayload = Record; /** * Verify a Standard Webhooks spec signature (Formbricks Hub). * * Formbricks implements the Standard Webhooks spec: * - `webhook-id` – unique event ID * - `webhook-timestamp` – unix seconds of the delivery * - `webhook-signature` – space-separated `v1,` digests * * The signed content is `${id}.${timestamp}.${rawBody}`; the digest is * base64(HMAC-SHA256(secretBytes, content)) where secretBytes is the * `whsec_`-prefixed secret with the prefix stripped and base64-decoded * (falls back to the raw string if it doesn't round-trip). At least one * of the presented signatures must match; deliveries older than 5 minutes * are rejected to blunt replay of captured webhooks. */ function verifyStandardWebhooksSignature( rawBody: Buffer | undefined, signatureHeader: string | undefined, idHeader: string | undefined, timestampHeader: string | undefined, secret: string, ): boolean { if (!rawBody || !signatureHeader || !idHeader || !timestampHeader || !secret) { return false; } const nowSec = Math.floor(Date.now() / 1000); const ts = Number(timestampHeader); if (!Number.isFinite(ts) || Math.abs(nowSec - ts) > 300) return false; let secretBytes: Buffer; const stripped = secret.replace(/^whsec_/i, ""); try { const decoded = Buffer.from(stripped, "base64"); // Guard against silent truncation of non-base64 secrets: only use the // decoded bytes if the round-trip is lossless. if (decoded.toString("base64").replace(/=+$/, "") === stripped.replace(/=+$/, "")) { secretBytes = decoded; } else { secretBytes = Buffer.from(secret, "utf8"); } } catch { secretBytes = Buffer.from(secret, "utf8"); } const signedContent = `${idHeader}.${timestampHeader}.${rawBody.toString("utf8")}`; const expected = crypto .createHmac("sha256", secretBytes) .update(signedContent) .digest("base64"); return signatureHeader.split(/\s+/).some((entry) => { const [version, sig] = entry.split(",", 2); if (version !== "v1" || !sig) return false; return timingSafeCompare(sig, expected); }); } @Controller("integrations/webhooks") export class IntegrationsWebhookController { private readonly logger = new Logger(IntegrationsWebhookController.name); constructor( private readonly configService: ConfigService, private readonly prisma: PrismaService, private readonly idempotency: WebhookIdempotencyService, private readonly selfHostedClients: SelfHostedClientsService, private readonly chatwootSync: ChatwootSyncService, private readonly eventsGateway: EventsGateway, ) {} /** * SECURITY: fail closed. When a webhook secret is not configured we reject * the request in production instead of silently accepting unauthenticated * webhook deliveries (which would let attackers forge booking/cancel/ * sync events). */ private requireSecret(envName: string, configured: string): void { if (!configured && process.env.NODE_ENV === "production") { this.logger.error(`${envName} is not configured — rejecting webhook`); throw new UnauthorizedException(`${envName} is not configured`); } } private async checkIdempotency(provider: string, body: Record | ChatwootWebhookPayload | DocumensoWebhookPayload | FormbricksWebhookPayload | ListmonkWebhookPayload | LiveKitWebhookPayload): Promise { const eventKey = this.idempotency.generateEventKey(provider, body); const isDuplicate = await this.idempotency.checkAndMark(eventKey); if (isDuplicate) { this.logger.warn(`Duplicate webhook ignored: ${provider}:${eventKey}`); } return isDuplicate; } /** * Verify a LiveKit webhook JWT: HS256-signed with LIVEKIT_API_SECRET, * with `iss` = LIVEKIT_API_KEY, rejecting expired tokens. When the raw * body is available (Nest `rawBody: true`), it also verifies the JWT's * `sha256` claim — standard base64 of sha256(raw body) — matching the * reference implementation in `livekit-server-sdk`'s WebhookReceiver. * This binds the signed token to the exact delivered bytes, so a captured * JWT cannot be replayed against a different payload. */ private verifyLiveKitJwt( authorization: string | undefined, apiKey: string, apiSecret: string, rawBody?: Buffer | string, ): boolean { if (!authorization?.startsWith("Bearer ")) return false; const token = authorization.slice("Bearer ".length).trim(); const parts = token.split("."); if (parts.length !== 3) return false; const [headerB64, payloadB64, signatureB64] = parts; try { const payload = JSON.parse( Buffer.from(payloadB64, "base64url").toString("utf8"), ); if (payload.iss !== apiKey) return false; const nowSec = Math.floor(Date.now() / 1000); if (typeof payload.exp === "number" && payload.exp < nowSec) return false; if (typeof payload.nbf === "number" && payload.nbf > nowSec + 60) return false; const signingInput = `${headerB64}.${payloadB64}`; const expected = crypto .createHmac("sha256", apiSecret) .update(signingInput) .digest("base64url"); if (!timingSafeCompare(expected, signatureB64)) return false; // LiveKit always includes the body hash; require it and verify it. if (typeof payload.sha256 !== "string") return false; if (rawBody === undefined) { // Cannot bind the token to the delivered bytes without the raw // body. Fail closed in production, tolerate in dev (the body hash // feature depends on Nest rawBody:true, which is configured). return process.env.NODE_ENV !== "production"; } const bodyBuf = typeof rawBody === "string" ? Buffer.from(rawBody, "utf8") : rawBody; const bodyHash = crypto .createHash("sha256") .update(bodyBuf) .digest("base64"); return timingSafeCompare(payload.sha256, bodyHash); } catch { return false; } } @Post("calcom") @HttpCode(HttpStatus.OK) async handleCalComWebhook( @Body() body: CalComWebhookPayload, @Headers("x-cal-secret-key") secret?: string, ) { const expected = this.configService.get("CALCOM_WEBHOOK_SECRET", ""); this.requireSecret("CALCOM_WEBHOOK_SECRET", expected); if (expected && !timingSafeCompare(secret, expected)) { throw new UnauthorizedException("Invalid Cal.com webhook secret"); } // Idempotency check if (await this.checkIdempotency("calcom", body)) { return { received: true, duplicate: true }; } const eventType = String(body?.triggerEvent || body?.event || "unknown"); const payload = (body?.payload || body) as CalComWebhookPayload; this.logger.log(`Cal.com webhook received: ${eventType}`); // Handle all Cal.com event types await this.handleCalComEvent(eventType, payload); return { received: true, event: eventType }; } private async handleCalComEvent(eventType: string, payload: CalComWebhookPayload) { const uid = String(payload?.uid || payload?.bookingId || ""); if (!uid) return; const booking = await this.prisma.booking.findFirst({ where: { OR: [{ calendarEventId: uid }, { meetLink: { contains: uid } }] }, include: { episode: { include: { podcast: { select: { workspaceId: true }, }, }, }, }, }); if (!booking) { this.logger.warn(`No booking found for Cal.com UID: ${uid}`); return; } switch (eventType) { case "BOOKING_CREATED": case "BOOKING_CONFIRMED": await this.prisma.booking.update({ where: { id: booking.id }, data: { isConfirmed: true, updatedAt: new Date() }, }); this.logger.log(`Booking ${booking.id} confirmed via Cal.com`); if (booking.episode?.podcast?.workspaceId) { this.eventsGateway.notifyCalComBookingCreated( booking.episode.podcast.workspaceId, { id: booking.id, isConfirmed: true }, ); } break; case "BOOKING_CANCELLED": case "BOOKING_REJECTED": await this.prisma.booking.update({ where: { id: booking.id }, data: { isConfirmed: false, updatedAt: new Date() }, }); this.logger.log(`Booking ${booking.id} cancelled via Cal.com`); if (booking.episode?.podcast?.workspaceId) { this.eventsGateway.notifyCalComBookingCancelled( booking.episode.podcast.workspaceId, booking.id, ); } break; case "BOOKING_RESCHEDULED": const newStart = String(payload?.startTime || payload?.rescheduledStartTime || ""); const newEnd = String(payload?.endTime || payload?.rescheduledEndTime || ""); if (newStart && newEnd) { await this.prisma.booking.update({ where: { id: booking.id }, data: { startTime: new Date(newStart), endTime: new Date(newEnd), updatedAt: new Date(), }, }); this.logger.log(`Booking ${booking.id} rescheduled to ${newStart}`); if (booking.episode?.podcast?.workspaceId) { this.eventsGateway.notifyCalComBookingRescheduled( booking.episode.podcast.workspaceId, booking.id, ); } } break; case "BOOKING_NO_SHOW": this.logger.log(`Booking ${booking.id} no-show (no status field available)`); break; case "BOOKING_ATTENDEE_ADDED": case "BOOKING_ATTENDEE_REMOVED": // Attendee changes - could sync to episode guests this.logger.log(`Attendee change for booking ${booking.id}: ${eventType}`); break; default: this.logger.debug(`Unhandled Cal.com event type: ${eventType}`); } } @Post("chatwoot/:accountId") @HttpCode(HttpStatus.OK) async handleChatwootWebhook( @Param("accountId") accountId: string, @Req() req: RawBodyRequest, @Body() body: ChatwootWebhookPayload, @Headers("x-chatwoot-signature") signature?: string, @Headers("x-chatwoot-timestamp") timestamp?: string, ) { const expected = this.configService.get("CHATWOOT_WEBHOOK_SECRET", ""); this.requireSecret("CHATWOOT_WEBHOOK_SECRET", expected); // Chatwoot signs the raw body (HMAC-SHA1 or SHA256), with some // versions prefixing the timestamp into the signed content. if ( expected && !verifyRawBodyHmac(req.rawBody, signature, expected, timestamp, true) ) { throw new UnauthorizedException("Invalid Chatwoot webhook signature"); } // Idempotency check if (await this.checkIdempotency("chatwoot", body)) { return { received: true, duplicate: true }; } const event = body?.event || "unknown"; this.logger.log(`Chatwoot webhook: ${event} for account ${accountId}`); await this.handleChatwootEvent(event, body, accountId); return { received: true, event, accountId }; } private async handleChatwootEvent(event: string, body: ChatwootWebhookPayload, accountId: string) { switch (event) { case "message_created": if (body?.message_type === "incoming" && body?.content) { const conversationId = body?.conversation?.id; const contactId = body?.conversation?.contact_inbox?.contact_id; const messageId = body?.message?.id; this.logger.log(`Incoming Chatwoot message in conversation ${conversationId} from contact ${contactId}`); // Sync contact to Guest record (existing behavior) if (contactId) { await this.syncChatwootContact(contactId, accountId); } // Sync message into internal support ticket/record (new) if (conversationId && messageId && body.content) { await this.chatwootSync.syncIncomingMessage( accountId, conversationId, messageId, body.content, "incoming", ); } } break; case "conversation_created": this.logger.log(`New Chatwoot conversation: ${body?.conversation?.id}`); break; case "conversation_status_changed": { const newStatus = body?.conversation?.status; const conversationId = body?.conversation?.id; this.logger.log(`Conversation ${conversationId} status changed to ${newStatus}`); if (conversationId && newStatus) { await this.chatwootSync.syncConversationStatus(accountId, conversationId, newStatus); } break; } case "conversation_assigned": { const conversationId = body?.conversation?.id; const assigneeId = body?.conversation?.assignee?.id; this.logger.log(`Conversation ${conversationId} assigned to ${assigneeId}`); if (conversationId) { await this.chatwootSync.syncConversationAssignment(accountId, conversationId, assigneeId); } break; } case "contact_created": case "contact_updated": const contactId = body?.contact?.id; if (contactId) { await this.syncChatwootContact(contactId, accountId); } break; default: this.logger.debug(`Unhandled Chatwoot event: ${event}`); } } private async syncChatwootContact(contactId: number, accountId: string) { try { // Find the workspace that has this Chatwoot integration const integration = await this.prisma.workspaceIntegration.findFirst({ where: { provider: "chatwoot" as IntegrationProvider, status: "connected", metadata: { path: ["accountId"], equals: accountId }, }, }); if (!integration) { this.logger.warn(`No workspace integration found for Chatwoot account ${accountId}`); return; } // Get the API key from the integration const apiKey = integration.accessToken; if (!apiKey) { this.logger.warn(`No API key found for Chatwoot integration ${integration.id}`); return; } // Fetch contact details from Chatwoot const contact = await this.selfHostedClients.getChatwootContact(apiKey, contactId); if (!contact?.email) { this.logger.warn(`Chatwoot contact ${contactId} has no email, skipping sync`); return; } // Find the first podcast in the workspace to associate the guest const workspace = await this.prisma.workspace.findUnique({ where: { id: integration.workspaceId }, include: { podcasts: { take: 1 } }, }); const podcastId = workspace?.podcasts?.[0]?.id; if (!podcastId) { this.logger.warn(`No podcast found for workspace ${integration.workspaceId}, cannot create guest`); return; } // Upsert the guest record const existingGuest = await this.prisma.guest.findFirst({ where: { podcastId, email: contact.email }, }); if (existingGuest) { // Update existing guest with Chatwoot metadata const currentCustomFields = (existingGuest.customFields as Record) || {}; await this.prisma.guest.update({ where: { id: existingGuest.id }, data: { fullName: contact.name || existingGuest.fullName, lastContactAt: new Date(), customFields: { ...currentCustomFields, chatwootContactId: contactId, chatwootAccountId: accountId, chatwootPhone: contact.phone_number, chatwootCustomAttributes: contact.custom_attributes, }, }, }); this.logger.log(`Updated guest ${existingGuest.id} from Chatwoot contact ${contactId}`); } else { // Create new guest from Chatwoot contact await this.prisma.guest.create({ data: { podcastId, email: contact.email, fullName: contact.name || contact.email.split("@")[0], status: "prospect", customFields: { chatwootContactId: contactId, chatwootAccountId: accountId, chatwootPhone: contact.phone_number, chatwootCustomAttributes: contact.custom_attributes, source: "chatwoot", }, }, }); this.logger.log(`Created guest from Chatwoot contact ${contactId} (${contact.email})`); } } catch (error: unknown) { const message = error instanceof Error ? error.message : String(error); this.logger.error(`Failed to sync Chatwoot contact ${contactId}: ${message}`); } } @Post("documenso") @HttpCode(HttpStatus.OK) async handleDocumensoWebhook( @Body() body: DocumensoWebhookPayload, @Headers("x-documenso-secret") secret?: string, ) { const expected = this.configService.get("DOCUMENSO_WEBHOOK_SECRET", ""); this.requireSecret("DOCUMENSO_WEBHOOK_SECRET", expected); // Documenso passes the webhook secret plainly in `X-Documenso-Secret`; // verify with a constant-time comparison. if (expected && !timingSafeCompare(secret, expected)) { throw new UnauthorizedException("Invalid Documenso webhook secret"); } // Idempotency check if (await this.checkIdempotency("documenso", body)) { return { received: true, duplicate: true }; } const event = body?.event || "unknown"; const documentId = body?.documentId || body?.data?.id; this.logger.log(`Documenso webhook: ${event} docId=${documentId ?? ""}`); // Handle document status changes if (documentId && (event === "document_completed" || event === "document_declined")) { // Could update local document record status this.logger.log(`Document ${documentId} status: ${event}`); } return { received: true, event, documentId }; } @Post("formbricks") @HttpCode(HttpStatus.OK) async handleFormbricksWebhook( @Req() req: RawBodyRequest, @Body() body: FormbricksWebhookPayload, @Headers("webhook-signature") signature?: string, @Headers("webhook-id") webhookId?: string, @Headers("webhook-timestamp") webhookTimestamp?: string, ) { const expected = this.configService.get("FORMBRICKS_WEBHOOK_SECRET", ""); this.requireSecret("FORMBRICKS_WEBHOOK_SECRET", expected); // Formbricks (Hub) implements the Standard Webhooks spec: the signature // is base64 HMAC-SHA256 over `id.timestamp.rawBody`. if ( expected && !verifyStandardWebhooksSignature( req.rawBody, signature, webhookId, webhookTimestamp, expected, ) ) { throw new UnauthorizedException("Invalid Formbricks webhook signature"); } // Idempotency check if (await this.checkIdempotency("formbricks", body)) { return { received: true, duplicate: true }; } const event = body?.event || body?.type || "unknown"; const surveyId = body?.surveyId || body?.data?.surveyId; const responseId = body?.responseId || body?.data?.id; this.logger.log(`Formbricks webhook: ${event} surveyId=${surveyId ?? ""} responseId=${responseId ?? ""}`); // If survey response is submitted, log journey event or store metadata if (body?.data?.userId) { const guest = await this.prisma.guest.findFirst({ where: { id: body.data.userId }, }); if (guest) { await this.prisma.guestJourneyEvent.create({ data: { guestId: guest.id, eventType: "follow_up", title: "Completed Survey", details: `Completed survey ${surveyId || ""}`, metadata: JSON.parse(JSON.stringify(body?.data || {})), }, }); this.logger.log(`Recorded survey completion journey event for guest ${guest.id}`); } } return { received: true, event, surveyId, responseId }; } @Post("listmonk") @HttpCode(HttpStatus.OK) async handleListmonkWebhook( @Body() body: ListmonkWebhookPayload, @Headers("x-listmonk-secret") secret?: string, ) { const expected = this.configService.get("LISTMONK_WEBHOOK_SECRET", ""); this.requireSecret("LISTMONK_WEBHOOK_SECRET", expected); if (expected && !timingSafeCompare(secret, expected)) { throw new UnauthorizedException("Invalid Listmonk webhook secret"); } // Idempotency check if (await this.checkIdempotency("listmonk", body)) { return { received: true, duplicate: true }; } const event = body?.event || "unknown"; const subscriberId = body?.data?.subscriber?.id || body?.subscriber_id; this.logger.log(`Listmonk webhook: ${event} subscriberId=${subscriberId ?? ""}`); await this.handleListmonkEvent(event, body); return { received: true, event, subscriberId }; } private async handleListmonkEvent(event: string, body: ListmonkWebhookPayload) { const subscriber = body?.data?.subscriber || body?.subscriber; if (!subscriber) return; const email = subscriber.email; const name = subscriber.name || ""; const status = subscriber.status || "enabled"; const lists = subscriber.lists || []; switch (event) { case "subscriber.created": case "subscriber.updated": // Upsert contact/guest in our system await this.syncListmonkSubscriber(subscriber); break; case "subscriber.deleted": // Mark as unsubscribed in our system await this.unsubscribeListmonkSubscriber(email); break; case "subscriber.subscribed": case "subscriber.unsubscribed": // Handle list-specific subscription changes await this.handleListmonkSubscriptionChange(email, lists, event); break; default: this.logger.debug(`Unhandled Listmonk event: ${event}`); } } private async syncListmonkSubscriber(subscriber: { id: number; email: string; name?: string; status?: string; lists?: Array<{ id: number; name: string }> }) { const email = subscriber.email; if (!email) return; try { const guest = await this.prisma.guest.findFirst({ where: { email }, }); if (guest) { await this.prisma.guest.update({ where: { id: guest.id }, data: { customFields: { listmonkSubscriberId: subscriber.id, listmonkStatus: subscriber.status, listmonkLists: subscriber.lists || [], }, }, }); this.logger.log(`Synced Listmonk subscriber ${subscriber.id} to guest ${guest.id}`); } else { this.logger.log(`Listmonk subscriber ${email} not found in guests, would create contact`); } } catch (error: unknown) { const message = error instanceof Error ? error.message : String(error); this.logger.error(`Failed to sync Listmonk subscriber ${subscriber.id}: ${message}`); } } private async unsubscribeListmonkSubscriber(email: string) { try { const guest = await this.prisma.guest.findFirst({ where: { email } }); if (guest) { await this.prisma.guest.update({ where: { id: guest.id }, data: { customFields: { listmonkUnsubscribedAt: new Date().toISOString(), }, }, }); this.logger.log(`Marked guest ${guest.id} as unsubscribed via Listmonk webhook`); } } catch (error: unknown) { const message = error instanceof Error ? error.message : String(error); this.logger.error(`Failed to unsubscribe guest for email ${email}: ${message}`); } } private async handleListmonkSubscriptionChange(email: string, lists: Array<{ id: number; name: string }>, event: string) { this.logger.log(`Listmonk subscription change for ${email}: ${event} on lists ${lists.map(l => l.id).join(", ")}`); // Could map specific list IDs to podcast/newsletter subscriptions } @Post("livekit") @HttpCode(HttpStatus.OK) async handleLiveKitWebhook( @Req() req: RawBodyRequest, @Body() body: LiveKitWebhookPayload, @Headers("authorization") authorization?: string, ) { const apiKey = this.configService.get("LIVEKIT_API_KEY", ""); const apiSecret = this.configService.get("LIVEKIT_API_SECRET", ""); const rawBody = (req as RawBodyRequest).rawBody; // SECURITY: validate the LiveKit webhook JWT (HS256 signed with // LIVEKIT_API_SECRET) plus its `sha256` body claim over the raw body. // Fail closed in production when not configured. if (process.env.NODE_ENV === "production") { if ( !apiKey || !apiSecret || !this.verifyLiveKitJwt(authorization, apiKey, apiSecret, rawBody) ) { throw new UnauthorizedException("Invalid LiveKit webhook authorization"); } } else if (apiKey && apiSecret && !this.verifyLiveKitJwt(authorization, apiKey, apiSecret, rawBody)) { throw new UnauthorizedException("Invalid LiveKit webhook authorization"); } else if (!apiKey || !apiSecret) { this.logger.warn("LIVEKIT_API_KEY / LIVEKIT_API_SECRET not configured — LiveKit webhook verification skipped (dev)"); } // Idempotency check if (await this.checkIdempotency("livekit", body)) { return { received: true, duplicate: true }; } const event = String(body?.event || "unknown"); const roomName = body?.room?.name || body?.egressInfo?.roomName || String(body?.roomName || ""); this.logger.log(`LiveKit webhook: ${event} room=${roomName ?? ""}`); // Rooms are created as `episode-{episodeId}` (see LivekitService.roomName). if (event === "room_finished" && roomName) { const episodeId = this.extractEpisodeIdFromRoom(roomName); if (episodeId) { const session = await this.prisma.recordingSession.findUnique({ where: { episodeId }, }); if (session && session.status === "active") { await this.prisma.recordingSession.update({ where: { id: session.id }, data: { status: "completed", endedAt: new Date() }, }); this.logger.log(`Marked recording session ${session.id} as completed via LiveKit webhook`); } } else { this.logger.warn(`Could not extract episode ID from room name: ${roomName}`); } } // Server-side recording / egress finished. LiveKit emits egress_ended // with egressInfo; accept the legacy dotted variants too. if ( event === "egress_ended" || event === "egress_completed" || event === "recording.completed" || event === "recording_completed" ) { await this.handleLiveKitRecordingFinished(body); } // Handle participant events if (event === "participant_joined" || event === "participant_left") { const participant = body?.participant; this.logger.log(`Participant ${participant?.identity} ${event} in room ${roomName}`); } return { received: true, event, roomName }; } private extractEpisodeIdFromRoom(roomName: string): string | null { const match = roomName.match(/^episode-(.+)$/); return match ? match[1] : null; } private async handleLiveKitRecordingFinished(body: LiveKitWebhookPayload): Promise { const egressInfo = body?.recording; const roomName = egressInfo?.name || body?.room?.name; if (!roomName) { this.logger.warn("LiveKit recording/egress event missing room name"); return; } const episodeId = this.extractEpisodeIdFromRoom(roomName); if (!episodeId) { this.logger.warn(`Could not extract episode ID from room name: ${roomName}`); return; } const session = await this.prisma.recordingSession.findUnique({ where: { episodeId }, }); if (!session) { this.logger.warn(`No recording session found for episode ${episodeId}`); return; } if (session.status === "completed" || session.status === "processing") { this.logger.log(`Session ${session.id} already ${session.status}, skipping`); return; } await this.prisma.recordingSession.update({ where: { id: session.id }, data: { status: "processing", endedAt: new Date() }, }); this.logger.log( `Session ${session.id} marked as processing after LiveKit recording finished (${egressInfo?.status ?? "unknown"})`, ); } }