"""Deterministic Gitea issue-to-release execution engine."""
from __future__ import annotations

import argparse
import errno
import fcntl
import json
import os
import re
import signal
import shlex
import subprocess
import tempfile
import time
from dataclasses import asdict, dataclass
from pathlib import Path
from typing import Any, Iterable, Protocol
from urllib import error, request


PRIORITY = {"P0": 0, "P1": 1, "P2": 2}
RUNTIME_ENV_NAMES = (
    "HOME", "LANG", "LC_ALL", "LC_CTYPE", "LOGNAME", "SHELL",
    "SSH_AUTH_SOCK", "TMPDIR", "USER",
)


@dataclass(frozen=True)
class Issue:
    number: int
    title: str
    body: str
    state: str
    labels: list[str]
    assignee: str | None


@dataclass(frozen=True)
class CommandResult:
    returncode: int
    stdout: str
    stderr: str
    timed_out: bool = False


@dataclass(frozen=True)
class EngineConfig:
    repo: str
    agent: str
    repo_path: Path
    state_path: Path
    agent_command: str
    test_command: str
    base_branch: str = "main"
    agent_timeout_seconds: float = 1800.0
    test_timeout_seconds: float = 900.0
    termination_grace_seconds: float = 5.0
    agent_env_vars: tuple[str, ...] = ()


@dataclass(frozen=True)
class RunState:
    issue: int
    branch: str
    status: str
    evidence: str = ""
    pr_number: int | None = None
    pr_url: str = ""
    schema_version: int = 1
    repo: str = ""
    head_sha: str = ""


class Client(Protocol):
    def list_open_issues(self, repo: str) -> list[Issue]: ...
    def claim_issue(self, repo: str, number: int, assignee: str) -> None: ...
    def get_issue(self, repo: str, number: int) -> Issue: ...
    def find_open_pr(self, repo: str, head: str) -> dict[str, Any] | None: ...
    def create_pr(self, repo: str, head: str, base: str, title: str, body: str) -> dict[str, Any]: ...


class Runner(Protocol):
    def run(
        self,
        command: list[str],
        cwd: Path,
        env: dict[str, str] | None = None,
        timeout_seconds: float | None = None,
        termination_grace_seconds: float = 5.0,
    ) -> CommandResult: ...


class StateStore:
    def __init__(self, path: Path):
        self.path = Path(path)

    def load(self) -> RunState | None:
        if not self.path.exists():
            return None
        return RunState(**json.loads(self.path.read_text()))

    def save(self, state: RunState) -> None:
        self.path.parent.mkdir(parents=True, exist_ok=True)
        temporary_path = None
        try:
            with tempfile.NamedTemporaryFile(
                mode="w",
                encoding="utf-8",
                dir=self.path.parent,
                prefix=self.path.name + ".",
                suffix=".tmp",
                delete=False,
            ) as temporary:
                temporary_path = Path(temporary.name)
                temporary.write(json.dumps(asdict(state), indent=2) + "\n")
                temporary.flush()
                os.fsync(temporary.fileno())
            os.replace(temporary_path, self.path)
            directory = os.open(self.path.parent, os.O_RDONLY | os.O_DIRECTORY)
            try:
                os.fsync(directory)
            finally:
                os.close(directory)
        finally:
            if temporary_path is not None:
                temporary_path.unlink(missing_ok=True)


class RepositoryLease:
    """Non-blocking, process-backed single-writer lease for one worktree."""

    def __init__(self, repo_path: Path):
        self.path = Path(repo_path).resolve() / ".release-engine" / "run.lock"
        self._file = None

    def __enter__(self):
        self.path.parent.mkdir(parents=True, exist_ok=True)
        lock_file = self.path.open("a+", encoding="utf-8")
        try:
            fcntl.flock(lock_file.fileno(), fcntl.LOCK_EX | fcntl.LOCK_NB)
        except OSError as exc:
            if exc.errno in (errno.EACCES, errno.EAGAIN):
                detail = ""
                try:
                    lock_file.seek(0)
                    metadata = json.loads(lock_file.read(512))
                    owner_pid = int(metadata["pid"])
                    acquired_at = float(metadata["acquired_at"])
                    detail = f" (owner pid={owner_pid}, acquired_at={acquired_at:.3f})"
                except (KeyError, TypeError, ValueError, json.JSONDecodeError):
                    pass
                lock_file.close()
                raise RuntimeError(f"release run already active{detail}") from exc
            lock_file.close()
            raise
        self._file = lock_file
        lock_file.seek(0)
        lock_file.truncate()
        lock_file.write(json.dumps({"pid": os.getpid(), "acquired_at": time.time()}) + "\n")
        lock_file.flush()
        os.fsync(lock_file.fileno())
        return self

    def __exit__(self, exc_type, exc, traceback):
        if self._file is not None:
            fcntl.flock(self._file.fileno(), fcntl.LOCK_UN)
            self._file.close()
            self._file = None


