"""Result persistence: one write path for every lane.

Writes the timestamped raw JSONL (the durable source of truth) first, then

validates rows against the lane's row contract, and only then writes the

Markdown summary and the latest.* copies. Rows are always persisted before a

contract violation is raised, so a contract bug never destroys collected

measurements; it fails the run loudly instead (data first, then the alarm).

The latest.* pointers are written strictly after validation so they never

point at rows that violate the contract.

"""

from __future__ import annotations

import json

import os

import shutil

from pathlib import Path

from typing import Any, Optional

from oakbench.rows import RowContractViolation, validate_rows

try:

    import fcntl

except ImportError:  # pragma: no cover - platforms without fcntl

    fcntl = None  # type: ignore[assignment]

def _write_all(fd: int, data: bytes) -> None:

    view = memoryview(data)

    written_total = 0

    while written_total < len(view):

        written = os.write(fd, view[written_total:])

        if written == 0:

            raise OSError("os.write returned 0 bytes")

        written_total += written

def _atomic_copy(src: Path, dst: Path) -> None:

    tmp = dst.with_name(f"{dst.name}.tmp-{os.getpid()}")

    try:

        shutil.copyfile(src, tmp)

        os.replace(tmp, dst)

    finally:

        if tmp.exists():

            tmp.unlink()

class ResultsStore:

    def __init__(

        self, results_dir: Path, lane: str, filename_suffix: str = "", flush_partial: bool = False

    ) -> None:

        """filename_suffix distinguishes lanes sharing one results dir:

        suffix "mount" writes <id>.mount.jsonl and latest.mount.jsonl so the

        mount lane never clobbers the core lane's latest.jsonl.

        flush_partial (opt-in, default off so other lanes are untouched)

        enables crash insurance for long campaigns: completed rows are appended

        incrementally to <id><suffix>.partial.jsonl via append_partial(), and

        write() finalizes by rewriting the canonical file and removing the

        partial only after the canonical write succeeded. A campaign killed

        hours in keeps every flushed row."""

        self.results_dir = results_dir

        self.lane = lane

        self.suffix = f".{filename_suffix}" if filename_suffix else ""

        self.flush_partial = flush_partial

    def partial_path(self, bench_id: str) -> Path:

        return self.results_dir / f"{bench_id}{self.suffix}.partial.jsonl"

    def append_partial(self, bench_id: str, rows: list[dict[str, Any]]) -> Path:

        """Append completed rows to the partial JSONL (crash insurance).

        Appends are flock-guarded os.write() loops of complete buffers (the

        oakbench/vcs_shim.py sidecar pattern): with O_APPEND plus the

        exclusive lock, concurrent flushers cannot interleave rows even if the

        OS accepts only part of the buffer per syscall. Lines use the same

        json.dumps(sort_keys=True) encoding as the canonical file, so a

        surviving partial is directly consumable by row readers.

        """

        if not self.flush_partial:

            raise RuntimeError(

                "partial flushing is disabled; construct ResultsStore(..., flush_partial=True)"

            )

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

        path = self.partial_path(bench_id)

        data = "".join(json.dumps(row, sort_keys=True) + "\n" for row in rows).encode("utf-8")

        fd = os.open(path, os.O_WRONLY | os.O_APPEND | os.O_CREAT, 0o644)

        try:

            if fcntl is not None:

                fcntl.flock(fd, fcntl.LOCK_EX)

            _write_all(fd, data)

        finally:

            try:

                if fcntl is not None:

                    fcntl.flock(fd, fcntl.LOCK_UN)

            finally:

                os.close(fd)

        return path

    def write(

        self, bench_id: str, rows: list[dict[str, Any]], summary_text: Optional[str] = None

    ) -> tuple[Path, Optional[Path]]:

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

        raw_path = self.results_dir / f"{bench_id}{self.suffix}.jsonl"

        summary_path: Optional[Path] = None

        with raw_path.open("w") as fh:

            for row in rows:

                fh.write(json.dumps(row, sort_keys=True) + "\n")

        # The raw {bench_id} file is intentionally written BEFORE validation

        # (the exception message points emitter debugging at it), but the

        # latest.* pointers must only ever reference contract-clean rows, so

        # validation happens here — after the raw write, before any latest copy.

        errors = validate_rows(rows, self.lane)

        if errors:

            preview = "\n  ".join(errors[:20])

            more = f"\n  ... and {len(errors) - 20} more" if len(errors) > 20 else ""

            raise RowContractViolation(

                f"Row contract violation ({self.lane} lane). Rows were written to {raw_path} "

                f"but the emitter has a bug:\n  {preview}{more}"

            )

        _atomic_copy(raw_path, self.results_dir / f"latest{self.suffix}.jsonl")

        if summary_text is not None:

            summary_path = self.results_dir / f"{bench_id}{self.suffix}.summary.md"

            summary_path.write_text(summary_text)

            _atomic_copy(summary_path, self.results_dir / f"latest{self.suffix}.summary.md")

        if self.flush_partial:

            # Finalize only after the canonical write AND validation succeeded;

            # on any failure the partial survives as the crash-insurance copy.

            self.partial_path(bench_id).unlink(missing_ok=True)

        return raw_path, summary_path