// Voice Call plugin module implements webhook behavior.
import http from "node:http";
import { URL } from "node:url";
import type { OpenClawConfig } from "openclaw/plugin-sdk/config-contracts";
import { createLazyRuntimeModule } from "openclaw/plugin-sdk/lazy-runtime";
import {
  asDateTimestampMs,
  resolveExpiresAtMsFromDurationMs,
} from "openclaw/plugin-sdk/number-runtime";
import { resolveConfiguredCapabilityProvider } from "openclaw/plugin-sdk/provider-selection-runtime";
import type { TalkEvent } from "openclaw/plugin-sdk/realtime-voice";
import {
  normalizeOptionalString,
  normalizeStringEntries,
} from "openclaw/plugin-sdk/string-coerce-runtime";
import {
  createWebhookInFlightLimiter,
  normalizeWebhookPath,
  WEBHOOK_BODY_READ_DEFAULTS,
} from "openclaw/plugin-sdk/webhook-ingress";
import {
  isRequestBodyLimitError,
  readRequestBodyWithLimit,
  requestBodyErrorToText,
} from "../api.js";
import type { OpenClawPluginApi } from "../api.js";
import { isAllowlistedCaller, normalizePhoneNumber } from "./allowlist.js";
import {
  normalizeVoiceCallConfig,
  resolveVoiceCallEffectiveConfig,
  resolveVoiceCallNumberRouteKeyForCall,
  type VoiceCallConfig,
} from "./config.js";
import { getHeader } from "./http-headers.js";
import type { CallManager } from "./manager.js";
import type { MediaStreamConfig } from "./media-stream.js";
import { MediaStreamHandler } from "./media-stream.js";
import type { VoiceCallProvider } from "./providers/base.js";
import { isProviderStatusTerminal } from "./providers/shared/call-status.js";
import type { TwilioProvider } from "./providers/twilio.js";
import { normalizeProxyIp } from "./proxy-ip.js";
import { resolveCallAgentId } from "./resolve-call-agent-id.js";
import type { CallRecord, NormalizedEvent, WebhookContext } from "./types.js";
import type { WebhookResponsePayload } from "./webhook.types.js";
import type { RealtimeCallHandler } from "./webhook/realtime-handler.js";
import { startStaleCallReaper } from "./webhook/stale-call-reaper.js";
import {
  StreamDisconnectGrace,
  type StreamDisconnectLifecycle,
} from "./webhook/stream-disconnect-grace.js";

const MAX_WEBHOOK_BODY_BYTES = WEBHOOK_BODY_READ_DEFAULTS.preAuth.maxBytes;
const WEBHOOK_BODY_TIMEOUT_MS = WEBHOOK_BODY_READ_DEFAULTS.preAuth.timeoutMs;
const MISSING_REMOTE_ADDRESS_IN_FLIGHT_KEY = "__voice_call_no_remote__";

type Logger = {
  info: (message: string) => void;
  warn: (message: string) => void;
  error: (message: string) => void;
  debug?: (message: string) => void;
};

const loadRealtimeTranscriptionRuntime = createLazyRuntimeModule(
  () => import("./realtime-transcription.runtime.js"),
);

const loadResponseGeneratorModule = createLazyRuntimeModule(
  () => import("./response-generator.js"),
);

type WebhookHeaderGateResult =
  | { ok: true }
  | {
      ok: false;
      reason: string;
    };

function appendRecentTalkEventMetadata(call: CallRecord, event: TalkEvent): void {
  const metadata = call.metadata ?? {};
  const recent = Array.isArray(metadata.recentTalkEvents)
    ? metadata.recentTalkEvents.filter(
        (entry): entry is { at: string; type: string; sessionId: string; turnId?: string } =>
          Boolean(entry) && typeof entry === "object" && !Array.isArray(entry),
      )
    : [];
  recent.push({
    at: event.timestamp,
    type: event.type,
    sessionId: event.sessionId,
    turnId: event.turnId,
  });
  call.metadata = {
    ...metadata,
    lastTalkEventAt: event.timestamp,
    lastTalkEventType: event.type,
    recentTalkEvents: recent.slice(-10),
  };
}

function buildRequestUrl(requestUrl: string | undefined): URL {
  return new URL(requestUrl ?? "/", "http://localhost");
}

function resolveForwardedClientIp(
  request: http.IncomingMessage,
  trustedProxyIPs: readonly string[],
): string | undefined {
  const normalizedTrustedProxyIps = new Set(
    trustedProxyIPs.map((ip) => normalizeProxyIp(ip)).filter((ip): ip is string => Boolean(ip)),
  );
  const forwardedFor = getHeader(request.headers, "x-forwarded-for");
  if (forwardedFor) {
    const forwardedIps = normalizeStringEntries(forwardedFor.split(","));
    if (forwardedIps.length > 0) {
      if (normalizedTrustedProxyIps.size === 0) {
        return forwardedIps[0];
      }
      for (let index = forwardedIps.length - 1; index >= 0; index -= 1) {
        const hop = forwardedIps[index];
        if (!normalizedTrustedProxyIps.has(normalizeProxyIp(hop) ?? "")) {
          return hop;
        }
      }
      return forwardedIps[0];
    }
  }

  const realIp = getHeader(request.headers, "x-real-ip")?.trim();
  return realIp || undefined;
}

