"""Adversarial authority-child and durable-identity tests."""

from __future__ import annotations

import fcntl
import os
from concurrent.futures import ThreadPoolExecutor
from dataclasses import replace
from datetime import date, datetime
from pathlib import Path
from threading import Barrier
from typing import Protocol
from unittest import TestCase
from unittest.mock import patch

from checkin_cli.weekly_operations import CanonicalPin, CustomerKey, DayState, WeeklyOperationInput, customer_identity_digest, WeeklyOperationRow, WeeklyOperationsConflict, WeeklyOperationsCorruption, WeeklyOperationsError, canonical_weekly_row, weekly_row_digest
from checkin_cli.weekly_operations_authority import AuthorityId, WeeklyOperationsAuthorityRoot
from checkin_cli.weekly_operations_layout import MARKER_NAME, customer_data_name, customer_lock_name, customer_storage_digest
from checkin_cli.weekly_operations_store import WeeklyOperationsStore

from tests._weekly_operations_support import initialize_at

CUSTOMER = CustomerKey("client_001")
OTHER = CustomerKey("client_002")
ASSERTIONS = TestCase()


class HasFileno(Protocol):
    def fileno(self) -> int: ...


def _operation() -> WeeklyOperationInput:
    return WeeklyOperationInput.for_customer(CUSTOMER, date(2026, 8, 17), DayState.SUBMITTED, CanonicalPin(1, "a" * 64), datetime.fromisoformat("2026-08-17T20:00:00+09:00"))


def _authority(root: Path) -> WeeklyOperationsAuthorityRoot:
    path = root / "profile-authority"
    path.mkdir(parents=True, mode=0o700)
    path.chmod(0o700)
    return initialize_at(path, AuthorityId("7" * 64))


def _data(authority: WeeklyOperationsAuthorityRoot, customer: CustomerKey = CUSTOMER) -> Path:
    return authority.observed_path / customer_data_name(customer_identity_digest(customer))


def _store(root: Path) -> WeeklyOperationsStore:
    return WeeklyOperationsStore.for_authority(_authority(root), CUSTOMER)


def test_removed_customers_directory_name_is_rejected_without_outside_mutation(tmp_path: Path) -> None:
    authority = _authority(tmp_path)
    outside = tmp_path / "outside"
    outside.mkdir(mode=0o700)
    (authority.observed_path / "customers-v1").symlink_to(outside, target_is_directory=True)
    with ASSERTIONS.assertRaises(WeeklyOperationsError):
        _ = WeeklyOperationsStore.for_authority(authority, CUSTOMER).append(_operation())
    assert tuple(outside.iterdir()) == ()


def test_integrity_hardlinked_sidecar_is_rejected(tmp_path: Path) -> None:
    authority = _authority(tmp_path)
    _ = WeeklyOperationsStore.for_authority(authority, CUSTOMER).append(_operation())
    other_data = authority.observed_path / customer_data_name(customer_identity_digest(OTHER))
    os.link(_data(authority), other_data)
    with ASSERTIONS.assertRaises(WeeklyOperationsCorruption):
        _ = WeeklyOperationsStore.for_authority(authority, OTHER).read()


def test_integrity_global_canonical_reversal_conflicts(tmp_path: Path) -> None:
    store = _store(tmp_path)
    _ = store.append(replace(_operation(), canonical=CanonicalPin(5, "e" * 64)))
    reversed_pin = replace(_operation(), kst_day=date(2026, 8, 18), canonical=CanonicalPin(4, "d" * 64), occurred_at=datetime.fromisoformat("2026-08-18T20:00:00+09:00"))
    with ASSERTIONS.assertRaises(WeeklyOperationsConflict):
        _ = store.append(reversed_pin)


