import json
import multiprocessing
from pathlib import Path
import subprocess
import sys


def _process_reserve(state_path, start_event, result_queue):
    from services.nemotron_dispatch_reservation_service import (
        SharedFileReservationStore,
    )

    store = SharedFileReservationStore(
        state_path,
        committed_ttl_sec=3600,
        lease_ttl_sec=60,
    )
    start_event.wait(timeout=5)
    result_queue.put(store.reserve("SKU-PROCESS-SHARED"))


def test_shared_store_coordinates_independent_instances_and_ttl(tmp_path):
    from services.nemotron_dispatch_reservation_service import (
        SharedFileReservationStore,
    )

    clock = {"now": 1000.0}
    path = tmp_path / "dedupe" / "state.json"
    first = SharedFileReservationStore(
        path,
        committed_ttl_sec=100,
        lease_ttl_sec=20,
        clock=lambda: clock["now"],
    )
    second = SharedFileReservationStore(
        path,
        committed_ttl_sec=100,
        lease_ttl_sec=20,
        clock=lambda: clock["now"],
    )

    token = first.reserve("SKU-SHARED")
    assert token
    assert second.reserve("SKU-SHARED") is None
    assert second.is_duplicate("SKU-SHARED") is True
    assert "SKU-SHARED" not in path.read_text(encoding="utf-8")
    assert second.release("SKU-SHARED", "stale-token") is False
    assert first.refresh("SKU-SHARED", token) is True
    assert first.commit("SKU-SHARED", token) is True
    assert second.reserve("SKU-SHARED") is None

    clock["now"] += 101
    next_token = second.reserve("SKU-SHARED")
    assert next_token and next_token != token
    assert second.release("SKU-SHARED", next_token) is True


def test_begin_side_effect_persists_extended_quarantine(tmp_path):
    from services.nemotron_dispatch_reservation_service import (
        SharedFileReservationStore,
    )

    clock = {"now": 1000.0}
    path = tmp_path / "state.json"
    first = SharedFileReservationStore(
        path,
        committed_ttl_sec=100,
        lease_ttl_sec=20,
        clock=lambda: clock["now"],
    )
    second = SharedFileReservationStore(
        path,
        committed_ttl_sec=100,
        lease_ttl_sec=20,
        clock=lambda: clock["now"],
    )

    token = first.reserve("SKU-SIDE-EFFECT")
    assert token
    assert first.begin_side_effect("SKU-SIDE-EFFECT", token) is True

    payload = json.loads(path.read_text(encoding="utf-8"))
    lease = next(iter(payload["leases"].values()))
    assert lease == {
        "token": token,
        "expires_at": 1100.0,
        "phase": "side_effect_started",
    }

    clock["now"] += 21
    assert second.reserve("SKU-SIDE-EFFECT") is None
    clock["now"] += 80
    assert second.reserve("SKU-SIDE-EFFECT")


def test_shared_store_is_atomic_across_processes(tmp_path):
    state_path = str(tmp_path / "process" / "state.json")
    context = multiprocessing.get_context("fork")
    start_event = context.Event()
    result_queue = context.Queue()
    processes = [
        context.Process(
            target=_process_reserve,
            args=(state_path, start_event, result_queue),
        )
        for _index in range(6)
    ]
    for process in processes:
        process.start()
    start_event.set()
    tokens = [result_queue.get(timeout=10) for _process in processes]
    for process in processes:
        process.join(timeout=10)
        assert process.exitcode == 0

    owners = [token for token in tokens if token]
    assert len(owners) == 1


def test_shared_store_corruption_fails_closed(tmp_path):
    from services.nemotron_dispatch_reservation_service import (
        SharedFileReservationStore,
    )

    path = tmp_path / "state.json"
    path.write_text("{not-json", encoding="utf-8")
    store = SharedFileReservationStore(
        path,
        committed_ttl_sec=3600,
        lease_ttl_sec=900,
    )

    assert store.reserve("SKU-CORRUPT") is None
    assert store.refresh("SKU-CORRUPT", "token") is False
    assert store.commit("SKU-CORRUPT", "token") is False
    assert store.release("SKU-CORRUPT", "token") is False
    assert store.is_duplicate("SKU-CORRUPT") is True
    assert path.read_text(encoding="utf-8") == "{not-json"


