"""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 errno

import os

import platform

import re

import shutil

import stat

import subprocess

import time

from pathlib import Path

from typing import Optional

from . import FAULT_NOT_FIRED, 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|out of disk ?space|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

FILL_CHUNKS = (1 << 20, 64 << 10, 4 << 10, 512, 1)

def free_bytes(directory: Path) -> int:

    """Bytes an unprivileged writer can still allocate on directory's filesystem."""

    st = os.statvfs(directory)

    return st.f_bavail * st.f_frsize

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

    """Fill the filesystem until writes fail at 1-byte granularity, then free ``leave_kib``.

    Writes shrink from 1 MiB to 64 KiB, 4 KiB, 512 B and 1 B, each until ENOSPC,

    with an fsync between rounds so delayed allocation cannot hide free space.

    Returns (filler, measured free bytes after the optional release).

    """

    filler = directory / "p4-filler.bin"

    fd = os.open(filler, os.O_WRONLY | os.O_CREAT | os.O_TRUNC, 0o644)

    try:

        for _round in range(3):

            for size in FILL_CHUNKS:

                buf = b"\0" * size

                while True:

                    try:

                        written = os.write(fd, buf)

                    except OSError as exc:

                        if exc.errno in (errno.ENOSPC, errno.EDQUOT):

                            break

                        raise

                    if written < size:

                        break

            try:

                os.fsync(fd)

            except OSError:

                pass

            if free_bytes(directory) == 0:

                break

        if leave_kib:

            size = os.fstat(fd).st_size

            os.ftruncate(fd, max(0, size - leave_kib * 1024))

            os.fsync(fd)

    finally:

        os.close(fd)

    return filler, free_bytes(directory)

def allocated_bytes(root: Path) -> int:

    total = 0

    for path in [root, *root.rglob("*")]:

        try:

            total += path.lstat().st_blocks * 512

        except OSError:

            pass

    return total

def op_need_bytes(fixture, subject_bin: str, work: Path) -> Optional[int]:

    """Net bytes the clean op allocates (client + server data) on an ordinary disk."""

    trial = Trial(fixture, subject_bin, work, "need")

    try:

        trial.start_server()

        trial.pre()

        before = allocated_bytes(trial.dir)

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

        trial.settle()

        if res.returncode != 0:

            return None  # calibration failed: the need is unmeasured, never zero

        return max(0, allocated_bytes(trial.dir) - before)

    finally:

        trial.cleanup()

ENOSPC_RE = re.compile(r"no space left|os error 28|ENOSPC|disk (is )?full|database or disk is full|SQLITE_FULL",

                       re.I)

# Evidence that a write under a read-only path was refused.

EACCES_RE = re.compile(r"permission denied|os error 13|os error 30|EACCES|EROFS|EPERM|read-?only (?:file ?system|"

                       r"database)|attempt to write a readonly database|SQLITE_READONLY|Operation not permitted", re.I)

def op_failed(op_res) -> bool:

    """The op errored, crashed or hung (returncode None = killed by the harness on timeout)."""

    return op_res is None or op_res.timed_out or op_res.returncode != 0

def disk_full_fired(free: int, need_bytes: int, op_res) -> tuple[bool, str]:

    """Did the disk-full fault take effect for this op? (fired, evidence).

    The disk was really filled before the op, so any op that fails, errors or

    hangs there is JUDGED (fired): an opaque error on a full disk (fb-548's

    "unable to open database file", a panic) must surface as a violation of

    op_failure_actionable, never as a skip. Only a SUCCESSFUL op that had room

    (measured free >= its calibrated need, no ENOSPC) is fault_not_fired."""

    if op_res is not None and ENOSPC_RE.search(op_res.text()):

        return True, "op reported ENOSPC"

    if free < need_bytes:

        return True, f"measured free {free} B < op need {need_bytes} B"

    if op_failed(op_res):

        rc = None if op_res is None else ("hang" if op_res.timed_out else op_res.returncode)

        return True, (f"op failed (rc={rc}) on the filled disk (free {free} B >= need {need_bytes} B, no ENOSPC "

                      "text): judged")

    return False, f"op succeeded with measured free {free} B >= op need {need_bytes} B"

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,

                    need_bytes: Optional[int]) -> dict:

    """Run the op with the trial's filesystem filled to ``leave_kib`` free (measured).

    See disk_full_fired: a failing / hanging op is always judged; only a

    success with room to spare is ``fault_not_fired``. need_bytes None (the

    clean calibration run failed) makes the row a harness error.

    """

    started = time.perf_counter()

    if need_bytes is None:

        return {"fault_point": f"free={leave_kib}KiB", "classification": HARNESS_ERROR, "invariants": {},

                "observations": {}, "fault_fired": None,

                "harness_error": "op need unmeasured: the clean calibration run of the op failed"}

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

    op_res = None

    fired = None

    try:

        with RamDisk(mount):

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

            try:

                trial.start_server()

                trial.pre()

                trial.adapter.env()  # create the subject HOME before the disk is full (harness setup)

                filler, free = fill(mount, leave_kib)

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

                enospc = bool(ENOSPC_RE.search(op_res.text()))

                fired, fired_evidence = disk_full_fired(free, need_bytes, op_res)

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

                obs.update(free_bytes_measured_before_op=free, op_need_bytes=need_bytes, enospc_reported=enospc,

                           fault_fired_evidence=fired_evidence,

                           fill_granularity_bytes=FILL_CHUNKS[-1], requested_free_kib=leave_kib)

                classification = classify(inv, fault_fired=fired, reader_errors=obs.get("reader_errors"))

                error = 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)

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

    row["fault_fired"] = fired

    if "free_bytes_measured_before_op" in obs:

        row["free_bytes_measured"] = obs["free_bytes_measured_before_op"]

    if classification == FAULT_NOT_FIRED:

        row["skip_reason"] = (f"disk not full for this op: it succeeded with "

                              f"{obs.get('free_bytes_measured_before_op')} B free >= {need_bytes} B needed")

    return row

