"""Independent readers: judge committed state without trusting the subject.

Oak: the local ``.oak/oak.db`` and the loopback server's ``<owner>/<repo>.oakdb``

are SQLite files. We read them with Python's stdlib sqlite3 on a private copy

(WAL checkpointed into the copy by SQLite itself, the standard recovery path;

the original files are never opened for write). Committed trees are decoded

and hash-verified by ``scripts/oakbench/oak_sqlite_reader.py`` (P1's

independent reader, run under a pinned reader Python with blake3+zstandard).

Git: the real Git plumbing is the oracle (for-each-ref, merge-base, fsck,

archive), per the program's "git fine as fixture/oracle" rule.

"""

from __future__ import annotations

import io

import os

import shutil

import sqlite3

import subprocess

import tarfile

import tempfile

from pathlib import Path

from typing import Iterable, Optional

from . import REPO_ROOT

from oakbench.conformance import tree_manifest

OAK_READER = REPO_ROOT / "scripts" / "oakbench" / "oak_sqlite_reader.py"

READER_PY_ENV = "OAK_BENCH_OAK_READER_PYTHON"

def worktree_manifest(root: Path) -> Optional[dict]:

    try:

        return tree_manifest(root)

    except (OSError, AssertionError):

        return None

def snapshot_sqlite(db_file: Path, dest_dir: Path) -> Path:

    """Copy a (possibly WAL-mode, possibly crashed) SQLite DB and fold its WAL in."""

    dest_dir.mkdir(parents=True, exist_ok=True)

    target = dest_dir / db_file.name

    shutil.copyfile(db_file, target)

    for suffix in ("-wal", "-shm", "-journal"):

        side = Path(str(db_file) + suffix)

        if side.exists():

            shutil.copyfile(side, Path(str(target) + suffix))

    conn = sqlite3.connect(str(target), timeout=5)

    try:

        conn.execute("PRAGMA wal_checkpoint(TRUNCATE)")

        conn.execute("PRAGMA journal_mode=DELETE")

    finally:

        conn.close()

    return target

class OakGraph:

    """Stdlib view of an Oak SQLite store's refs and commit graph."""

    def __init__(self, db_file: Path):

        self._tmp = tempfile.TemporaryDirectory(prefix="p4-oakgraph-")

        self.path = snapshot_sqlite(Path(db_file), Path(self._tmp.name))

        self.conn = sqlite3.connect(self.path.as_uri() + "?mode=ro", uri=True, timeout=5)

    def close(self) -> None:

        self.conn.close()

        self._tmp.cleanup()

    def __enter__(self) -> "OakGraph":

        return self

    def __exit__(self, *exc) -> None:

        self.close()

    def integrity(self) -> str:

        rows = self.conn.execute("PRAGMA integrity_check").fetchall()

        return "ok" if rows == [("ok",)] else "; ".join(str(r[0]) for r in rows[:5])

    def _table_exists(self, name: str) -> bool:

        return self.conn.execute("SELECT 1 FROM sqlite_master WHERE type='table' AND name=?", (name,)).fetchone() is not None

    def metadata(self, key: str) -> Optional[str]:

        if not self._table_exists("metadata"):

            return None

        row = self.conn.execute("SELECT value FROM metadata WHERE key=?", (key,)).fetchone()

        return row[0] if row else None

    def branch_heads(self) -> dict[str, str]:

        if not self._table_exists("branch_heads"):

            return {}

        return {name: head for name, head in self.conn.execute("SELECT branch_name, head_hash FROM branch_heads")

                if head}

    def ref_heads(self) -> set[str]:

        """Every commit a ref-like row names: branch heads, HEAD, tags."""

        heads = set(self.branch_heads().values())

        head = self.metadata("head")

        if head:

            heads.add(head)

        if self._table_exists("tags"):

            cols = {r[1] for r in self.conn.execute("PRAGMA table_info(tags)")}

            for col in ("commit_hash", "target_hash", "hash"):

                if col in cols:

                    heads.update(r[0] for r in self.conn.execute(f"SELECT {col} FROM tags") if r[0])

                    break

        return heads

    def commit_row(self, commit: str) -> Optional[tuple]:

        return self.conn.execute("SELECT parent_hash, merge_parent_hash FROM commits WHERE hash=?", (commit,)).fetchone()

    def has_commit(self, commit: str) -> bool:

        return self.commit_row(commit) is not None

    def ancestors(self, heads: Iterable[str]) -> set[str]:

        seen: set[str] = set()

        stack = [h for h in heads if h]

        while stack:

            current = stack.pop()

            if current in seen:

                continue

            seen.add(current)

            row = self.commit_row(current)

            if row:

                stack.extend(p for p in row if p)

        return seen

    def reachable(self) -> set[str]:

        return self.ancestors(self.ref_heads())

    def is_ancestor(self, ancestor: str, descendant: str) -> bool:

        return ancestor in self.ancestors([descendant])

    def missing_parents(self, heads: Iterable[str]) -> list[str]:

        """Commits reachable by hash links whose rows are absent (a torn graph).

        A shallow store legitimately lacks history below its boundary; callers

        pass only heads whose full history they created in the fixture."""

        missing = []

        for commit in self.ancestors(heads):

            if not self.has_commit(commit):

                missing.append(commit)

        return missing