function normalizeWebhookResponse(parsed: {
  statusCode?: number;
  providerResponseHeaders?: Record<string, string>;
  providerResponseBody?: string;
}): WebhookResponsePayload {
  return {
    statusCode: parsed.statusCode ?? 200,
    headers: parsed.providerResponseHeaders,
    body: parsed.providerResponseBody ?? "OK",
  };
}

function buildRealtimeRejectedTwiML(): WebhookResponsePayload {
  return {
    statusCode: 200,
    headers: { "Content-Type": "text/xml" },
    body: '<?xml version="1.0" encoding="UTF-8"?><Response><Reject reason="rejected" /></Response>',
  };
}

function buildTwilioReplayTwiML(): WebhookResponsePayload {
  return {
    statusCode: 200,
    headers: { "Content-Type": "text/xml" },
    body: '<?xml version="1.0" encoding="UTF-8"?><Response></Response>',
  };
}

const WEBHOOK_REPLAY_RESPONSE_TTL_MS = 10 * 60 * 1000;
const WEBHOOK_REPLAY_RESPONSE_MAX_ENTRIES = 10_000;
const WEBHOOK_REPLAY_RESPONSE_PRUNE_INTERVAL = 64;

type CachedWebhookResponse = {
  expiresAt: number;
  response: Promise<WebhookResponsePayload>;
};

function cloneWebhookResponsePayload(payload: WebhookResponsePayload): WebhookResponsePayload {
  return {
    statusCode: payload.statusCode,
    headers: payload.headers ? { ...payload.headers } : undefined,
    body: payload.body,
  };
}

/**
 * HTTP server for receiving voice call webhooks from providers.
 * Supports WebSocket upgrades for media streams when streaming is enabled.
 */
export class VoiceCallWebhookServer {
  private server: http.Server | null = null;
  private listeningUrl: string | null = null;
  private startPromise: Promise<string> | null = null;
  private stopPromise: Promise<void> | null = null;
  private config: VoiceCallConfig;
  private manager: CallManager;
  private provider: VoiceCallProvider;
  private coreConfig: OpenClawConfig | null;
  private fullConfig: OpenClawConfig | null;
  private agentRuntime: OpenClawPluginApi["runtime"]["agent"] | null;
  private logger: Logger;
  private stopStaleCallReaper: (() => void) | null = null;
  private readonly webhookInFlightLimiter = createWebhookInFlightLimiter();

  /** Media stream handler for bidirectional audio (when streaming enabled) */
  private mediaStreamHandler: MediaStreamHandler | null = null;
  private readonly streamDisconnectGrace: StreamDisconnectGrace;
  /** Realtime voice handler for duplex provider bridges. */
  private realtimeHandler: RealtimeCallHandler | null = null;
  private replayResponses = new Map<string, CachedWebhookResponse>();
  private replayResponseCacheCalls = 0;

  constructor(
    config: VoiceCallConfig,
    manager: CallManager,
    provider: VoiceCallProvider,
    coreConfig?: OpenClawConfig,
    fullConfig?: OpenClawConfig,
    agentRuntime?: OpenClawPluginApi["runtime"]["agent"],
    logger?: Logger,
  ) {
    this.config = normalizeVoiceCallConfig(config);
    this.manager = manager;
    this.provider = provider;
    this.coreConfig = coreConfig ?? null;
    this.fullConfig = fullConfig ?? null;
    this.agentRuntime = agentRuntime ?? null;
    this.logger = logger ?? {
      info: console.log,
      warn: console.warn,
      error: console.error,
      debug: console.debug,
    };
    // Route all webhook diagnostics through a single logging path with
    // consistent [voice-call] attribution so operational tooling can
    // identify voice-call messages on the shared plugin logger.
    const rawLogger = this.logger;
    const rawDebug = rawLogger.debug;
    this.logger = {
      info: (msg: string) => rawLogger.info(`[voice-call] ${msg}`),
      warn: (msg: string) => rawLogger.warn(`[voice-call] ${msg}`),
      error: (msg: string) => rawLogger.error(`[voice-call] ${msg}`),
      debug: rawDebug ? (msg: string) => rawDebug(`[voice-call] ${msg}`) : undefined,
    };
    this.streamDisconnectGrace = new StreamDisconnectGrace(({ providerCallId }) => {
      const call = this.manager.getCallByProviderCallId(providerCallId);
      if (!call) {
        return;
      }
      this.logger.info(
        `Call finalization requested reason=stream-disconnect-grace-expired callId=${call.callId} providerCallId=${providerCallId}`,
      );
      void this.manager
        .endCall(call.callId)
        .then((result) => {
          if (!result.success) {
            this.logger.warn(
              `Failed to auto-end call ${call.callId}: ${result.error ?? "unknown error"}`,
            );
          }
        })
        .catch((err: unknown) => {
          this.logger.warn(`Failed to auto-end call ${call.callId}: ${String(err)}`);
        });
    });
  }

