"""Client-facing job endpoints."""

from __future__ import annotations

import json
import uuid
from pathlib import Path
from typing import Annotated

from fastapi import APIRouter, Depends, Query
from fastapi.responses import FileResponse, JSONResponse

from hive_compute.config import get_settings
from hive_compute.coordinator.auth import require_client_key
from hive_compute.coordinator.billing import (
    estimate_billing_amount_cents,
    get_partner_budget_snapshot,
    get_partner_credit_balance,
)
from hive_compute.coordinator.database import get_db_ctx, record_event
from hive_compute.coordinator.metrics import increment_job_submitted
from hive_compute.coordinator.normalizer import (
    download_remote_audio,
    download_supported_media,
    is_supported_remote_media_source,
    normalize_audio,
    probe_audio_duration,
)
from hive_compute.coordinator.planner import create_shards_for_job
from hive_compute.models import (
    JobListItemResponse,
    JobListResponse,
    JobResultResponse,
    JobStatusResponse,
    JobSubmitRequest,
    JobSubmitResponse,
)
from hive_compute.url_safety import validate_outbound_http_url

router = APIRouter(tags=["jobs"])
ClientAuth = Annotated[dict, Depends(require_client_key)]


def _partner_filter(auth: dict) -> tuple[str, list]:
    """Build SQL WHERE clause for partner tenant isolation.

    Partners see only their own jobs. The default client key (no partner) sees all jobs.
    """
    if auth.get("partner_id"):
        return "AND partner_id = ?", [auth["partner_id"]]
    return "", []


@router.post("/jobs", response_model=JobSubmitResponse, status_code=201)
def submit_job(req: JobSubmitRequest, auth: ClientAuth):
    """Submit a transcription job."""
    settings = get_settings()
    job_id = f"job_{uuid.uuid4().hex[:12]}"
    resolved_model, model_source = settings.resolve_job_model(req.model, req.quality)
    billing_quality_tier = req.quality.value if model_source == "quality" else None
    input_config_payload = {
        "language": req.language,
        "model": resolved_model,
        "quality": req.quality.value,
        "model_source": model_source,
        "billing_quality_tier": billing_quality_tier,
    }
    partner_id = auth.get("partner_id")

    try:
        if not _is_local_audio_input(req.audio_url):
            validate_outbound_http_url(req.audio_url, purpose="audio_url")
        if req.callback_url:
            validate_outbound_http_url(req.callback_url, purpose="callback_url")
    except ValueError as exc:
        return JSONResponse(status_code=422, content={"error": str(exc)})

    # Normalize coordinator-ingested sources to mono 16kHz PCM WAV. Use measured
    # duration for sharding. Direct remote audio URLs still skip normalization and
    # are fetched by workers as before.
    try:
        norm_meta = _try_normalize(job_id, req.audio_url, auth, settings)
    except RuntimeError as exc:
        return JSONResponse(status_code=422, content={"error": f"Audio normalization failed: {exc}"})
    effective_duration = req.audio_duration
    effective_url = req.audio_url
    measured_duration = None
    estimated_cost_cents = None
    if norm_meta:
        effective_duration = norm_meta["duration_seconds"]
        effective_url = norm_meta["normalized_url"]
        measured_duration = norm_meta["duration_seconds"]
        if norm_meta.get("source_audio_url"):
            input_config_payload["source_audio_url"] = norm_meta["source_audio_url"]
    elif settings.remote_audio_probe_enabled:
        measured_duration = probe_audio_duration(req.audio_url)
        if measured_duration:
            effective_duration = measured_duration

    input_config = json.dumps(input_config_payload)

    initial_shard_count = 0 if (effective_duration and effective_duration > 0) else None
    max_attempts = settings.validated_fallback_max_attempts()

    if partner_id:
        budget = get_partner_budget_snapshot(partner_id)
        if effective_duration is not None:
            estimated_cost_cents = estimate_billing_amount_cents(
                effective_duration,
                budget["rate_per_minute"],
                billing_quality_tier,
            )

        budget_cents = budget["monthly_budget_cents"]
        if budget_cents is not None:
            if estimated_cost_cents is None:
                return JSONResponse(
                    status_code=422,
                    content={"error": "audio_duration is required when a partner monthly budget cap is enabled"},
                )

            remaining_cents = budget_cents - budget["billed_cents"] - budget["reserved_cents"]
            if estimated_cost_cents > remaining_cents:
                return JSONResponse(
                    status_code=403,
                    content={
                        "error": (
                            "Partner monthly budget cap exceeded: "
                            f"remaining={max(remaining_cents, 0)}¢ estimated={estimated_cost_cents}¢"
                        )
                    },
                )

        credits_usd = get_partner_credit_balance(partner_id)
        if credits_usd is not None:
            if estimated_cost_cents is None:
                return JSONResponse(
                    status_code=422,
                    content={"error": "audio_duration is required when partner prepaid credits are enabled"},
                )

            available_cents = int(round(float(credits_usd) * 100))
            if estimated_cost_cents > available_cents:
                return JSONResponse(
                    status_code=402,
                    content={
                        "error": (
                            "Partner prepaid credits exhausted: "
                            f"available=${float(credits_usd):.2f} estimated=${estimated_cost_cents / 100.0:.2f}"
                        )
                    },
                )

    with get_db_ctx() as db:
        db.execute(
            """
            INSERT INTO jobs (job_id, job_type, status, priority, audio_url, input_config,
                              callback_url, audio_duration, shard_count, max_attempts, partner_id,
                              estimated_cost_cents,
                              content_hash, measured_duration, channel_count, original_format,
                              normalized_path, audio_token)
            VALUES (?, 'transcribe', 'pending', ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
            """,
            (
                job_id,
                req.priority,
                effective_url,
                input_config,
                req.callback_url,
                effective_duration,
                initial_shard_count,
                max_attempts,
                partner_id,
                estimated_cost_cents,
                norm_meta["content_hash"] if norm_meta else None,
                measured_duration,
                _normalized_channel_count(norm_meta) if norm_meta else None,
                _normalized_original_format(req.audio_url, norm_meta) if norm_meta else None,
                norm_meta["normalized_path"] if norm_meta else None,
                norm_meta["audio_token"] if norm_meta else None,
            ),
        )
        record_event(db, job_id, "created", to_status="pending", actor=partner_id or "client")

        if initial_shard_count is not None:
            create_shards_for_job(job_id, effective_duration, db=db)

        row = db.execute("SELECT created_at FROM jobs WHERE job_id = ?", (job_id,)).fetchone()

    increment_job_submitted()

    return JobSubmitResponse(
        job_id=job_id,
        status="pending",
        created_at=row["created_at"],
        resolved_model=resolved_model,
    )


