# retoor <retoor@molodetz.nl>

import asyncio
import json
import logging
from datetime import datetime, timedelta, timezone

from devplacepy.database import get_table, get_int_setting
from devplacepy.services.base import BaseService, ConfigField
from devplacepy.services.jobs import queue

logger = logging.getLogger(__name__)

DEFAULT_RETENTION_SECONDS = 7 * 24 * 60 * 60
DEFAULT_MAX_CONCURRENT = 2
DEFAULT_JOB_TIMEOUT_SECONDS = 600


class JobService(BaseService):
    kind = ""
    min_interval = 1
    max_retries = 3

    def __init__(self, name: str, interval_seconds: int = 2):
        super().__init__(name=name, interval_seconds=interval_seconds)
        self.kind = self.kind or name
        self.retention_key = f"{name}_retention_seconds"
        self.concurrency_key = f"{name}_max_concurrent"
        self.timeout_key = f"{name}_job_timeout_seconds"
        self.config_fields = [
            ConfigField(
                self.retention_key,
                "Artifact retention (seconds)",
                type="int",
                default=DEFAULT_RETENTION_SECONDS,
                minimum=60,
                help="Finished artifacts are deleted once unused for this long.",
                group="Jobs",
            ),
            ConfigField(
                self.concurrency_key,
                "Max concurrent jobs",
                type="int",
                default=DEFAULT_MAX_CONCURRENT,
                minimum=1,
                maximum=16,
                help="Number of jobs processed in parallel.",
                group="Jobs",
            ),
            ConfigField(
                self.timeout_key,
                "Job timeout (seconds)",
                type="int",
                default=DEFAULT_JOB_TIMEOUT_SECONDS,
                minimum=10,
                help="A running job with no live task after this long is retried.",
                group="Jobs",
            ),
        ]
        self._inflight: dict = {}

    def retention_seconds(self) -> int:
        return max(60, get_int_setting(self.retention_key, DEFAULT_RETENTION_SECONDS))

    def max_concurrent(self) -> int:
        return max(1, get_int_setting(self.concurrency_key, DEFAULT_MAX_CONCURRENT))

    def job_timeout_seconds(self) -> int:
        return max(10, get_int_setting(self.timeout_key, DEFAULT_JOB_TIMEOUT_SECONDS))

    async def process(self, job: dict) -> dict:
        raise NotImplementedError

    def cleanup(self, job: dict) -> None:
        pass

    async def on_enable(self) -> None:
        self._recover_orphans(min_age_seconds=0)

    async def on_disable(self) -> None:
        for entry in list(self._inflight.values()):
            entry["task"].cancel()
        self._inflight.clear()

    async def run_once(self) -> None:
        self._reap()
        self._recover_orphans(min_age_seconds=self.job_timeout_seconds())
        self._refill()
        await self._sweep_expired()

    def _reap(self) -> None:
        for uid in list(self._inflight):
            entry = self._inflight[uid]
            task = entry["task"]
            if not task.done():
                continue
            del self._inflight[uid]
            duration_ms = int(
                (datetime.now(timezone.utc) - entry["started"]).total_seconds() * 1000
            )
            exc = task.exception() if not task.cancelled() else asyncio.CancelledError()
            if exc is not None:
                self._finish_failed(
                    uid, str(exc) or exc.__class__.__name__, duration_ms
                )
            else:
                self._finish_done(uid, task.result() or {}, duration_ms)

    def _finish_done(self, uid: str, job_result: dict, duration_ms: int) -> None:
        now = datetime.now(timezone.utc)
        get_table("jobs").update(
            {
                "uid": uid,
                "status": queue.DONE,
                "result": json.dumps(job_result),
                "error": "",
                "completed_at": now.isoformat(),
                "updated_at": now.isoformat(),
                "duration_ms": duration_ms,
                "last_accessed_at": now.isoformat(),
                "expires_at": (
                    now + timedelta(seconds=self.retention_seconds())
                ).isoformat(),
                "bytes_in": int(job_result.get("bytes_in", 0)),
                "bytes_out": int(job_result.get("bytes_out", 0)),
                "item_count": int(job_result.get("item_count", 0)),
            },
            ["uid"],
        )
        self.log(f"Job {uid} done in {duration_ms}ms")

    def _finish_failed(self, uid: str, error: str, duration_ms: int) -> None:
        now = datetime.now(timezone.utc)
        get_table("jobs").update(
            {
                "uid": uid,
                "status": queue.FAILED,
                "error": error[:2000],
                "completed_at": now.isoformat(),
                "updated_at": now.isoformat(),
                "duration_ms": duration_ms,
            },
            ["uid"],
        )
        self.log(f"Job {uid} failed: {error}")

    def _refill(self) -> None:
        capacity = self.max_concurrent() - len(self._inflight)
        if capacity <= 0:
            return
        table = get_table("jobs")
        pending = list(
            table.find(
                kind=self.kind, status=queue.PENDING, order_by=["uid"], _limit=capacity
            )
        )
        for row in pending:
            uid = row["uid"]
            if uid in self._inflight:
                continue
            now = datetime.now(timezone.utc)
            table.update(
                {
                    "uid": uid,
                    "status": queue.RUNNING,
                    "started_at": now.isoformat(),
                    "updated_at": now.isoformat(),
                    "error": "",
                },
                ["uid"],
            )
            job = queue.get_job(uid)
            self._inflight[uid] = {
                "task": asyncio.create_task(self._run_job(job)),
                "started": now,
            }
            self.log(f"Job {uid} started")

    async def _run_job(self, job: dict) -> dict:
        return await self.process(job)

    def _recover_orphans(self, min_age_seconds: int) -> None:
        table = get_table("jobs")
        now = datetime.now(timezone.utc)
        for row in table.find(kind=self.kind, status=queue.RUNNING):
            uid = row["uid"]
            if uid in self._inflight:
                continue
            if min_age_seconds > 0 and not self._older_than(
                row.get("started_at"), now, min_age_seconds
            ):
                continue
            retry_count = int(row.get("retry_count") or 0)
            if retry_count >= self.max_retries:
                self._finish_failed(
                    uid, "exceeded retry limit after orphan recovery", 0
                )
                continue
            table.update(
                {
                    "uid": uid,
                    "status": queue.PENDING,
                    "retry_count": retry_count + 1,
                    "started_at": "",
                    "updated_at": now.isoformat(),
                },
                ["uid"],
            )
            self.log(f"Recovered orphaned job {uid} (retry {retry_count + 1})")

    async def _sweep_expired(self) -> None:
        table = get_table("jobs")
        now = datetime.now(timezone.utc)
        for row in table.find(kind=self.kind, status=queue.DONE):
            expires_at = row.get("expires_at")
            if not expires_at or not self._is_past(expires_at, now):
                continue
            job = queue.get_job(row["uid"])
            try:
                await asyncio.to_thread(self.cleanup, job)
            except Exception as exc:
                self.log(f"Cleanup failed for {row['uid']}: {exc}")
            table.delete(uid=row["uid"])
            self.log(f"Pruned expired job {row['uid']}")

    def _older_than(self, stamp: str, now: datetime, seconds: int) -> bool:
        moment = self._parse(stamp)
        if moment is None:
            return True
        return (now - moment).total_seconds() >= seconds

    def _is_past(self, stamp: str, now: datetime) -> bool:
        moment = self._parse(stamp)
        return moment is not None and moment < now

    def _parse(self, stamp: str):
        if not stamp:
            return None
        try:
            return datetime.fromisoformat(stamp)
        except (ValueError, TypeError):
            return None

    def collect_metrics(self) -> dict:
        metrics = queue.job_metrics(self.kind)
        by_status = metrics["by_status"]
        stats = [
            {"label": "Total jobs", "value": metrics["total"]},
            {"label": "Pending", "value": by_status[queue.PENDING]},
            {"label": "Running", "value": by_status[queue.RUNNING]},
            {"label": "Done", "value": by_status[queue.DONE]},
            {"label": "Failed", "value": by_status[queue.FAILED]},
            {"label": "Items zipped", "value": metrics["items"]},
            {"label": "Bytes in", "value": _human_bytes(metrics["bytes_in"])},
            {"label": "Bytes out", "value": _human_bytes(metrics["bytes_out"])},
            {"label": "Avg duration", "value": f"{metrics['avg_ms']} ms"},
        ]
        table = {
            "columns": ["Job", "Status", "Name", "Out", "Items"],
            "rows": [
                [
                    (row.get("uid") or "")[:8],
                    row.get("status", ""),
                    (row.get("preferred_name") or "")[:32],
                    _human_bytes(int(row.get("bytes_out") or 0)),
                    int(row.get("item_count") or 0),
                ]
                for row in metrics["recent"]
            ],
        }
        return {"stats": stats, "table": table}


def _human_bytes(size: int) -> str:
    value = float(size)
    for unit in ("B", "KB", "MB", "GB", "TB"):
        if value < 1024 or unit == "TB":
            return f"{value:.0f} {unit}" if unit == "B" else f"{value:.1f} {unit}"
        value /= 1024
    return f"{value:.1f} TB"
