"""Untimed fixture preparation: generate, characterize, ingest, serve.

Everything here happens OUTSIDE the timed path and is recorded as

``ingest.*`` diagnostic rows (setup cost is itself scale evidence: e.g. how

long Oak needs to ingest a 100k-commit Git history).

Layout under ``<work>/<fixture_id>/``::

    fixture/src.git              generated bare repository (the fixture)

    fixture/GENERATOR_MANIFEST.json, fixture/SIZER.json

    git/work                     git working repository (all refs local)

    git/served/bench/fx.git      bare repository behind the smart-HTTP server

    git/clientA, git/clientB     sync clients (clone over HTTP)

    oak/work                     oak repository ingested with `oak clone <src.git>`

    oak/serve-data               `oak serve` data directory

    oak/clientA, oak/clientB     sync clients (clone from the loopback server)

Transport: both subjects sync over loopback HTTP (git: the stdlib smart-HTTP

server in oakbench.gitserver, CGI `git http-backend`; oak: `oak serve`).

Rows name the transport; neither side uses a local-file shortcut.

"""

from __future__ import annotations

import json

import os

import re

import shutil

import subprocess

import sys

import time

from dataclasses import dataclass, field

from pathlib import Path

from typing import Any, Dict, List, Optional, Tuple

from . import generator as gen

from . import sizer as sz

from .tiers import Tier

SERVE_REPO = "bench/fx"

TRANSPORT_GIT = "loopback_http_git_http_backend"

TRANSPORT_OAK = "loopback_http_oak_serve"

def _say(message: str) -> None:

    print(message, file=sys.stderr, flush=True)

class PrepError(RuntimeError):

    pass

def run(cmd: List[str], cwd: Optional[Path] = None, env: Optional[Dict[str, str]] = None,

        check: bool = True, input_bytes: Optional[bytes] = None, timeout: Optional[float] = None

        ) -> subprocess.CompletedProcess:

    proc = subprocess.run(cmd, cwd=str(cwd) if cwd else None, env=env, capture_output=True,

                          input=input_bytes, timeout=timeout)

    if check and proc.returncode != 0:

        raise PrepError(f"{' '.join(cmd)} (cwd={cwd}) exited {proc.returncode}: "

                        f"{proc.stderr.decode(errors='replace')[-2000:]}")

    return proc

def dir_bytes(root: Path) -> int:

    return gen.dir_bytes(root) if root.exists() else 0

@dataclass

class Expected:

    """Oracle facts read from the fixture with git (independent of Oak)."""

    main_file_count: int

    sample: Dict[str, str]  # path -> git blob id at main

    probe_path: str

    probe_bytes: bytes

    probe_lines: int

    probe_commit_count: int

    main_commits: int

    branch: str

    branch_paths: List[str]

    branch_bytes: Dict[str, bytes]

    main_bytes_of_branch_paths: Dict[str, bytes]

    conflict: Dict[str, Any]

    protected: List[str]

    all_paths_sample_size: int

@dataclass

class Prepared:

    tier: Tier

    root: Path

    fixture_dir: Path

    manifest: Dict[str, Any]

    sizer: Dict[str, Any]

    expected: Expected

    git_bin: str

    oak_bin: str

    env: Dict[str, str]

    phases: Dict[str, float] = field(default_factory=dict)

    kinds: Tuple[str, ...] = ()

    git_url: Optional[str] = None

    oak_url: Optional[str] = None

    oak_base_branch: Optional[str] = None

    oak_sync_branch: Optional[str] = None

    oak_branches_created: int = 0

    oak_seed: Optional[str] = None

    oak_seed_refusal: Optional[str] = None

    oak_remote_unavailable: Optional[str] = None

    oak_identity: Dict[str, Any] = field(default_factory=dict)

    git_server: Any = None

    oak_server: Optional[subprocess.Popen] = None

    notes: List[str] = field(default_factory=list)

    # Mutable harness bookkeeping shared by ops.

    state: Dict[str, Any] = field(default_factory=dict)

    def path(self, *parts: str) -> Path:

        return self.root.joinpath(*parts)

    def fixture_fields(self) -> Dict[str, Any]:

        spec = self.tier.spec

        return {

            "tier": self.tier.tier_id,

            "tier_kind": self.tier.kind,

            "fixture_id": self.tier.fixture_id,

            "fixture_generator": "suites/scale/scalesuite/generator.py",

            "fixture_version": self.manifest["generator_version"],

            "fixture_seed": spec.seed,

            "fixture_manifest_sha256": self.manifest["refs_sha256"],

            "fixture_spec_sha256": self.manifest["spec_sha256"],

            "fixture_head_commit": self.manifest["head_commit"],

            "fixture_stats": sz.compact(self.sizer),

        }

    def close(self) -> None:

        if self.git_server is not None:

            try:

                self.git_server.stop()

            except Exception:  # noqa: BLE001 - best-effort teardown

                pass

            self.git_server = None

        if self.oak_server is not None:

            self.oak_server.terminate()

            try:

                self.oak_server.wait(timeout=10)

            except subprocess.TimeoutExpired:

                self.oak_server.kill()

                self.oak_server.wait()

            self.oak_server = None

