"""Zero-inotify, event-driven receipts for authoritative Linux file commits."""
from __future__ import annotations

import ctypes
import errno
import fcntl
import hashlib
import os
import select
import signal
import stat
import struct
import sys
import threading
import time
from pathlib import Path

from gateway._commit_receipts import CommitObservationError, CommitReceiptLog

# Kept as the public commit-interest contract used by existing callers.  The
# dnotify backend maps it to directory mutation notifications below.
IN_CLOSE_WRITE = 0x08
IN_MOVED_TO = 0x80
IN_CREATE = 0x100
COMMIT_MASK = IN_CLOSE_WRITE | IN_MOVED_TO | IN_CREATE
BACKEND = "dnotify_signalfd_v1"
_DNOTIFY_MASK = (
    fcntl.DN_MODIFY
    | fcntl.DN_CREATE
    | fcntl.DN_DELETE
    | fcntl.DN_RENAME
    | fcntl.DN_ATTRIB
    | fcntl.DN_MULTISHOT
)
_F_SETOWN_EX = 15
_F_OWNER_TID = 0
_SIGNAL = signal.SIGRTMIN + 7
_SIGNALFD_RECORD_SIZE = 128
_SIGNALFD_PREFIX = struct.Struct("IiiIIiIIIIii")
_OWNER = struct.Struct("ii")
_EMPTY_SHA256 = hashlib.sha256(b"").hexdigest()
_LIBC = ctypes.CDLL(None, use_errno=True)


class _SigSet(ctypes.Structure):
    _fields_ = [("bits", ctypes.c_ulong * 16)]


_LIBC.sigemptyset.argtypes = [ctypes.POINTER(_SigSet)]
_LIBC.sigemptyset.restype = ctypes.c_int
_LIBC.sigaddset.argtypes = [ctypes.POINTER(_SigSet), ctypes.c_int]
_LIBC.sigaddset.restype = ctypes.c_int
_LIBC.signalfd.argtypes = [ctypes.c_int, ctypes.POINTER(_SigSet), ctypes.c_int]
_LIBC.signalfd.restype = ctypes.c_int


class _BackendSetupError(OSError):
    def __init__(self, code: str, error: OSError) -> None:
        super().__init__(error.errno, error.strerror)
        self.code = code


