"""Subject adapters: the same robustness operations for Oak and Git.

Remotes: Oak uses a loopback ``oak serve`` on a fixed per-fixture port (the

client stores the remote URL, so every trial restarts serve on the same port

from a copied data dir). Git uses a local bare repository addressed by the

relative path ``../srv/remote.git`` (file transport), or the stdlib smart-HTTP

server behind the same fault proxy for network-phase cases.

"""

from __future__ import annotations

import json

import subprocess

import time

import urllib.error

import urllib.request

from dataclasses import dataclass, field

from pathlib import Path

from typing import Optional

from .procs import Result, Spawned, run, subject_env, wait_until

from .readers import GitOracle, OakGraph, oak_commit_tree, worktree_manifest

OPS = ("commit", "push", "pull_ff", "pull_diverged", "clone", "switch", "merge_local", "merge_hosted", "mount_finish")

TOKEN = "p4-loopback-token"

REPO_ID = "p4/robust"

OTHER_BRANCH = "p4side"

def payload(root: Path, tag: str, count: int = 200, start: int = 0, prefix: str = "f") -> None:

    for i in range(start, start + count):

        path = root / f"d{i % 10}" / f"{prefix}{i}.txt"

        path.parent.mkdir(parents=True, exist_ok=True)

        path.write_text((f"{tag} line for {prefix}{i}\n" * 40))

@dataclass

class Server:

    """Loopback oak serve bound to a fixed port (restartable on copied data)."""

    oak_bin: str

    data_dir: Path

    port: int

    log: Path

    spawned: Optional[Spawned] = None

    @property

    def url(self) -> str:

        return f"http://127.0.0.1:{self.port}"

    def start(self, port: Optional[int] = None) -> "Server":

        bind = port or self.port

        self.log.parent.mkdir(parents=True, exist_ok=True)

        handle = self.log.open("ab")

        proc = subprocess.Popen([self.oak_bin, "serve", "--dir", str(self.data_dir), "--port", str(bind), "--token", TOKEN],

                                stdout=handle, stderr=subprocess.STDOUT, stdin=subprocess.DEVNULL,

                                env=subject_env(self.log.parent / "serve-home"), start_new_session=True)

        self.spawned = Spawned(proc, self.log)

        def ready() -> bool:

            if proc.poll() is not None:

                raise RuntimeError(f"oak serve exited early (rc={proc.returncode}); see {self.log}")

            try:

                with urllib.request.urlopen(f"http://127.0.0.1:{bind}/api/capabilities", timeout=1) as r:

                    return r.status == 200

            except urllib.error.HTTPError:

                return True  # any HTTP answer means the listener is up

            except OSError:

                return False

        if not wait_until(ready, 15):

            self.stop()

            raise RuntimeError(f"oak serve not ready on {bind}")

        return self

    def stop(self) -> None:

        if self.spawned:

            self.spawned.stop()

            self.spawned = None

    def db(self) -> Path:

        owner, name = REPO_ID.split("/")

        return self.data_dir / owner / f"{name}.oakdb"

class Adapter:

    kind = "?"

    subject = "?"

    def __init__(self, binary: str, home: Path):

        self.bin = binary

        self.home = home

    def env(self, extra: Optional[dict] = None) -> dict:

        raise NotImplementedError

class OakAdapter(Adapter):

    kind = "oak"

    subject = "oak_installed"

    def env(self, extra: Optional[dict] = None) -> dict:

        env = subject_env(self.home, {"OAK_API_KEY": TOKEN, "OAK_AUTHOR": "p4"})

        if extra:

            env.update(extra)

        return env

    # commands -----------------------------------------------------------------

    def argv(self, op: str, **kw) -> list[str]:

        b = kw.get("bin") or self.bin

        if op == "commit":

            return [b, "commit"]

        if op == "push":

            return [b, "push"]

        if op in ("pull_ff", "pull_diverged", "pull"):

            return [b, "pull"]

        if op == "clone":

            return [b, "clone", REPO_ID, str(kw["dest"]), "-r", kw["url"], "--branch", kw["branch"]]

        if op == "switch":

            return [b, "switch", kw["target"]]

        if op == "merge_hosted":

            return [b, "merge", "--json"]

        raise KeyError(op)

    def pre_argv(self, op: str) -> list[list[str]]:

        return []

    def run(self, argv: list[str], cwd: Path, extra_env: Optional[dict] = None, timeout: float = 90) -> Result:

        return run(argv, cwd, self.env(extra_env), timeout=timeout)

    # fixture ------------------------------------------------------------------

    def status_json(self, repo: Path) -> tuple[Optional[dict], Result]:

        res = self.run([self.bin, "status", "--json"], repo, timeout=60)

        try:

            return json.loads(res.stdout), res

        except (ValueError, TypeError):

            return None, res

    def must(self, argv: list[str], cwd: Path) -> Result:

        res = self.run(argv, cwd)

        if res.returncode != 0:

            raise RuntimeError(f"fixture command failed: {argv} rc={res.returncode}: {res.stderr[-400:]}")

        return res

    def db(self, repo: Path) -> Path:

        return repo / ".oak" / "oak.db"

    # independent observations ------------------------------------------------

    def local_graph(self, repo: Path) -> OakGraph:

        return OakGraph(self.db(repo))

    def head(self, repo: Path) -> Optional[str]:

        with self.local_graph(repo) as g:

            return g.metadata("head")

    def current_branch(self, repo: Path) -> Optional[str]:

        with self.local_graph(repo) as g:

            return g.metadata("current_branch")

    def tree(self, repo: Path, commit: Optional[str] = None) -> tuple[Optional[dict], str]:

        return oak_commit_tree(self.db(repo), commit)

    def status_state(self, repo: Path) -> tuple[Optional[bool], bool, Result]:

        """(lists uncommitted changes | None if status refused, operation in progress, result)."""

        data, res = self.status_json(repo)

        if res.returncode != 0 or data is None:

            return None, False, res

        changes = data.get("working_changes", {}).get("changes", data.get("changes", []))

        in_progress = bool(data.get("merge_in_progress") or data.get("sync_in_progress")

                           or (data.get("progress_state") or {}).get("in_progress"))

        return bool(changes), in_progress, res

    def status_dirty(self, repo: Path) -> tuple[Optional[bool], Result]:

        changes, in_progress, res = self.status_state(repo)

        return (None if changes is None else changes or in_progress), res

