"""Descriptor-bound coordinator authority for weekly generation storage."""

from __future__ import annotations

import hashlib
import json
import os
import stat
from dataclasses import dataclass
from pathlib import Path
from typing import Final, override

_DIRECTORY_FLAGS: Final = os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW | os.O_CLOEXEC


class WeeklyOwnerStorageError(RuntimeError):
    def __init__(self, reason: str) -> None:
        super().__init__(reason)
        self.reason: str = reason

    @override
    def __str__(self) -> str:
        return f"weekly owner storage authority is invalid: {self.reason}"


@dataclass(frozen=True, slots=True)
class WeeklyOwnerStorageAuthority:
    """Pinned owner-actions directory supplied by the canonical coordinator."""

    descriptor: int
    acquisition_path: Path
    device: int
    inode: int
    mode: int
    owner: int
    links: int
    binding_digest: str

    def verify(self) -> None:
        try:
            opened = os.fstat(self.descriptor)
            named = self.acquisition_path.stat(follow_symlinks=False)
        except OSError as error:
            raise WeeklyOwnerStorageError("unavailable") from error
        observed = opened.st_dev, opened.st_ino, stat.S_IMODE(opened.st_mode), opened.st_uid
        expected = self.device, self.inode, self.mode, self.owner
        named_identity = named.st_dev, named.st_ino
        if (
            not stat.S_ISDIR(opened.st_mode)
            or stat.S_ISLNK(named.st_mode)
            or observed != expected
            or named_identity != (self.device, self.inode)
            or self.binding_digest != _binding_digest((*expected, self.links))
        ):
            raise WeeklyOwnerStorageError("descriptor or inode drift")

    def open_file(self, name: str, flags: int, mode: int = 0o600) -> int:
        if "/" in name or name in {"", ".", ".."}:
            raise WeeklyOwnerStorageError("child name")
        self.verify()
        try:
            descriptor = os.open(
                name, flags | os.O_NOFOLLOW | os.O_CLOEXEC,
                mode, dir_fd=self.descriptor,
            )
            opened = os.fstat(descriptor)
            named = os.stat(name, dir_fd=self.descriptor, follow_symlinks=False)
        except OSError as error:
            raise WeeklyOwnerStorageError("child open") from error
        if (
            not stat.S_ISREG(opened.st_mode)
            or (opened.st_dev, opened.st_ino) != (named.st_dev, named.st_ino)
            or opened.st_uid != os.geteuid()
            or opened.st_nlink != 1
        ):
            os.close(descriptor)
            raise WeeklyOwnerStorageError("child inode")
        self.verify()
        return descriptor

    def verify_child(self, name: str, descriptor: int) -> None:
        """Fence a retained child descriptor against its coordinator-owned name."""
        self.verify()
        try:
            opened = os.fstat(descriptor)
            named = os.stat(name, dir_fd=self.descriptor, follow_symlinks=False)
        except OSError as error:
            raise WeeklyOwnerStorageError("child verification") from error
        if (
            not stat.S_ISREG(opened.st_mode)
            or (opened.st_dev, opened.st_ino) != (named.st_dev, named.st_ino)
            or stat.S_IMODE(opened.st_mode) != 0o600
            or opened.st_uid != os.geteuid()
            or opened.st_nlink != 1
        ):
            raise WeeklyOwnerStorageError("child drift")


def bind_weekly_owner_storage(path: Path) -> WeeklyOwnerStorageAuthority:
    """Create or acquire the coordinator-owned private storage directory."""
    absolute = path.absolute()
    absolute.mkdir(parents=True, exist_ok=True, mode=0o700)
    absolute.chmod(0o700)
    try:
        descriptor = os.open(absolute, _DIRECTORY_FLAGS)
        info = os.fstat(descriptor)
        named = absolute.stat(follow_symlinks=False)
    except OSError as error:
        raise WeeklyOwnerStorageError("acquisition") from error
    values = info.st_dev, info.st_ino, stat.S_IMODE(info.st_mode), info.st_uid, info.st_nlink
    if (
        not stat.S_ISDIR(info.st_mode)
        or stat.S_ISLNK(named.st_mode)
        or (info.st_dev, info.st_ino) != (named.st_dev, named.st_ino)
        or info.st_uid != os.geteuid()
        or stat.S_IMODE(info.st_mode) != 0o700
    ):
        os.close(descriptor)
        raise WeeklyOwnerStorageError("directory inode")
    return WeeklyOwnerStorageAuthority(
        descriptor, absolute, *values, _binding_digest(values),
    )


def _binding_digest(values: tuple[int, int, int, int, int]) -> str:
    encoded = json.dumps(("weekly-owner-storage-v1", values), separators=(",", ":")).encode()
    return hashlib.sha256(encoded).hexdigest()