class ProductionCommitObserver:
    """Observe authoritative parent directories through dnotify and signalfd."""

    def __init__(self, authoritative_roots: tuple[Path, ...], receipt_path: Path) -> None:
        if not authoritative_roots:
            raise CommitObservationError("path_outside_roots")
        self._roots = tuple(self._normalize(path) for path in authoritative_roots)
        self._receipt_path = self._normalize(receipt_path)
        self._receipts = CommitReceiptLog(self._receipt_path)
        self._signal_fd = -1
        self._directory_fds: dict[int, Path] = {}
        self._paths_by_parent: dict[Path, tuple[Path, ...]] = {}
        self._old_signal_mask: set[int] | None = None
        self._owner_thread = -1
        self._armed = False
        self._watcher_id = ""
        self._cursor = -1
        self._states: dict[Path, tuple[str, str, int]] = {}

    def __enter__(self) -> ProductionCommitObserver:
        return self

    def __exit__(self, *_args: object) -> None:
        self.close()

    def close(self) -> None:
        for descriptor in tuple(self._directory_fds):
            try:
                fcntl.fcntl(descriptor, fcntl.F_NOTIFY, 0)
            except OSError:
                pass
            try:
                os.close(descriptor)
            except OSError:
                pass
        self._directory_fds.clear()
        self._paths_by_parent.clear()
        if self._signal_fd >= 0:
            try:
                while True:
                    try:
                        if not os.read(self._signal_fd, _SIGNALFD_RECORD_SIZE * 256):
                            break
                    except BlockingIOError:
                        break
                os.close(self._signal_fd)
            finally:
                self._signal_fd = -1
        if self._old_signal_mask is not None:
            signal.pthread_sigmask(signal.SIG_SETMASK, self._old_signal_mask)
            self._old_signal_mask = None
        self._owner_thread = -1
        self._armed = False

    def arm(
        self,
        *,
        candidate_id: str,
        profile_id: str,
        watched_paths: tuple[Path, ...],
        watcher_id: str,
        mask: int,
        starting_cursor: int,
        update_offset: int,
        armed_at: str,
    ) -> dict[str, object]:
        if self._armed or not all((candidate_id, profile_id, watcher_id, armed_at)):
            raise CommitObservationError("duplicate_watcher_identity")
        if type(starting_cursor) is not int or starting_cursor < 0:
            raise CommitObservationError("sequence_not_forward")
        if type(update_offset) is not int or update_offset < 0:
            raise CommitObservationError("invalid_subscription")
        if type(mask) is not int or mask <= 0 or not watched_paths:
            raise CommitObservationError("invalid_subscription")
        if sys.platform != "linux" or _SIGNAL > signal.SIGRTMAX:
            raise CommitObservationError("backend_unavailable")
        for root in self._roots:
            self._validate_root(root)
        trusted_paths: set[Path] = {self._trusted(path) for path in watched_paths}
        paths: tuple[Path, ...] = tuple(
            sorted(trusted_paths, key=lambda path: path.as_posix())
        )
        self._prepare_receipt_parent()
        try:
            self._setup_backend(paths)
            # Subscription is live before this snapshot.  A commit racing the
            # snapshot is either represented here or remains queued in signalfd.
            self._states = {path: self._state(path) for path in paths}
            initial = [
                {"path": str(path), "byte_hash": value[0], "inode_hash": value[1]}
                for path, value in self._states.items()
            ]
            resources = self._resource_receipt()
            record = self._receipts.append(
                {
                    "kind": "subscription_armed",
                    "candidate_id": candidate_id,
                    "profile_id": profile_id,
                    "watched_paths": [str(path) for path in paths],
                    "watcher_id": watcher_id,
                    "mask": mask,
                    "starting_cursor": starting_cursor,
                    "update_offset": update_offset,
                    "initial_states": initial,
                    "armed_at": armed_at,
                    **resources,
                },
                reject_watcher=watcher_id,
            )
        except _BackendSetupError as exc:
            try:
                self._receipts.append(
                    {
                        "kind": "subscription_error",
                        "candidate_id": candidate_id,
                        "profile_id": profile_id,
                        "watched_paths": [str(path) for path in paths],
                        "watcher_id": watcher_id,
                        "backend": BACKEND,
                        "errno": exc.errno,
                        "error": exc.code,
                    },
                    reject_watcher=watcher_id,
                )
            finally:
                self.close()
            raise CommitObservationError(exc.code) from exc
        except BaseException:
            self.close()
            raise
        self._watcher_id = watcher_id
        self._cursor = starting_cursor
        self._armed = True
        return record

    def observe(self, *, sequence: int, timeout: float) -> dict[str, object]:
        if not self._armed:
            raise CommitObservationError("subscription_not_armed")
        if threading.get_native_id() != self._owner_thread:
            raise CommitObservationError("observer_thread_changed")
        if type(sequence) is not int or sequence <= self._cursor:
            raise CommitObservationError("sequence_not_forward")
        if sequence != self._cursor + 1:
            raise CommitObservationError("sequence_not_contiguous")
        if timeout < 0:
            raise ValueError("timeout must be non-negative")
        deadline = time.monotonic() + timeout
        candidates = self._states
        while True:
            changed = self._changed(candidates)
            if changed:
                path, value = changed[0]
                record = self._receipts.append(
                    {
                        "kind": "commit_observed",
                        "watcher_id": self._watcher_id,
                        "path": str(path),
                        "inode": value[2],
                        "sequence": sequence,
                        "commit_byte_hash": value[0],
                        **self._resource_receipt(),
                    }
                )
                self._states[path] = value
                self._cursor = sequence
                return record
            remaining = deadline - time.monotonic()
            if remaining <= 0:
                raise CommitObservationError("event_timeout")
            parents = self._next_kernel_parents(remaining)
            candidates = {
                path: self._states[path]
                for parent in parents
                for path in self._paths_by_parent[parent]
            }

    def _setup_backend(self, paths: tuple[Path, ...]) -> None:
        current_mask = signal.pthread_sigmask(signal.SIG_BLOCK, set())
        if _SIGNAL in current_mask:
            raise _BackendSetupError(
                "signal_unavailable", OSError(errno.EBUSY, os.strerror(errno.EBUSY))
            )
        try:
            self._old_signal_mask = signal.pthread_sigmask(signal.SIG_BLOCK, {_SIGNAL})
        except OSError as exc:
            raise _BackendSetupError("signal_mask_failed", exc) from exc
        self._owner_thread = threading.get_native_id()
        try:
            self._signal_fd = self._create_signalfd()
        except OSError as exc:
            raise _BackendSetupError("signalfd_setup_failed", exc) from exc
        grouped: dict[Path, list[Path]] = {}
        for path in paths:
            grouped.setdefault(path.parent, []).append(path)
        for parent, children in sorted(
            grouped.items(), key=lambda item: item[0].as_posix()
        ):
            try:
                descriptor = os.open(
                    parent,
                    os.O_RDONLY | os.O_DIRECTORY | os.O_CLOEXEC | os.O_NOFOLLOW,
                )
            except OSError as exc:
                raise _BackendSetupError("directory_open_failed", exc) from exc
            self._directory_fds[descriptor] = parent
            self._paths_by_parent[parent] = tuple(
                sorted(children, key=lambda path: path.as_posix())
            )
            try:
                fcntl.fcntl(
                    descriptor,
                    _F_SETOWN_EX,
                    _OWNER.pack(_F_OWNER_TID, self._owner_thread),
                )
                fcntl.fcntl(descriptor, fcntl.F_SETSIG, _SIGNAL)
                fcntl.fcntl(descriptor, fcntl.F_NOTIFY, _DNOTIFY_MASK)
            except OSError as exc:
                raise _BackendSetupError("dnotify_setup_failed", exc) from exc

    def _create_signalfd(self) -> int:
        mask = _SigSet()
        if _LIBC.sigemptyset(ctypes.byref(mask)) != 0:
            error = ctypes.get_errno()
            raise OSError(error, os.strerror(error))
        if _LIBC.sigaddset(ctypes.byref(mask), _SIGNAL) != 0:
            error = ctypes.get_errno()
            raise OSError(error, os.strerror(error))
        descriptor = _LIBC.signalfd(
            -1, ctypes.byref(mask), os.O_CLOEXEC | os.O_NONBLOCK
        )
        if descriptor < 0:
            error = ctypes.get_errno()
            raise OSError(error, os.strerror(error))
        return descriptor

    def _resource_receipt(self) -> dict[str, object]:
        return {
            "backend": BACKEND,
            "directory_resource_count": len(self._directory_fds),
            "signal_fd_count": 1 if self._signal_fd >= 0 else 0,
            "inotify_watch_count": 0,
        }

    def _next_kernel_parents(self, timeout: float) -> tuple[Path, ...]:
        self._wait_signal(timeout)
        descriptors = self._read_notifications()
        parents = {self._directory_fds[descriptor] for descriptor in descriptors}
        return tuple(sorted(parents, key=lambda path: path.as_posix()))

    def _wait_signal(self, timeout: float) -> None:
        ready, _, exceptional = select.select(
            (self._signal_fd,), (), (self._signal_fd,), timeout
        )
        if exceptional:
            raise CommitObservationError("notification_lost")
        if not ready:
            raise CommitObservationError("event_timeout")

    def _read_notifications(self) -> tuple[int, ...]:
        try:
            data = os.read(self._signal_fd, _SIGNALFD_RECORD_SIZE * 256)
        except BlockingIOError as exc:
            raise CommitObservationError("notification_malformed") from exc
        if not data or len(data) % _SIGNALFD_RECORD_SIZE:
            raise CommitObservationError("notification_malformed")
        descriptors: set[int] = set()
        for offset in range(0, len(data), _SIGNALFD_RECORD_SIZE):
            values = _SIGNALFD_PREFIX.unpack_from(data, offset)
            signo, descriptor, overrun = values[0], values[5], values[8]
            if signo != _SIGNAL or descriptor not in self._directory_fds:
                raise CommitObservationError("notification_malformed")
            if overrun:
                raise CommitObservationError("notification_lost")
            descriptors.add(descriptor)
        if not descriptors:
            raise CommitObservationError("notification_malformed")
        return tuple(sorted(descriptors))

    def _changed(
        self, prior: dict[Path, tuple[str, str, int]]
    ) -> list[tuple[Path, tuple[str, str, int]]]:
        changed = []
        for path, old in prior.items():
            try:
                current = self._state(path)
            except CommitObservationError as exc:
                if str(exc) != "state_not_regular":
                    raise
                current = self._state(path)
            if current != old and current[0] != _EMPTY_SHA256:
                changed.append((path, current))
        return changed

    def _state(self, path: Path) -> tuple[str, str, int]:
        self._no_symlinks(path)
        try:
            descriptor = os.open(path, os.O_RDONLY | os.O_CLOEXEC | os.O_NOFOLLOW)
        except OSError as exc:
            raise CommitObservationError("state_unavailable") from exc
        try:
            metadata = os.fstat(descriptor)
            if not stat.S_ISREG(metadata.st_mode) or metadata.st_nlink != 1:
                raise CommitObservationError("state_not_regular")
            if stat.S_IMODE(metadata.st_mode) & 0o077:
                raise CommitObservationError("state_not_private")
            digest = hashlib.sha256()
            while chunk := os.read(descriptor, 65536):
                digest.update(chunk)
        finally:
            os.close(descriptor)
        inode_hash = hashlib.sha256(
            f"{metadata.st_dev}:{metadata.st_ino}".encode()
        ).hexdigest()
        return digest.hexdigest(), inode_hash, metadata.st_ino

    def _trusted(self, path: Path) -> Path:
        normalized = self._normalize(path)
        if not any(self._within(normalized, root) for root in self._roots):
            raise CommitObservationError("path_outside_roots")
        self._no_symlinks(normalized)
        return normalized

    def _prepare_receipt_parent(self) -> None:
        root = next(
            (root for root in self._roots if self._within(self._receipt_path, root)),
            None,
        )
        if root is None:
            raise CommitObservationError("path_outside_roots")
        current = root
        for part in self._receipt_path.parent.relative_to(root).parts:
            current /= part
            try:
                current.mkdir(mode=0o700)
            except FileExistsError:
                pass
            self._validate_directory(current)
        if os.path.lexists(self._receipt_path):
            self._no_symlinks(self._receipt_path)

    def _no_symlinks(self, path: Path) -> None:
        root = next((root for root in self._roots if self._within(path, root)), None)
        if root is None:
            raise CommitObservationError("path_outside_roots")
        current = root
        for part in ("", *path.relative_to(root).parts):
            current = current if not part else current / part
            try:
                if stat.S_ISLNK(os.lstat(current).st_mode):
                    raise CommitObservationError("symlink_rejected")
            except FileNotFoundError:
                return

    def _validate_root(self, path: Path) -> None:
        current = Path(path.anchor)
        for part in path.parts[1:]:
            current /= part
            try:
                if stat.S_ISLNK(os.lstat(current).st_mode):
                    raise CommitObservationError("symlink_rejected")
            except OSError as exc:
                raise CommitObservationError("state_unavailable") from exc
        self._validate_directory(path)

    @staticmethod
    def _validate_directory(path: Path) -> None:
        try:
            metadata = os.lstat(path)
        except OSError as exc:
            raise CommitObservationError("state_unavailable") from exc
        if stat.S_ISLNK(metadata.st_mode):
            raise CommitObservationError("symlink_rejected")
        if not stat.S_ISDIR(metadata.st_mode) or stat.S_IMODE(metadata.st_mode) & 0o077:
            raise CommitObservationError("state_not_private")

    @staticmethod
    def _normalize(path: Path) -> Path:
        return Path(os.path.abspath(os.path.normpath(os.fspath(path))))

    @staticmethod
    def _within(path: Path, root: Path) -> bool:
        try:
            return os.path.commonpath((path, root)) == str(root)
        except ValueError:
            return False