def _label_names(labels: Iterable[Any]) -> list[str]:
    return [label.get("name", "") if isinstance(label, dict) else str(label) for label in labels]


def _priority(issue: Issue) -> tuple[int, int]:
    rank = min((PRIORITY.get(name, 3) for name in issue.labels), default=3)
    return rank, issue.number


def select_issue(issues: Iterable[Issue], agent: str, issue_number: int | None = None) -> Issue:
    open_issues = [i for i in issues if i.state == "open"]
    if issue_number is not None:
        match = next((i for i in open_issues if i.number == issue_number), None)
        if match is None:
            raise ValueError(f"open issue #{issue_number} not found")
        if match.assignee not in (None, agent):
            raise ValueError(f"issue #{issue_number} assigned to {match.assignee}")
        return match

    eligible = [i for i in open_issues if i.assignee in (None, agent)]
    if not eligible:
        raise ValueError("no eligible open issues")
    return sorted(eligible, key=_priority)[0]


def branch_name(agent: str, issue: Issue) -> str:
    slug = re.sub(r"[^a-z0-9]+", "-", issue.title.lower()).strip("-")[:48]
    actor = re.sub(r"[^a-z0-9]+", "-", agent.lower()).strip("-")
    return f"{actor}/{issue.number}-{slug or 'issue'}"


def parse_command(command: str, name: str) -> list[str]:
    try:
        arguments = shlex.split(command)
    except ValueError as exc:
        raise ValueError(f"invalid {name}: {exc}") from exc
    if not arguments:
        raise ValueError(f"invalid {name}: command is empty")
    return arguments


def subprocess_environment(
    allowed_names: Iterable[str] = (),
    overrides: dict[str, str] | None = None,
) -> dict[str, str]:
    """Build an explicit child environment without inheriting parent secrets."""
    environment = {"PATH": os.environ.get("PATH", os.defpath)}
    for name in (*RUNTIME_ENV_NAMES, *allowed_names):
        if name in os.environ:
            environment[name] = os.environ[name]
    if overrides:
        environment.update(overrides)
    return environment


def validate_agent_env_vars(names: Iterable[str]) -> None:
    for name in names:
        if (
            re.fullmatch(r"[A-Za-z_][A-Za-z0-9_]*", name) is None
            or name == "GITEA_TOKEN"
            or name.startswith("RELEASE_")
        ):
            raise ValueError(f"invalid agent environment variable: {name}")


class ShellRunner:
    def run(
        self,
        command: list[str],
        cwd: Path,
        env: dict[str, str] | None = None,
        timeout_seconds: float | None = None,
        termination_grace_seconds: float = 5.0,
    ) -> CommandResult:
        process = subprocess.Popen(
            command,
            cwd=str(cwd),
            env=dict(env or {}),
            shell=False,
            text=True,
            stdout=subprocess.PIPE,
            stderr=subprocess.PIPE,
            start_new_session=True,
        )
        try:
            stdout, stderr = process.communicate(timeout=timeout_seconds)
            return CommandResult(process.returncode, stdout, stderr)
        except subprocess.TimeoutExpired:
            os.killpg(process.pid, signal.SIGTERM)
            try:
                stdout, stderr = process.communicate(timeout=termination_grace_seconds)
            except subprocess.TimeoutExpired:
                os.killpg(process.pid, signal.SIGKILL)
                stdout, stderr = process.communicate()
            return CommandResult(process.returncode, stdout, stderr, timed_out=True)


