"""Environment-fault family: disk full, read-only directories, fsync load.

- disk_full   the trial (client + server data) lives on a small RAM disk

              (macOS: ``hdiutil attach -nomount ram://`` + ``newfs_hfs`` mounted

              inside the work dir; no sudo) filled to leave ``free_kib`` free.

              The operation must fail with an actionable error or succeed with

              the exact outcome; the store must open; pre-op commits survive;

              after the filler is removed the next operation must converge.

- readonly    the store directory (``.oak`` / ``.git``) or one worktree subdir is

              made read-only for the operation, then restored.

- fsync_load  (see runner) the same_branch concurrency scenario under a

              background fsync storm (4 writers x 256 KiB + F_FULLFSYNC).

"""

from __future__ import annotations

import os

import platform

import re

import shutil

import stat

import subprocess

import time

from pathlib import Path

from typing import Optional

from . import HARNESS_ERROR

from .crash import evaluate

from .engine import Calibration, Trial, actionable, classify, converge_check

FREE_KIB = (0, 256, 4096)

# An environment fault is actionable when the error names the OS cause the user must fix.

CAUSE_RE = re.compile(r"no space left|os error 28|ENOSPC|permission denied|os error 13|read-?only|disk I/O error|"

                      r"disk (is )?full|quota", re.I)

def env_actionable(res) -> bool:

    return actionable(res) or (res.returncode not in (None, 0) and res.returncode > 0

                               and "panicked at" not in res.text() and bool(CAUSE_RE.search(res.text())))

class RamDisk:

    def __init__(self, mountpoint: Path, mib: int = 96):

        self.mountpoint = mountpoint

        self.mib = mib

        self.device: Optional[str] = None

    def __enter__(self) -> "RamDisk":

        if platform.system() != "Darwin":

            raise NotImplementedError("disk-full fixture implemented with macOS hdiutil RAM disks (Linux: small tmpfs pending)")

        sectors = self.mib * 2048

        out = subprocess.run(["hdiutil", "attach", "-nomount", f"ram://{sectors}"], capture_output=True, text=True,

                             timeout=60)

        if out.returncode != 0:

            raise NotImplementedError("hdiutil ram disk unavailable: " + out.stderr[-200:])

        self.device = out.stdout.split()[0]

        subprocess.run(["newfs_hfs", "-v", "p4full", self.device], capture_output=True, check=True, timeout=60)

        self.mountpoint.mkdir(parents=True, exist_ok=True)

        subprocess.run(["diskutil", "mount", "-mountPoint", str(self.mountpoint), self.device], capture_output=True,

                       check=True, timeout=60)

        return self

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

        if self.device:

            subprocess.run(["diskutil", "unmount", "force", str(self.mountpoint)], capture_output=True, timeout=60)

            subprocess.run(["hdiutil", "detach", self.device, "-force"], capture_output=True, timeout=60)

            self.device = None

def fill(directory: Path, leave_kib: int) -> Path:

    filler = directory / "p4-filler.bin"

    chunk = b"\0" * (1 << 20)

    with filler.open("wb") as fh:

        try:

            while True:

                fh.write(chunk)

                fh.flush()

        except OSError:

            pass

    size = filler.stat().st_size

    with filler.open("r+b") as fh:

        fh.truncate(max(0, size - leave_kib * 1024))

    return filler

def _judge_env(trial: Trial, calib: Calibration, op_res, restore) -> tuple[dict, dict]:

    """Common judgement: op either converged or failed actionably; then restore and re-run."""

    inv: dict = {"op_no_hang": not op_res.timed_out}

    obs: dict = {"op": op_res.brief(300)}

    if op_res.returncode == 0:

        ok, detail = converge_check(trial, calib, trial.target_repo())

        inv["op_success_is_true"] = ok

        obs["op_converge"] = detail

    else:

        inv["op_failure_actionable"] = env_actionable(op_res)

    restore()

    more_inv, more_obs = evaluate(trial, calib, op_res, True)

    more_inv.pop("op_no_hang", None)

    inv.update(more_inv)

    obs.update({k: v for k, v in more_obs.items() if k != "op"})

    return inv, obs

