"""Conservative release eligibility; unknown process resources keep the kernel."""

from __future__ import annotations

import asyncio
from collections.abc import Callable
import hashlib
import os
import threading


def environment_hash() -> bytes:
    digest = hashlib.sha256()
    for key, value in sorted(os.environ.items()):
        digest.update(os.fsencode(key))
        digest.update(b"\0")
        digest.update(os.fsencode(value))
        digest.update(b"\0")
    return digest.digest()


def descriptors() -> set[tuple[int, int, int, int]] | None:
    for directory in ("/proc/self/fd", "/dev/fd"):
        try:
            names = os.listdir(directory)
        except OSError:
            continue
        result = set()
        for name in names:
            if not name.isdecimal():
                continue
            fd = int(name)
            try:
                stat = os.fstat(fd)
            except OSError:
                continue  # listdir's own descriptor already closed
            result.add((fd, stat.st_dev, stat.st_ino, stat.st_mode))
        return result
    return None


class Quiescence:
    def __init__(
        self,
        loop: asyncio.AbstractEventLoop,
        owns: list[Callable[[], bool]],
        internal: Callable[[], tuple[set[threading.Thread], set[int]]],
        monitor: Callable[[], None],
    ) -> None:
        self.loop = loop
        self.owns = owns
        self.internal = internal
        self.monitor = monitor  # the periodic sampler's own callback
        self.environment = environment_hash()
        self.threads = set(threading.enumerate())
        initial = descriptors()
        _, internal_fds = internal()
        self.descriptors = (
            None
            if initial is None
            else {fd for fd in initial if fd[0] not in internal_fds}
        )
        self.tasks = asyncio.all_tasks(loop)
        self.timers = set(getattr(loop, "_scheduled", ()))

    def ready(self) -> bool:
        try:
            if environment_hash() != self.environment:
                return False
            internal_threads, internal_fds = self.internal()
            if set(threading.enumerate()) - self.threads - internal_threads:
                return False
            if asyncio.all_tasks(self.loop) - self.tasks:
                return False
            timers = getattr(self.loop, "_scheduled", None)
            if timers is None or any(
                timer not in self.timers
                and not timer.cancelled()
                and getattr(getattr(timer, "_callback", None), "__name__", "")
                not in ("refresh_work", "recheck_work")
                for timer in timers
            ):
                return False
            current = descriptors()
            if (
                current is None
                or self.descriptors is None
                or {fd for fd in current if fd[0] not in internal_fds}
                - self.descriptors
            ):
                return False
            return not any(check() for check in self.owns)
        except Exception:
            return False