from __future__ import annotations

import json
import multiprocessing
import os
import time
from datetime import datetime, timezone
from pathlib import Path

import pytest

import gyxx_flow.core.locks as locks_module
from gyxx_flow.core.context import RunContext
from gyxx_flow.core.layout import DataLayout
from gyxx_flow.core.locks import LockManager, ResourceBusyError
from gyxx_flow.core.records import RunJournal


def _crash_while_holding_reclaim_guard(path: str) -> None:
    with locks_module._exclusive_reclaim_guard(
        Path(path),
        resource="module:product",
        deadline=time.monotonic() + 5,
        poll_seconds=0.01,
    ):
        os._exit(0)


def _acquire_shared_resource(root: str, worker: int, queue) -> None:  # type: ignore[no-untyped-def]
    try:
        with LockManager(Path(root)).acquire(
            "module:product",
            owner=f"worker-{worker}",
            timeout_seconds=10,
            poll_seconds=0.01,
        ):
            time.sleep(0.03)
    except Exception as exc:
        queue.put((worker, type(exc).__name__))
    else:
        queue.put((worker, "ok"))


def _context() -> RunContext:
    return RunContext.create(
        "shop.weekly",
        "2026-07-27",
        now=datetime(2026, 7, 27, 4, 0, tzinfo=timezone.utc),
        random_suffix="abc123",
    )


def test_run_journal_tracks_steps_and_final_status(tmp_path: Path) -> None:
    context = _context()
    journal = RunJournal.create(DataLayout(tmp_path), context)

    journal.start_step("collect.jd", attempt=1)
    journal.finish_step("collect.jd", status="success", exit_code=0)
    journal.start_step("publish.feishu", attempt=1)
    journal.finish_step(
        "publish.feishu",
        status="failed",
        exit_code=3,
        error="credential binding missing",
    )
    journal.finalize("failed", error="publish.feishu failed")

    payload = json.loads(journal.path.read_text(encoding="utf-8"))
    assert payload["run_id"] == context.run_id
    assert payload["status"] == "failed"
    assert payload["steps"]["collect.jd"]["status"] == "success"
    assert payload["steps"]["publish.feishu"]["exit_code"] == 3
    assert payload["steps"]["publish.feishu"]["error"] == "credential binding missing"
    assert payload["error"] == "publish.feishu failed"
    assert payload["trace"]["paths"]["run"].startswith("runs/")
    assert payload["trace"]["paths"]["log"].startswith("logs/")
    assert payload["trace"]["paths"]["evidence"].startswith("data/evidence/")


def test_run_journal_records_deduplicated_trace_references(tmp_path: Path) -> None:
    journal = RunJournal.create(DataLayout(tmp_path), _context())

    journal.record_input("artifact:raw-001")
    journal.record_input("artifact:raw-001")
    journal.record_output("artifact:normalized-001")
    journal.record_external_write("outbox:msg-001")

    payload = json.loads(journal.path.read_text(encoding="utf-8"))
    assert payload["trace"]["inputs"] == ["artifact:raw-001"]
    assert payload["trace"]["outputs"] == ["artifact:normalized-001"]
    assert payload["trace"]["external_writes"] == ["outbox:msg-001"]


def test_run_journal_rejects_invalid_transitions(tmp_path: Path) -> None:
    journal = RunJournal.create(DataLayout(tmp_path), _context())

    with pytest.raises(ValueError, match="not running"):
        journal.finish_step("collect.jd", status="success")

    journal.start_step("collect.jd", attempt=1)
    with pytest.raises(ValueError, match="already running"):
        journal.start_step("collect.jd", attempt=2)


def test_named_resource_lock_blocks_concurrent_owner_and_releases(tmp_path: Path) -> None:
    manager = LockManager(tmp_path)

    with manager.acquire("browser:xingtu", owner="run-001") as first:
        assert first.path.exists()
        with pytest.raises(ResourceBusyError, match="browser:xingtu"):
            with manager.acquire("browser:xingtu", owner="run-002", timeout_seconds=0):
                pass

    assert not first.path.exists()
    with manager.acquire("browser:xingtu", owner="run-002", timeout_seconds=0) as second:
        metadata = json.loads(second.path.read_text(encoding="utf-8"))
        assert metadata["resource"] == "browser:xingtu"
        assert metadata["owner"] == "run-002"