def _git(prepared_git: str, repo: Path, *args: str, env: Dict[str, str], check: bool = True) -> str:

    return run([prepared_git, "-C", str(repo), *args], env=env, check=check).stdout.decode(errors="replace")

def compute_expected(tier: Tier, fixture: Path, manifest: Dict[str, Any], git_bin: str,

                     env: Dict[str, str], sample_size: int = 400) -> Expected:

    repo = fixture / "src.git"

    listing = run([git_bin, "-C", str(repo), "ls-tree", "-r", "-z", "main"], env=env).stdout

    entries: List[Tuple[str, str]] = []

    for item in listing.split(b"\0"):

        if not item:

            continue

        meta, path = item.split(b"\t", 1)

        entries.append((path.decode(), meta.split()[2].decode()))

    import random

    rng = random.Random(f"{tier.spec.seed}:oracle-sample")

    chosen = rng.sample(entries, min(sample_size, len(entries)))

    probe = manifest["probe_path"]

    probe_bytes = run([git_bin, "-C", str(repo), "cat-file", "blob", f"main:{probe}"], env=env).stdout

    branch = manifest["branch_sample"].get("name") or ""

    branch_paths = list(manifest["branch_sample"].get("changed_paths") or [])

    branch_bytes = {p: run([git_bin, "-C", str(repo), "cat-file", "blob", f"{branch}:{p}"], env=env).stdout

                    for p in branch_paths} if branch else {}

    main_bytes = {p: run([git_bin, "-C", str(repo), "cat-file", "blob", f"main:{p}"], env=env).stdout

                  for p in branch_paths}

    protected = sorted({probe, *branch_paths, *(manifest.get("conflict", {}).get("paths_sample") or []),

                        *manifest.get("binary_paths", []), ".gitignore"})

    return Expected(

        main_file_count=len(entries), sample=dict(chosen), probe_path=probe, probe_bytes=probe_bytes,

        probe_lines=probe_bytes.count(b"\n"), probe_commit_count=manifest["probe_path_commit_count"],

        main_commits=manifest["main_commits"], branch=branch, branch_paths=branch_paths,

        branch_bytes=branch_bytes, main_bytes_of_branch_paths=main_bytes,

        conflict=manifest.get("conflict") or {}, protected=protected, all_paths_sample_size=len(chosen))

def write_ignored_extras(work: Path, tier: Tier) -> None:

    if not tier.ignored_extras:

        return

    base = work / "build" / "cache"

    for index in range(tier.ignored_extras):

        target = base / f"g{index // 500}" / f"x{index}.o"

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

        target.write_bytes(b"ignored build product %d\n" % index)

def _one_line_error(text: str) -> str:

    try:

        return str(json.loads(text.strip().splitlines()[0])["error"]["message"])[:400]

    except (ValueError, KeyError, IndexError, TypeError):

        return " ".join(text.split())[:400]

