from __future__ import annotations

import io
import json
import os
import subprocess
import sys
from pathlib import Path

import pytest

import game_data_capture.ingest.spool_helper as spool_helper
from game_data_capture.ingest.spool_helper import (
    PNG_SIGNATURE,
    SpoolError,
    acknowledge_receipt,
    durable_receipts,
    receipt_name,
    reconcile_spool,
    spool_png,
)


def png(payload: bytes = b"data") -> bytes:
    return PNG_SIGNATURE + payload


def fill_receipt_capacity(spool: Path) -> dict[str, object]:
    seed = spool_png(
        io.BytesIO(png(b"seed")),
        spool=spool,
        max_bytes=100,
        receipt_token="seed",
        policy_version="v1",
    )
    for index in range(255):
        token = f"pre{index}"
        sidecar = spool / receipt_name(token)
        sidecar.write_text(
            json.dumps(dict(seed, receipt_token=token), sort_keys=True, separators=(",", ":"))
        )
        sidecar.chmod(0o600)
    return seed


def test_png_spool_is_private_atomic_and_deduplicated(
    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
    spool = (tmp_path / "spool").resolve()
    replaced: list[tuple[Path, Path]] = []
    real_replace = os.replace

    def replace(source: Path, destination: Path) -> None:
        replaced.append((source, destination))
        real_replace(source, destination)

    monkeypatch.setattr(os, "replace", replace)
    first = spool_png(
        io.BytesIO(png()), spool=spool, max_bytes=100, receipt_token="one", policy_version="v1"
    )
    second = spool_png(
        io.BytesIO(png()), spool=spool, max_bytes=100, receipt_token="two", policy_version="v1"
    )
    assert first["sha256"] == second["sha256"]
    assert (spool.stat().st_mode & 0o777) == 0o700
    installed = spool / str(first["spool_name"])
    assert installed.read_bytes() == png() and (installed.stat().st_mode & 0o777) == 0o600
    assert len(replaced) == 1 and replaced[0][1] == installed
    sidecars = sorted(spool.glob("*.receipt.json"))
    assert [sidecar.name for sidecar in sidecars] == ["one.receipt.json", "two.receipt.json"]
    assert all((sidecar.stat().st_mode & 0o777) == 0o600 for sidecar in sidecars)
    assert all(set(json.loads(sidecar.read_text())) == set(first) for sidecar in sidecars)
    assert not list(spool.glob("*.partial"))


@pytest.mark.parametrize("value", [b"", b"text", PNG_SIGNATURE[:4]])
def test_empty_text_and_malformed_png_leave_no_partial(tmp_path: Path, value: bytes) -> None:
    spool = (tmp_path / "spool").resolve()
    with pytest.raises(SpoolError):
        spool_png(
            io.BytesIO(value), spool=spool, max_bytes=100, receipt_token="one", policy_version="v1"
        )
    assert not list(spool.glob("*.partial"))


def test_unsupported_mime_leaves_no_partial(tmp_path: Path) -> None:
    spool = (tmp_path / "spool").resolve()
    with pytest.raises(SpoolError):
        spool_png(
            io.BytesIO(png()),
            spool=spool,
            max_bytes=100,
            receipt_token="one",
            policy_version="v1",
            media_type="text/plain",
        )
    assert not spool.exists()


def test_short_reads_accumulate_the_png_signature(tmp_path: Path) -> None:
    class ShortReads:
        def __init__(self) -> None:
            self.parts = [PNG_SIGNATURE[:3], PNG_SIGNATURE[3:] + b"payload", b""]

        def read(self, _size: int) -> bytes:
            return self.parts.pop(0)

    receipt = spool_png(
        ShortReads(),
        spool=(tmp_path / "spool").resolve(),
        max_bytes=100,
        receipt_token="one",
        policy_version="v1",
    )
    assert receipt["byte_size"] == len(PNG_SIGNATURE + b"payload")


def test_size_bound_and_crash_reconciliation_remove_partials(tmp_path: Path) -> None:
    spool = (tmp_path / "spool").resolve()
    with pytest.raises(SpoolError):
        spool_png(
            io.BytesIO(png(b"x" * 100)),
            spool=spool,
            max_bytes=10,
            receipt_token="one",
            policy_version="v1",
        )
    leftover = spool / ".crash.partial"
    leftover.write_bytes(b"incomplete")
    assert reconcile_spool(spool) == [".crash.partial"]
    assert not leftover.exists()


def test_pending_receipt_is_promoted_only_when_its_png_is_verified(tmp_path: Path) -> None:
    spool = (tmp_path / "spool").resolve()
    receipt = spool_png(
        io.BytesIO(png()), spool=spool, max_bytes=100, receipt_token="one", policy_version="v1"
    )
    final = spool / receipt_name("one")
    pending = spool / ".one.receipt.pending"
    os.replace(final, pending)
    assert reconcile_spool(spool) == []
    assert final.exists() and not pending.exists()
    final.unlink()
    pending.write_text(json.dumps(receipt))
    assert reconcile_spool(spool) == [pending.name]
    assert not pending.exists()


def test_receipt_token_replay_cannot_replace_the_original_sidecar(tmp_path: Path) -> None:
    spool = (tmp_path / "spool").resolve()
    original = spool_png(
        io.BytesIO(png(b"original")),
        spool=spool,
        max_bytes=100,
        receipt_token="same-token",
        policy_version="v1",
    )
    final = spool / receipt_name("same-token")
    original_bytes = final.read_bytes()
    assert (
        spool_png(
            io.BytesIO(png(b"original")),
            spool=spool,
            max_bytes=100,
            receipt_token="same-token",
            policy_version="v1",
        )
        == original
    )
    with pytest.raises(SpoolError, match="receipt token conflict"):
        spool_png(
            io.BytesIO(png(b"different")),
            spool=spool,
            max_bytes=100,
            receipt_token="same-token",
            policy_version="v1",
        )
    assert final.read_bytes() == original_bytes
    conflicting = dict(original, sha256="0" * 64, spool_name=f"{'0' * 64}.png")
    pending = spool / ".same-token.receipt.pending"
    pending.write_text(json.dumps(conflicting, sort_keys=True, separators=(",", ":")))
    pending.chmod(0o600)
    assert pending.name in reconcile_spool(spool)
    assert final.read_bytes() == original_bytes and not pending.exists()


def test_sidecars_are_bounded_validated_and_acknowledged_exactly_once(tmp_path: Path) -> None:
    spool = (tmp_path / "spool").resolve()
    first = spool_png(
        io.BytesIO(png(b"one")),
        spool=spool,
        max_bytes=100,
        receipt_token="one",
        policy_version="v1",
    )
    second = spool_png(
        io.BytesIO(png(b"two")),
        spool=spool,
        max_bytes=100,
        receipt_token="two",
        policy_version="v1",
    )
    malformed = spool / "bad.receipt.json"
    malformed.write_bytes(b"{" + b"x" * 1024)
    malformed.chmod(0o600)
    linked = spool / "linked.receipt.json"
    linked.symlink_to(malformed)
    assert {item["receipt_token"] for item in durable_receipts(spool, max_items=2)} == {
        "one",
        "two",
    }
    assert acknowledge_receipt(spool, first)
    assert not acknowledge_receipt(spool, first)
    assert (spool / receipt_name("two")).exists()
    assert not acknowledge_receipt(spool, dict(second, receipt_token="one"))


def test_malformed_sidecars_do_not_starve_valid_receipts_or_partial_cleanup(tmp_path: Path) -> None:
    spool = (tmp_path / "spool").resolve()
    spool.mkdir(mode=0o700)
    for index in range(256):
        malformed = spool / f"{index:03d}.receipt.json"
        malformed.write_bytes(b"{")
        malformed.chmod(0o600)
    valid = spool_png(
        io.BytesIO(png()), spool=spool, max_bytes=100, receipt_token="valid", policy_version="v1"
    )
    assert list(durable_receipts(spool, max_items=1)) == [valid]
    assert list(spool.glob("*.receipt.json")) == [spool / "valid.receipt.json"]
    for index in range(256):
        pending = spool / f".pending{index}.receipt.pending"
        pending.write_text(
            json.dumps(
                dict(valid, receipt_token=f"pending{index}"), sort_keys=True, separators=(",", ":")
            )
        )
        pending.chmod(0o600)
    partial = spool / ".after-pending.partial"
    partial.write_bytes(b"incomplete")
    assert partial.name in reconcile_spool(spool)
    assert not partial.exists()


def test_receipt_capacity_rejects_new_tokens_but_accepts_exact_replay(tmp_path: Path) -> None:
    spool = (tmp_path / "spool").resolve()
    seed = fill_receipt_capacity(spool)
    evidence = (spool / receipt_name("seed")).read_bytes()
    png_names = {path.name for path in spool.glob("*.png")}
    with pytest.raises(SpoolError, match="capacity"):
        spool_png(
            io.BytesIO(png(b"new")),
            spool=spool,
            max_bytes=100,
            receipt_token="new",
            policy_version="v1",
        )
    assert (spool / receipt_name("seed")).read_bytes() == evidence
    assert {path.name for path in spool.glob("*.png")} == png_names
    assert (
        spool_png(
            io.BytesIO(png(b"seed")),
            spool=spool,
            max_bytes=100,
            receipt_token="seed",
            policy_version="v1",
        )
        == seed
    )


def test_receipt_capacity_reconciles_malformed_files_before_counting(tmp_path: Path) -> None:
    spool = (tmp_path / "spool").resolve()
    spool.mkdir(mode=0o700)
    for index in range(256):
        malformed = spool / f"bad{index}.receipt.json"
        malformed.write_bytes(b"{")
        malformed.chmod(0o600)
    accepted = spool_png(
        io.BytesIO(png()), spool=spool, max_bytes=100, receipt_token="accepted", policy_version="v1"
    )
    assert accepted["receipt_token"] == "accepted"
    assert list(spool.glob("bad*.receipt.json")) == []


def test_receipt_capacity_is_serialized_across_helper_processes(tmp_path: Path) -> None:
    spool = (tmp_path / "spool").resolve()
    fill_receipt_capacity(spool)
    for index in range(1):
        (spool / receipt_name(f"pre{index}")).unlink()
    arguments = [
        sys.executable,
        "-m",
        "game_data_capture.ingest.spool_helper",
        "--spool",
        str(spool),
        "--max-bytes",
        "100",
        "--policy-version",
        "v1",
    ]
    first = subprocess.Popen([*arguments, "--receipt-token", "race-one"], stdin=subprocess.PIPE)
    second = subprocess.Popen([*arguments, "--receipt-token", "race-two"], stdin=subprocess.PIPE)
    assert first.stdin is not None and second.stdin is not None
    first.stdin.write(png(b"one"))
    first.stdin.close()
    second.stdin.write(png(b"two"))
    second.stdin.close()
    assert sorted((first.wait(), second.wait())) == [0, 2]
    assert len(list(spool.glob("*.receipt.json"))) == 256


def test_spool_lock_is_private_rejects_links_and_releases_on_failure(
    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
    spool = (tmp_path / "spool").resolve()
    spool.mkdir(mode=0o700)
    (spool / ".spool.lock").symlink_to(tmp_path / "outside")
    with pytest.raises(SpoolError, match="unsafe spool lock"):
        spool_png(
            io.BytesIO(png()), spool=spool, max_bytes=100, receipt_token="link", policy_version="v1"
        )
    (spool / ".spool.lock").unlink()
    with monkeypatch.context() as context:
        context.setattr(
            spool_helper,
            "_install_pending_sidecar",
            lambda *_args: (_ for _ in ()).throw(OSError()),
        )
        with pytest.raises(OSError):
            spool_png(
                io.BytesIO(png()),
                spool=spool,
                max_bytes=100,
                receipt_token="failure",
                policy_version="v1",
            )
    accepted = spool_png(
        io.BytesIO(png()),
        spool=spool,
        max_bytes=100,
        receipt_token="recovered",
        policy_version="v1",
    )
    assert accepted["receipt_token"] == "recovered"
    assert (spool / ".spool.lock").stat().st_mode & 0o777 == 0o600


def test_short_write_removes_partial(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
    spool = (tmp_path / "spool").resolve()
    monkeypatch.setattr(os, "write", lambda _fd, _data: 0)
    with pytest.raises(OSError):
        spool_png(
            io.BytesIO(png()), spool=spool, max_bytes=100, receipt_token="one", policy_version="v1"
        )
    assert not list(spool.glob("*.partial"))


def test_helper_source_has_no_service_or_storage_imports() -> None:
    source = Path(__file__).parents[3] / "src/game_data_capture/ingest/spool_helper.py"
    text = source.read_text(encoding="utf-8")
    for forbidden in ("fastapi", "sqlite", "openai", "game_data_capture.storage"):
        assert forbidden not in text.lower()


def test_stdin_entrypoint_returns_only_bounded_json_receipt(tmp_path: Path) -> None:
    spool = (tmp_path / "spool").resolve()
    completed = subprocess.run(
        [
            sys.executable,
            "-m",
            "game_data_capture.ingest.spool_helper",
            "--spool",
            str(spool),
            "--max-bytes",
            "100",
            "--receipt-token",
            "receipt",
            "--policy-version",
            "v1",
        ],
        input=png(),
        capture_output=True,
        check=False,
        env={"PATH": os.environ.get("PATH", "")},
    )
    assert completed.returncode == 0 and len(completed.stdout) <= 1024 and not completed.stderr
    receipt = json.loads(completed.stdout)
    assert receipt["media_type"] == "image/png" and receipt["receipt_token"] == "receipt"