  /**
   * Get the media stream handler (for wiring to provider).
   */
  getMediaStreamHandler(): MediaStreamHandler | null {
    return this.mediaStreamHandler;
  }

  getRealtimeHandler(): RealtimeCallHandler | null {
    return this.realtimeHandler;
  }

  speakRealtime(callId: string, instructions: string): { success: boolean; error?: string } {
    if (!this.realtimeHandler) {
      return { success: false, error: "Realtime voice handler is not configured" };
    }
    return this.realtimeHandler.speak(callId, instructions);
  }

  setRealtimeHandler(handler: RealtimeCallHandler): void {
    this.realtimeHandler = handler;
  }

  getStreamDisconnectLifecycle(): StreamDisconnectLifecycle {
    return this.streamDisconnectGrace;
  }

  private resolveMediaStreamClientIp(request: http.IncomingMessage): string | undefined {
    const remoteIp = request.socket.remoteAddress ?? undefined;
    const trustedProxyIPs = this.config.webhookSecurity.trustedProxyIPs.filter(Boolean);
    const normalizedTrustedProxyIps = new Set(
      trustedProxyIPs.map((ip) => normalizeProxyIp(ip)).filter((ip): ip is string => Boolean(ip)),
    );
    const normalizedRemoteIp = normalizeProxyIp(remoteIp);
    const fromTrustedProxy =
      normalizedTrustedProxyIps.size > 0 &&
      normalizedRemoteIp !== undefined &&
      normalizedTrustedProxyIps.has(normalizedRemoteIp);
    const shouldTrustForwardingHeaders =
      this.config.webhookSecurity.trustForwardingHeaders && fromTrustedProxy;

    if (shouldTrustForwardingHeaders) {
      const forwardedIp = resolveForwardedClientIp(request, trustedProxyIPs);
      if (forwardedIp) {
        return forwardedIp;
      }
    }

    return remoteIp;
  }

  private shouldSuppressBargeInForInitialMessage(call: CallRecord | undefined): boolean {
    if (!call || call.direction !== "outbound") {
      return false;
    }

    // Suppress only while the initial greeting is actively being played.
    // If playback fails and the call leaves "speaking", do not block auto-response.
    if (call.state !== "speaking") {
      return false;
    }

    const mode = (call.metadata?.mode as string | undefined) ?? "conversation";
    if (mode !== "conversation") {
      return false;
    }

    const initialMessage = normalizeOptionalString(call.metadata?.initialMessage) ?? "";
    return initialMessage.length > 0;
  }