class GiteaClient:
    def __init__(self, api_url: str, token: str):
        self.api_url = api_url.rstrip("/")
        self.token = token

    def _request(self, method: str, path: str, body: dict[str, Any] | None = None) -> Any:
        data = json.dumps(body).encode() if body is not None else None
        req = request.Request(f"{self.api_url}/{path.lstrip('/')}", data=data, method=method)
        req.add_header("Accept", "application/json")
        req.add_header("Authorization", f"token {self.token}")
        if data is not None:
            req.add_header("Content-Type", "application/json")
        try:
            with request.urlopen(req, timeout=20) as response:
                raw = response.read()
                return json.loads(raw) if raw else None
        except error.HTTPError as exc:
            detail = exc.read().decode(errors="replace")
            if exc.code == 401:
                detail += " (authentication rejected; refresh the cron GITEA_TOKEN)"
            raise RuntimeError(f"Gitea {method} {path} failed: HTTP {exc.code}: {detail}") from exc

    @staticmethod
    def _issue(item: dict[str, Any]) -> Issue:
        assignee = (item.get("assignee") or {}).get("login")
        return Issue(
            number=item["number"],
            title=item["title"],
            body=item.get("body", ""),
            state=item["state"],
            labels=_label_names(item.get("labels", [])),
            assignee=assignee,
        )

    def list_open_issues(self, repo: str) -> list[Issue]:
        items = self._request("GET", f"repos/{repo}/issues?state=open&limit=100")
        return [self._issue(item) for item in items if not item.get("pull_request")]

    def claim_issue(self, repo: str, number: int, assignee: str) -> None:
        self._request("PATCH", f"repos/{repo}/issues/{number}", {"assignee": assignee})

    def get_issue(self, repo: str, number: int) -> Issue:
        return self._issue(self._request("GET", f"repos/{repo}/issues/{number}"))

    def find_open_pr(self, repo: str, head: str) -> dict[str, Any] | None:
        pulls = self._request("GET", f"repos/{repo}/pulls?state=open&limit=100")
        for pull in pulls:
            if pull.get("head", {}).get("label") == head or pull.get("head", {}).get("ref") == head:
                return pull
        return None

    def create_pr(self, repo: str, head: str, base: str, title: str, body: str) -> dict[str, Any]:
        return self._request("POST", f"repos/{repo}/pulls", {
            "head": head,
            "base": base,
            "title": title,
            "body": body,
        })