def _is_local_audio_input(audio_url: str) -> bool:
    return audio_url.startswith("/") or audio_url.startswith("file:")


def _try_normalize(job_id: str, audio_url: str, auth: dict, settings) -> dict | None:
    """Normalize coordinator-ingested audio into an internal fetchable asset."""
    if _is_local_audio_input(audio_url):
        return _normalize_local_audio(job_id, audio_url, auth, settings)
    if is_supported_remote_media_source(audio_url):
        return _normalize_remote_media(job_id, audio_url, settings)
    if settings.remote_audio_ingest_enabled:
        return _normalize_remote_audio(job_id, audio_url, settings)
    return None


def _normalize_local_audio(job_id: str, audio_url: str, auth: dict, settings) -> dict:
    import shutil

    if auth.get("partner_id"):
        raise RuntimeError("Local audio inputs are not allowed for partner API keys")

    if not settings.allow_local_audio_jobs:
        raise RuntimeError(
            "Local audio inputs are disabled on this coordinator. "
            "Upload the file to HTTPS storage or set HIVE_ALLOW_LOCAL_AUDIO_JOBS=true for trusted internal use."
        )

    input_path = Path(audio_url.removeprefix("file://"))
    if not input_path.exists():
        raise RuntimeError(f"Local audio file not found: {input_path}")

    if not shutil.which("ffmpeg"):
        raise RuntimeError("ffmpeg is required to process local audio files but was not found")

    meta = normalize_audio(input_path, _normalized_audio_output_path(settings, job_id))
    meta["source_audio_url"] = audio_url
    return _finalize_normalized_audio(job_id, meta, settings)