  /**
   * Initialize media streaming with the selected realtime transcription provider.
   */
  private async initializeMediaStreaming(): Promise<void> {
    const streaming = this.config.streaming;
    const pluginConfig =
      this.fullConfig ?? (this.coreConfig as unknown as OpenClawConfig | undefined);
    const { getRealtimeTranscriptionProvider, listRealtimeTranscriptionProviders } =
      await loadRealtimeTranscriptionRuntime();
    const resolution = resolveConfiguredCapabilityProvider({
      configuredProviderId: streaming.provider,
      providerConfigs: streaming.providers,
      cfg: pluginConfig,
      cfgForResolve: pluginConfig ?? ({} as OpenClawConfig),
      getConfiguredProvider: (providerId) =>
        getRealtimeTranscriptionProvider(providerId, pluginConfig),
      listProviders: () => listRealtimeTranscriptionProviders(pluginConfig),
      resolveProviderConfig: ({ provider, cfg, rawConfig }) =>
        provider.resolveConfig?.({ cfg, rawConfig }) ?? rawConfig,
      isProviderConfigured: ({ provider, cfg, providerConfig }) =>
        provider.isConfigured({ cfg, providerConfig }),
    });
    if (!resolution.ok && resolution.code === "missing-configured-provider") {
      this.logger.warn(
        `Streaming enabled but realtime transcription provider "${resolution.configuredProviderId}" is not registered`,
      );
      return;
    }
    if (!resolution.ok && resolution.code === "no-registered-provider") {
      this.logger.warn("Streaming enabled but no realtime transcription provider is registered");
      return;
    }
    if (!resolution.ok) {
      this.logger.warn(
        `Streaming enabled but provider "${resolution.provider?.id}" is not configured`,
      );
      return;
    }
    const provider = resolution.provider;
    const providerConfig = resolution.providerConfig;

    const streamConfig: MediaStreamConfig = {
      transcriptionProvider: provider,
      providerConfig,
      cfg: this.fullConfig ?? (this.coreConfig as OpenClawConfig | null) ?? undefined,
      preStartTimeoutMs: streaming.preStartTimeoutMs,
      maxPendingConnections: streaming.maxPendingConnections,
      maxPendingConnectionsPerIp: streaming.maxPendingConnectionsPerIp,
      maxConnections: streaming.maxConnections,
      resolveClientIp: (request) => this.resolveMediaStreamClientIp(request),
      shouldAcceptStream: ({ callId, token }) => {
        // The classic media handler parses Twilio frames and only Twilio issues
        // its per-call token. Other carriers use their separate realtime path.
        if (this.provider.name !== "twilio") {
          this.logger.warn(
            `Rejecting media stream: provider ${this.provider.name} does not support authenticated classic streaming`,
          );
          return false;
        }
        const call = this.manager.getCallByProviderCallId(callId);
        if (!call) {
          return false;
        }
        const twilio = this.provider as TwilioProvider;
        if (!twilio.isValidStreamToken(callId, token)) {
          this.logger.warn(`Rejecting media stream: invalid token for ${callId}`);
          return false;
        }
        return true;
      },
      onTranscript: (providerCallId, transcript) => {
        this.logger.info(`Transcript received ${providerCallId} chars=${transcript.length}`);
        const call = this.manager.getCallByProviderCallId(providerCallId);
        if (!call) {
          this.logger.warn(`No active call found for provider ID: ${providerCallId}`);
          return;
        }
        const suppressBargeIn = this.shouldSuppressBargeInForInitialMessage(call);
        if (suppressBargeIn) {
          this.logger.info(
            `Ignoring barge transcript while initial message is still playing (${providerCallId})`,
          );
          return;
        }

        // Clear TTS queue on barge-in (user started speaking, interrupt current playback)
        if (this.provider.name === "twilio") {
          (this.provider as TwilioProvider).clearTtsQueue(providerCallId);
        }

        // Create a speech event and process it through the manager
        const event: NormalizedEvent = {
          id: `stream-transcript-${Date.now()}`,
          type: "call.speech",
          callId: call.callId,
          providerCallId,
          timestamp: Date.now(),
          transcript,
          isFinal: true,
        };
        this.processEventWithAutoResponse(event);
      },
      onSpeechStart: (providerCallId) => {
        if (this.provider.name !== "twilio") {
          return;
        }
        const call = this.manager.getCallByProviderCallId(providerCallId);
        if (this.shouldSuppressBargeInForInitialMessage(call)) {
          return;
        }
        (this.provider as TwilioProvider).clearTtsQueue(providerCallId);
      },
      onPartialTranscript: (callId, partial) => {
        this.logger.info(`Partial transcript ${callId} chars=${partial.length}`);
      },
      onTalkEvent: (providerCallId, _streamSid, event) => {
        const call = this.manager.getCallByProviderCallId(providerCallId);
        if (call) {
          appendRecentTalkEventMetadata(call, event);
        }
      },
      onConnect: (callId, streamSid) => {
        this.logger.info(`Media stream connected: ${callId} -> ${streamSid}`);
        this.streamDisconnectGrace.connect(callId, streamSid);

        // Register stream with provider for TTS routing
        if (this.provider.name === "twilio") {
          (this.provider as TwilioProvider).registerCallStream(callId, streamSid);
        }
      },
      onTranscriptionReady: (callId) => {
        this.manager.speakInitialMessage(callId).catch((err: unknown) => {
          this.logger.warn(`Failed to speak initial message: ${String(err)}`);
        });
      },
      onDisconnect: (callId, streamSid) => {
        this.logger.info(`Media stream disconnected: ${callId} (${streamSid})`);
        if (this.provider.name === "twilio") {
          (this.provider as TwilioProvider).unregisterCallStream(callId, streamSid);
        }

        this.streamDisconnectGrace.disconnect(callId, streamSid);
      },
    };

    this.mediaStreamHandler = new MediaStreamHandler(streamConfig);
    this.logger.info("Media streaming initialized");
  }

