/**
 * Remote message handler for B2B webhook and WebSocket integration.
 *
 * Allows onMessage to be a URL string instead of a callable:
 * - HTTP webhook: onMessage="https://api.customer.com/patter/message"
 * - WebSocket: onMessage="ws://localhost:9000/stream"
 */

import crypto from 'node:crypto';
import { getLogger } from './logger';
import { validateWebhookUrl } from './server';

const MAX_RESPONSE_BYTES = 64 * 1024;

/**
 * Validate a WebSocket URL against the same SSRF blocklist used for HTTP
 * webhooks. Translates ws(s):// to http(s):// before delegating to
 * validateWebhookUrl so the scheme/hostname/IP checks apply uniformly.
 */
function validateWebSocketUrl(url: string): void {
  let translated = url;
  if (url.startsWith('ws://')) {
    translated = 'http://' + url.slice('ws://'.length);
  } else if (url.startsWith('wss://')) {
    translated = 'https://' + url.slice('wss://'.length);
  }
  validateWebhookUrl(translated);
}

/** Dispatches per-turn messages to a remote HTTP webhook or WebSocket endpoint. */
export class RemoteMessageHandler {
  private readonly webhookSecret: string | undefined;

  /**
   * @param webhookSecret Optional HMAC secret. When provided, outgoing webhook
   *   requests include an `X-Patter-Signature` header so the receiver can
   *   verify the payload originated from Patter.
   */
  constructor(webhookSecret?: string) {
    this.webhookSecret = webhookSecret;
  }

  /**
   * Compute HMAC-SHA256 hex digest for the given body.
   */
  private signPayload(body: string): string {
    if (!this.webhookSecret) {
      throw new Error('Cannot sign without a webhookSecret');
    }
    return crypto.createHmac('sha256', this.webhookSecret).update(body).digest('hex');
  }

  /**
   * Release resources held by this handler.
   */
  close(): void {
    // No persistent HTTP client in the TS SDK (fetch is stateless),
    // but provided for API parity with the Python SDK.
  }

  /**
   * POST transcript to HTTP webhook, return response text.
   *
   * The webhook receives a JSON payload:
   *   { text, call_id, caller, callee, history }
   *
   * The response can be plain text or JSON { text: "..." }.
   *
   * When `webhookSecret` was provided at construction time, the request
   * includes an `X-Patter-Signature` header with the HMAC-SHA256 hex
   * digest of the JSON body.
   */
  async callWebhook(url: string, data: Record<string, unknown>): Promise<string> {
    try {
      validateWebhookUrl(url);
    } catch (e) {
      getLogger().warn(`Webhook URL rejected by SSRF guard: ${String(e)}`);
      return '';
    }
    if (url.startsWith('http://')) {
      getLogger().warn(
        'Webhook URL uses unencrypted http:// — call transcripts ' +
        'and phone numbers will be sent in plaintext. Use https:// in production.'
      );
    }
    const body = JSON.stringify(data);
    const headers: Record<string, string> = { 'Content-Type': 'application/json' };
    if (this.webhookSecret) {
      headers['X-Patter-Signature'] = this.signPayload(body);
    }
    const response = await fetch(url, {
      method: 'POST',
      // Fail closed on 3xx so a redirect can't bypass the SSRF guard that
      // validated ``url``. Matches Python httpx (follow_redirects=False).
      redirect: 'error',
      headers,
      body,
      signal: AbortSignal.timeout(30_000),
    });

    if (!response.ok) {
      throw new Error(`Webhook returned HTTP ${response.status}`);
    }

    // Reject an honestly-declared oversized body before buffering it; chunked
    // bodies stay bounded by the AbortSignal.timeout above.
    const declaredLen = response.headers?.get('content-length');
    if (declaredLen && /^\d+$/.test(declaredLen) && Number(declaredLen) > MAX_RESPONSE_BYTES) {
      throw new Error(`Webhook response too large: ${declaredLen} bytes (max ${MAX_RESPONSE_BYTES})`);
    }
    const text = await response.text();
    if (text.length > MAX_RESPONSE_BYTES) {
      throw new Error(`Webhook response too large: ${text.length} bytes (max ${MAX_RESPONSE_BYTES})`);
    }

    const contentType = response.headers.get('content-type') || '';
    if (contentType.includes('application/json')) {
      try {
        const body = JSON.parse(text);
        if (typeof body === 'object' && body !== null && 'text' in body) {
          return String(body.text);
        }
        return String(body);
      } catch {
        return text;
      }
    }
    return text;
  }

