"""Server-driven bulk job runner — the fix for the "submitting client is gone"
failure class (10 of the last 15 failed CSV runs on 2026-10-20/21).

Before this module the BROWSER orchestrated every bulk run: it chunked the
file, submitted each chunk to /validate/bulk, polled the engine, mirrored rows
into verifier_checks and finalized the job. If the tab closed, throttled or
slept mid-run, the job orphaned in 'processing' until the reaper closed it and
refunded the unprocessed remainder. The engine itself was healthy — its chunk
task even completed 500/500 while the browser row froze at 0/3000.

Contract (v2):
  POST /api/v1/jobs  {job_id, emails, deep?, final?}
    - JWT required (productFetch always attaches the Supabase session token).
    - verifier_jobs row must exist, belong to the JWT subject and still be
      open (queued/processing).
    - A background task validates the list in chunks (bulk_sync_chunk_size),
      upserts verifier_checks rows after EVERY chunk, heartbeats verifier_jobs
      (processed_count + counts + updated_at) so the dashboard and the reaper
      both see live progress, and — on the final batch — flips the job to
      'completed' and refunds risky rows server-side.
    - Any chunk-level failure finalizes the job as 'error' and refunds BOTH
      the risky rows and the unprocessed remainder (same ledger mechanics as
      the client finalize and the pg_cron reaper; idempotent via
      refunded_credits).
  The client polls verifier_jobs directly after accepting 202 — it no longer
  needs to stay alive for the run to finish.
"""
import asyncio
import json
import logging
import time
from typing import Dict, List, Optional

from fastapi import APIRouter, HTTPException, Request
from pydantic import BaseModel

from ..auth import verify_supabase_jwt
from ..config import settings
from ..db import repo
from ..models import ValidationResult
from ..validator.engine import engine

logger = logging.getLogger("emailvalidator.jobs")

router = APIRouter()

# Strong references so the event loop cannot GC an in-flight runner, and a
# semaphore so concurrent job submissions cannot spawn unbounded SMTP work.
_job_tasks: set[asyncio.Task] = set()
_job_semaphore = asyncio.Semaphore(settings.bulk_concurrency)

GROUPS = ("good", "risky", "bad")


class JobRequest(BaseModel):
    job_id: str
    emails: List[str]
    deep: bool = True
    final: bool = True
    user_id: Optional[str] = None  # informational; ownership comes from the JWT
    # Per-email upload metadata the CSV carried (frontend submitServerJob sends
    # it keyed by lowercase email). Optional + additive: predating clients omit it.
    pass_through: Optional[Dict[str, Dict[str, str]]] = None


# ---- classification (mirror of src/lib/verifier.ts classify()) ------------


def classify_engine_result(r: ValidationResult) -> dict:
    """Map an engine ValidationResult to the verifier_checks row shape the
    dashboard renders. Mirrors the frontend classify() branch order so a row
    never disagrees with a client-side re-classification:
      disposable → bad | dead domain → bad | dns inconclusive → risky |
      valid+smtp → good | engine invalid → bad | catch-all → risky | else risky.
    """
    disposable = bool(getattr(r.disposable, "is_disposable", False))
    catch_all = bool(getattr(r.catch_all, "is_catch_all", False))
    dns = r.dns
    status = (r.status.value if hasattr(r.status, "value") else str(r.status)) or "unknown"

    checks_run = r.checks_run or []
    dns_error = (dns.error or "") if dns else ""
    dns_failure = dns_error == "dns_timeout" or "timeout" in dns_error.lower()
    a_fallback_only = bool(dns and dns.domain_exists and not dns.has_mx_records)
    smtp_verified = bool(r.smtp and r.smtp.verified is True)
    reason = r.reason_code

    if disposable:
        out_status, result, grp = "disposable", "Disposable", "bad"
        reason = reason or "disposable"
    elif dns is not None and dns.domain_exists is False:
        # NXDOMAIN is definitive (the stale-build regression that showed a
        # 0% Bad column on a 3000-row list must never resurrect server-side).
        out_status, result, grp = "invalid", "Invalid", "bad"
        reason = reason or "nxdomain"
    elif dns_failure or a_fallback_only:
        out_status, result, grp = "unknown", "Unknown", "risky"
        reason = reason or ("dns_timeout" if dns_failure else "a_fallback_only")
    elif status == "valid" and smtp_verified:
        out_status, result, grp = "valid", "Ok", "good"
    elif status == "invalid":
        out_status, result, grp = "invalid", "Invalid", "bad"
    elif catch_all:
        out_status, result, grp = "catch_all", "Catch-all", "risky"
        reason = reason or "catch_all_detected"
    else:
        out_status, result, grp = "unknown", "Unknown", "risky"
        reason = reason or "smtp_unavailable"

    score = r.scoring.score if r.scoring is not None else None
    return {
        "email": r.email,
        "status": out_status,
        "result": result,
        "grp": grp,
        "score": score,
        "reason": reason,
        "pass_through": None,
        "checks_run": checks_run,
        "verified_at": r.validated_at,
        "run_id": r.run_id,
        "validator_version": r.validator_version,
    }