def _snapshot_seed_repo(p: "Prepared", src: Path, dest: Path) -> Path:

    """A bare git repo whose main is ONE parentless commit with the fixture's main tree."""

    git, env = p.git_bin, dict(p.env)

    env.update(GIT_AUTHOR_DATE="1700000000 +0000", GIT_COMMITTER_DATE="1700000000 +0000")

    if dest.exists():

        shutil.rmtree(dest)

    run([git, "init", "-q", "--bare", "-b", "main", str(dest)], env=env)

    run([git, "-C", str(dest), "fetch", "-q", str(src), "main:refs/heads/fixture-main"], env=env)

    tree = _git(git, dest, "rev-parse", "refs/heads/fixture-main^{tree}", env=env).strip()

    commit = run([git, "-C", str(dest), "commit-tree", tree, "-m", "snapshot of fixture main"], env=env

                 ).stdout.decode().strip()

    _git(git, dest, "update-ref", "refs/heads/main", commit, env=env)

    _git(git, dest, "update-ref", "-d", "refs/heads/fixture-main", env=env)

    return dest

def _oak_serve(oak_bin: str, data_dir: Path, env: Dict[str, str], log_path: Path) -> Tuple[subprocess.Popen, str]:

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

    log = log_path.open("wb")

    proc = subprocess.Popen([oak_bin, "serve", "--port", "0", "--dir", str(data_dir)], stdout=log,

                            stderr=subprocess.STDOUT, env=env)

    deadline = time.time() + 30

    while time.time() < deadline:

        text = log_path.read_text(errors="replace")

        match = re.search(r"listening on (http://127\.0\.0\.1:\d+)", text)

        if match:

            return proc, match.group(1)

        if proc.poll() is not None:

            break

        time.sleep(0.1)

    proc.terminate()

    raise PrepError(f"oak serve did not report a listening port: {log_path.read_text(errors='replace')[-500:]}")

def oak_status(oak_bin: str, cwd: Path, env: Dict[str, str]) -> Dict[str, Any]:

    out = run([oak_bin, "status", "--json"], cwd=cwd, env=env).stdout

    return json.loads(out)

def prepare(tier: Tier, work_root: Path, *, git_bin: str, oak_bin: str, env: Dict[str, str],

            kinds: Tuple[str, ...] = ("git", "oak"), oak_branch_cap: Optional[int] = None,

            log=None) -> Prepared:

    """Generate (or reuse) the fixture and build every workspace the ops need."""

    root = work_root / tier.fixture_id

    fixture = root / "fixture"

    phases: Dict[str, float] = {}

    def phase(name: str, started: float) -> None:

        phases[name] = round(time.perf_counter() - started, 3)

        (log or _say)(f"[scale] {tier.tier_id}: {name} {phases[name]:.1f}s")

    started = time.perf_counter()

    manifest = gen.generate(tier.spec, fixture, git_bin)

    phases["fixture.generate"] = manifest["generation_elapsed_s"] if not manifest.get("reused") else 0.0

    phases["fixture.generate_reused"] = 1.0 if manifest.get("reused") else 0.0

    sizer_path = fixture / "SIZER.json"

    if sizer_path.exists() and json.loads(sizer_path.read_text()).get("refs_sha256") == manifest["refs_sha256"]:

        sizer = json.loads(sizer_path.read_text())["stats"]

    else:

        t0 = time.perf_counter()

        sizer = sz.measure(fixture / "src.git", git_bin)

        sizer_path.write_text(json.dumps({"refs_sha256": manifest["refs_sha256"], "stats": sizer}, indent=2) + "\n")

        phase("fixture.sizer", t0)

    expected = compute_expected(tier, fixture, manifest, git_bin, env)

    prepared = Prepared(tier=tier, root=root, fixture_dir=fixture, manifest=manifest, sizer=sizer,

                        expected=expected, git_bin=git_bin, oak_bin=oak_bin, env=env, phases=phases,

                        kinds=kinds)

    src = fixture / "src.git"

    try:

        if "git" in kinds:

            _prepare_git(prepared, src, phase)

        if "oak" in kinds:

            _prepare_oak(prepared, src, phase, oak_branch_cap)

    except Exception:

        prepared.close()

        raise

    phases["prepare.total"] = round(time.perf_counter() - started, 3)

    return prepared

