"""Owner-only filesystem primitives for nutrition onboarding."""

from __future__ import annotations

import json
import os
import stat
import tempfile
from pathlib import Path


def validate_profile_path(path: Path, profile_root: Path) -> None:
    root = Path(profile_root).absolute()
    target = Path(path).absolute()
    try:
        root_info = root.lstat()
    except FileNotFoundError:
        root_info = None
    if root_info is not None:
        if stat.S_ISLNK(root_info.st_mode):
            raise ValueError("onboarding profile root must not be a symlink")
        if not stat.S_ISDIR(root_info.st_mode) or root_info.st_uid != os.getuid():
            raise ValueError("onboarding profile root must be owner-controlled")
    try:
        relative = target.relative_to(root)
    except ValueError as exc:
        raise ValueError("onboarding path escapes the profile") from exc
    current = root
    for part in relative.parts:
        current /= part
        try:
            info = current.lstat()
        except FileNotFoundError:
            continue
        if stat.S_ISLNK(info.st_mode):
            raise ValueError("onboarding path ancestor must not be a symlink")
        if info.st_uid != os.getuid():
            raise ValueError("onboarding path ancestor must be owner-controlled")
        if current != target and not stat.S_ISDIR(info.st_mode):
            raise ValueError("onboarding path ancestor must be a directory")


def ensure_private_profile_directory(path: Path, profile_root: Path) -> None:
    """Create/verify a profile-relative directory tree through retained no-follow fds."""
    root = Path(profile_root).absolute()
    target = Path(path).absolute()
    try:
        relative = target.relative_to(root)
    except ValueError as exc:
        raise ValueError("onboarding directory escapes the profile") from exc
    flags = os.O_RDONLY | os.O_DIRECTORY | os.O_CLOEXEC | getattr(os, "O_NOFOLLOW", 0)
    try:
        root_fd = os.open(root, flags)
    except OSError as exc:
        raise ValueError("onboarding profile root cannot be opened safely") from exc
    descriptors: list[int] = []
    parent_fd = root_fd
    try:
        root_info = os.fstat(root_fd)
        if (
            root_info.st_uid != os.getuid()
            or not stat.S_ISDIR(root_info.st_mode)
            or stat.S_IMODE(root_info.st_mode) & 0o022
        ):
            raise ValueError("onboarding profile root must be owner-only")
        for component in relative.parts:
            created = False
            try:
                before = os.stat(component, dir_fd=parent_fd, follow_symlinks=False)
            except FileNotFoundError:
                created = True
                os.mkdir(component, 0o700, dir_fd=parent_fd)
                child_fd = os.open(component, flags, dir_fd=parent_fd)
                os.fchmod(child_fd, 0o700)
                os.fsync(parent_fd)
                before = os.stat(component, dir_fd=parent_fd, follow_symlinks=False)
            else:
                if stat.S_ISLNK(before.st_mode):
                    raise ValueError("onboarding directory ancestor must not be a symlink")
                child_fd = os.open(component, flags, dir_fd=parent_fd)
            opened = os.fstat(child_fd)
            if (
                (before.st_dev, before.st_ino) != (opened.st_dev, opened.st_ino)
                or not stat.S_ISDIR(opened.st_mode)
                or opened.st_uid != os.getuid()
                or stat.S_IMODE(opened.st_mode) & 0o022
                or (created and stat.S_IMODE(opened.st_mode) != 0o700)
            ):
                os.close(child_fd)
                raise ValueError("onboarding directory ancestor must be owner-only 0700")
            descriptors.append(child_fd)
            parent_fd = child_fd
    finally:
        for descriptor in reversed(descriptors):
            os.close(descriptor)
        os.close(root_fd)


def validate_private_directory(path: Path) -> None:
    info = path.lstat()
    if stat.S_ISLNK(info.st_mode) or not stat.S_ISDIR(info.st_mode):
        raise ValueError("onboarding path must be a regular directory, not a symlink")
    if info.st_uid != os.getuid() or stat.S_IMODE(info.st_mode) != 0o700:
        raise ValueError("onboarding directory must be owner-only")


def validate_private_file(path: Path) -> None:
    info = path.lstat()
    if stat.S_ISLNK(info.st_mode) or not stat.S_ISREG(info.st_mode):
        raise ValueError("onboarding file must be regular, not a symlink")
    if info.st_nlink != 1:
        raise ValueError("onboarding file must not be hard-linked")
    if info.st_uid != os.getuid() or stat.S_IMODE(info.st_mode) != 0o600:
        raise ValueError("onboarding file must be owner-only")


def create_private_file(path: Path, content: bytes) -> None:
    if path.exists():
        validate_private_file(path)
        return
    descriptor = os.open(path, os.O_CREAT | os.O_EXCL | os.O_WRONLY, 0o600)
    with os.fdopen(descriptor, "wb") as handle:
        handle.write(content)
        handle.flush()
        os.fsync(handle.fileno())


def fsync_directory(path: Path) -> None:
    descriptor = os.open(path, os.O_RDONLY | os.O_DIRECTORY)
    try:
        os.fsync(descriptor)
    finally:
        os.close(descriptor)


def read_private_json(path: Path) -> dict[str, object]:
    validate_private_file(path)
    value = json.loads(path.read_text(encoding="utf-8"))
    if not isinstance(value, dict):
        raise ValueError("private JSON document must be an object")
    return value


def atomic_write_private_json(path: Path, document: dict[str, object]) -> None:
    payload = json.dumps(
        document,
        ensure_ascii=False,
        sort_keys=True,
        separators=(",", ":"),
    ).encode("utf-8")
    fd, temporary = tempfile.mkstemp(prefix=f".{path.name}.", dir=path.parent)
    temp_path = Path(temporary)
    try:
        os.fchmod(fd, 0o600)
        with os.fdopen(fd, "wb") as handle:
            handle.write(payload)
            handle.flush()
            os.fsync(handle.fileno())
        os.replace(temp_path, path)
        path.chmod(0o600)
        fsync_directory(path.parent)
    finally:
        temp_path.unlink(missing_ok=True)