"""Strict issuance and consumption of selected torn-tail repair authority."""

from __future__ import annotations

import os
from datetime import date, datetime
from pathlib import Path
from unittest import TestCase
from unittest.mock import patch

from checkin_cli.weekly_operations import CanonicalPin, CustomerKey, DayState, WeeklyOperationInput, WeeklyOperationsCorruption, WeeklyOperationsInputError, canonical_weekly_row, customer_identity_digest, weekly_row_digest
from checkin_cli.weekly_operations_authority import AuthorityId, WeeklyOperationsAuthorityRoot, issue_repair_authority, open_authority_root_for_repair
from checkin_cli.weekly_operations_layout import customer_data_name, customer_lock_name
from tests._weekly_operations_support import initialize_at
from checkin_cli.weekly_operations_store import WeeklyOperationsStore

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


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(tmp_path: Path) -> tuple[Path, WeeklyOperationsAuthorityRoot]:
    path = tmp_path / "authority"
    path.mkdir(parents=True, mode=0o700)
    authority = initialize_at(path, AuthorityId("5" * 64))
    return path, authority


def _snapshot(path: Path) -> tuple[tuple[str, bytes], ...]:
    return tuple(sorted((str(item.relative_to(path)), item.read_bytes()) for item in path.rglob("*") if item.is_file()))


def _data(path: Path) -> Path:
    return path / customer_data_name(customer_identity_digest(CUSTOMER))


def _lock(path: Path) -> Path:
    return path / customer_lock_name(customer_identity_digest(CUSTOMER))


def _make_torn(path: Path, authority: WeeklyOperationsAuthorityRoot) -> WeeklyOperationsStore:
    store = WeeklyOperationsStore.for_authority(authority, CUSTOMER)
    _ = store.append(_operation())
    with _data(path).open("ab") as handle:
        _ = handle.write(b"torn-tail")
    return store


def test_r8_red_healthy_history_cannot_issue_repair_authority(tmp_path: Path) -> None:
    path, authority = _authority(tmp_path)
    _ = WeeklyOperationsStore.for_authority(authority, CUSTOMER).append(_operation())
    binding = authority.binding
    authority.close()
    before = _snapshot(path)
    with ASSERTIONS.assertRaises(WeeklyOperationsCorruption):
        _ = open_authority_root_for_repair(authority.parent, binding, customer_identity_digest(CUSTOMER))
    assert _snapshot(path) == before


def test_r8_red_absent_history_cannot_issue_or_create_repair_files(tmp_path: Path) -> None:
    path, authority = _authority(tmp_path)
    binding = authority.binding
    authority.close()
    before = _snapshot(path)
    with ASSERTIONS.assertRaises(WeeklyOperationsCorruption):
        _ = open_authority_root_for_repair(authority.parent, binding, customer_identity_digest(CUSTOMER))
    assert _snapshot(path) == before


def test_empty_or_zero_prefix_fragment_cannot_issue_repair(tmp_path: Path) -> None:
    for index, payload in enumerate((b"", b"only-an-incomplete-fragment")):
        path, authority = _authority(tmp_path / str(index))
        _ = _data(path).write_bytes(payload)
        _ = _lock(path).write_bytes(b"")
        _data(path).chmod(0o600)
        _lock(path).chmod(0o600)
        before = _snapshot(path)
        with ASSERTIONS.assertRaises(WeeklyOperationsCorruption):
            _ = issue_repair_authority(authority, customer_identity_digest(CUSTOMER))
        assert _snapshot(path) == before


def test_interior_corruption_and_forged_transition_cannot_issue_repair(tmp_path: Path) -> None:
    for index, forged_transition in enumerate((False, True)):
        path, authority = _authority(tmp_path / str(index))
        store = WeeklyOperationsStore.for_authority(authority, CUSTOMER)
        row = store.append(_operation()).row
        if forged_transition:
            changed = row.model_copy(update={"state": DayState.LATE_SUBMITTED, "row_digest": "0" * 64})
            changed = changed.model_copy(update={"row_digest": weekly_row_digest(changed)})
            payload = canonical_weekly_row(changed, include_digest=True) + b"\n"
        else:
            payload = _data(path).read_bytes().replace(b'"state":"submitted"', b'"state":"forgedxxx"')
        _ = _data(path).write_bytes(payload)
        _data(path).chmod(0o600)
        before = _snapshot(path)
        with ASSERTIONS.assertRaises(WeeklyOperationsCorruption):
            _ = issue_repair_authority(authority, customer_identity_digest(CUSTOMER))
        assert _snapshot(path) == before


def test_torn_target_without_existing_lock_cannot_issue_or_create_lock(tmp_path: Path) -> None:
    path, authority = _authority(tmp_path)
    _ = _make_torn(path, authority)
    _lock(path).unlink()
    before = _snapshot(path)
    with ASSERTIONS.assertRaises(WeeklyOperationsCorruption):
        _ = issue_repair_authority(authority, customer_identity_digest(CUSTOMER))
    assert _snapshot(path) == before and not _lock(path).exists()


