from __future__ import annotations

import os
import shutil
from concurrent.futures import ThreadPoolExecutor
from datetime import date
from pathlib import Path
from collections.abc import Callable
from typing import TypeVar
from tempfile import TemporaryDirectory
from threading import Barrier

from checkin_cli.models import EventType
from checkin_cli.store import CanonicalEventTransaction
from checkin_cli.weekly_operations import (
    CustomerKey,
    WeeklyOperationsConflict,
    WeeklyOperationsError,
)
from checkin_cli.weekly_operations_canonical_registry_history import (
    CanonicalAuthorityRegistrationRow,
    canonical_registration_row,
    registration_row_digest,
)
from checkin_cli.weekly_operations_correlation import (
    CanonicalCheckinCorrelationTransaction,
    CorrelationAction,
    CorrelationRequest,
    CorrelationScope,
    CorrelationResult,
)
from checkin_cli.weekly_operations_customer_authority import CanonicalCheckinCustomerAuthority
from checkin_cli.weekly_operations_customer_authority_factory import (
    open_canonical_checkin_customer_authority,
)
from checkin_cli.weekly_operations_store import WeeklyOperationsStore
from tests._weekly_operations_correlation_support import (
    CUSTOMER,
    DAY,
    StoreFixture,
    UnregisteredCanonicalFixture,
    event,
    register_runtime,
    store_at,
    unregistered_source_at,
)

_REGISTRY = "canonical-authorities-v1.jsonl"
_T = TypeVar("_T")


def _commit(store: WeeklyOperationsStore, source: CanonicalCheckinCustomerAuthority) -> CorrelationResult:
    scope = CorrelationScope(
        source.customer_identity_digest,
        date.fromisoformat(DAY),
        CorrelationAction.CHECKIN,
    )
    return CanonicalCheckinCorrelationTransaction(
        store,
        CorrelationRequest(scope, source),
    ).commit()


def _conflict(operation: Callable[[], _T], reason: str | None = None) -> None:
    try:
        _ = operation()
    except WeeklyOperationsError as error:
        if reason is not None:
            assert reason in str(error)
    else:
        raise AssertionError("typed conflict was not raised")


def test_rogue_copyback_cannot_reregister_or_append_twice() -> None:
    with TemporaryDirectory(prefix="task4-r5-rogue-") as raw:
        root = Path(raw)
        base = event("r5_original_valid_root01")
        pending = unregistered_source_at(root / "profile", CUSTOMER, base)
        fixture = store_at(root / "authority")
        source = register_runtime(
            pending.runtime, fixture.authority
        )
        _ = _commit(fixture.store, source)
        expected = source.registered_binding
        source.close()
        customer_root = pending.runtime.data_root
        sealed = customer_root.with_name("sealed-original")
        _ = customer_root.rename(sealed)
        _ = shutil.copytree(sealed, customer_root)
        rogue = CanonicalEventTransaction.for_customer_runtime(pending.runtime)
        _ = rogue.append_one(
            event("r5_rogue_correction001", EventType.CORRECTION, supersedes=base.event_id)
        )

        _conflict(
            lambda: open_canonical_checkin_customer_authority(
                pending.runtime, expected, fixture.authority
            )
        )
        for _attempt in range(2):
            _conflict(
                lambda: register_runtime(
                    pending.runtime, fixture.authority
                ),
                "already_registered",
            )
        assert len(fixture.store.read()) == 1
        fixture.close()


def test_concurrent_registration_has_one_row_one_binding() -> None:
    with TemporaryDirectory(prefix="task4-r5-race-") as raw:
        root = Path(raw)
        pending = unregistered_source_at(
            root / "profile", CUSTOMER, event("r5_registration_race01")
        )
        fixture = store_at(root / "authority")
        barrier = Barrier(3)

        def register() -> CanonicalCheckinCustomerAuthority | WeeklyOperationsConflict:
            _ = barrier.wait(timeout=5)
            try:
                return register_runtime(
                    pending.runtime, fixture.authority
                )
            except WeeklyOperationsConflict as error:
                return error

        with ThreadPoolExecutor(max_workers=2) as executor:
            futures = [executor.submit(register) for _ in range(2)]
            _ = barrier.wait(timeout=5)
            outcomes = tuple(future.result(timeout=5) for future in futures)
        winners = tuple(
            result for result in outcomes if isinstance(result, CanonicalCheckinCustomerAuthority)
        )
        losers = tuple(result for result in outcomes if isinstance(result, WeeklyOperationsConflict))
        assert len(winners) == 1 and len(losers) == 1
        assert "already_registered" in str(losers[0])
        assert (fixture.authority.observed_path / _REGISTRY).read_bytes().count(b"\n") == 1
        winners[0].close()
        fixture.close()