def test_named_resource_lock_quarantines_provably_dead_owner(
    monkeypatch: pytest.MonkeyPatch,
    tmp_path: Path,
) -> None:
    manager = LockManager(tmp_path)
    stale = manager.acquire("module:product", owner="run-dead")
    stale.path.parent.mkdir(parents=True, exist_ok=True)
    stale.path.write_text(
        json.dumps(
            {
                "resource": "module:product",
                "owner": "run-dead",
                "acquired_at": "2026-08-01T10:00:00+00:00",
                "pid": 424242,
            }
        ),
        encoding="utf-8",
    )
    monkeypatch.setattr(locks_module, "_process_is_running", lambda _pid: False)

    with manager.acquire(
        "module:product", owner="run-new", timeout_seconds=0
    ) as acquired:
        metadata = json.loads(acquired.path.read_text(encoding="utf-8"))
        assert metadata["owner"] == "run-new"

    quarantined = list((tmp_path / "stale").glob("*.lock"))
    assert len(quarantined) == 1
    assert json.loads(quarantined[0].read_text(encoding="utf-8"))["owner"] == "run-dead"


@pytest.mark.parametrize("liveness", [True, None])
def test_named_resource_lock_keeps_active_or_unknown_owner(
    monkeypatch: pytest.MonkeyPatch,
    tmp_path: Path,
    liveness: bool | None,
) -> None:
    manager = LockManager(tmp_path)
    occupied = manager.acquire("module:product", owner="run-existing")
    occupied.path.parent.mkdir(parents=True, exist_ok=True)
    occupied.path.write_text(
        json.dumps(
            {
                "resource": "module:product",
                "owner": "run-existing",
                "acquired_at": "2026-08-01T10:00:00+00:00",
                "pid": 424242,
            }
        ),
        encoding="utf-8",
    )
    monkeypatch.setattr(locks_module, "_process_is_running", lambda _pid: liveness)

    with pytest.raises(ResourceBusyError, match="module:product"):
        with manager.acquire("module:product", owner="run-new", timeout_seconds=0):
            pass

    assert occupied.path.exists()


def test_named_resource_lock_does_not_reclaim_recent_malformed_lock(
    tmp_path: Path,
) -> None:
    manager = LockManager(tmp_path)
    occupied = manager.acquire("module:product", owner="run-existing")
    occupied.path.parent.mkdir(parents=True, exist_ok=True)
    occupied.path.write_text('{"resource": "module:product"', encoding="utf-8")

    with pytest.raises(ResourceBusyError, match="module:product"):
        with manager.acquire("module:product", owner="run-new", timeout_seconds=0):
            pass

    assert occupied.path.exists()


def test_named_resource_lock_quarantines_old_malformed_lock_without_pid(
    tmp_path: Path,
) -> None:
    manager = LockManager(tmp_path)
    occupied = manager.acquire("module:product", owner="run-existing")
    occupied.path.parent.mkdir(parents=True, exist_ok=True)
    occupied.path.write_text('{"resource": "module:product"', encoding="utf-8")
    old = time.time() - locks_module._MALFORMED_LOCK_GRACE_SECONDS - 5
    os.utime(occupied.path, (old, old))

    with manager.acquire(
        "module:product", owner="run-new", timeout_seconds=0
    ) as acquired:
        metadata = json.loads(acquired.path.read_text(encoding="utf-8"))
        assert metadata["owner"] == "run-new"

    quarantined = list((tmp_path / "stale").glob("*.lock"))
    assert len(quarantined) == 1


def test_named_resource_lock_keeps_old_malformed_lock_with_active_pid(
    tmp_path: Path,
) -> None:
    manager = LockManager(tmp_path)
    occupied = manager.acquire("module:product", owner="run-existing")
    occupied.path.parent.mkdir(parents=True, exist_ok=True)
    occupied.path.write_text(f'{{"pid": {os.getpid()},', encoding="utf-8")
    old = time.time() - locks_module._MALFORMED_LOCK_GRACE_SECONDS - 5
    os.utime(occupied.path, (old, old))

    with pytest.raises(ResourceBusyError, match="module:product"):
        with manager.acquire("module:product", owner="run-new", timeout_seconds=0):
            pass

    assert occupied.path.exists()


