Spaces:

Benaah
/

MicProtocol

Running

MicProtocol / apps /api /src /integrations /integrations-webhook.controller.ts
Upload folder using huggingface_hub
6abfd9d verified about 21 hours ago
33.1 kB
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=<hex>` (older
 * installs may use `sha1=<hex>`). Some versions include the
 * `X-Chatwoot-Timestamp` in the signed content (`<ts>.<rawBody>`) — 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<string, unknown>;
  };
}

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<string, unknown>;

/**
 * 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,<base64>` 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<string, unknown> | ChatwootWebhookPayload | DocumensoWebhookPayload | FormbricksWebhookPayload | ListmonkWebhookPayload | LiveKitWebhookPayload): Promise<boolean> {
    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<string>("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<string>("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<string, any>) || {};
        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<string>("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<string>("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<string>("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<string>("LIVEKIT_API_KEY", "");
    const apiSecret = this.configService.get<string>("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<void> {
    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"})`,
    );
  }
}