  /**
   * Start the webhook server.
   * Idempotent: returns immediately if the server is already listening.
   */
  async start(): Promise<string> {
    if (this.stopPromise) {
      await this.stopPromise;
    }

    const { port, bind, path: webhookPath } = this.config.serve;
    const streamPath = this.config.streaming.streamPath;

    // Guard: if a server is already listening, return the existing URL.
    // This prevents EADDRINUSE when start() is called more than once on the
    // same instance (e.g. during config hot-reload or concurrent ensureRuntime).
    if (this.server?.listening) {
      return this.listeningUrl ?? this.resolveListeningUrl(bind, webhookPath);
    }

    if (this.config.streaming.enabled && !this.mediaStreamHandler) {
      await this.initializeMediaStreaming();
    }

    if (this.startPromise) {
      return this.startPromise;
    }

    this.startPromise = new Promise((resolve, reject) => {
      this.server = http.createServer((req, res) => {
        this.handleRequest(req, res, webhookPath).catch((err: unknown) => {
          this.logger.error(`Webhook error: ${String(err)}`);
          res.statusCode = 500;
          res.end("Internal Server Error");
        });
      });

      // Handle WebSocket upgrades for realtime voice and media streams.
      if (this.realtimeHandler || this.mediaStreamHandler) {
        this.server.on("upgrade", (request, socket, head) => {
          if (this.realtimeHandler && this.isRealtimeWebSocketUpgrade(request)) {
            this.realtimeHandler.handleWebSocketUpgrade(request, socket, head);
            return;
          }
          const path = this.getUpgradePathname(request);
          if (path === streamPath && this.mediaStreamHandler) {
            this.mediaStreamHandler?.handleUpgrade(request, socket, head);
          } else {
            socket.destroy();
          }
        });
      }

      this.server.on("error", (err) => {
        this.server = null;
        this.listeningUrl = null;
        this.startPromise = null;
        reject(err);
      });

      this.server.listen(port, bind, () => {
        const url = this.resolveListeningUrl(bind, webhookPath);
        this.listeningUrl = url;
        this.startPromise = null;
        this.logger.info(`Webhook server listening on ${url}`);
        if (this.mediaStreamHandler) {
          const address = this.server?.address();
          const actualPort =
            address && typeof address === "object" ? address.port : this.config.serve.port;
          this.logger.info(`Media stream WebSocket on ws://${bind}:${actualPort}${streamPath}`);
        }
        resolve(url);

        // Start the stale call reaper if configured
        this.stopStaleCallReaper = startStaleCallReaper({
          manager: this.manager,
          staleCallReaperSeconds: this.config.staleCallReaperSeconds,
        });
      });
    });

    return this.startPromise;
  }

  /**
   * Stop the webhook server.
   */
  stop(): Promise<void> {
    if (this.stopPromise) {
      return this.stopPromise;
    }

    const server = this.server;
    const serverClosePromise = new Promise<void>((resolve, reject) => {
      if (!server) {
        resolve();
        return;
      }
      server.close((error) => {
        if (error) {
          reject(error);
          return;
        }
        resolve();
      });
    });
    this.startPromise = null;
    this.streamDisconnectGrace.close();
    if (this.stopStaleCallReaper) {
      this.stopStaleCallReaper();
      this.stopStaleCallReaper = null;
    }
    this.webhookInFlightLimiter.clear();

    this.stopPromise = (async () => {
      const results = await Promise.allSettled([
        serverClosePromise,
        this.mediaStreamHandler?.close(serverClosePromise) ?? Promise.resolve(),
        this.realtimeHandler?.close(serverClosePromise) ?? Promise.resolve(),
      ]);

      this.streamDisconnectGrace.close();
      if (this.server === server) {
        this.server = null;
      }
      this.listeningUrl = null;

      const failure = results.find(
        (result): result is PromiseRejectedResult => result.status === "rejected",
      );
      if (failure) {
        throw failure.reason;
      }
    })().finally(() => {
      this.stopPromise = null;
    });
    return this.stopPromise;
  }

  private resolveListeningUrl(bind: string, webhookPath: string): string {
    const address = this.server?.address();
    if (address && typeof address === "object") {
      const host = address.address && address.address.length > 0 ? address.address : bind;
      const normalizedHost = host.includes(":") && !host.startsWith("[") ? `[${host}]` : host;
      return `http://${normalizedHost}:${address.port}${webhookPath}`;
    }
    return `http://${bind}:${this.config.serve.port}${webhookPath}`;
  }

  private getUpgradePathname(request: http.IncomingMessage): string | null {
    try {
      return buildRequestUrl(request.url).pathname;
    } catch {
      return null;
    }
  }

  private isWebhookPathMatch(requestPath: string, configuredPath: string): boolean {
    return normalizeWebhookPath(requestPath) === normalizeWebhookPath(configuredPath);
  }

  /**
   * Handle incoming HTTP request.
   */
  private async handleRequest(
    req: http.IncomingMessage,
    res: http.ServerResponse,
    webhookPath: string,
  ): Promise<void> {
    const payload = await this.runWebhookPipeline(req, webhookPath);
    this.writeWebhookResponse(res, payload);
  }