class ReleaseEngine:
    def __init__(self, config: EngineConfig, client: Client, runner: Runner):
        self.config = config
        self.client = client
        self.runner = runner
        self.store = StateStore(config.state_path)

    def execute(self, issue_number: int | None = None, dry_run: bool = False) -> dict[str, Any]:
        with RepositoryLease(self.config.repo_path):
            return self._execute_locked(issue_number=issue_number, dry_run=dry_run)

    def _execute_locked(self, issue_number: int | None = None, dry_run: bool = False) -> dict[str, Any]:
        deadlines = {
            "agent timeout seconds": self.config.agent_timeout_seconds,
            "test timeout seconds": self.config.test_timeout_seconds,
            "termination grace seconds": self.config.termination_grace_seconds,
        }
        for name, value in deadlines.items():
            if value <= 0:
                raise ValueError(f"{name} must be positive")
        validate_agent_env_vars(self.config.agent_env_vars)

        issue = select_issue(self.client.list_open_issues(self.config.repo), self.config.agent, issue_number)
        branch = branch_name(self.config.agent, issue)
        head_sha = ""

        existing_pr = self.client.find_open_pr(self.config.repo, branch)
        if existing_pr:
            return {
                "issue": issue.number,
                "branch": branch,
                "pr_number": existing_pr["number"],
                "pr_url": existing_pr.get("html_url", ""),
                "reused": True,
            }

        if dry_run:
            return {
                "dry_run": True,
                "issue": issue.number,
                "title": issue.title,
                "branch": branch,
                "repo": self.config.repo,
                "agent": self.config.agent,
            }

        agent_command = parse_command(self.config.agent_command, "agent command")
        test_command = parse_command(self.config.test_command, "test command")
        runtime_env = subprocess_environment()
        agent_env = subprocess_environment(
            self.config.agent_env_vars,
            {
                "RELEASE_ISSUE_NUMBER": str(issue.number),
                "RELEASE_ISSUE_TITLE": issue.title,
                "RELEASE_ISSUE_BODY": issue.body,
                "RELEASE_REPO": self.config.repo,
                "RELEASE_BRANCH": branch,
            },
        )
        checkpoint = self.store.load()
        resumable_statuses = {"agent_complete", "tests_passed", "pushed"}
        if checkpoint is not None and checkpoint.status in resumable_statuses:
            if (
                checkpoint.schema_version != 1
                or checkpoint.repo != self.config.repo
                or checkpoint.issue != issue.number
                or checkpoint.branch != branch
                or not checkpoint.head_sha
            ):
                raise RuntimeError("release checkpoint does not match this run")
            branch_result = self.runner.run(
                ["git", "checkout", branch], self.config.repo_path, runtime_env
            )
            if branch_result.returncode != 0:
                raise RuntimeError(f"checkpoint branch checkout failed: {branch_result.stderr.strip()}")
            head_result = self.runner.run(
                ["git", "rev-parse", "HEAD"], self.config.repo_path, runtime_env
            )
            if head_result.returncode != 0 or head_result.stdout.strip() != checkpoint.head_sha:
                raise RuntimeError("release checkpoint commit does not match the local branch")
            status_result = self.runner.run(
                ["git", "status", "--porcelain"], self.config.repo_path, runtime_env
            )
            if status_result.returncode != 0 or status_result.stdout.strip():
                raise RuntimeError("release checkpoint worktree is not clean")
            head_sha = checkpoint.head_sha
            evidence = checkpoint.evidence
        else:
            checkpoint = None
            evidence = ""
            self.client.claim_issue(self.config.repo, issue.number, self.config.agent)
            verified = self.client.get_issue(self.config.repo, issue.number)
            if verified.assignee != self.config.agent:
                raise RuntimeError(
                    f"claim verification failed: expected {self.config.agent}, got {verified.assignee}"
                )
            self.store.save(RunState(issue.number, branch, "claimed"))

            branch_result = self.runner.run(
                ["git", "checkout", "-B", branch], self.config.repo_path, runtime_env
            )
            if branch_result.returncode != 0:
                raise RuntimeError(f"branch creation failed: {branch_result.stderr.strip()}")

            agent_result = self.runner.run(
                agent_command,
                self.config.repo_path,
                agent_env,
                timeout_seconds=self.config.agent_timeout_seconds,
                termination_grace_seconds=self.config.termination_grace_seconds,
            )
            if agent_result.timed_out:
                evidence = (agent_result.stdout + "\n" + agent_result.stderr).strip()[-2000:]
                self.store.save(RunState(issue.number, branch, "agent_timed_out", evidence))
                raise RuntimeError(f"coding agent timed out: {evidence}")
            if agent_result.returncode != 0:
                self.store.save(RunState(issue.number, branch, "agent_failed", agent_result.stderr[-2000:]))
                raise RuntimeError(f"coding agent failed: {agent_result.stderr.strip()}")
            head_result = self.runner.run(
                ["git", "rev-parse", "HEAD"], self.config.repo_path, runtime_env
            )
            head_sha = head_result.stdout.strip()
            if head_result.returncode != 0 or not head_sha:
                raise RuntimeError("could not identify the release checkpoint commit")
            self.store.save(RunState(
                issue.number, branch, "agent_complete",
                repo=self.config.repo, head_sha=head_sha,
            ))

        if checkpoint is None or checkpoint.status == "agent_complete":
            test_result = self.runner.run(
                test_command,
                self.config.repo_path,
                runtime_env,
                timeout_seconds=self.config.test_timeout_seconds,
                termination_grace_seconds=self.config.termination_grace_seconds,
            )
            evidence = (test_result.stdout + "\n" + test_result.stderr).strip()[-4000:]
            if test_result.timed_out:
                self.store.save(RunState(
                    issue.number, branch, "tests_timed_out", evidence,
                    repo=self.config.repo, head_sha=head_sha,
                ))
                raise RuntimeError(f"tests timed out: {evidence}")
            if test_result.returncode != 0:
                self.store.save(RunState(
                    issue.number, branch, "tests_failed", evidence,
                    repo=self.config.repo, head_sha=head_sha,
                ))
                raise RuntimeError(f"tests failed: {evidence}")
            self.store.save(RunState(
                issue.number, branch, "tests_passed", evidence,
                repo=self.config.repo, head_sha=head_sha,
            ))

        if checkpoint is not None and checkpoint.status == "pushed":
            remote_result = self.runner.run(
                ["git", "ls-remote", "--heads", "origin", branch],
                self.config.repo_path,
                runtime_env,
            )
            expected_remote = f"{checkpoint.head_sha}\trefs/heads/{branch}"
            if remote_result.returncode != 0 or remote_result.stdout.strip() != expected_remote:
                raise RuntimeError("release checkpoint commit does not match the remote branch")
        else:
            push_result = self.runner.run(
                ["git", "push", "-u", "origin", branch], self.config.repo_path, runtime_env
            )
            if push_result.returncode != 0:
                self.store.save(RunState(
                    issue.number, branch, "push_failed", push_result.stderr[-2000:],
                    repo=self.config.repo, head_sha=head_sha,
                ))
                raise RuntimeError(f"push failed: {push_result.stderr.strip()}")
            self.store.save(RunState(
                issue.number, branch, "pushed", evidence,
                repo=self.config.repo, head_sha=head_sha,
            ))

        pr_body = (
            f"Closes #{issue.number}\n\n"
            "## Automated validation\n"
            f"- Command: `{self.config.test_command}`\n"
            f"- Result: PASS\n\n"
            "```text\n"
            f"{evidence[-2000:]}\n"
            "```"
        )
        pr = self.client.create_pr(
            self.config.repo,
            branch,
            self.config.base_branch,
            f"feat: {issue.title}",
            pr_body,
        )
        final = RunState(
            issue.number,
            branch,
            "pr_opened",
            evidence,
            pr_number=pr["number"],
            pr_url=pr.get("html_url", ""),
            repo=self.config.repo,
            head_sha=head_sha,
        )
        self.store.save(final)
        return {
            "issue": issue.number,
            "branch": branch,
            "pr_number": pr["number"],
            "pr_url": pr.get("html_url", ""),
            "tests": "passed",
        }


