"""What the container's kernel files say: memory, a process, the deploy freeze marker.

Pure over a root path (`/sys/fs/cgroup`, `/proc`, the uploads directory), so every reader is
testable on a temporary directory and no caller ever guesses. Nothing here is stored.

Why: the 19 Sep sweep OOM-killed four collectors in one container; `guard.sh` now sheds the
largest collector at `SHED_AT_BYTES`, and the page refuses a Start above `START_HEADROOM_BYTES`,
400 MiB under that line, so a Start from the page never walks the container into the shed. A pid
is only ever believed when its command line is a collector of the named source (`alive`): the
host dev server sees another pid namespace, and a reused pid would otherwise be signalled.
"""

from __future__ import annotations

from dataclasses import dataclass
from datetime import UTC, datetime, timedelta
from pathlib import Path

#: A Start from the page is refused above this: 400 MiB under the guard's shed line.
START_HEADROOM_BYTES = 3 * 1024 ** 3
#: Where `guard.sh` sheds the largest collector (`SHED_MIB=3400`); shown so the two agree.
SHED_AT_BYTES = 3400 * 1024 ** 2
#: A freeze marker older than this is a deploy that never removed it; the page says so.
FREEZE_STALE = timedelta(hours=6)

CGROUP = "/sys/fs/cgroup"
PROC = "/proc"


@dataclass(frozen=True, slots=True)
class ContainerMemory:
    used_bytes: int | None
    limit_bytes: int | None
    readable: bool


def container_memory(cgroup: str | Path = CGROUP) -> ContainerMemory:
    """The cgroup v2 `memory.current` and `memory.max` (None when `max`), or unreadable."""
    root = Path(cgroup)
    try:
        used = int(root.joinpath("memory.current").read_text().strip())
    except (OSError, ValueError):
        return ContainerMemory(None, None, False)
    try:
        raw = root.joinpath("memory.max").read_text().strip()
        limit = None if raw == "max" else int(raw)
    except (OSError, ValueError):
        limit = None
    return ContainerMemory(used, limit, True)


def rss_of(pid: int, proc: str | Path = PROC) -> int | None:
    """Resident set size in bytes from `/proc/<pid>/status`, or None."""
    try:
        for line in Path(proc, str(pid), "status").read_text().splitlines():
            if line.startswith("VmRSS:"):
                return int(line.split()[1]) * 1024
    except (OSError, ValueError, IndexError):
        return None
    return None


def cmdline_of(pid: int, proc: str | Path = PROC) -> list[str] | None:
    try:
        raw = Path(proc, str(pid), "cmdline").read_bytes()
    except OSError:
        return None
    return [part for part in raw.decode("utf-8", "replace").split("\0") if part]


def collects(argv: list[str], slug: str) -> bool:
    """Whether a command line is a collector process of `slug`: `app.cli collect` with no
    `--source` at all (the documented default runs every enabled collector, this one among
    them), or with `--source <slug>` or `--source=<slug>`. Why: `alive` once demanded the
    literal items `--source` and `<slug>`, so a healthy run started as `python -m app.cli
    collect` read `alive is False`, the page called it "stalled: process gone" ahead of its
    fresh heartbeat, and Stop now on that row would have marked a live run ended."""
    subcommand = next((argv[i + 1] for i, part in enumerate(argv[:-1]) if "app.cli" in part), None)
    if subcommand != "collect":
        return False
    for i, part in enumerate(argv):
        if part == "--source":
            return i + 1 < len(argv) and argv[i + 1] == slug
        if part.startswith("--source="):
            return part[len("--source="):] == slug
    return True


def alive(pid: int | None, slug: str, proc: str | Path = PROC) -> bool | None:
    """Whether `pid` is a live collector of `slug`: None when the process table is unreadable,
    False for pid <= 1, a zombie, or a command line that is not an `app.cli collect` of this
    source (`collects`). A pid is never believed on its own: pids are reused, and the host dev
    server's `/proc` is another namespace entirely."""
    if pid is None or pid <= 1:
        return False
    root = Path(proc)
    if not root.is_dir():
        return None
    argv = cmdline_of(pid, proc)
    if argv is None:
        return False if root.joinpath("1").exists() else None
    try:
        status = root.joinpath(str(pid), "status").read_text()
    except OSError:
        return False
    for line in status.splitlines():
        if line.startswith("State:") and "Z" in line.split()[1:2]:
            return False
    return collects(argv, slug)


@dataclass(frozen=True, slots=True)
class Freeze:
    frozen: bool
    stamp: str | None
    written_at: datetime | None
    stale: bool

    def age(self, now: datetime) -> timedelta | None:
        return None if self.written_at is None else now - self.written_at


def freeze_state(marker: str | Path, now: datetime | None = None) -> Freeze:
    """The deploy freeze marker (`D3 <iso> by <user>`, written before the gate check and removed
    after health). Its stamp line is what the page shows; past `FREEZE_STALE` it is called stale."""
    path = Path(marker)
    try:
        stamp = path.read_text().strip().splitlines()[0] if path.read_text().strip() else ""
        written = datetime.fromtimestamp(path.stat().st_mtime, UTC)
    except OSError:
        return Freeze(False, None, None, False)
    now = now or datetime.now(UTC)
    return Freeze(True, stamp or None, written, now - written > FREEZE_STALE)