"""Atomic canonical data replacement for a proven torn-tail repair."""

from __future__ import annotations

import os
import secrets
import signal
import stat
from threading import current_thread, main_thread
from types import FrameType
from typing import Final

from checkin_cli.weekly_operations import WeeklyOperationsAuthorityCompromise, WeeklyOperationsCorruption
from checkin_cli.weekly_operations_signals import DeferredTerminationSection

_FILE_FLAGS: Final = os.O_NOFOLLOW | os.O_CLOEXEC | os.O_NONBLOCK
FileIdentity = tuple[int, int]


def _identity(info: os.stat_result) -> FileIdentity:
    return info.st_dev, info.st_ino


def _read_all(descriptor: int) -> bytes:
    _ = os.lseek(descriptor, 0, os.SEEK_SET)
    chunks: list[bytes] = []
    while chunk := os.read(descriptor, 65536):
        chunks.append(chunk)
    return b"".join(chunks)


def _verify_created_temp(info: os.stat_result) -> None:
    safe = stat.S_ISREG(info.st_mode) and stat.S_IMODE(info.st_mode) == 0o600 and info.st_uid == os.geteuid() and info.st_nlink == 1
    if not safe:
        raise WeeklyOperationsAuthorityCompromise("repair created temp is unsafe")


def _verify_named(customers: int, name: str, expected: FileIdentity, reason: str) -> None:
    try:
        named = os.stat(name, dir_fd=customers, follow_symlinks=False)
    except OSError as error:
        raise WeeklyOperationsAuthorityCompromise(reason) from error
    if not stat.S_ISREG(named.st_mode) or _identity(named) != expected:
        raise WeeklyOperationsAuthorityCompromise(reason)


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


def _reconcile(customers: int, data_name: str, expected: FileIdentity, prefix: bytes) -> None:
    try:
        descriptor = os.open(data_name, os.O_RDONLY | _FILE_FLAGS, dir_fd=customers)
    except OSError as error:
        raise WeeklyOperationsAuthorityCompromise("repaired canonical data is missing") from error
    try:
        info = os.fstat(descriptor)
        payload = _read_all(descriptor)
        _verify_named(customers, data_name, expected, "repaired canonical data identity changed")
        if _identity(info) != expected or not stat.S_ISREG(info.st_mode) or stat.S_IMODE(info.st_mode) != 0o600 or info.st_uid != os.geteuid() or info.st_nlink != 1 or payload != prefix:
            raise WeeklyOperationsAuthorityCompromise("repaired canonical data reconciliation failed")
    finally:
        os.close(descriptor)


def _cleanup_temp(customers: int, name: str, expected: FileIdentity) -> None:
    _verify_named(customers, name, expected, "repair temp identity changed before cleanup")
    os.unlink(name, dir_fd=customers)


class _RepairTerminated(WeeklyOperationsCorruption):
    pass


def _terminate(_signal_number: int, _frame: FrameType | None) -> None:
    raise _RepairTerminated("repair publication terminated")


def _cleanup_after_interruption(customers: int, name: str, expected: FileIdentity) -> None:
    try:
        _cleanup_temp(customers, name, expected)
    except (_RepairTerminated, KeyboardInterrupt, SystemExit) as interruption:
        try:
            _ = os.stat(name, dir_fd=customers, follow_symlinks=False)
        except FileNotFoundError:
            raise interruption
        _cleanup_temp(customers, name, expected)
        raise interruption


def publish_repaired_prefix(customers: int, data_name: str, prefix: bytes, original: FileIdentity) -> None:
    """Atomically replace the canonical name without truncating its old inode."""
    temp_name = f".weekly-repair-{secrets.token_hex(12)}.tmp"
    temp: int | None = None
    observed_temp_identity: FileIdentity | None = None
    temp_identity: FileIdentity | None = None
    attempted = False
    install_handler = current_thread() is main_thread()
    previous_handler = signal.getsignal(signal.SIGTERM)
    if install_handler:
        _ = signal.signal(signal.SIGTERM, _terminate)
    try:
        _verify_named(customers, data_name, original, "repair data name changed before staging")
        with DeferredTerminationSection():
            temp = os.open(temp_name, os.O_WRONLY | os.O_CREAT | os.O_EXCL | _FILE_FLAGS, 0o600, dir_fd=customers)
            temp_info = os.fstat(temp)
            observed_temp_identity = _identity(temp_info)
            _verify_created_temp(temp_info)
            _verify_named(customers, temp_name, observed_temp_identity, "repair created temp identity changed")
            temp_identity = observed_temp_identity
        _write_all(temp, prefix)
        os.fsync(temp)
        _verify_named(customers, temp_name, temp_identity, "repair temp identity changed before publication")
        _verify_named(customers, data_name, original, "repair data name changed before publication")
        attempted = True
        try:
            os.replace(temp_name, data_name, src_dir_fd=customers, dst_dir_fd=customers)
        except (OSError, WeeklyOperationsCorruption, KeyboardInterrupt, SystemExit):
            try:
                _reconcile(customers, data_name, temp_identity, prefix)
            except (OSError, WeeklyOperationsCorruption):
                raise
        _reconcile(customers, data_name, temp_identity, prefix)
        os.fsync(customers)
        os.close(temp)
        temp = None
    except (OSError, WeeklyOperationsCorruption, KeyboardInterrupt, SystemExit) as error:
        cleanup_identity = temp_identity or observed_temp_identity
        if temp is not None and cleanup_identity is not None:
            try:
                _reconcile(customers, data_name, cleanup_identity, prefix)
            except (OSError, WeeklyOperationsCorruption):
                try:
                    _cleanup_after_interruption(customers, temp_name, cleanup_identity)
                finally:
                    os.close(temp)
                    temp = None
                if attempted:
                    raise WeeklyOperationsAuthorityCompromise("repair publication namespace changed") from error
                raise error
            os.fsync(customers)
            os.close(temp)
            temp = None
            return
        raise
    finally:
        if install_handler:
            _ = signal.signal(signal.SIGTERM, previous_handler)