def test_integrity_rehashed_logical_key_is_rejected(tmp_path: Path) -> None:
    store = _store(tmp_path)
    _ = store.append(_operation())
    path = _data(store.authority)
    row = WeeklyOperationRow.model_validate_json(path.read_bytes())
    changed = row.model_copy(update={"logical_key": "f" * 64, "row_digest": "0" * 64})
    changed = changed.model_copy(update={"row_digest": weekly_row_digest(changed)})
    _ = path.write_bytes(canonical_weekly_row(changed, include_digest=True) + b"\n")
    path.chmod(0o600)
    with ASSERTIONS.assertRaises(WeeklyOperationsCorruption):
        _ = store.read()


def test_integrity_data_and_lock_symlinks_fail_without_target_mutation(tmp_path: Path) -> None:
    data_authority, lock_authority = _authority(tmp_path / "data"), _authority(tmp_path / "lock")
    outside = tmp_path / "outside"
    outside.mkdir(mode=0o700)
    data_target, lock_target = outside / "data", outside / "lock"
    _ = data_target.write_bytes(b"data")
    _ = lock_target.write_bytes(b"lock")
    data_target.chmod(0o600)
    lock_target.chmod(0o600)
    data_name = customer_data_name(customer_identity_digest(CUSTOMER))
    lock_name = customer_lock_name(customer_identity_digest(CUSTOMER))
    for authority, name, target in ((data_authority, data_name, data_target), (lock_authority, lock_name, lock_target)):
        (authority.observed_path / name).symlink_to(target)
    with ASSERTIONS.assertRaises(WeeklyOperationsError):
        _ = WeeklyOperationsStore.for_authority(data_authority, CUSTOMER).read()
    with ASSERTIONS.assertRaises(WeeklyOperationsError):
        _ = WeeklyOperationsStore.for_authority(lock_authority, CUSTOMER).read()
    assert (data_target.read_bytes(), lock_target.read_bytes()) == (b"data", b"lock")


def test_integrity_nonregular_data_and_lock_fail_typed(tmp_path: Path) -> None:
    names = customer_data_name(customer_identity_digest(CUSTOMER)), customer_lock_name(customer_identity_digest(CUSTOMER))
    for suffix, name in zip(("data", "lock"), names, strict=True):
        authority = _authority(tmp_path / suffix)
        (authority.observed_path / name).mkdir(mode=0o600)
        with ASSERTIONS.assertRaises(WeeklyOperationsError):
            _ = WeeklyOperationsStore.for_authority(authority, CUSTOMER).read()


def test_integrity_unsafe_data_and_lock_modes_are_preserved(tmp_path: Path) -> None:
    data_store, lock_store = _store(tmp_path / "data"), _store(tmp_path / "lock")
    _ = data_store.append(_operation())
    _ = lock_store.append(_operation())
    data_path = _data(data_store.authority)
    lock_path = _data(lock_store.authority).with_name(customer_lock_name(customer_identity_digest(CUSTOMER)))
    data_path.chmod(0o644)
    lock_path.chmod(0o644)
    with ASSERTIONS.assertRaises(WeeklyOperationsError):
        _ = data_store.read()
    with ASSERTIONS.assertRaises(WeeklyOperationsError):
        _ = lock_store.read()
    assert (data_path.stat().st_mode & 0o777, lock_path.stat().st_mode & 0o777) == (0o644, 0o644)


def _assert_file_replacement(root: Path, *, lock: bool) -> None:
    store = _store(root)
    _ = store.append(_operation())
    target = _data(store.authority).with_name(customer_lock_name(customer_identity_digest(CUSTOMER))) if lock else _data(store.authority)
    original = target.read_bytes()
    acquired, replaced = Barrier(2), Barrier(2)
    real_flock = fcntl.flock

    def gated(descriptor: int | HasFileno, operation: int, /) -> None:
        real_flock(descriptor, operation)
        if operation == fcntl.LOCK_SH:
            _ = acquired.wait(timeout=5)
            _ = replaced.wait(timeout=5)

    with patch("fcntl.flock", side_effect=gated):
        with ThreadPoolExecutor(max_workers=1) as executor:
            result = executor.submit(store.read)
            _ = acquired.wait(timeout=5)
            _ = target.replace(target.with_suffix(".old"))
            _ = target.write_bytes(original)
            target.chmod(0o600)
            _ = replaced.wait(timeout=5)
            with ASSERTIONS.assertRaises(WeeklyOperationsError):
                _ = result.result(timeout=5)