def _prepare_git(p: Prepared, src: Path, phase) -> None:

    git, env = p.git_bin, p.env

    gdir = p.path("git")

    if gdir.exists():

        shutil.rmtree(gdir)

    gdir.mkdir(parents=True)

    work = gdir / "work"

    t0 = time.perf_counter()

    # All fixture refs become local branches (like the oak work repo, which

    # creates them locally), objects hardlinked from the read-only fixture.

    run([git, "clone", "-q", "--bare", str(src), str(work / ".git")], env=env)

    for key, value in (("core.bare", "false"), ("gc.auto", "0"), ("maintenance.auto", "false"),

                       ("user.name", "Oak Bench"), ("user.email", "[email protected]")):

        _git(git, work, "config", key, value, env=env)

    _git(git, work, "reset", "-q", "--hard", "main", env=env)

    write_ignored_extras(work, p.tier)

    phase("ingest.git_worktree", t0)

    t0 = time.perf_counter()

    served = gdir / "served" / (SERVE_REPO + ".git")

    served.parent.mkdir(parents=True)

    run([git, "clone", "-q", "--bare", str(src), str(served)], env=env)

    _git(git, served, "config", "http.receivepack", "true", env=env)

    _git(git, served, "config", "gc.auto", "0", env=env)

    import sys

    scripts = str(Path(__file__).resolve().parents[3] / "scripts")

    if scripts not in sys.path:

        sys.path.insert(0, scripts)

    from oakbench.gitserver import GitHTTPServer

    p.git_server = GitHTTPServer(gdir / "served", git_bin=git).start()

    p.git_url = f"{p.git_server.url}/{SERVE_REPO}.git"

    phase("ingest.git_serve_seed", t0)

    if not p.tier.sync:

        return

    t0 = time.perf_counter()

    for name in ("clientA", "clientB"):

        run([git, "clone", "-q", p.git_url, str(gdir / name)], env=env)

        for key, value in (("gc.auto", "0"), ("maintenance.auto", "false"), ("pull.rebase", "false")):

            _git(git, gdir / name, "config", key, value, env=env)

    phase("ingest.git_sync_clients", t0)

def _commit_file(p: Prepared, kind: str, repo: Path, rel: str, content: bytes) -> None:

    target = repo / rel

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

    target.write_bytes(content)

    if kind == "git":

        _git(p.git_bin, repo, "add", "-A", env=p.env)

        _git(p.git_bin, repo, "commit", "-q", "--no-verify", "-m", f"harness {rel}", env=p.env)

    else:

        run([p.oak_bin, "commit", "--json"], cwd=repo, env=p.env)