  /**
   * Send transcript via WebSocket, yield response chunks.
   *
   * Sends the message data as JSON. Receives one or more JSON frames
   * with { text: "..." } - multiple frames enable streaming.
   * A frame with { done: true } signals end of response.
   */
  async *callWebSocket(url: string, data: Record<string, unknown>): AsyncGenerator<string, void, unknown> {
    try {
      validateWebSocketUrl(url);
    } catch (e) {
      getLogger().warn(`WebSocket URL rejected by SSRF guard: ${String(e)}`);
      return;
    }
    if (url.startsWith('ws://')) {
      getLogger().warn(
        'WebSocket URL uses unencrypted ws:// — call transcripts ' +
        'and phone numbers will be sent in plaintext. Use wss:// in production.'
      );
    }
    // Dynamic import to avoid hard dependency on ws
    const { WebSocket } = await import('ws');
    const ws = new WebSocket(url);

    const chunks: string[] = [];
    let done = false;
    let error: Error | null = null;
    let resolveNext: ((value: string | null) => void) | null = null;

    // Register message / close / error handlers BEFORE awaiting 'open' so
    // frames delivered immediately after open are not lost in the gap.
    ws.on('message', (raw: Buffer | string) => {
      const rawStr = raw.toString();
      let text: string | null = null;

      try {
        const frame = JSON.parse(rawStr);
        if (typeof frame === 'object' && frame !== null) {
          if (frame.done) {
            done = true;
            ws.close();
            if (resolveNext) {
              const r = resolveNext;
              resolveNext = null;
              r(null);
            }
            return;
          }
          text = frame.text || null;
        } else {
          text = String(frame);
        }
      } catch {
        text = rawStr;
      }

      if (text && resolveNext) {
        const r = resolveNext;
        resolveNext = null;
        r(text);
      } else if (text) {
        chunks.push(text);
      }
    });

    ws.on('close', () => {
      done = true;
      if (resolveNext) {
        const r = resolveNext;
        resolveNext = null;
        r(null);
      }
    });

    ws.on('error', (err: Error) => {
      error = err;
      done = true;
      if (resolveNext) {
        const r = resolveNext;
        resolveNext = null;
        r(null);
      }
    });

    try {
      await new Promise<void>((resolve, reject) => {
        ws.on('open', () => {
          ws.send(JSON.stringify(data));
          resolve();
        });
        ws.on('error', (err: Error) => {
          reject(err);
        });
      });

      // Yield buffered chunks first
      while (chunks.length > 0) {
        yield chunks.shift()!;
      }

      // Then wait for new messages (with a per-read timeout to avoid
      // hanging forever if the server connects but then goes silent).
      const READ_TIMEOUT_MS = 30_000;
      // Drain buffered chunks even after ``done``/close: the consumer spends
      // seconds in TTS between yields, so a server streaming
      // ``{text:a}{text:b}{done}`` back-to-back buffered b (and the done)
      // while we were suspended mid-yield — the old ``!done`` condition then
      // silently dropped every buffered chunk after the first.
      while (!error && (chunks.length > 0 || !done)) {
        const messagePromise = new Promise<string | null>((resolve) => {
          if (chunks.length > 0) {
            resolve(chunks.shift()!);
          } else {
            resolveNext = resolve;
          }
        });
        let timeoutHandle: ReturnType<typeof setTimeout> | undefined;
        const timeoutPromise = new Promise<never>((_, reject) => {
          timeoutHandle = setTimeout(
            () => reject(new Error('WebSocket read timeout: no frame received within 30 s')),
            READ_TIMEOUT_MS
          );
        });
        let text: string | null;
        try {
          text = await Promise.race([messagePromise, timeoutPromise]);
        } catch (timeoutErr) {
          // Clean up the pending resolve so the message handler doesn't call it
          // after we've already given up.
          resolveNext = null;
          throw timeoutErr;
        } finally {
          clearTimeout(timeoutHandle);
        }
        if (text === null) break;
        yield text;
      }

      if (error) throw error;
    } finally {
      if (ws.readyState !== ws.CLOSED && ws.readyState !== ws.CLOSING) {
        ws.close();
      }
    }
  }
}

/** Check if onMessage is a remote URL string. */
export function isRemoteUrl(onMessage: unknown): onMessage is string {
  if (typeof onMessage !== 'string') return false;
  return onMessage.startsWith('http://') ||
    onMessage.startsWith('https://') ||
    onMessage.startsWith('ws://') ||
    onMessage.startsWith('wss://');
}

/** Check if a URL is a WebSocket URL. */
export function isWebSocketUrl(url: string): boolean {
  return url.startsWith('ws://') || url.startsWith('wss://');
}