def test_integrity_data_replacement_race_is_detected(tmp_path: Path) -> None:
    _assert_file_replacement(tmp_path, lock=False)


def test_integrity_lock_replacement_race_is_detected(tmp_path: Path) -> None:
    _assert_file_replacement(tmp_path, lock=True)


def test_integrity_hardlinked_lock_is_rejected(tmp_path: Path) -> None:
    authority = _authority(tmp_path)
    customers = authority.observed_path
    outside = tmp_path / "outside-lock"
    _ = outside.write_bytes(b"")
    outside.chmod(0o600)
    os.link(outside, customers / customer_lock_name(customer_identity_digest(CUSTOMER)))
    with ASSERTIONS.assertRaises(WeeklyOperationsError):
        _ = WeeklyOperationsStore.for_authority(authority, CUSTOMER).read()


def _assert_directory_replacement(root: Path) -> None:
    store = _store(root)
    _ = store.append(_operation())
    target = store.authority.observed_path
    acquired, replaced = Barrier(2), Barrier(2)
    real_flock = fcntl.flock

    def gated(descriptor: int | HasFileno, operation: int, /) -> None:
        real_flock(descriptor, operation)
        if operation == fcntl.LOCK_SH:
            _ = acquired.wait(timeout=5)
            _ = replaced.wait(timeout=5)

    with patch("fcntl.flock", side_effect=gated):
        with ThreadPoolExecutor(max_workers=1) as executor:
            result = executor.submit(store.read)
            _ = acquired.wait(timeout=5)
            detached = target.with_name(target.name + "-old")
            _ = target.replace(detached)
            target.mkdir(mode=0o700)
            _ = replaced.wait(timeout=5)
            assert len(result.result(timeout=5)) == 1
            assert tuple(target.iterdir()) == () and (detached / MARKER_NAME).is_file()


def test_integrity_flat_customer_directory_injection_is_detected(tmp_path: Path) -> None:
    authority = _authority(tmp_path)
    injected = authority.observed_path / customer_storage_digest(customer_identity_digest(CUSTOMER))
    injected.mkdir(mode=0o700)
    with ASSERTIONS.assertRaises(WeeklyOperationsError):
        _ = WeeklyOperationsStore.for_authority(authority, CUSTOMER).read()


def test_authority_bootstrap_path_replacement_cannot_redirect_read(tmp_path: Path) -> None:
    _assert_directory_replacement(tmp_path)


def test_integrity_unsafe_authority_and_customer_modes_fail(tmp_path: Path) -> None:
    authority = _authority(tmp_path)
    store = WeeklyOperationsStore.for_authority(authority, CUSTOMER)
    authority.observed_path.chmod(0o755)
    with ASSERTIONS.assertRaises(WeeklyOperationsError):
        _ = store.read()


def test_integrity_unsafe_opaque_customer_mode_is_preserved(tmp_path: Path) -> None:
    authority = _authority(tmp_path)
    lock = authority.observed_path / customer_lock_name(customer_identity_digest(CUSTOMER))
    _ = lock.write_bytes(b"")
    lock.chmod(0o644)
    with ASSERTIONS.assertRaises(WeeklyOperationsError):
        _ = WeeklyOperationsStore.for_authority(authority, CUSTOMER).read()
    assert lock.stat().st_mode & 0o777 == 0o644


def test_integrity_fifo_data_entry_fails_typed_without_blocking(tmp_path: Path) -> None:
    authority = _authority(tmp_path)
    customers = authority.observed_path
    fifo = customers / customer_data_name(customer_identity_digest(CUSTOMER))
    os.mkfifo(fifo, 0o600)
    with ASSERTIONS.assertRaises(WeeklyOperationsError):
        _ = WeeklyOperationsStore.for_authority(authority, CUSTOMER).read()