def _prepare_oak(p: Prepared, src: Path, phase, branch_cap: Optional[int]) -> None:

    oak, env = p.oak_bin, p.env

    odir = p.path("oak")

    if odir.exists():

        shutil.rmtree(odir)

    odir.mkdir(parents=True)

    work = odir / "work"

    t0 = time.perf_counter()

    # `oak clone <path>.git` is Oak's own git ingest: it runs `git clone` and

    # replays the HEAD history into Oak storage (other refs are not imported).

    ingest_env = dict(env)

    ingest_env["PATH"] = str(Path(p.git_bin).parent) + os.pathsep + env.get("PATH", "")

    run([oak, "clone", str(src), str(work)], env=ingest_env, input_bytes=b"")

    shutil.rmtree(work / ".git", ignore_errors=True)

    phase("ingest.oak_import", t0)

    p.oak_base_branch = oak_status(oak, work, env)["branch"]

    t0 = time.perf_counter()

    spec = p.tier.spec

    factory = gen.ContentFactory(spec)

    cache = gen.layout(spec)

    total = spec.branches if branch_cap is None else min(spec.branches, branch_cap)

    created = 0

    for number in range(total):

        name = gen.branch_name(number)

        run([oak, "switch", "-c", name], cwd=work, env=env)

        for index, path in gen.branch_changes(spec, number, cache):

            (work / path).write_bytes(factory.text(index, name))

        run([oak, "commit", "--json"], cwd=work, env=env)

        created += 1

    if spec.conflict_files:

        for side in ("bench-ours", "bench-theirs"):

            run([oak, "switch", "-c", side], cwd=work, env=env)

            for i in range(spec.conflict_files):

                (work / gen.file_path(spec, i, cache)).write_bytes(factory.text(i, side))

            run([oak, "commit", "--json"], cwd=work, env=env)

    run([oak, "switch", p.oak_base_branch], cwd=work, env=env)

    p.oak_branches_created = created

    if total < spec.branches:

        p.notes.append(f"oak branches capped at {total} of {spec.branches} (--oak-branch-cap)")

    write_ignored_extras(work, p.tier)

    phase("ingest.oak_branches", t0)

    if not p.tier.sync:

        p.oak_remote_unavailable = "tier_excludes_sync_clients_disk_budget"

        return

    t0 = time.perf_counter()

    p.oak_server, p.oak_url = _oak_serve(oak, odir / "serve-data", env, odir / "serve.log")

    # Seed from a SEPARATE import: pushing from the work repo would give it a

    # remote, after which Oak refreshes main from the server during local ops

    # (network inside timed local ops; and with a snapshot seed, an unrelated

    # main replaces the imported one). The git work repo never contacts its

    # origin during local ops either.

    ingest_env = dict(env)

    ingest_env["PATH"] = str(Path(p.git_bin).parent) + os.pathsep + env.get("PATH", "")

    full_seed = odir / "seed-full"

    run([oak, "clone", str(src), str(full_seed)], env=ingest_env, input_bytes=b"")

    shutil.rmtree(full_seed / ".git", ignore_errors=True)

    seeded = run([oak, "push", "-r", p.oak_url, "--repo", SERVE_REPO, "--json"], cwd=full_seed, env=env,

                 check=False)

    shutil.rmtree(full_seed, ignore_errors=True)

    if seeded.returncode == 0:

        p.oak_seed = "full_history"

        phase("ingest.oak_serve_seed", t0)

    else:

        # Oak's staged push has fixed per-operation caps (oak_core::protocol

        # STAGED_*: 1M resolved manifest entries summed over every pushed

        # commit, 50k blobs, 100k commits, ...), so a history that exceeds them

        # cannot be published at all. Record the refusal, then seed the

        # server with a one-commit snapshot of main (like `git clone

        # --depth 1` content) so the sync ops still run, labelled as such.

        message = (seeded.stdout + seeded.stderr).decode(errors="replace")

        p.oak_seed_refusal = _one_line_error(message)

        p.notes.append(f"oak serve could not accept the full history: {p.oak_seed_refusal}")

        phase("ingest.oak_serve_seed_full_refused", t0)

        t0 = time.perf_counter()

        snapshot = _snapshot_seed_repo(p, src, odir / "seed-src.git")

        seed_work = odir / "seed-work"

        run([oak, "clone", str(snapshot), str(seed_work)], env=ingest_env, input_bytes=b"")

        shutil.rmtree(seed_work / ".git", ignore_errors=True)

        second = run([oak, "push", "-r", p.oak_url, "--repo", SERVE_REPO, "--json"], cwd=seed_work, env=env,

                     check=False)

        shutil.rmtree(seed_work, ignore_errors=True)

        shutil.rmtree(snapshot, ignore_errors=True)

        if second.returncode != 0:

            p.oak_remote_unavailable = "oak_push_admission_cap: " + _one_line_error(

                (second.stdout + second.stderr).decode(errors="replace"))

            p.notes.append(f"oak snapshot seed also refused; oak sync ops skipped: {p.oak_remote_unavailable}")

            phase("ingest.oak_serve_seed_snapshot_refused", t0)

            return

        p.oak_seed = "snapshot_1_commit"

        phase("ingest.oak_serve_seed_snapshot", t0)

    t0 = time.perf_counter()

    client_a = odir / "clientA"

    run([oak, "clone", "-r", p.oak_url, SERVE_REPO, str(client_a)], env=env)

    _commit_file(p, "oak", client_a, "sync/seed.txt", b"seed\n")

    run([oak, "push", "--json"], cwd=client_a, env=env)

    p.oak_sync_branch = oak_status(oak, client_a, env)["branch"]

    run([oak, "clone", "-r", p.oak_url, SERVE_REPO, str(odir / "clientB"), "--branch", p.oak_sync_branch], env=env)

    phase("ingest.oak_sync_clients", t0)