def test_changed_content_after_issuance_fails_byte_identically(tmp_path: Path) -> None:
    path, authority = _authority(tmp_path)
    _ = _make_torn(path, authority)
    repair_authority = issue_repair_authority(authority, customer_identity_digest(CUSTOMER))
    with _data(path).open("ab") as handle:
        _ = handle.write(b"changed")
    before = _snapshot(path)
    repair_store = WeeklyOperationsStore.for_authority(repair_authority, CUSTOMER)
    with ASSERTIONS.assertRaises(WeeklyOperationsCorruption):
        _ = repair_store.repair_torn_tail()
    assert _snapshot(path) == before


def test_changed_data_inode_after_issuance_fails_byte_identically(tmp_path: Path) -> None:
    path, authority = _authority(tmp_path)
    _ = _make_torn(path, authority)
    repair_authority = issue_repair_authority(authority, customer_identity_digest(CUSTOMER))
    payload = _data(path).read_bytes()
    _ = _data(path).replace(tmp_path / "old-data")
    _ = _data(path).write_bytes(payload)
    _data(path).chmod(0o600)
    before = _snapshot(path)
    repair_store = WeeklyOperationsStore.for_authority(repair_authority, CUSTOMER)
    with ASSERTIONS.assertRaises(WeeklyOperationsCorruption):
        _ = repair_store.repair_torn_tail()
    assert _snapshot(path) == before


def test_lock_removed_after_issuance_is_not_recreated(tmp_path: Path) -> None:
    path, authority = _authority(tmp_path)
    _ = _make_torn(path, authority)
    repair_authority = issue_repair_authority(authority, customer_identity_digest(CUSTOMER))
    _lock(path).unlink()
    before = _snapshot(path)
    repair_store = WeeklyOperationsStore.for_authority(repair_authority, CUSTOMER)
    with ASSERTIONS.assertRaises(WeeklyOperationsCorruption):
        _ = repair_store.repair_torn_tail()
    assert _snapshot(path) == before and not _lock(path).exists()


def test_normal_authority_cannot_repair_or_create_side_effects(tmp_path: Path) -> None:
    path, authority = _authority(tmp_path)
    store = _make_torn(path, authority)
    before = _snapshot(path)
    with ASSERTIONS.assertRaisesRegex(WeeklyOperationsInputError, "repair authority required"):
        _ = store.repair_torn_tail()
    assert _snapshot(path) == before


def test_changed_lock_inode_after_issuance_fails_byte_identically(tmp_path: Path) -> None:
    path, authority = _authority(tmp_path)
    _ = _make_torn(path, authority)
    repair_authority = issue_repair_authority(authority, customer_identity_digest(CUSTOMER))
    _ = _lock(path).replace(tmp_path / "old-lock")
    _ = _lock(path).write_bytes(b"")
    _lock(path).chmod(0o600)
    before = _snapshot(path)
    repair_store = WeeklyOperationsStore.for_authority(repair_authority, CUSTOMER)
    with ASSERTIONS.assertRaises(WeeklyOperationsCorruption):
        _ = repair_store.repair_torn_tail()
    assert _snapshot(path) == before


def test_r9_red_data_name_substitution_never_truncates_detached_inode(tmp_path: Path) -> None:
    path, authority = _authority(tmp_path)
    _ = _make_torn(path, authority)
    repair_authority = issue_repair_authority(authority, customer_identity_digest(CUSTOMER))
    repair_store = WeeklyOperationsStore.for_authority(repair_authority, CUSTOMER)
    original = _data(path).read_bytes()
    detached = tmp_path / "detached-original-data"
    repair_binding = repair_authority.repair_binding
    assert repair_binding is not None
    real_replace = os.replace

    def substitute_then_replace(source: str, destination: str, *, src_dir_fd: int | None = None, dst_dir_fd: int | None = None) -> None:
        real_replace(_data(path), detached)
        _ = _data(path).write_bytes(original)
        _data(path).chmod(0o600)
        real_replace(source, destination, src_dir_fd=src_dir_fd, dst_dir_fd=dst_dir_fd)

    with patch("os.replace", side_effect=substitute_then_replace):
        result = repair_store.repair_torn_tail()
    assert detached.read_bytes() == original
    assert _data(path).read_bytes() == original[: repair_binding.valid_prefix_offset]
    assert result.removed_bytes == repair_binding.torn_tail_length


def test_r9_red_same_inode_lock_byte_mutation_blocks_repair(tmp_path: Path) -> None:
    path, authority = _authority(tmp_path)
    _ = _make_torn(path, authority)
    repair_authority = issue_repair_authority(authority, customer_identity_digest(CUSTOMER))
    repair_store = WeeklyOperationsStore.for_authority(repair_authority, CUSTOMER)
    data_before = _data(path).read_bytes()
    lock_inode = _lock(path).stat().st_ino
    _ = _lock(path).write_bytes(b"forged-lock-bytes")
    _lock(path).chmod(0o600)
    assert _lock(path).stat().st_ino == lock_inode
    with ASSERTIONS.assertRaises(WeeklyOperationsCorruption):
        _ = repair_store.repair_torn_tail()
    assert _data(path).read_bytes() == data_before