  private async runWebhookPipeline(
    req: http.IncomingMessage,
    webhookPath: string,
  ): Promise<WebhookResponsePayload> {
    const url = buildRequestUrl(req.url);

    if (url.pathname === "/voice/hold-music") {
      return {
        statusCode: 200,
        headers: { "Content-Type": "text/xml" },
        body: `<?xml version="1.0" encoding="UTF-8"?>
<Response>
  <Say voice="alice">All agents are currently busy. Please hold.</Say>
  <Play loop="0">https://s3.amazonaws.com/com.twilio.music.classical/BusyStrings.mp3</Play>
</Response>`,
      };
    }

    if (!this.isWebhookPathMatch(url.pathname, webhookPath)) {
      return { statusCode: 404, body: "Not Found" };
    }

    if (req.method !== "POST") {
      return { statusCode: 405, body: "Method Not Allowed" };
    }

    const headerGate = this.verifyPreAuthWebhookHeaders(req.headers);
    if (!headerGate.ok) {
      this.logger.warn(`Webhook rejected before body read: ${headerGate.reason}`);
      return { statusCode: 401, body: "Unauthorized" };
    }

    // createWebhookInFlightLimiter intentionally treats an empty key as fail-open.
    // Missing socket metadata must still share one bucket instead of bypassing
    // the pre-auth limiter entirely.
    const remoteAddress = req.socket.remoteAddress;
    if (!remoteAddress) {
      this.logger.warn(
        `Webhook accepted with no remote address; using shared fallback in-flight key`,
      );
    }
    const inFlightKey = remoteAddress || MISSING_REMOTE_ADDRESS_IN_FLIGHT_KEY;
    if (!this.webhookInFlightLimiter.tryAcquire(inFlightKey)) {
      this.logger.warn(`Webhook rejected before body read: too many in-flight requests`);
      return { statusCode: 429, body: "Too Many Requests" };
    }

    try {
      let body = "";
      try {
        body = await this.readBody(req, MAX_WEBHOOK_BODY_BYTES, WEBHOOK_BODY_TIMEOUT_MS);
      } catch (err) {
        if (isRequestBodyLimitError(err, "PAYLOAD_TOO_LARGE")) {
          return { statusCode: 413, body: "Payload Too Large" };
        }
        if (isRequestBodyLimitError(err, "REQUEST_BODY_TIMEOUT")) {
          return { statusCode: 408, body: requestBodyErrorToText("REQUEST_BODY_TIMEOUT") };
        }
        throw err;
      }

      const ctx: WebhookContext = {
        headers: req.headers as Record<string, string | string[] | undefined>,
        rawBody: body,
        url: url.toString(),
        method: "POST",
        query: Object.fromEntries(url.searchParams),
        remoteAddress: req.socket.remoteAddress ?? undefined,
      };

      const verification = this.provider.verifyWebhook(ctx);
      if (!verification.ok) {
        this.logger.warn(`Webhook verification failed: ${verification.reason}`);
        return { statusCode: 401, body: "Unauthorized" };
      }
      if (!verification.verifiedRequestKey) {
        this.logger.warn("Webhook verification succeeded without request identity key");
        return { statusCode: 401, body: "Unauthorized" };
      }

      const isReplay = Boolean(verification.isReplay);
      if (isReplay) {
        this.logger.warn("Replay detected; skipping event side effects");
        const cachedResponse = await this.getCachedReplayResponse(verification.verifiedRequestKey);
        if (cachedResponse) {
          return cachedResponse;
        }
        if (this.provider.name === "twilio") {
          return buildTwilioReplayTwiML();
        }
      }

      const buildResponse = async (): Promise<WebhookResponsePayload> => {
        const initialTwiML = this.provider.consumeInitialTwiML?.(ctx);
        if (initialTwiML !== undefined && initialTwiML !== null) {
          const params = new URLSearchParams(ctx.rawBody);
          this.logger.info(
            `Serving provider initial TwiML before realtime handling (callSid=${params.get("CallSid") ?? "unknown"}, direction=${params.get("Direction") ?? "unknown"})`,
          );
          return {
            statusCode: 200,
            headers: { "Content-Type": "application/xml" },
            body: initialTwiML,
          };
        }

        const realtimeParams = this.getRealtimeTwimlParams(ctx);
        if (realtimeParams) {
          const direction = realtimeParams.get("Direction");
          const isInboundRealtimeRequest = !direction || direction === "inbound";
          if (
            isInboundRealtimeRequest &&
            !this.shouldAcceptRealtimeInboundRequest(realtimeParams)
          ) {
            this.logger.info("Realtime inbound call rejected before stream setup");
            return buildRealtimeRejectedTwiML();
          }
          this.logger.info(
            `Serving realtime TwiML for Twilio call ${realtimeParams.get("CallSid") ?? "unknown"} (direction=${direction ?? "unknown"})`,
          );
          return this.realtimeHandler!.buildTwiMLPayload(req, realtimeParams);
        }

        const parsed = this.provider.parseWebhookEvent(ctx, {
          verifiedRequestKey: verification.verifiedRequestKey,
        });
        if (!isReplay && this.processParsedEvents(parsed.events)) {
          verification.releaseReplay?.();
        }

        return normalizeWebhookResponse(parsed);
      };

      if (isReplay) {
        return await buildResponse();
      }

      return await this.cacheReplayResponse(
        verification.verifiedRequestKey,
        buildResponse,
        verification.releaseReplay,
      );
    } finally {
      this.webhookInFlightLimiter.release(inFlightKey);
    }
  }