def disk_full_trial(fixture, subject_bin: str, work: Path, name: str, calib: Calibration, leave_kib: int) -> dict:

    started = time.perf_counter()

    mount = work / f"{name}-mnt"

    op_res = None

    try:

        with RamDisk(mount):

            trial = Trial(fixture, subject_bin, mount, "t")

            try:

                trial.start_server()

                trial.pre()

                filler = fill(mount, leave_kib)

                op_res = trial.adapter.run(trial.op_argv(), trial.op_cwd())

                inv, obs = _judge_env(trial, calib, op_res, lambda: filler.unlink(missing_ok=True))

                obs["free_kib_before_op"] = leave_kib

                classification, error = classify(inv), None

            finally:

                trial.stop_server()

    except NotImplementedError as exc:

        return {"unsupported": str(exc)}

    except Exception as exc:  # noqa: BLE001

        inv, obs, classification, error = {}, {}, HARNESS_ERROR, f"{type(exc).__name__}: {exc}"

    finally:

        shutil.rmtree(mount, ignore_errors=True)

    return _row(f"free={leave_kib}KiB", inv, obs, classification, error, op_res, started)

def readonly_trial(fixture, subject_bin: str, work: Path, name: str, calib: Calibration, target: str) -> dict:

    """target: 'store' (.oak/.git) or 'subdir' (worktree d3/)."""

    started = time.perf_counter()

    trial = Trial(fixture, subject_bin, work, name)

    op_res = None

    changed: list[tuple[Path, int]] = []

    def lock_down() -> None:

        repo = trial.client

        base = repo / (".oak" if trial.kind == "oak" else ".git") if target == "store" else repo / "d3"

        for path in [base, *base.rglob("*")] if target == "store" else [base]:

            mode = path.stat().st_mode

            changed.append((path, mode))

            os.chmod(path, mode & ~(stat.S_IWUSR | stat.S_IWGRP | stat.S_IWOTH))

    def restore() -> None:

        for path, mode in reversed(changed):

            try:

                os.chmod(path, mode)

            except OSError:

                pass

        changed.clear()

    try:

        trial.start_server()

        trial.pre()

        lock_down()

        op_res = trial.adapter.run(trial.op_argv(), trial.op_cwd())

        inv, obs = _judge_env(trial, calib, op_res, restore)

        classification, error = classify(inv), None

    except Exception as exc:  # noqa: BLE001

        restore()

        inv, obs, classification, error = {}, {}, HARNESS_ERROR, f"{type(exc).__name__}: {exc}"

    finally:

        restore()

        trial.cleanup()

    return _row(f"readonly={target}", inv, obs, classification, error, op_res, started)

def _row(point, inv, obs, classification, error, op_res, started) -> dict:

    row = {"fault_point": point, "invariants": inv, "violations": sorted(k for k, v in inv.items() if v is False),

           "observations": obs, "classification": classification,

           "elapsed_ms": round(op_res.elapsed_ms if op_res else (time.perf_counter() - started) * 1000.0, 3),

           "command": op_res.argv if op_res else []}

    if error:

        row["harness_error"] = error

    return row

class FsyncLoad:

    """Background fsync storm (idea from the L8 fsyncload.py harness)."""

    SCRIPT = r"""

import os, sys, fcntl, time

d, stop = sys.argv[1], sys.argv[2]

buf = os.urandom(256 * 1024)

p = os.path.join(d, f"load-{os.getpid()}")

while not os.path.exists(stop):

    with open(p, "wb") as f:

        f.write(buf); f.flush()

        try: fcntl.fcntl(f.fileno(), 51)

        except OSError: os.fsync(f.fileno())

os.unlink(p)

"""

    def __init__(self, directory: Path, workers: int = 4):

        self.dir = directory

        self.workers = workers

        self.procs: list[subprocess.Popen] = []

    def __enter__(self) -> "FsyncLoad":

        self.dir.mkdir(parents=True, exist_ok=True)

        self.stop_file = self.dir / "stop"

        self.stop_file.unlink(missing_ok=True)

        script = self.dir / "fsyncload.py"

        script.write_text(self.SCRIPT)

        for _ in range(self.workers):

            self.procs.append(subprocess.Popen(["python3", str(script), str(self.dir), str(self.stop_file)]))

        return self

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

        self.stop_file.write_text("stop")

        for proc in self.procs:

            try:

                proc.wait(timeout=20)

            except subprocess.TimeoutExpired:

                proc.kill()