def test_atomic_write_fsyncs_file_and_parent_directory(tmp_path, monkeypatch):
    from services import nemotron_dispatch_reservation_service as service

    fsynced = []
    real_fsync = service.os.fsync

    def record_fsync(fd):
        fsynced.append(fd)
        return real_fsync(fd)

    monkeypatch.setattr(service.os, "fsync", record_fsync)
    store = service.SharedFileReservationStore(
        tmp_path / "nested" / "state.json",
        committed_ttl_sec=3600,
        lease_ttl_sec=900,
    )

    assert store.reserve("SKU-FSYNC")
    assert len(fsynced) >= 2


def test_missing_process_shared_lock_fails_closed(monkeypatch, tmp_path):
    from services import nemotron_dispatch_reservation_service as service

    monkeypatch.setattr(service, "fcntl", None)
    monkeypatch.delenv("FLASK_ENV", raising=False)
    monkeypatch.delenv("USE_POSTGRESQL", raising=False)

    assert service.build_production_shared_reservation_store(
        committed_ttl_sec=3600,
        lease_ttl_sec=900,
    ) is None

    store = service.SharedFileReservationStore(
        tmp_path / "state.json",
        committed_ttl_sec=3600,
        lease_ttl_sec=900,
    )
    assert store.reserve("SKU-NO-FCNTL") is None
    assert store.is_duplicate("SKU-NO-FCNTL") is True


def test_production_builder_forces_canonical_shared_data_path(monkeypatch):
    from services import nemotron_dispatch_reservation_service as service

    monkeypatch.setenv("FLASK_ENV", "production")
    store = service.build_production_shared_reservation_store(
        committed_ttl_sec=3600,
        lease_ttl_sec=900,
    )

    assert store is not None
    assert store.state_path == Path(
        "/app/data/ai_automation/nemotron_dispatch_dedupe_state.json"
    )
    assert store.lock_path == Path(
        "/app/data/ai_automation/nemotron_dispatch_dedupe_state.json.lock"
    )


def test_postgres_runtime_cannot_fall_back_to_process_local_memory(monkeypatch):
    from services import nemotron_dispatch_reservation_service as service

    monkeypatch.delenv("FLASK_ENV", raising=False)
    monkeypatch.setenv("USE_POSTGRESQL", "true")

    assert service.production_shared_reservation_required() is True
    assert service.build_production_shared_reservation_store(
        committed_ttl_sec=3600,
        lease_ttl_sec=900,
    ) is not None


def test_shared_state_contains_only_hashed_sku_and_timing_metadata(tmp_path):
    from services.nemotron_dispatch_reservation_service import (
        SharedFileReservationStore,
    )

    path = tmp_path / "state.json"
    store = SharedFileReservationStore(
        path,
        committed_ttl_sec=3600,
        lease_ttl_sec=900,
    )
    token = store.reserve("RAW-SKU-MUST-NOT-APPEAR")
    payload = json.loads(path.read_text(encoding="utf-8"))

    assert token
    assert "RAW-SKU-MUST-NOT-APPEAR" not in path.read_text(encoding="utf-8")
    assert set(payload) == {"version", "leases", "committed"}
    assert len(next(iter(payload["leases"]))) == 64


def test_production_containers_share_data_mount_and_force_shared_backend():
    compose = (
        Path(__file__).resolve().parents[1] / "docker-compose.yml"
    ).read_text(encoding="utf-8")

    assert compose.count("- ./data:/app/data") >= 3
    assert compose.count("- FLASK_ENV=production") >= 3


def test_shared_reservation_canary_has_no_business_side_effects(tmp_path):
    from services.nemotron_dispatch_reservation_service import (
        run_shared_reservation_canary,
    )

    payload = run_shared_reservation_canary(
        state_path=tmp_path / "canary" / "state.json",
    )

    assert payload["success"] is True
    assert payload["check_pass_count"] == payload["check_count"] == 6
    assert payload["controlled_apply"] == {
        "writes_shared_reservation_state": True,
        "terminal_state_clear": True,
        "writes_database": False,
        "sends_telegram": False,
        "executes_business_tool": False,
        "reads_secret": False,
    }


def test_shared_reservation_canary_cli_is_machine_readable(tmp_path):
    completed = subprocess.run(
        [
            sys.executable,
            "scripts/ops/run_nemotron_dispatch_reservation_canary.py",
            "--state-path",
            str(tmp_path / "cli" / "state.json"),
        ],
        cwd=Path(__file__).resolve().parents[1],
        capture_output=True,
        check=False,
        text=True,
    )

    assert completed.returncode == 0, completed.stderr
    payload = json.loads(completed.stdout)
    assert payload["status"] == "shared_reservation_canary_passed"
    assert payload["controlled_apply"]["writes_database"] is False