class GitAdapter(Adapter):

    kind = "git"

    subject = "git"

    def env(self, extra: Optional[dict] = None) -> dict:

        env = subject_env(self.home)

        if extra:

            env.update(extra)

        return env

    def argv(self, op: str, **kw) -> list[str]:

        b = kw.get("bin") or self.bin

        if op == "commit":

            return [b, "commit", "-q", "--no-verify", "-m", "p4 commit"]

        if op == "push":

            return [b, "push", "-q", "origin", "HEAD"]

        if op == "pull_ff":

            return [b, "pull", "-q", "--ff-only"]

        if op in ("pull_diverged", "pull"):

            return [b, "pull", "-q", "--no-rebase", "--no-edit"]

        if op == "clone":

            return [b, "clone", "-q", "--branch", kw["branch"], kw["url"], str(kw["dest"])]

        if op == "switch":

            return [b, "switch", "-q", kw["target"]]

        if op == "merge_local":

            return [b, "merge", "-q", "--no-edit", OTHER_BRANCH]

        raise KeyError(op)

    def pre_argv(self, op: str) -> list[list[str]]:

        # Git's commit is add+commit; staging is untimed setup (not a crash target).

        return [[self.bin, "add", "-A"]] if op == "commit" else []

    def run(self, argv: list[str], cwd: Path, extra_env: Optional[dict] = None, timeout: float = 90) -> Result:

        return run(argv, cwd, self.env(extra_env), timeout=timeout)

    def must(self, argv: list[str], cwd: Path) -> Result:

        res = self.run(argv, cwd)

        if res.returncode != 0:

            raise RuntimeError(f"fixture command failed: {argv} rc={res.returncode}: {res.stderr[-400:]}")

        return res

    @property

    def oracle(self) -> GitOracle:

        return GitOracle(self.bin, self.env())

    def head(self, repo: Path) -> Optional[str]:

        return self.oracle.head(repo)

    def current_branch(self, repo: Path) -> Optional[str]:

        res = self.run([self.bin, "symbolic-ref", "-q", "--short", "HEAD"], repo)

        return res.stdout.strip() or None

    def tree(self, repo: Path, commit: Optional[str] = None) -> tuple[Optional[dict], str]:

        commit = commit or self.head(repo)

        if not commit:

            return None, "corrupt:no HEAD"

        return self.oracle.commit_tree(repo, commit)

    def status_state(self, repo: Path) -> tuple[Optional[bool], bool, Result]:

        res = self.run([self.bin, "status", "--porcelain"], repo, timeout=60)

        if res.returncode != 0:

            return None, False, res

        # A crashed git can leave a stale lock that makes `git commit` refuse ("remove the file

        # manually") until the user intervenes, so the torn state cannot be committed as-is.

        in_progress = (any((repo / ".git" / name).exists()

                           for name in ("MERGE_HEAD", "rebase-merge", "rebase-apply", "CHERRY_PICK_HEAD"))

                       or bool(self.blocking_locks(repo)))

        return bool(res.stdout.strip()), in_progress, res

    @staticmethod

    def blocking_locks(repo: Path) -> list[str]:

        """Stale git lock files that make the next `git commit` refuse ("remove the file manually").

        index.lock (index write), HEAD.lock (HEAD reflog/update; left by a switch killed between

        its renames) and the lock of the ref HEAD points to (refs/heads/<current>.lock; left by a

        killed commit/merge/pull). Other *.lock files (config.lock after a killed clone, locks of

        other refs, packed-refs.lock, ORIG_HEAD.lock) do not stop a commit - verified against real

        git in tests - so a torn worktree next to them can still be committed as user work."""

        gitdir = repo / ".git"

        candidates = ["index.lock", "HEAD.lock"]

        try:

            head = (gitdir / "HEAD").read_text().strip()

        except OSError:

            head = ""

        if head.startswith("ref: refs/"):

            candidates.append(head[len("ref: "):] + ".lock")

        return [name for name in candidates if (gitdir / name).is_file()]

    def status_dirty(self, repo: Path) -> tuple[Optional[bool], Result]:

        changes, in_progress, res = self.status_state(repo)

        return (None if changes is None else changes or in_progress), res

def manifest(repo: Path) -> Optional[dict]:

    return worktree_manifest(repo)