"""Durable job queue on PostgreSQL (SELECT … FOR UPDATE SKIP LOCKED). Survives reboots; retries with
exponential backoff + jitter; dead-letter after max_attempts; checkpoints keyed by scope."""
from __future__ import annotations

import json
import random
import socket
from datetime import UTC, datetime, timedelta
from typing import Any

from findata_core.config import get_settings
from findata_core.db import connection, transaction
from findata_core.redact import redact

WORKER_ID = f"{socket.gethostname()}:{random.randint(1000, 9999)}"


def enqueue(kind: str, dataset_id: str | None = None, params: dict[str, Any] | None = None, priority: int = 50,
            dedupe_key: str | None = None, max_attempts: int = 5, created_by: str = "system",
            run_at: datetime | None = None) -> str | None:
    """Insert a job; returns id, or None if a job with the same dedupe_key is already queued/running."""
    with transaction() as conn:
        if dedupe_key:
            r = conn.execute("SELECT id FROM ingest_jobs WHERE dedupe_key=%s", (dedupe_key,)).fetchone()
            if r:
                st = conn.execute("SELECT status FROM ingest_jobs WHERE id=%s", (r["id"],)).fetchone()["status"]
                if st in ("queued", "running"):
                    return None
                conn.execute("UPDATE ingest_jobs SET dedupe_key=NULL WHERE id=%s", (r["id"],))
        r = conn.execute(
            "INSERT INTO ingest_jobs(kind, dataset_id, params, priority, dedupe_key, max_attempts, created_by, next_run_at) "
            "VALUES (%s,%s,%s,%s,%s,%s,%s,%s) RETURNING id",
            (kind, dataset_id, json.dumps(params or {}), priority, dedupe_key, max_attempts, created_by,
             run_at or datetime.now(UTC))).fetchone()
        return str(r["id"])


def claim_next(kinds: list[str] | None = None) -> dict[str, Any] | None:
    with transaction() as conn:
        sql = ("SELECT * FROM ingest_jobs WHERE status='queued' AND next_run_at <= now() "
               + ("AND kind = ANY(%s) " if kinds else "") +
               "ORDER BY priority, next_run_at LIMIT 1 FOR UPDATE SKIP LOCKED")
        job = conn.execute(sql, (kinds,) if kinds else None).fetchone()
        if not job:
            return None
        conn.execute("UPDATE ingest_jobs SET status='running', attempts=attempts+1, locked_by=%s, locked_at=now(), "
                     "started_at=COALESCE(started_at, now()) WHERE id=%s", (WORKER_ID, job["id"]))
        run = conn.execute("INSERT INTO ingest_runs(job_id, worker, code_version) VALUES (%s,%s,%s) RETURNING id",
                           (job["id"], WORKER_ID, _code_version())).fetchone()
        job = dict(job)
        job["attempts"] += 1
        job["run_id"] = str(run["id"])
        return job


def finish(job: dict[str, Any], ok: bool, result: dict[str, Any] | None = None, error: str | None = None) -> None:
    secrets = get_settings().secret_values()
    with transaction() as conn:
        if ok:
            conn.execute("UPDATE ingest_jobs SET status='succeeded', finished_at=now(), locked_by=NULL, result=%s, last_error=NULL WHERE id=%s",
                         (json.dumps(result or {}, default=str), job["id"]))
            conn.execute("UPDATE ingest_runs SET status='succeeded', finished_at=now(), stats=%s WHERE id=%s",
                         (json.dumps(result or {}, default=str), job["run_id"]))
            return
        msg = redact(error or "unknown error", secrets)[:4000]
        dead = job["attempts"] >= job["max_attempts"]
        if dead:
            conn.execute("UPDATE ingest_jobs SET status='dead', finished_at=now(), locked_by=NULL, last_error=%s WHERE id=%s", (msg, job["id"]))
        else:
            delay = min(3600, 30 * 2 ** job["attempts"]) + random.uniform(0, 30)
            conn.execute("UPDATE ingest_jobs SET status='queued', locked_by=NULL, last_error=%s, next_run_at=%s WHERE id=%s",
                         (msg, datetime.now(UTC) + timedelta(seconds=delay), job["id"]))
        conn.execute("UPDATE ingest_runs SET status=%s, finished_at=now(), stats=%s WHERE id=%s",
                     ("dead" if dead else "failed", json.dumps({"error": msg}), job["run_id"]))


def progress(job_id: str, data: dict[str, Any]) -> None:
    with transaction() as conn:
        conn.execute("UPDATE ingest_jobs SET progress=%s WHERE id=%s", (json.dumps(data, default=str), job_id))


def set_checkpoint(scope: str, key: str, value: dict[str, Any]) -> None:
    with transaction() as conn:
        conn.execute("INSERT INTO checkpoints(scope, key, value) VALUES (%s,%s,%s) ON CONFLICT (scope, key) DO UPDATE SET value=EXCLUDED.value, updated_at=now()",
                     (scope, key, json.dumps(value, default=str)))


def get_checkpoint(scope: str, key: str) -> dict[str, Any] | None:
    with connection() as conn:
        r = conn.execute("SELECT value FROM checkpoints WHERE scope=%s AND key=%s", (scope, key)).fetchone()
        return r["value"] if r else None


def requeue_stale(max_age_minutes: int = 240) -> int:
    """Jobs left 'running' by a crashed worker are re-queued (idempotent tasks + checkpoints make this safe)."""
    with transaction() as conn:
        r = conn.execute("UPDATE ingest_jobs SET status='queued', locked_by=NULL, next_run_at=now() WHERE status='running' "
                         "AND locked_at < now() - make_interval(mins => %s) RETURNING id", (max_age_minutes,)).fetchall()
        return len(r)


def _code_version() -> str:
    import subprocess
    try:
        return subprocess.run(["git", "rev-parse", "--short", "HEAD"], capture_output=True, text=True, timeout=5,
                              cwd=str(get_settings().root)).stdout.strip() or "unknown"
    except Exception:
        return "unknown"