def _normalize_remote_media(job_id: str, audio_url: str, settings) -> dict:
    if not settings.remote_media_ingest_enabled:
        raise RuntimeError(
            "Remote media ingest is disabled on this coordinator. "
            "Provide a direct audio file URL or enable HIVE_REMOTE_MEDIA_INGEST_ENABLED."
        )

    download_dir = _normalized_audio_root(settings) / "downloads"
    downloaded_path = download_supported_media(audio_url, download_dir, stem=job_id)
    try:
        meta = normalize_audio(downloaded_path, _normalized_audio_output_path(settings, job_id))
    finally:
        downloaded_path.unlink(missing_ok=True)

    meta["source_audio_url"] = audio_url
    return _finalize_normalized_audio(job_id, meta, settings)


def _normalize_remote_audio(job_id: str, audio_url: str, settings) -> dict:
    download_dir = _normalized_audio_root(settings) / "downloads"
    downloaded_path = download_remote_audio(audio_url, download_dir, stem=job_id)
    try:
        meta = normalize_audio(downloaded_path, _normalized_audio_output_path(settings, job_id))
    finally:
        downloaded_path.unlink(missing_ok=True)

    meta["source_audio_url"] = audio_url
    return _finalize_normalized_audio(job_id, meta, settings)


def _normalized_audio_output_path(settings, job_id: str) -> Path:
    return _normalized_audio_root(settings) / f"{job_id}.wav"


def _normalized_audio_root(settings) -> Path:
    db_path = Path(settings.database_path).resolve().parent
    norm_dir = db_path / "normalized_audio"
    norm_dir.mkdir(parents=True, exist_ok=True)
    return norm_dir


def _finalize_normalized_audio(job_id: str, meta: dict, settings) -> dict:
    import secrets

    output_path = _normalized_audio_output_path(settings, job_id)
    meta["normalized_path"] = str(output_path)
    token = secrets.token_urlsafe(32)
    meta["audio_token"] = token
    origin = settings.get_manager_origins()[0].rstrip("/")
    meta["normalized_url"] = f"{origin}/api/v1/audio/{job_id}/{token}"
    return meta


def _normalized_channel_count(norm_meta: dict) -> int | None:
    return norm_meta.get("channel_count", norm_meta.get("channels"))


def _normalized_original_format(audio_url: str, norm_meta: dict) -> str | None:
    if norm_meta.get("original_format"):
        return norm_meta["original_format"]
    suffix = Path(audio_url.removeprefix("file://")).suffix.lower().lstrip(".")
    return suffix or None


@router.get("/audio/{job_id}/{token}")
def serve_normalized_audio(job_id: str, token: str):
    """Serve normalized audio. Random token in URL provides access control.

    No auth header required (workers download via raw GET). The token is
    a random secret generated at normalization time — not derivable from content.
    """
    with get_db_ctx() as db:
        row = db.execute(
            "SELECT normalized_path, audio_token FROM jobs WHERE job_id = ?", (job_id,),
        ).fetchone()

    if not row or not row["normalized_path"]:
        return JSONResponse(status_code=404, content={"error": "No normalized audio"})

    if not row["audio_token"] or token != row["audio_token"]:
        return JSONResponse(status_code=403, content={"error": "Invalid audio access token"})

    path = Path(row["normalized_path"])
    if not path.exists():
        return JSONResponse(status_code=404, content={"error": "Normalized audio file missing"})

    return FileResponse(path, media_type="audio/wav", filename=f"{job_id}.wav")


@router.get("/jobs", response_model=JobListResponse)
def list_jobs(
    auth: ClientAuth,
    status: str | None = Query(default=None, description="Filter by status"),
    limit: int = Query(default=50, ge=1, le=200),
    offset: int = Query(default=0, ge=0),
):
    """List jobs with optional status filter. Partners only see their own jobs."""
    pf, pf_params = _partner_filter(auth)

    with get_db_ctx() as db:
        if status:
            rows = db.execute(
                f"SELECT * FROM jobs WHERE status = ? {pf} ORDER BY created_at DESC LIMIT ? OFFSET ?",
                (status, *pf_params, limit, offset),
            ).fetchall()
            total = db.execute(
                f"SELECT COUNT(*) as cnt FROM jobs WHERE status = ? {pf}",
                (status, *pf_params),
            ).fetchone()["cnt"]
        else:
            rows = db.execute(
                f"SELECT * FROM jobs WHERE 1=1 {pf} ORDER BY created_at DESC LIMIT ? OFFSET ?",
                (*pf_params, limit, offset),
            ).fetchall()
            total = db.execute(
                f"SELECT COUNT(*) as cnt FROM jobs WHERE 1=1 {pf}",
                pf_params,
            ).fetchone()["cnt"]

    return JobListResponse(
        jobs=[
            JobListItemResponse(
                job_id=row["job_id"],
                job_type=row["job_type"],
                status=row["status"],
                priority=row["priority"],
                claimed_by=row["claimed_by"],
                created_at=row["created_at"],
                completed_at=row["completed_at"],
                attempts=row["attempts"],
            )
            for row in rows
        ],
        total=total,
        limit=limit,
        offset=offset,
    )