# ---- persistence helpers (service-role writes, best-effort like repo) ------


async def _write_checks(rows: List[dict]) -> None:
    if not rows:
        return
    try:
        await asyncio.to_thread(repo.write_verifier_checks, rows)
    except Exception as exc:  # never lose a run to one write blip
        logger.warning("job_checks_write_failed: %s", exc)


async def _update_job(job_id: str, **fields) -> None:
    try:
        await asyncio.to_thread(repo.update_verifier_job, job_id, fields)
    except Exception as exc:
        logger.warning("job_update_failed job=%s: %s", job_id, exc)


async def _finalize_failed(job_id: str, error: str, counts: dict, processed: int) -> None:
    """Close a failed job and refund risky + unprocessed server-side.

    Mirrors the ledger mechanics of public.refund_job_risky() (migration
    20261020000000) minus its auth.uid() guard — the engine writes with the
    service-role key, under which auth.uid() is null and the SQL function
    would raise. Idempotency comes from verifier_jobs.refunded_credits.
    """
    await _update_job(
        job_id,
        status="error",
        error=error[:500],
        processed_count=processed,
        valid_count=counts.get("good", 0),
        risky_count=counts.get("risky", 0),
        bad_count=counts.get("bad", 0),
    )
    try:
        job = await asyncio.to_thread(repo.get_verifier_job, job_id)
        if not job:
            return
        total = int(job.get("total_count") or 0)
        charged = int(job.get("credits_charged") or 0)
        refunded = int(job.get("refunded_credits") or 0)
        risky = int(job.get("risky_count") or 0)
        unprocessed = max(total - int(job.get("processed_count") or 0), 0)
        due = min(risky + unprocessed, charged) - refunded
        if due > 0:
            await asyncio.to_thread(repo.refund_credits, job["user_id"], due,
                                    "Auto-refund — risky and unprocessed results are not billed")
            await _update_job(job_id, refunded_credits=refunded + due)
    except Exception as exc:
        logger.warning("job_refund_failed job=%s: %s", job_id, exc)


# ---- the runner ------------------------------------------------------------


async def _run_verifier_job(job_id: str, user_id: str, emails: List[str], deep: bool,
                            final: bool, pass_through: Optional[dict] = None) -> None:
    started = time.monotonic()
    counts = {g: 0 for g in GROUPS}
    processed = 0
    # Defensive dedupe (the frontend already deduped; a client bug must not
    # double-write rows — (job_id, email) upsert would silently dedupe them
    # while counts drift).
    seen: set[str] = set()
    ordered: List[str] = []
    for e in emails:
        e = (e or "").strip().lower()
        if e and "@" in e and e not in seen:
            seen.add(e)
            ordered.append(e)
    emails = ordered
    total = len(emails)
    if total == 0:
        await _finalize_failed(job_id, "No valid addresses reached the engine.", counts, 0)
        return

    chunk_size = max(int(settings.bulk_sync_chunk_size or 50), 10)
    async with _job_semaphore:
        try:
            await _update_job(job_id, status="processing")
            for i in range(0, total, chunk_size):
                chunk = emails[i:i + chunk_size]
                try:
                    results = await engine.validate_batch(chunk, deep=deep)
                except Exception as exc:  # one bad stage must not kill the job
                    logger.warning("job_chunk_failed job=%s: %s", job_id, exc)
                    await _finalize_failed(
                        job_id,
                        f"Engine chunk failed after {processed}/{total} addresses: {exc}",
                        counts, processed,
                    )
                    return
                if len(results) != len(chunk):
                    await _finalize_failed(
                        job_id,
                        f"Engine returned {len(results)} of {len(chunk)} results for one chunk",
                        counts, processed,
                    )
                    return
                # Stamp ownership columns: verifier_checks.user_id/job_id are
                # NOT NULL — rows without them fail the upsert (23502) and the
                # run completes with counts but zero detail rows (2026-11-01).
                rows = []
                for r in results:
                    row = classify_engine_result(r)
                    row["job_id"] = job_id
                    row["user_id"] = user_id
                    if pass_through:
                        row["pass_through"] = pass_through.get((r.email or "").lower())
                    rows.append(row)
                await _write_checks(rows)
                for row in rows:
                    counts[row["grp"]] = counts.get(row["grp"], 0) + 1
                processed += len(chunk)
                # Heartbeat + live progress in one write; the dashboard polls
                # this row instead of the engine, and the reaper reads it.
                await _update_job(
                    job_id,
                    processed_count=processed,
                    valid_count=counts["good"],
                    risky_count=counts["risky"],
                    bad_count=counts["bad"],
                )
            if final:
                await _update_job(
                    job_id,
                    status="completed",
                    error=None,
                    processed_count=processed,
                    valid_count=counts["good"],
                    risky_count=counts["risky"],
                    bad_count=counts["bad"],
                )
                # Risky rows are never billed; the completed-job refund covers
                # exactly the risky_count (unprocessed term is zero by design).
                try:
                    job = await asyncio.to_thread(repo.get_verifier_job, job_id)
                    if job:
                        charged = int(job.get("credits_charged") or 0)
                        refunded = int(job.get("refunded_credits") or 0)
                        due = min(counts["risky"], charged) - refunded
                        if due > 0:
                            await asyncio.to_thread(
                                repo.refund_credits, job["user_id"], due,
                                "Auto-refund — risky results are not billed")
                            await _update_job(job_id, refunded_credits=refunded + due)
                except Exception as exc:
                    logger.warning("job_risky_refund_failed job=%s: %s", job_id, exc)
            logger.info("job_batch_done job=%s processed=%s/%s final=%s secs=%.1f",
                        job_id, processed, total, final, time.monotonic() - started)
        except Exception as exc:  # absolute safety net
            logger.exception("job_runner_crashed job=%s", job_id)
            await _finalize_failed(job_id, f"Runner crashed: {exc}", counts, processed)


