"""Invocation cgroups inside a dedicated, delegated USER service.

This owns ordinary descendants, including setsid/double forks. It is deliberately
not a sandbox against the runtime UID: that UID can ask its user manager to own
persistent services/Podman scopes. Those are outside invocation cancellation.
Never point this class at the manager/authority groups or the whole user slice.
"""
import os
import resource
from pathlib import Path
import uuid


def require(value, message):
    if not value:
        raise ValueError(message)


def current_group(pid='self'):
    rows = Path(f'/proc/{pid}/cgroup').read_text().splitlines()
    require(len(rows) == 1 and rows[0].startswith('0::/'), 'unified cgroup required')
    return Path('/sys/fs/cgroup') / rows[0][4:]


class Invocation:
    RETIRE_STEPS = 64
    RETIRE_DEPTH = 64
    # One process-wide traversal, not one depth-sized FD stack per job.
    walker = None

    def __init__(self, path):
        self.path = path
        # Pin this reservation, never traverse a replacement pathname or sibling.
        self.fd = os.open(path, os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW | os.O_CLOEXEC)
        self.walk = []

    def enter(self):
        self.write_control('cgroup.procs', str(os.getpid()))
        require(current_group() == self.path, 'invocation cgroup entry failed')

    def write_control(self, name, value):
        fd = os.open(name, os.O_WRONLY | os.O_NOFOLLOW | os.O_CLOEXEC, dir_fd=self.fd)
        with os.fdopen(fd, 'w') as stream:
            stream.write(value)

    def kill(self):
        self.write_control('cgroup.kill', '1')

    def empty(self):
        fd = os.open('cgroup.events', os.O_RDONLY | os.O_NOFOLLOW | os.O_CLOEXEC, dir_fd=self.fd)
        with os.fdopen(fd) as stream:
            events = dict(row.split() for row in stream.read().splitlines())
        return events['populated'] == '0'

    def release_walk(self):
        """Allocation-free pressure recovery; retain the reservation itself."""
        for fd, name, entries in self.walk:
            entries.close()
            if name is not None:
                os.close(fd)
        self.walk.clear()
        if Invocation.walker is self:
            Invocation.walker = None

    def close(self):
        """Bounded, resumable postorder retirement of this empty subtree only.

        False means more ticks are required. Errors retain the root capability
        and reset the traversal for retry; no populated/unreadable tree is lost.
        cgroupfs has no symlinks, but dirfd/no-follow also prevents following one
        in a malformed fixture or an unexpectedly replaced reservation.
        """
        try:
            require(self.empty(), 'invocation retirement unresolved')
            if Invocation.walker not in (None, self):
                return False
            Invocation.walker = self
            if not self.walk:
                # scandir(fd) duplicates the descriptor but shares its directory
                # offset. A failed traversal must start over, not skip a subtree.
                os.lseek(self.fd, 0, os.SEEK_SET)
                self.walk.append((self.fd, None, os.scandir(self.fd)))
            for _ in range(self.RETIRE_STEPS):
                fd, name, entries = self.walk[-1]
                entry = next(entries, None)
                if entry is not None:
                    if entry.is_dir(follow_symlinks=False):
                        require(len(self.walk) < self.RETIRE_DEPTH, 'invocation retirement depth bound')
                        child = os.open(entry.name, os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW | os.O_CLOEXEC,
                                        dir_fd=fd)
                        try:
                            scan = os.scandir(child)
                        except BaseException:
                            os.close(child)
                            raise
                        self.walk.append((child, entry.name, scan))
                    continue
                entries.close()
                if name is None:
                    before, after = os.fstat(self.fd), self.path.lstat()
                    require((before.st_dev, before.st_ino) == (after.st_dev, after.st_ino),
                            'invocation reservation replaced')
                    self.path.rmdir()
                    self.walk.pop()
                    os.close(self.fd)
                    self.fd = None
                    Invocation.walker = None
                    return True
                # rmdir is the kernel's final empty/no-children barrier. Do not
                # unlink files, recurse outside this dirfd, or use rmtree.
                parent = self.walk[-2][0]
                before, after = os.fstat(fd), os.stat(name, dir_fd=parent, follow_symlinks=False)
                require((before.st_dev, before.st_ino) == (after.st_dev, after.st_ino),
                        'invocation descendant replaced')
                os.rmdir(name, dir_fd=parent)
                self.walk.pop()
                os.close(fd)
            return False
        except (OSError, ValueError):
            self.release_walk()
            raise


class Groups:
    def __init__(self):
        # At most 32 jobs (root + connection), 32 pending admissions (four
        # rights each), one 128-FD walk, and 128 spare observation/control FDs.
        # Bind this reserve to the actual inherited soft limit; do not raise it.
        soft, _ = resource.getrlimit(resource.RLIMIT_NOFILE)
        require(soft == resource.RLIM_INFINITY or
                soft >= len(os.listdir('/proc/self/fd')) + 512,
                'runtime descriptor reserve unavailable')
        self.root = current_group()
        require(self.root.name == 'archipelago-runtime-worker.service',
                'worker must run in its dedicated USER unit')
        require('/user' in str(self.root) or 'runtime-user' in str(self.root),
                'runtime user hierarchy required')
        require((self.root / 'cgroup.kill').is_file(), 'cgroup.kill unavailable')
        # No adoption of a previous worker's jobs after crash. systemd must first
        # retire its service cgroup. Refuse stale leaves, including empty ones.
        require(not any(p.is_dir() for p in self.root.iterdir()), 'prior worker groups remain')
        control = self.root / 'owner'
        control.mkdir()
        (control / 'cgroup.procs').write_text(str(os.getpid()))
        self.control = control

    def reserve(self):
        path = self.root / ('job-' + uuid.uuid4().hex)
        path.mkdir()
        try:
            return Invocation(path)
        except BaseException:
            path.rmdir()  # No child was launched into this empty reservation.
            raise