import { Injectable, Logger, ServiceUnavailableException } from '@nestjs/common';
import { ConfigService } from '@nestjs/config';
import * as http from 'http';
import * as https from 'https';
import axios from 'axios';

export interface SynthesizedSpeech {
  stream: NodeJS.ReadableStream;
  contentType: string;
}

export interface ExtractedResume {
  text: string;
  sections: {
    contact?: Record<string, string>;
    experience?: string[];
    education?: string[];
    skills?: string[];
    summary?: string;
  };
  metadata: {
    hasTables: boolean;
    hasImages: boolean;
    pageCount: number;
    confidence: number;
    hasMultipleColumns?: boolean;
    fontCount?: number;
    minFontSize?: number | null;
    averageFontSize?: number | null;
    averageCharsPerPage?: number;
    longLineCount?: number;
    bulletCount?: number;
  };
}

/**
 * Upper bound for extraction. The Python service is a remote dependency that
 * can sleep (HF Spaces idle) or wedge; without a timeout axios waits forever
 * and every upload pins a connection until the process dies.
 */
const EXTRACT_TIMEOUT_MS = 60_000;

/**
 * The Python Space sleeps when idle, and waking it takes most of a minute — a
 * cold start was measured answering its own root path in 58s. That means the
 * first upload after an idle period reliably failed and the user was told the
 * extraction service was unavailable, when in truth their request had just woken
 * it up and the next one would have worked.
 *
 * The first attempt is therefore treated as the wake-up call: on an error that
 * looks like "not awake yet" we wait and try once more.
 */
const EXTRACT_ATTEMPTS = 2;
const WAKE_RETRY_DELAY_MS = 15_000;

/**
 * Does this failure look like a sleeping or still-booting service rather than a
 * genuine rejection? A 4xx means the service answered and said no — retrying
 * that would just waste the user's time.
 */
function looksAsleep(error: unknown): boolean {
  if (!axios.isAxiosError(error)) return false;

  const status = error.response?.status;
  if (status !== undefined) {
    // 503/504 from the Space itself or the gateway in front of it while booting.
    return status === 502 || status === 503 || status === 504;
  }
  // No response at all: timeout, refused, reset, DNS not ready.
  return ['ECONNABORTED', 'ETIMEDOUT', 'ECONNREFUSED', 'ECONNRESET', 'ENOTFOUND', 'EAI_AGAIN']
    .includes(error.code ?? '');
}

const sleep = (ms: number) => new Promise<void>(resolve => setTimeout(resolve, ms));

@Injectable()
export class PythonBridgeService {
  private readonly logger = new Logger(PythonBridgeService.name);
  private readonly baseUrl: string;
  private readonly serviceKey: string | undefined;

  constructor(private readonly configService: ConfigService) {
    this.baseUrl =
      this.configService.get<string>('PYTHON_SERVICE_URL') ??
      'http://localhost:8000';
    this.serviceKey = this.configService.get<string>('SERVICE_API_KEY');
    if (!this.serviceKey) {
      this.logger.warn(
        'SERVICE_API_KEY is not set: calls to the Python service will be rejected.',
      );
    }
  }

  /** Shared-secret header the Python service checks on every compute route. */
  private authHeaders(): Record<string, string> {
    return this.serviceKey ? { 'x-service-key': this.serviceKey } : {};
  }

