services/workers.py

from __future__ import annotations

import asyncio
import tempfile
import httpx
from concurrent.futures import ThreadPoolExecutor
from datetime import datetime, timezone
from pathlib import Path
from typing import Optional
from dataclasses import dataclass

from services import job_store
from services.cv_chunker import chunk_cv
from services.cv_converter import CVConverter
from services.job_matcher import JobMatcher

# ---------------------------------------------------------------------------
# Dedicated single-thread executor for ML work.
#
# Why not asyncio's default thread pool?
#   asyncio.to_thread() uses the loop's default ThreadPoolExecutor which is
#   shared with ALL other coroutines in the process (file I/O, HTTP clients,
#   etc.).  When two heavy ML tasks run simultaneously they can saturate that
#   shared pool, starving incoming HTTP requests of threads and making the
#   server appear frozen even though the event loop is technically free.
#
#   A dedicated pool with max_workers=1 means:
#   • ML work is 100% isolated — never competes with HTTP request threads.
#   • Only one Marker/chunk_cv call runs at a time (model is not thread-safe).
#   • The asyncio default pool stays free for file reads, httpx, etc.
# ---------------------------------------------------------------------------

_ML_EXECUTOR = ThreadPoolExecutor(max_workers=1, thread_name_prefix="ml-worker")


class CvWorker:
    """Handles background tasks for CV processing."""

    def __init__(self, converter: CVConverter, matcher: JobMatcher):
        self.converter = converter
        self.matcher = matcher

    async def run_cv_processing(
        self,
        job_id: str,
        file_bytes: bytes,
        filename: str,
        callback_url: str,
        callback_secret: str = None,
    ) -> None:
        """Process a CV and POST the result back to the Java callback endpoint."""
        loop = asyncio.get_running_loop()
        tmp_path: Optional[Path] = None
        try:
            t_start = datetime.now(timezone.utc)

            suffix = Path(filename).suffix.lower() if filename else ".pdf"
            with tempfile.NamedTemporaryFile(delete=False, suffix=suffix) as tmp:
                tmp.write(file_bytes)
                tmp_path = Path(tmp.name)

            # Conversion — CPU-bound, isolated executor
            t_conv_start = datetime.now(timezone.utc)
            conversion = await loop.run_in_executor(
                _ML_EXECUTOR, self.converter.convert, tmp_path
            )
            t_conv_end = datetime.now(timezone.utc)
            conversion_ms = int((t_conv_end - t_conv_start).total_seconds() * 1000)

            if not conversion.success:
                error_msg = conversion.error or "Conversion failed"
                job_store.set_failed(job_id, error_msg)
                await self.post_callback(
                    callback_url,
                    {
                        "message": error_msg,
                        "statusCode": 422,
                        "payload": None,
                    },
                    callback_secret,
                )
                return

            # chunk_cv with embedder — CPU-bound, isolated executor
            t_chunk_start = datetime.now(timezone.utc)
            chunks = await loop.run_in_executor(
                _ML_EXECUTOR,
                lambda: chunk_cv(conversion.markdown, embedder=self.matcher._embed),
            )
            t_chunk_end = datetime.now(timezone.utc)
            chunking_ms = int((t_chunk_end - t_chunk_start).total_seconds() * 1000)

            t_end = datetime.now(timezone.utc)
            total_ms = int((t_end - t_start).total_seconds() * 1000)

            payload = {
                "markdown": conversion.markdown,
                "cv_title": chunks["cv_title"],
                "seniority": chunks["seniority"],
                "years_experience": chunks["years_experience"],
                "category": chunks["category"],
                "completeness_score": chunks["completeness_score"],
                "chunks": {
                    "summary": chunks["chunks"]["summary"],
                    "contact": chunks["chunks"]["contact"],
                    "links": chunks["chunks"]["links"],
                    "skills": chunks["chunks"]["skills"],
                    "experience": chunks["chunks"]["experience"],
                    "education": chunks["chunks"]["education"],
                    "projects": chunks["chunks"]["projects"],
                    "awards": chunks["chunks"]["awards"],
                },
                "section_embeddings": chunks.get("section_embeddings"),
                "file_type": conversion.file_type,
                "method_used": conversion.method_used,
                "is_scanned": conversion.is_scanned,
                "page_count": conversion.page_count,
                "warnings": conversion.warnings,
                "timing": {
                    "conversion_ms": conversion_ms,
                    "chunking_ms": chunking_ms,
                    "total_ms": total_ms,
                },
                "processed_at": t_end.isoformat(),
            }

            job_store.set_completed(job_id, payload)

            await self.post_callback(
                callback_url,
                {
                    "message": "CV processed successfully",
                    "statusCode": 200,
                    "payload": payload,
                },
                callback_secret,
            )

        except Exception as exc:
            error_msg = f"Processing error: {exc}"
            job_store.set_failed(job_id, error_msg)
            await self.post_callback(
                callback_url,
                {
                    "message": error_msg,
                    "statusCode": 500,
                    "payload": None,
                },
                callback_secret,
            )

        finally:
            if tmp_path is not None:
                try:
                    tmp_path.unlink(missing_ok=True)
                except Exception:
                    pass

    async def post_callback(
        self,
        callback_url: str,
        body: dict,
        callback_secret: str = None,
    ) -> None:
        """POST the processing result back to Java."""
        try:
            headers = {}
            if callback_secret:
                headers["X-Callback-Secret"] = callback_secret
            async with httpx.AsyncClient(timeout=30.0) as client:
                response = await client.post(callback_url, json=body, headers=headers)
                response.raise_for_status()
        except Exception as exc:
            print(f"[CV-ASYNC] Failed to POST callback to {callback_url}: {exc}")