def readonly_base(repo: Path, kind: str, target: str) -> Path:

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

def write_snapshot(base: Path, target: str) -> Optional[dict]:

    """What a write under ``base`` would change, as lock_down restricts it.

    store: every path below is read-only -> {relpath: (inode, size, mtime_ns)} of

    the whole tree. subdir: only the directory itself is read-only -> its entry

    set {name: inode} plus the directory's own mtime (entry create/rename/delete)."""

    if not base.exists():

        return None

    if target == "store":

        out = {}

        for path in [base, *base.rglob("*")]:

            try:

                st = path.lstat()

            except OSError:

                continue

            is_dir = stat.S_ISDIR(st.st_mode)

            out[str(path.relative_to(base))] = (st.st_ino, 0 if is_dir else st.st_size,

                                                0 if is_dir else st.st_mtime_ns)

        return out

    out = {"<dir-mtime>": base.stat().st_mtime_ns}

    for child in base.iterdir():

        out[child.name] = child.lstat().st_ino

    return out

def readonly_exercised(fixture, subject_bin: str, work: Path, name: str, target: str) -> tuple[Optional[bool], dict]:

    """Does a clean run of the op write under the path readonly_trial locks down?"""

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

    try:

        trial.start_server()

        trial.pre()

        base = readonly_base(trial.client, trial.kind, target)

        before = write_snapshot(base, target)

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

        trial.settle()

        after = write_snapshot(base, target)

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

            return None, {"clean_rc": res.returncode, "detail": "clean run failed or path missing"}

        changed = sorted(k for k in set(before) | set(after or {}) if before.get(k) != (after or {}).get(k))

        return bool(changed), {"clean_run_changed": len(changed), "sample": changed[:5]}

    finally:

        trial.stop_server()

        trial.cleanup()

def readonly_fired(writable_after_lockdown: bool, exercised: Optional[bool], op_res) -> tuple[Optional[bool], str]:

    """Did the read-only fault take effect? (fired, evidence).

    A failing / hanging op is always judged (fired): its error must be

    actionable whatever caused it. A successful op counts as fired only if the

    path really was unwritable (not the case as root) and the op attempted a

    write there: it reported a permission/read-only error, or a clean run of

    the same op writes under that path. Otherwise 'not exercised'."""

    if op_failed(op_res):

        return True, "op failed with the path read-only: judged"

    if writable_after_lockdown:

        return False, "not exercised: path still writable after chmod (running as root?)"

    if op_res is not None and EACCES_RE.search(op_res.text()):

        return True, "op reported a permission/read-only error"

    if exercised:

        return True, "a clean run of the op writes under the read-only path"

    if exercised is None:

        return None, "exercise probe failed; whether the op wrote under the path is unmeasured"

    return False, "not exercised: a clean run of the op writes nothing under the read-only path"

def readonly_trial(fixture, subject_bin: str, work: Path, name: str, calib: Calibration, target: str,

                   exercised: Optional[bool] = None, exercise_detail: Optional[dict] = None) -> dict:

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

    exercised: whether a clean run of the op writes under the locked path

    (readonly_exercised); decides fault_fired together with the op's output."""

    started = time.perf_counter()

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

    op_res = None

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

    fired, fired_evidence = None, "not run"

    def lock_down() -> None:

        repo = trial.client

        base = readonly_base(repo, trial.kind, target)

        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()

        writable = os.access(readonly_base(trial.client, trial.kind, target), os.W_OK)

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

        fired, fired_evidence = readonly_fired(writable, exercised, op_res)

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

        obs.update(fault_fired_evidence=fired_evidence, readonly_exercise=exercise_detail)

        if fired is None:

            # A successful op whose exposure to the fault is unmeasured is neither

            # held nor a skip: the instrument could not decide.

            classification, error = HARNESS_ERROR, "readonly: " + fired_evidence

        else:

            classification, error = classify(inv, fault_fired=fired, reader_errors=obs.get("reader_errors")), None

    except Exception as exc:  # noqa: BLE001

        restore()

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

    finally:

        restore()

        trial.cleanup()

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

    row["fault_fired"] = fired

    if classification == FAULT_NOT_FIRED:

        row["skip_reason"] = fired_evidence

    return row

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

    elif classification == HARNESS_ERROR and obs.get("reader_errors"):

        row["harness_error"] = "independent reader failed: " + "; ".join(obs["reader_errors"][:2])

    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()