  /**
   * Call Python service to extract text and structure from a PDF resume buffer
   */
  async extractResume(fileBuffer: Buffer, originalFileName: string): Promise<ExtractedResume> {
    const isDocx = originalFileName.toLowerCase().endsWith('.docx');
    const contentType = isDocx
      ? 'application/vnd.openxmlformats-officedocument.wordprocessingml.document'
      : 'application/pdf';

    for (let attempt = 1; attempt <= EXTRACT_ATTEMPTS; attempt++) {
      try {
        // Rebuilt per attempt: a FormData that has already been streamed cannot
        // be replayed.
        const formData = new FormData();
        const blob = new Blob([new Uint8Array(fileBuffer)], { type: contentType });
        formData.append('file', blob, originalFileName);

        const response = await axios.post<ExtractedResume>(
          `${this.baseUrl}/extract-resume`,
          formData,
          {
            headers: {
              'Content-Type': 'multipart/form-data',
              ...this.authHeaders(),
            },
            timeout: EXTRACT_TIMEOUT_MS,
          },
        );
        return response.data;
      } catch (error) {
        // Log status/message only — the full axios error carries the request
        // config, including the service key and the uploaded resume body.
        const status = axios.isAxiosError(error) ? error.response?.status : undefined;
        const message = error instanceof Error ? error.message : String(error);

        const retryable = attempt < EXTRACT_ATTEMPTS && looksAsleep(error);
        this.logger.error(
          `Resume extraction failed (attempt ${attempt}/${EXTRACT_ATTEMPTS})` +
          `${status ? ` (HTTP ${status})` : ''}: ${message}` +
          `${retryable ? `: service looks asleep, retrying in ${WAKE_RETRY_DELAY_MS / 1000}s` : ''}`,
        );

        if (!retryable) {
          throw new ServiceUnavailableException(
            'PDF extraction service is currently unavailable. Please try again later.',
          );
        }
        await sleep(WAKE_RETRY_DELAY_MS);
      }
    }

    // Unreachable: the final attempt either returns or throws above.
    throw new ServiceUnavailableException(
      'PDF extraction service is currently unavailable. Please try again later.',
    );
  }

  /**
   * Call the Python service to synthesize speech (Kokoro TTS) for the given text.
   * Returns the upstream audio as a stream (piped straight to the client instead
   * of buffering the whole clip in memory, so first audio reaches the browser as
   * soon as Python starts sending). Throws ServiceUnavailableException if TTS is
   * down, so callers can fall back to browser speech synthesis.
   */
  async synthesizeSpeech(text: string, voice?: string, format?: string): Promise<SynthesizedSpeech> {
    try {
      const response = await axios.post<NodeJS.ReadableStream>(
        `${this.baseUrl}/tts`,
        { text, voice, format },
        { responseType: 'stream', timeout: 30000, headers: this.authHeaders() },
      );
      return {
        stream: response.data,
        contentType: (response.headers['content-type'] as string) ?? 'audio/wav',
      };
    } catch {
      throw new ServiceUnavailableException('Speech synthesis is currently unavailable.');
    }
  }

  /**
   * Fire a synthesis request purely to warm the Python service's audio cache
   * (e.g. pre-generating question audio right after /generate, so the browser's
   * later requests are cache hits instead of multi-second live syntheses).
   * The response body is discarded; failures propagate so callers can stop
   * warming when TTS is down.
   */
  async warmSpeech(text: string, format: string = 'ogg'): Promise<void> {
    await axios.post(
      `${this.baseUrl}/tts`,
      { text, format },
      { responseType: 'arraybuffer', timeout: 60000, headers: this.authHeaders() },
    );
  }

  /**
   * Generic POST helper — keeps all HTTP logic in one place
   */
  private post<T>(path: string, body: Record<string, any>): Promise<T> {
    return new Promise((resolve, reject) => {
      const payload = JSON.stringify(body);
      const url = new URL(this.baseUrl + path);
      const isHttps = url.protocol === 'https:';
      const transport = isHttps ? https : http;

      const options = {
        hostname: url.hostname,
        port: url.port || (isHttps ? 443 : 80),
        path: url.pathname,
        method: 'POST',
        headers: {
          'Content-Type': 'application/json',
          'Content-Length': Buffer.byteLength(payload),
          ...this.authHeaders(),
        },
      };

      const req = transport.request(options, (res) => {
        let data = '';
        res.on('data', (chunk) => (data += chunk));
        res.on('end', () => {
          try {
            resolve(JSON.parse(data) as T);
          } catch {
            reject(new Error('Invalid JSON response from Python service'));
          }
        });
      });

      req.on('error', reject);
      req.setTimeout(30000, () => {
        req.destroy();
        reject(new Error('Python service request timed out'));
      });

      req.write(payload);
      req.end();
    });
  }
}