# ---- routes ----------------------------------------------------------------


@router.post("/jobs", tags=["Bulk Jobs"])
async def create_server_job(request: Request, payload: JobRequest):
    """Accept a bulk job for server-side processing (202 + DB polling)."""
    token = request.headers.get("authorization", "")
    token = token[7:].strip() if token.lower().startswith("bearer ") else ""
    if not token:
        raise HTTPException(401, "Missing bearer token — sign in first.")
    try:
        claims = await verify_supabase_jwt(token)
    except Exception as exc:
        raise HTTPException(401, f"Invalid or expired token: {exc}")
    user_id = claims.get("sub", "")
    if not user_id:
        raise HTTPException(401, "Token has no subject claim.")
    if not payload.emails:
        raise HTTPException(400, "emails must not be empty.")
    if not settings.supabase_configured:
        raise HTTPException(503, "Server persistence is not configured (SUPABASE_URL/SERVICE key).")

    job = await asyncio.to_thread(repo.get_verifier_job, payload.job_id)
    if not job:
        raise HTTPException(404, f"Job {payload.job_id} not found — create it first.")
    if str(job.get("user_id")) != user_id:
        # Ownership mismatch: identical response for missing/foreign jobs.
        raise HTTPException(404, f"Job {payload.job_id} not found for this account.")
    if job.get("status") not in ("queued", "processing"):
        raise HTTPException(409, f"Job {payload.job_id} is already {job.get('status')}.")
    if not payload.emails:
        raise HTTPException(400, "emails must not be empty.")

    # Cap one submission so a giant file cannot pin the runner forever: the
    # client splits into final-flagged batches (20k ≈ 600 KB JSON).
    MAX_PER_CALL = 20000
    emails = payload.emails[:MAX_PER_CALL]

    task = asyncio.create_task(
        _run_verifier_job(payload.job_id, user_id, emails, payload.deep,
                          payload.final, payload.pass_through)
    )
    _job_tasks.add(task)
    task.add_done_callback(_job_tasks.discard)

    return {
        "accepted": len(emails),
        "job_id": payload.job_id,
        "status": "processing",
        "final": payload.final,
        "message": "Processing server-side — poll verifier_jobs for progress.",
    }


@router.get("/jobs/{job_id}", tags=["Bulk Jobs"])
async def job_status(job_id: str, request: Request):
    """Lightweight engine-side echo of the job row (debug aid; the dashboard
    polls the DB directly and does not depend on this)."""
    token = request.headers.get("authorization", "")
    token = token[7:].strip() if token.lower().startswith("bearer ") else ""
    if not token:
        raise HTTPException(401, "Missing bearer token — sign in first.")
    try:
        claims = await verify_supabase_jwt(token)
    except Exception as exc:
        raise HTTPException(401, f"Invalid or expired token: {exc}")
    job = await asyncio.to_thread(repo.get_verifier_job, job_id)
    if not job or str(job.get("user_id")) != claims.get("sub", ""):
        raise HTTPException(404, f"Job {job_id} not found for this account.")
    return {
        "job_id": job_id,
        "status": job.get("status"),
        "total": job.get("total_count"),
        "processed": job.get("processed_count"),
        "valid": job.get("valid_count"),
        "risky": job.get("risky_count"),
        "bad": job.get("bad_count"),
    }