def _registered(
    root: Path,
) -> tuple[
    UnregisteredCanonicalFixture,
    StoreFixture,
    CanonicalCheckinCustomerAuthority,
    Path,
]:
    pending = unregistered_source_at(
        root / "profile", CUSTOMER, event("r5_integrity_source001")
    )
    fixture = store_at(root / "authority")
    source = register_runtime(
        pending.runtime, fixture.authority
    )
    return pending, fixture, source, fixture.authority.observed_path / _REGISTRY


def test_registry_corruption_torn_duplicate_and_mismatch_fail_closed() -> None:
    attacks = ("corrupt", "torn", "duplicate", "mismatch")
    for attack in attacks:
        with TemporaryDirectory(prefix=f"task4-r5-{attack}-") as raw:
            pending, fixture, source, path = _registered(Path(raw))
            original = path.read_bytes()
            assert fixture.store.read() == ()
            if attack == "corrupt":
                _ = path.write_bytes(
                    original.replace(b"schema_version", b"schema-versi0n", 1)
                )
            elif attack == "torn":
                _ = path.write_bytes(original[:-1])
            elif attack == "duplicate":
                _ = path.write_bytes(original + original)
            else:
                row = CanonicalAuthorityRegistrationRow.model_validate_json(original)
                provisional = row.model_copy(
                    update={"binding_digest": "f" * 64, "row_digest": "0" * 64}
                )
                mismatched = provisional.model_copy(
                    update={"row_digest": registration_row_digest(provisional)}
                )
                _ = path.write_bytes(canonical_registration_row(mismatched) + b"\n")
            _conflict(
                lambda: open_canonical_checkin_customer_authority(
                    pending.runtime, source.registered_binding, fixture.authority
                )
            )
            assert not any(
                child.name.endswith(".day-status-v1.jsonl")
                for child in fixture.authority.observed_path.iterdir()
            )
            source.close()
            fixture.close()


def test_registry_namespace_mode_link_and_fifo_fail_without_blocking() -> None:
    attacks = ("symlink", "hardlink", "mode", "fifo")
    for attack in attacks:
        with TemporaryDirectory(prefix=f"task4-r5-{attack}-") as raw:
            pending, fixture, source, path = _registered(Path(raw))
            assert fixture.store.read() == ()
            if attack == "symlink":
                sealed = Path(raw) / "sealed-registry"
                _ = path.rename(sealed)
                _ = path.symlink_to(sealed)
            elif attack == "hardlink":
                os.link(path, Path(raw) / "registry-hardlink")
            elif attack == "mode":
                path.chmod(0o640)
            else:
                _ = path.rename(Path(raw) / "sealed-registry")
                os.mkfifo(path, mode=0o600)
            _conflict(
                lambda: open_canonical_checkin_customer_authority(
                    pending.runtime, source.registered_binding, fixture.authority
                )
            )
            assert not any(
                child.name.endswith(".day-status-v1.jsonl")
                for child in fixture.authority.observed_path.iterdir()
            )
            source.close()
            fixture.close()


def test_cross_customer_substitution_and_privacy_fail_closed() -> None:
    sentinel = CustomerKey("raw-customer-sentinel-r5")
    with TemporaryDirectory(prefix="task4-r5-privacy-") as raw:
        root = Path(raw)
        first = unregistered_source_at(
            root / "first", sentinel, event("r5_privacy_source_0001")
        )
        second = unregistered_source_at(
            root / "second", CustomerKey("other_customer_r5"), event("r5_other_source_000001")
        )
        fixture = store_at(root / "authority", sentinel)
        second_fixture = store_at(
            root / "second-authority", CustomerKey("other_customer_r5")
        )
        source = register_runtime(
            first.runtime, fixture.authority
        )
        second_source = register_runtime(
            second.runtime, second_fixture.authority
        )
        second_binding = second_source.registered_binding
        second_source.close()
        _conflict(
            lambda: open_canonical_checkin_customer_authority(
                second.runtime, second_binding, fixture.authority
            ),
            "registry binding disagrees",
        )
        payload = (fixture.authority.observed_path / _REGISTRY).read_bytes()
        assert b"raw-customer-sentinel-r5" not in payload
        assert b"display_name" not in payload and b"telegram" not in payload
        source.close()
        second_fixture.close()
        fixture.close()