"""Unnamed-marker and retained repair cancellation boundaries."""

from __future__ import annotations

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

import checkin_cli.weekly_operations_publish as initializer
import checkin_cli.weekly_operations_repair_publish as repair
from checkin_cli.weekly_operations import CanonicalPin, CustomerKey, DayState, WeeklyOperationInput, WeeklyOperationsCorruption, customer_identity_digest
from checkin_cli.weekly_operations_authority import AuthorityId, begin_authority_initialization, issue_repair_authority, open_authority_root
from checkin_cli.weekly_operations_canonical_registry_history import REGISTRY_NAME
from checkin_cli.weekly_operations_layout import MARKER_NAME, customer_data_name
from checkin_cli.weekly_operations_link import link_unnamed_file
from checkin_cli.weekly_operations_parent import acquire_parent_authority
from checkin_cli.weekly_operations_signals import DeferredTerminationSection, InitializationCancellation
from checkin_cli.weekly_operations_store import WeeklyOperationsStore
from tests._weekly_operations_support import initialize_at

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


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 _empty_root(tmp_path: Path) -> Path:
    root = tmp_path / "authority"
    root.mkdir(parents=True, mode=0o700)
    return root


def _repair_setup(tmp_path: Path) -> tuple[Path, Path, WeeklyOperationsStore]:
    root = _empty_root(tmp_path)
    authority = initialize_at(root, AuthorityId("1" * 64))
    store = WeeklyOperationsStore.for_authority(authority, CUSTOMER)
    _ = store.append(_operation())
    data = root / customer_data_name(customer_identity_digest(CUSTOMER))
    with data.open("ab") as handle:
        _ = handle.write(b"torn-tail")
    repair_authority = issue_repair_authority(authority, customer_identity_digest(CUSTOMER))
    return root, data, WeeklyOperationsStore.for_authority(repair_authority, CUSTOMER)


def test_sigterm_immediately_before_unnamed_write_leaves_no_name(tmp_path: Path) -> None:
    root = _empty_root(tmp_path)

    def terminate(descriptor: int, payload: bytes) -> None:
        os.kill(os.getpid(), signal.SIGTERM)
        offset = 0
        while offset < len(payload):
            offset += os.write(descriptor, payload[offset:])

    parent = acquire_parent_authority(root)
    binding = None
    with patch.object(initializer, "_write_all", side_effect=terminate):
        with ASSERTIONS.assertRaises(InitializationCancellation):
            with begin_authority_initialization(parent, AuthorityId("2" * 64)) as transaction:
                binding = transaction.binding
                transaction.acknowledge_binding()
    assert binding is not None
    reopened = open_authority_root(parent, binding)
    reopened.close()


def test_sigterm_immediately_before_unnamed_fsync_leaves_no_name(tmp_path: Path) -> None:
    root = _empty_root(tmp_path)

    real_fsync = os.fsync

    def terminate(descriptor: int) -> None:
        os.kill(os.getpid(), signal.SIGTERM)
        real_fsync(descriptor)

    parent = acquire_parent_authority(root)
    binding = None
    with patch("os.fsync", side_effect=terminate):
        with ASSERTIONS.assertRaises(InitializationCancellation):
            with begin_authority_initialization(parent, AuthorityId("3" * 64)) as transaction:
                binding = transaction.binding
                transaction.acknowledge_binding()
    assert binding is not None
    reopened = open_authority_root(parent, binding)
    reopened.close()


def test_sigterm_immediately_after_real_unnamed_fsync_leaves_no_name(tmp_path: Path) -> None:
    root = _empty_root(tmp_path)
    real_fsync = os.fsync

    def fsync_then_terminate(descriptor: int) -> None:
        real_fsync(descriptor)
        os.kill(os.getpid(), signal.SIGTERM)

    parent = acquire_parent_authority(root)
    binding = None
    with patch("os.fsync", side_effect=fsync_then_terminate):
        with ASSERTIONS.assertRaises(InitializationCancellation):
            with begin_authority_initialization(parent, AuthorityId("4" * 64)) as transaction:
                binding = transaction.binding
                transaction.acknowledge_binding()
    assert binding is not None
    reopened = open_authority_root(parent, binding)
    reopened.close()


def test_sigint_immediately_before_descriptor_link_leaves_no_name(tmp_path: Path) -> None:
    root = _empty_root(tmp_path)

    def interrupt(source: int, directory: int, name: str) -> None:
        os.kill(os.getpid(), signal.SIGINT)
        link_unnamed_file(source, directory, name)

    parent = acquire_parent_authority(root)
    binding = None
    with patch.object(initializer, "link_unnamed_file", side_effect=interrupt):
        with ASSERTIONS.assertRaises(InitializationCancellation):
            with begin_authority_initialization(parent, AuthorityId("5" * 64)) as transaction:
                binding = transaction.binding
                transaction.acknowledge_binding()
    assert binding is not None
    reopened = open_authority_root(parent, binding)
    reopened.close()