def _parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--repo", required=True)
    parser.add_argument("--agent", default=os.getenv("RELEASE_AGENT", "timmy"))
    parser.add_argument("--repo-path", type=Path, default=Path.cwd())
    parser.add_argument("--state", type=Path, default=Path(".release-engine/state.json"))
    parser.add_argument("--issue", type=int)
    parser.add_argument("--dry-run", action="store_true")
    parser.add_argument("--api", default=os.getenv("GITEA_API", "http://127.0.0.1:3000/api/v1"))
    parser.add_argument("--agent-command", default=os.getenv("RELEASE_AGENT_COMMAND", ""))
    parser.add_argument("--test-command", default=os.getenv("RELEASE_TEST_COMMAND", "python3 -m pytest tests/ -q"))
    parser.add_argument("--agent-timeout", type=float, default=float(os.getenv("RELEASE_AGENT_TIMEOUT", "1800")))
    parser.add_argument("--test-timeout", type=float, default=float(os.getenv("RELEASE_TEST_TIMEOUT", "900")))
    parser.add_argument("--termination-grace", type=float, default=float(os.getenv("RELEASE_TERMINATION_GRACE", "5")))
    parser.add_argument(
        "--agent-env",
        action="append",
        default=[name for name in os.getenv("RELEASE_AGENT_ENV", "").split(",") if name],
        help="parent environment variable to pass to the coding agent; repeat as needed",
    )
    return parser.parse_args()


def main() -> int:
    args = _parse_args()
    token = os.getenv("GITEA_TOKEN", "")
    if not token:
        raise SystemExit("GITEA_TOKEN is required")
    if not args.dry_run and not args.agent_command:
        raise SystemExit("RELEASE_AGENT_COMMAND is required outside --dry-run")
    config = EngineConfig(
        repo=args.repo,
        agent=args.agent,
        repo_path=args.repo_path.resolve(),
        state_path=args.state,
        agent_command=args.agent_command,
        test_command=args.test_command,
        agent_timeout_seconds=args.agent_timeout,
        test_timeout_seconds=args.test_timeout,
        termination_grace_seconds=args.termination_grace,
        agent_env_vars=tuple(args.agent_env),
    )
    result = ReleaseEngine(config, GiteaClient(args.api, token), ShellRunner()).execute(
        issue_number=args.issue,
        dry_run=args.dry_run,
    )
    print(json.dumps(result, indent=2))
    return 0


if __name__ == "__main__":
    raise SystemExit(main())