@dataclass
class CvTask:
    job_id: str
    file_bytes: bytes
    filename: str
    callback_url: str
    callback_secret: str = None


class QueueManager:
    """
    Manages an asyncio Queue with a single background worker.

    concurrency=1 is intentional — the Marker ML model and the
    sentence-transformer are NOT thread-safe and must not run in
    parallel on the same process.  Jobs queue up and are processed
    one at a time.  The dedicated _ML_EXECUTOR above ensures ML work
    never blocks the asyncio event loop — HTTP endpoints (job-status,
    new submissions) stay responsive even while a CV is being processed.
    """

    def __init__(self, worker: CvWorker, concurrency: int = 1):
        self.worker = worker
        self.concurrency = concurrency
        self.queue: asyncio.Queue = asyncio.Queue()
        self.tasks: list = []
        self._active_ids: set[str] = set()

    async def start(self) -> None:
        for _ in range(self.concurrency):
            task = asyncio.create_task(self._worker_loop())
            self.tasks.append(task)
        print(
            f"[QUEUE] Started {self.concurrency} worker(s) — one CV processed at a time."
        )

    async def stop(self) -> None:
        for task in self.tasks:
            task.cancel()
        await asyncio.gather(*self.tasks, return_exceptions=True)
        _ML_EXECUTOR.shutdown(wait=False)
        print("[QUEUE] Stopped all workers.")

    async def _worker_loop(self) -> None:
        while True:
            try:
                task: CvTask = await self.queue.get()
                job_store.set_processing(task.job_id)
                self._active_ids.discard(task.job_id)
                print(
                    f"[QUEUE] Worker picked up job {task.job_id}. "
                    f"Remaining in queue: {self.queue.qsize()}"
                )
                await self.worker.run_cv_processing(
                    task.job_id,
                    task.file_bytes,
                    task.filename,
                    task.callback_url,
                    task.callback_secret,
                )
                self.queue.task_done()
            except asyncio.CancelledError:
                break
            except Exception as exc:
                print(f"[QUEUE] Unhandled error in worker loop: {exc}")

    async def enqueue(self, task: CvTask) -> None:
        if task.job_id in self._active_ids:
            print(
                f"[QUEUE] Duplicate submission rejected: job {task.job_id} is already queued."
            )
            raise ValueError(f"Job {task.job_id} is already in the processing queue.")
        self._active_ids.add(task.job_id)
        job_store.set_queued(task.job_id)
        await self.queue.put(task)
        print(f"[QUEUE] Job {task.job_id} enqueued. Queue depth: {self.queue.qsize()}")

    def get_queue_position(self, job_id: str) -> Optional[int]:
        try:
            for idx, task in enumerate(list(self.queue._queue)):
                if task.job_id == job_id:
                    return idx + 1
        except Exception as exc:
            print(f"[QUEUE] Error checking queue position: {exc}")
        return None