def reader_python() -> Optional[str]:

    configured = os.environ.get(READER_PY_ENV)

    if configured:

        return configured if Path(configured).exists() else None

    return None

def oak_commit_tree(db_file: Path, commit: Optional[str] = None) -> tuple[Optional[dict], str]:

    """Hash-verified manifest of an Oak commit's tree via the independent reader.

    Returns (manifest, status) with status one of: ok | corrupt:<reason> |

    unavailable:<reason>. ``commit`` defaults to the store's HEAD; otherwise the

    private copy's metadata is pointed at it (the copy is ours; the subject's

    store is never modified).

    """

    python = reader_python()

    if python is None:

        return None, f"unavailable:set {READER_PY_ENV} to a Python with blake3+zstandard"

    with tempfile.TemporaryDirectory(prefix="p4-oaktree-") as temp:

        repo = Path(temp) / "repo"

        snap = snapshot_sqlite(Path(db_file), repo / ".oak")

        if snap.name != "oak.db":

            snap.rename(repo / ".oak" / "oak.db")

        db = repo / ".oak" / "oak.db"

        if commit is not None:

            conn = sqlite3.connect(str(db))

            try:

                row = conn.execute("SELECT branch_name FROM commits WHERE hash=?", (commit,)).fetchone()

                if row is None:

                    return None, "corrupt:commit row missing"

                for key, value in (("head", commit), ("current_branch", row[0] or "p4-reader")):

                    conn.execute("DELETE FROM metadata WHERE key=?", (key,))

                    conn.execute("INSERT INTO metadata(key, value) VALUES (?, ?)", (key, value))

                conn.commit()

            finally:

                conn.close()

        out = Path(temp) / "out"

        proc = subprocess.run([python, str(OAK_READER), str(repo), str(out)], capture_output=True, text=True,

                              timeout=120)

        if proc.returncode == 0:

            return tree_manifest(out), "ok"

        tail = (proc.stdout or proc.stderr).strip()[-300:]

        if proc.returncode == 1:

            return None, "corrupt:" + tail

        return None, "unavailable:" + tail

class GitOracle:

    """Real-Git plumbing used as the Git subject's committed-state oracle."""

    def __init__(self, git_bin: str, env: dict):

        self.git = git_bin

        self.env = env

    def _run(self, repo: Path, *args: str, check: bool = False) -> subprocess.CompletedProcess:

        return subprocess.run([self.git, *args], cwd=str(repo), env=self.env, capture_output=True, timeout=120,

                              check=check)

    def refs(self, repo: Path) -> dict[str, str]:

        proc = self._run(repo, "for-each-ref", "--format=%(objectname) %(refname)")

        result = {}

        for line in proc.stdout.decode().splitlines():

            obj, _, name = line.partition(" ")

            result[name] = obj

        head = self._run(repo, "rev-parse", "--verify", "-q", "HEAD")

        if head.returncode == 0:

            result["HEAD"] = head.stdout.decode().strip()

        return result

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

        return self.refs(repo).get("HEAD")

    def reachable_from_refs(self, repo: Path, commit: str) -> bool:

        proc = self._run(repo, "for-each-ref", "--contains", commit, "--format=%(refname)")

        if proc.returncode == 0 and proc.stdout.strip():

            return True

        head = self.head(repo)

        return bool(head) and self.is_ancestor(repo, commit, head)

    def is_ancestor(self, repo: Path, ancestor: str, descendant: str) -> bool:

        return self._run(repo, "merge-base", "--is-ancestor", ancestor, descendant).returncode == 0

    def fsck(self, repo: Path) -> str:

        proc = self._run(repo, "fsck", "--no-progress", "--no-dangling")

        if proc.returncode == 0:

            return "ok"

        return (proc.stderr.decode("utf-8", "replace") or proc.stdout.decode("utf-8", "replace")).strip()[-400:] or "fsck failed"

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

        proc = self._run(repo, "archive", "--format=tar", commit)

        if proc.returncode != 0:

            return None, "corrupt:" + proc.stderr.decode("utf-8", "replace")[-300:]

        with tempfile.TemporaryDirectory(prefix="p4-gittree-") as temp:

            with tarfile.open(fileobj=io.BytesIO(proc.stdout)) as tar:

                tar.extractall(temp)  # noqa: S202 - our own fixture content

            return tree_manifest(Path(temp)), "ok"