  private pruneReplayResponses(rawNow: number): void {
    const now = asDateTimestampMs(rawNow);
    if (now !== undefined) {
      for (const [key, entry] of this.replayResponses) {
        if (entry.expiresAt <= now) {
          this.replayResponses.delete(key);
        }
      }
    }
    while (this.replayResponses.size > WEBHOOK_REPLAY_RESPONSE_MAX_ENTRIES) {
      const oldest = this.replayResponses.keys().next().value;
      if (!oldest) {
        break;
      }
      this.replayResponses.delete(oldest);
    }
  }

  private async getCachedReplayResponse(key: string): Promise<WebhookResponsePayload | null> {
    const now = asDateTimestampMs(Date.now());
    const entry = this.replayResponses.get(key);
    if (!entry || now === undefined) {
      return null;
    }
    if (entry.expiresAt <= now) {
      this.replayResponses.delete(key);
      return null;
    }
    return cloneWebhookResponsePayload(await entry.response);
  }

  private async cacheReplayResponse(
    key: string,
    buildResponse: () => Promise<WebhookResponsePayload>,
    releaseReplay?: () => void,
  ): Promise<WebhookResponsePayload> {
    const now = Date.now();
    const expiresAt = resolveExpiresAtMsFromDurationMs(WEBHOOK_REPLAY_RESPONSE_TTL_MS, {
      nowMs: now,
    });
    this.replayResponseCacheCalls += 1;
    if (this.replayResponseCacheCalls % WEBHOOK_REPLAY_RESPONSE_PRUNE_INTERVAL === 0) {
      this.pruneReplayResponses(now);
    }

    let cachedEntry: CachedWebhookResponse | undefined;
    const ownerResponse = buildResponse()
      .then(cloneWebhookResponsePayload)
      .catch((err: unknown) => {
        if (cachedEntry && this.replayResponses.get(key) === cachedEntry) {
          this.replayResponses.delete(key);
        }
        releaseReplay?.();
        throw err;
      });
    // Twilio owners receive the real one-time TwiML; waiters only see token-free XML.
    const response = ownerResponse.then((payload) =>
      this.provider.name === "twilio"
        ? buildTwilioReplayTwiML()
        : cloneWebhookResponsePayload(payload),
    );
    // Preserve rejection for concurrent waiters without creating an orphaned rejection.
    void response.catch(() => {});
    if (expiresAt !== undefined) {
      cachedEntry = {
        expiresAt,
        response,
      };
      this.replayResponses.set(key, cachedEntry);
    }
    if (this.replayResponses.size > WEBHOOK_REPLAY_RESPONSE_MAX_ENTRIES) {
      this.pruneReplayResponses(now);
    }
    return cloneWebhookResponsePayload(await ownerResponse);
  }

  private verifyPreAuthWebhookHeaders(headers: http.IncomingHttpHeaders): WebhookHeaderGateResult {
    if (this.config.skipSignatureVerification) {
      return { ok: true };
    }
    switch (this.provider.name) {
      case "telnyx": {
        const signature = getHeader(headers, "telnyx-signature-ed25519");
        const timestamp = getHeader(headers, "telnyx-timestamp");
        if (signature && timestamp) {
          return { ok: true };
        }
        return { ok: false, reason: "missing Telnyx signature or timestamp header" };
      }
      case "twilio":
        if (getHeader(headers, "x-twilio-signature")) {
          return { ok: true };
        }
        return { ok: false, reason: "missing X-Twilio-Signature header" };
      case "plivo": {
        const hasV3 =
          Boolean(getHeader(headers, "x-plivo-signature-v3")) &&
          Boolean(getHeader(headers, "x-plivo-signature-v3-nonce"));
        const hasV2 =
          Boolean(getHeader(headers, "x-plivo-signature-v2")) &&
          Boolean(getHeader(headers, "x-plivo-signature-v2-nonce"));
        if (hasV3 || hasV2) {
          return { ok: true };
        }
        return { ok: false, reason: "missing Plivo signature headers" };
      }
      default:
        return { ok: true };
    }
  }

  private isRealtimeWebSocketUpgrade(req: http.IncomingMessage): boolean {
    try {
      const pathname = buildRequestUrl(req.url).pathname;
      const pattern = this.realtimeHandler?.getStreamPathPattern();
      if (!pattern) {
        return false;
      }
      const normalizedPattern = normalizeWebhookPath(pattern);
      const normalizedPathname = normalizeWebhookPath(pathname);
      if (normalizedPattern === "/") {
        return true;
      }
      return (
        normalizedPathname === normalizedPattern ||
        normalizedPathname.startsWith(`${normalizedPattern}/`)
      );
    } catch {
      return false;
    }
  }