def test_sigterm_after_real_descriptor_link_returns_only_exact_committed_binding(tmp_path: Path) -> None:
    root = _empty_root(tmp_path)

    def link_then_terminate(source: int, directory: int, name: str) -> None:
        link_unnamed_file(source, directory, name)
        os.kill(os.getpid(), signal.SIGTERM)

    parent = acquire_parent_authority(root)
    binding = None
    with patch.object(initializer, "link_unnamed_file", side_effect=link_then_terminate):
        with ASSERTIONS.assertRaises(InitializationCancellation):
            with begin_authority_initialization(parent, AuthorityId("6" * 64)) as transaction:
                binding = transaction.binding
                transaction.acknowledge_binding()
    assert binding is not None
    assert binding.marker_inode == (root / MARKER_NAME).stat().st_ino
    reopened = open_authority_root(parent, binding)
    assert {path.name for path in root.iterdir()} == {MARKER_NAME, REGISTRY_NAME}
    reopened.close()


def test_sigterm_after_real_repair_temp_create_before_registration_twice(tmp_path: Path) -> None:
    for attempt in (1, 2):
        root, data, store = _repair_setup(tmp_path / f"repair-{attempt}")
        original = data.read_bytes()
        real_open = os.open

        def create_then_terminate(path: str | bytes, flags: int, mode: int = 0o777, *, dir_fd: int | None = None) -> int:
            descriptor = real_open(path, flags, mode, dir_fd=dir_fd)
            if isinstance(path, str) and path.startswith(".weekly-repair-") and flags & os.O_CREAT:
                os.kill(os.getpid(), signal.SIGTERM)
            return descriptor

        with patch("os.open", side_effect=create_then_terminate):
            with ASSERTIONS.assertRaisesRegex(WeeklyOperationsCorruption, "terminated"):
                _ = store.repair_torn_tail()
        assert data.read_bytes() == original and not tuple(root.glob(".weekly-repair-*.tmp"))


def test_sigterm_immediately_before_repair_critical_section_creates_nothing(tmp_path: Path) -> None:
    root, data, store = _repair_setup(tmp_path)
    original = data.read_bytes()

    def terminate_before(_section: DeferredTerminationSection) -> DeferredTerminationSection:
        os.kill(os.getpid(), signal.SIGTERM)
        raise AssertionError("SIGTERM handler did not interrupt repair")

    with patch.object(DeferredTerminationSection, "__enter__", new=terminate_before):
        with ASSERTIONS.assertRaisesRegex(WeeklyOperationsCorruption, "terminated"):
            _ = store.repair_torn_tail()
    assert data.read_bytes() == original and not tuple(root.glob(".weekly-repair-*.tmp"))


def test_sigterm_during_repair_cleanup_preserves_interruption(tmp_path: Path) -> None:
    root, data, store = _repair_setup(tmp_path)
    original = data.read_bytes()
    real_unlink = os.unlink
    sent = False

    def terminate_cleanup(name: str, *, dir_fd: int | None = None) -> None:
        nonlocal sent
        real_unlink(name, dir_fd=dir_fd)
        if not sent and name.startswith(".weekly-repair-"):
            sent = True
            os.kill(os.getpid(), signal.SIGTERM)

    with patch.object(repair, "_write_all", side_effect=OSError("trigger cleanup")), patch("os.unlink", side_effect=terminate_cleanup):
        with ASSERTIONS.assertRaisesRegex(WeeklyOperationsCorruption, "terminated"):
            _ = store.repair_torn_tail()
    assert data.read_bytes() == original and not tuple(root.glob(".weekly-repair-*.tmp"))


def test_sigterm_immediately_after_repair_critical_section_cleans_registered_inode(tmp_path: Path) -> None:
    root, data, store = _repair_setup(tmp_path)
    original = data.read_bytes()
    real_exit = DeferredTerminationSection.__exit__

    def terminate_after(section: DeferredTerminationSection, kind: type[BaseException] | None, error: BaseException | None, traceback: TracebackType | None) -> None:
        real_exit(section, kind, error, traceback)
        os.kill(os.getpid(), signal.SIGTERM)

    with patch.object(DeferredTerminationSection, "__exit__", new=terminate_after):
        with ASSERTIONS.assertRaisesRegex(WeeklyOperationsCorruption, "terminated"):
            _ = store.repair_torn_tail()
    assert data.read_bytes() == original and not tuple(root.glob(".weekly-repair-*.tmp"))


def test_repair_temp_validation_substitution_preserves_unowned_name(tmp_path: Path) -> None:
    root, data, store = _repair_setup(tmp_path)
    original = data.read_bytes()
    stolen = tmp_path / "retained-created-temp"

    def substitute_then_reject(_info: os.stat_result) -> None:
        temp = next(root.glob(".weekly-repair-*.tmp"))
        _ = temp.replace(stolen)
        temp.touch(mode=0o600)
        raise WeeklyOperationsCorruption("injected temp validation failure")

    with patch.object(repair, "_verify_created_temp", side_effect=substitute_then_reject):
        with ASSERTIONS.assertRaises(WeeklyOperationsCorruption):
            _ = store.repair_torn_tail()
    assert stolen.is_file() and next(root.glob(".weekly-repair-*.tmp")).is_file()
    assert data.read_bytes() == original


def test_repair_signals_are_unmasked_before_ordinary_write(tmp_path: Path) -> None:
    _, _, store = _repair_setup(tmp_path)

    def checked_write(descriptor: int, payload: bytes) -> None:
        current_mask = signal.pthread_sigmask(signal.SIG_BLOCK, frozenset())
        assert signal.SIGTERM not in current_mask and signal.SIGINT not in current_mask
        _ = os.write(descriptor, payload)

    with patch.object(repair, "_write_all", side_effect=checked_write):
        result = store.repair_torn_tail()
    assert result.removed_bytes == len(b"torn-tail")