def test_named_resource_lock_uses_process_start_to_detect_pid_reuse(
    monkeypatch: pytest.MonkeyPatch,
    tmp_path: Path,
) -> None:
    manager = LockManager(tmp_path)
    occupied = manager.acquire("module:product", owner="run-existing")
    occupied.path.parent.mkdir(parents=True, exist_ok=True)
    occupied.path.write_text(
        json.dumps(
            {
                "resource": "module:product",
                "owner": "run-existing",
                "acquired_at": "2026-08-01T10:00:00+00:00",
                "pid": 424242,
                "process_started_at": 100.0,
            }
        ),
        encoding="utf-8",
    )
    monkeypatch.setattr(locks_module, "_process_is_running", lambda _pid: True)
    monkeypatch.setattr(locks_module, "_process_started_at", lambda _pid: 100.0)

    with pytest.raises(ResourceBusyError, match="module:product"):
        with manager.acquire("module:product", owner="run-new", timeout_seconds=0):
            pass

    monkeypatch.setattr(locks_module, "_process_started_at", lambda _pid: 200.0)
    with manager.acquire(
        "module:product", owner="run-new", timeout_seconds=0
    ) as acquired:
        assert acquired.path.exists()


def test_named_resource_lock_guard_is_released_when_reclaimer_crashes(
    tmp_path: Path,
) -> None:
    manager = LockManager(tmp_path)
    lock = manager.acquire("module:product", owner="run-new")
    guard = lock.path.with_suffix(lock.path.suffix + ".reclaim")
    context = multiprocessing.get_context("spawn")
    process = context.Process(
        target=_crash_while_holding_reclaim_guard,
        args=(str(guard),),
    )

    process.start()
    process.join(timeout=15)

    assert process.exitcode == 0
    assert guard.exists()
    with manager.acquire(
        "module:product", owner="run-new", timeout_seconds=1
    ) as acquired:
        assert acquired.path.exists()


def test_named_resource_lock_serializes_concurrent_stale_recovery(
    tmp_path: Path,
) -> None:
    manager = LockManager(tmp_path)
    stale = manager.acquire("module:product", owner="run-dead")
    stale.path.parent.mkdir(parents=True, exist_ok=True)
    stale.path.write_text(
        json.dumps(
            {
                "resource": "module:product",
                "owner": "run-dead",
                "acquired_at": "2026-08-01T10:00:00+00:00",
                "pid": 2_147_483_647,
            }
        ),
        encoding="utf-8",
    )
    context = multiprocessing.get_context("spawn")
    queue = context.Queue()
    workers = [
        context.Process(
            target=_acquire_shared_resource,
            args=(str(tmp_path), worker, queue),
        )
        for worker in range(4)
    ]

    for process in workers:
        process.start()
    for process in workers:
        process.join(timeout=20)

    assert [process.exitcode for process in workers] == [0, 0, 0, 0]
    results = sorted(queue.get(timeout=2) for _ in workers)
    assert results == [(worker, "ok") for worker in range(4)]
    assert not stale.path.exists()
    assert len(list((tmp_path / "stale").glob("*.lock"))) == 1


def test_named_resource_lock_never_publishes_partial_metadata(
    monkeypatch: pytest.MonkeyPatch,
    tmp_path: Path,
) -> None:
    manager = LockManager(tmp_path)
    lock = manager.acquire("module:product", owner="run-new")

    def fail_link(_source: Path, _target: Path) -> None:
        raise OSError("injected link failure")

    monkeypatch.setattr(locks_module.os, "link", fail_link)

    with pytest.raises(OSError, match="injected link failure"):
        with lock:
            pass

    assert not lock.path.exists()
    assert list(tmp_path.glob(".*.tmp")) == []


@pytest.mark.parametrize("resource", ["", "../state", "resource\nname"])
def test_named_resource_lock_rejects_unsafe_name(tmp_path: Path, resource: str) -> None:
    manager = LockManager(tmp_path)

    with pytest.raises(ValueError, match="resource"):
        with manager.acquire(resource, owner="run-001"):
            pass