  private getRealtimeTwimlParams(ctx: WebhookContext): URLSearchParams | null {
    if (!this.realtimeHandler || this.provider.name !== "twilio") {
      return null;
    }

    const params = new URLSearchParams(ctx.rawBody);
    const direction = params.get("Direction");
    const isSupportedDirection =
      !direction || direction === "inbound" || direction.startsWith("outbound");
    if (!isSupportedDirection) {
      return null;
    }

    if (ctx.query?.type === "status") {
      return null;
    }

    const callStatus = params.get("CallStatus");
    if (callStatus && isProviderStatusTerminal(callStatus)) {
      return null;
    }

    // Initial TwiML fetches without gathered input may enter realtime handling.
    // Replay checks run before this helper so retries cannot mint new stream tokens.
    return !params.get("SpeechResult") && !params.get("Digits") ? params : null;
  }

  private shouldAcceptRealtimeInboundRequest(params: URLSearchParams): boolean {
    switch (this.config.inboundPolicy) {
      case "open":
        return true;
      case "allowlist":
      case "pairing":
        return isAllowlistedCaller(
          normalizePhoneNumber(params.get("From") ?? undefined),
          this.config.allowFrom,
        );
      default:
        return false;
    }
  }

  private processParsedEvents(events: NormalizedEvent[]): boolean {
    let replayable = false;
    for (const event of events) {
      try {
        replayable = this.processEventWithAutoResponse(event) || replayable;
      } catch (err) {
        this.logger.error(`Error processing event ${event.type}: ${String(err)}`);
        throw err;
      }
    }
    return replayable;
  }

  private processEventWithAutoResponse(event: NormalizedEvent): boolean {
    const result = this.manager.processEvent(event);
    if (result.kind !== "final-speech") {
      return result.replayable === true;
    }
    if (result.waiterResolved) {
      return false;
    }
    const callMode = result.call.metadata?.mode as string | undefined;
    if (result.call.direction !== "inbound" && callMode !== "conversation") {
      return false;
    }

    // Both media-stream and carrier-webhook transcripts share this handoff.
    // The manager result excludes replays and turn-token mismatches.
    void this.handleInboundResponse(result.call.callId, result.transcript).catch((err: unknown) => {
      this.logger.warn(`Failed to auto-respond: ${String(err)}`);
    });
    return false;
  }

  private writeWebhookResponse(res: http.ServerResponse, payload: WebhookResponsePayload): void {
    res.statusCode = payload.statusCode;
    if (payload.headers) {
      for (const [key, value] of Object.entries(payload.headers)) {
        res.setHeader(key, value);
      }
    }
    res.end(payload.body);
  }

  /**
   * Read request body as string with timeout protection.
   */
  private readBody(
    req: http.IncomingMessage,
    maxBytes: number,
    timeoutMs = WEBHOOK_BODY_TIMEOUT_MS,
  ): Promise<string> {
    return readRequestBodyWithLimit(req, { maxBytes, timeoutMs });
  }

  /**
   * Handle auto-response for inbound calls using the agent system.
   * Supports tool calling for richer voice interactions.
   */
  private async handleInboundResponse(callId: string, userMessage: string): Promise<void> {
    this.logger.info(`Auto-responding to inbound call ${callId} chars=${userMessage.length}`);

    // Get the persisted call context for routing and response delivery.
    const call = this.manager.getCall(callId);
    if (!call) {
      this.logger.warn(`Call ${callId} not found for auto-response`);
      return;
    }

    if (!this.coreConfig) {
      this.logger.warn("Core config missing; skipping auto-response");
      return;
    }
    if (!this.agentRuntime) {
      this.logger.warn("Agent runtime missing; skipping auto-response");
      return;
    }

    try {
      const { generateVoiceResponse } = await loadResponseGeneratorModule();
      const numberRouteKey = resolveVoiceCallNumberRouteKeyForCall(call);
      const effectiveConfig = resolveVoiceCallEffectiveConfig(this.config, numberRouteKey).config;

      const result = await generateVoiceResponse({
        voiceConfig: effectiveConfig,
        coreConfig: this.coreConfig,
        agentRuntime: this.agentRuntime,
        callId,
        sessionKey: call.sessionKey,
        from: call.from,
        senderIsOwner: call.direction === "inbound" ? false : undefined,
        agentId: resolveCallAgentId(call, effectiveConfig),
        transcript: call.transcript,
        userMessage,
        onEarlyText: async (text) => {
          this.logger.info(`Early AI response queued ${callId} chars=${text.length}`);
          const speakResult = await this.manager.speak(callId, text, { listenAfterPlayback: true });
          return speakResult.success;
        },
      });

      if (result.error) {
        this.logger.error(`Response generation error: ${result.error}`);
        return;
      }

      if (result.text && !result.deliveredEarly) {
        this.logger.info(`AI response delivered ${callId} chars=${result.text.length}`);
        await this.manager.speak(callId, result.text, { listenAfterPlayback: true });
      }
    } catch (err) {
      this.logger.error(`Auto-response error: ${String(err)}`);
    }
  }
}
/* oxlint-disable max-lines -- TODO: split this grandfathered oversized file. */