"""In-process async runs for plugin Action.on_mr with progress fan-out."""

from __future__ import annotations

import asyncio
import logging
import secrets
import threading
from collections.abc import AsyncIterator, Coroutine
from dataclasses import dataclass, field
from datetime import UTC, datetime
from typing import Any

from pygittools.merge import MergeRequest

from pygitweb.api.action import Action, ActionProgress, MrEvent, MrTrigger, ProgressSink

logger = logging.getLogger(__name__)

_MAX_HISTORY = 64


@dataclass
class ActionRun:
    run_id: str
    action_name: str
    project: str
    trigger: MrTrigger
    mr_ref: str
    mr_oid: str
    created_at: datetime = field(default_factory=lambda: datetime.now(UTC))
    history: list[ActionProgress] = field(default_factory=list)
    current: ActionProgress = field(
        default_factory=lambda: ActionProgress(message="queued", fraction=0.0),
    )
    _condition: threading.Condition = field(default_factory=threading.Condition)

    def snapshot(self) -> dict[str, object]:
        with self._condition:
            cur = self.current
            return {
                "run_id": self.run_id,
                "action": self.action_name,
                "project": self.project,
                "trigger": self.trigger.value,
                "mr_ref": self.mr_ref,
                "mr_oid": self.mr_oid,
                "created_at": self.created_at.isoformat(),
                "message": cur.message,
                "fraction": cur.fraction,
                "done": cur.done,
                "ok": cur.ok,
            }


class _RunProgressSink:
    def __init__(self, registry: ActionRunRegistry, run: ActionRun) -> None:
        self._registry = registry
        self._run = run

    async def report(self, progress: ActionProgress) -> None:
        self._registry.publish(self._run, progress)


class ActionRunRegistry:
    """Tracks MR-triggered action runs and wakes SSE/poll waiters on progress."""

    def __init__(self) -> None:
        self._runs: dict[str, ActionRun] = {}
        self._lock = threading.Lock()

    def get(self, run_id: str) -> ActionRun | None:
        with self._lock:
            return self._runs.get(run_id)

    def get_many(self, run_ids: list[str]) -> list[ActionRun]:
        with self._lock:
            return [self._runs[rid] for rid in run_ids if rid in self._runs]

    def start_mr_triggers(
        self,
        project: str,
        trigger: MrTrigger,
        mr: MergeRequest,
        *,
        actions: dict[str, Action] | None = None,
    ) -> list[str]:
        """Start matching plugin ``on_mr`` runners; return new run ids."""
        from pygitweb.plugin_loader import PLUGIN_ACTIONS

        plugins = actions if actions is not None else PLUGIN_ACTIONS
        mr_oid = ""
        try:
            from pygitweb.git_helpers import open_repo

            repo = open_repo(project)
            if mr.name in repo.references:
                mr_oid = str(repo.references[mr.name].resolve().target)
        except Exception:
            logger.exception("pygitweb: failed to resolve MR oid for triggers")

        event = MrEvent(
            project=project,
            trigger=trigger,
            mr_ref=mr.name,
            mr_oid=mr_oid,
            title=mr.title,
            ours=mr.ours,
            theirs=mr.theirs,
            head=str(mr.target),
        )
        run_ids: list[str] = []
        for name, action in plugins.items():
            try:
                wanted = action.mr_triggers()
            except Exception:
                logger.exception("pygitweb: mr_triggers failed for action %r", name)
                continue
            if trigger not in wanted:
                continue
            run_id = secrets.token_hex(8)
            run = ActionRun(
                run_id=run_id,
                action_name=name,
                project=project,
                trigger=trigger,
                mr_ref=mr.name,
                mr_oid=mr_oid,
            )
            with self._lock:
                self._runs[run_id] = run
            run_ids.append(run_id)
            self._spawn(_execute_run(self, run, action, event))
        return run_ids

    def _spawn(self, coro: Coroutine[Any, Any, None]) -> None:
        try:
            loop = asyncio.get_running_loop()
        except RuntimeError:
            thread = threading.Thread(
                target=_run_coro_in_thread,
                args=(coro,),
                daemon=True,
                name="pygitweb-action-run",
            )
            thread.start()
            return
        loop.create_task(coro)

    def publish(self, run: ActionRun, progress: ActionProgress) -> None:
        """Record progress and wake waiters (safe from any thread / event loop)."""
        with run._condition:
            run.current = progress
            run.history.append(progress)
            if len(run.history) > _MAX_HISTORY:
                run.history = run.history[-_MAX_HISTORY:]
            run._condition.notify_all()

    async def wait_for_update(
        self,
        run: ActionRun,
        after: ActionProgress,
        timeout: float = 30.0,
    ) -> ActionProgress:
        """Park until progress advances past ``after``, the run finishes, or ``timeout``."""

        def _wait() -> ActionProgress:
            with run._condition:
                if run.current is not after or run.current.done:
                    return run.current
                run._condition.wait(timeout=timeout)
                return run.current

        return await asyncio.to_thread(_wait)

    async def stream(self, run: ActionRun) -> AsyncIterator[ActionProgress]:
        """Yield the current snapshot, then each subsequent progress until done."""
        with run._condition:
            current = run.current
        yield current
        while not current.done:
            nxt = await self.wait_for_update(run, after=current)
            if nxt is current and not nxt.done:
                continue
            current = nxt
            yield current


def _run_coro_in_thread(coro: Coroutine[Any, Any, None]) -> None:
    try:
        asyncio.run(coro)
    except Exception:
        logger.exception("pygitweb: background action run failed")


async def _execute_run(
    registry: ActionRunRegistry,
    run: ActionRun,
    action: Action,
    event: MrEvent,
) -> None:
    sink: ProgressSink = _RunProgressSink(registry, run)
    registry.publish(run, ActionProgress(message="running", fraction=0.0))
    try:
        await action.on_mr(event, sink)
        if not run.current.done:
            registry.publish(
                run,
                ActionProgress(message="completed", fraction=1.0, done=True, ok=True),
            )
    except Exception as exc:
        logger.exception("pygitweb: action %r on_mr failed", run.action_name)
        registry.publish(
            run,
            ActionProgress(message=str(exc) or "failed", fraction=1.0, done=True, ok=False),
        )


ACTION_RUNS: ActionRunRegistry = ActionRunRegistry()


def start_mr_action_runs(
    project: str,
    trigger: MrTrigger,
    mr: MergeRequest,
    *,
    actions: dict[str, Action] | None = None,
) -> list[str]:
    """Convenience wrapper around ``ACTION_RUNS.start_mr_triggers``."""
    return ACTION_RUNS.start_mr_triggers(project, trigger, mr, actions=actions)


def parse_run_ids(raw: str | None) -> list[str]:
    if not raw or not raw.strip():
        return []
    return [part.strip() for part in raw.split(",") if part.strip()]