@router.get("/jobs/{job_id}", response_model=JobStatusResponse)
def get_job_status(job_id: str, auth: ClientAuth):
    """Get job status."""
    pf, pf_params = _partner_filter(auth)

    with get_db_ctx() as db:
        row = db.execute(
            f"SELECT * FROM jobs WHERE job_id = ? {pf}",
            (job_id, *pf_params),
        ).fetchone()

    if not row:
        return JSONResponse(status_code=404, content={"error": "Job not found"})

    return JobStatusResponse(
        job_id=row["job_id"],
        job_type=row["job_type"],
        status=row["status"],
        priority=row["priority"],
        claimed_by=row["claimed_by"],
        created_at=row["created_at"],
        claimed_at=row["claimed_at"],
        completed_at=row["completed_at"],
        attempts=row["attempts"],
        max_attempts=row["max_attempts"],
    )


@router.get("/jobs/{job_id}/result", response_model=JobResultResponse)
def get_job_result(job_id: str, auth: ClientAuth):
    """Get job result."""
    pf, pf_params = _partner_filter(auth)

    with get_db_ctx() as db:
        row = db.execute(
            f"SELECT * FROM jobs WHERE job_id = ? {pf}",
            (job_id, *pf_params),
        ).fetchone()

    if not row:
        return JSONResponse(status_code=404, content={"error": "Job not found"})

    if row["status"] != "completed":
        return JSONResponse(
            status_code=202,
            content={"job_id": job_id, "status": row["status"], "message": "Job not yet completed"},
        )

    result = json.loads(row["result_data"]) if row["result_data"] else None
    return JobResultResponse(
        job_id=row["job_id"],
        status=row["status"],
        result=result,
        completed_at=row["completed_at"],
    )


@router.get("/jobs/{job_id}/shards")
def get_job_shards(job_id: str, auth: ClientAuth):
    """Get shard breakdown for a sharded job."""
    pf, pf_params = _partner_filter(auth)

    with get_db_ctx() as db:
        # Verify job belongs to partner
        job = db.execute(
            f"SELECT job_id FROM jobs WHERE job_id = ? {pf}",
            (job_id, *pf_params),
        ).fetchone()
        if not job:
            return JSONResponse(status_code=404, content={"error": "Job not found"})

        rows = db.execute(
            "SELECT * FROM shards WHERE job_id = ? AND canary_id IS NULL ORDER BY shard_index ASC",
            (job_id,),
        ).fetchall()

    return {
        "job_id": job_id,
        "shards": [
            {
                "shard_id": row["shard_id"],
                "shard_index": row["shard_index"],
                "start_seconds": row["start_seconds"],
                "end_seconds": row["end_seconds"],
                "overlap_seconds": row["overlap_seconds"],
                "status": row["status"],
                "claimed_by": row["claimed_by"],
                "attempts": row["attempts"],
            }
            for row in rows
        ],
        "total": len(rows),
    }


@router.get("/jobs/{job_id}/events")
def get_job_events(job_id: str, auth: ClientAuth):
    """Get audit trail for a job."""
    pf, pf_params = _partner_filter(auth)

    with get_db_ctx() as db:
        job = db.execute(
            f"SELECT job_id FROM jobs WHERE job_id = ? {pf}",
            (job_id, *pf_params),
        ).fetchone()
        if not job:
            return JSONResponse(status_code=404, content={"error": "Job not found"})

        rows = db.execute(
            "SELECT * FROM job_events WHERE job_id = ? AND event_type NOT LIKE 'canary%' ORDER BY created_at ASC",
            (job_id,),
        ).fetchall()

    return [
        {
            "event_id": row["event_id"],
            "job_id": row["job_id"],
            "event_type": row["event_type"],
            "from_status": row["from_status"],
            "to_status": row["to_status"],
            "actor": row["actor"],
            "detail": row["detail"],
            "created_at": row["created_at"],
        }
        for row in rows
    ]
