"""Hygiene: the things that rot quietly between updates.

Everything here is readable without root, so the tool can report honestly before
it ever asks for a password. Where a check would need elevation it degrades to
"unknown" rather than guessing.
"""

from __future__ import annotations

import os
import platform
import time
from dataclasses import dataclass, field
from pathlib import Path

CACHE_DIR = Path("/var/cache/pacman/pkg")
ETC = Path("/etc")
MODULES = Path("/usr/lib/modules")
CONFIG_SUFFIXES = (".pacnew", ".pacsave", ".pacorig")

#: package name prefixes whose update means the running kernel or its modules no
#: longer match what is on disk
REBOOT_PREFIXES = ("linux", "nvidia", "virtualbox", "zfs", "openzfs", "bbswitch")


def parse_cache_name(filename: str) -> tuple[str, str] | None:
    """`(name, version)` for a cached package file, or None if unrecognised.

    `<name>-<pkgver>-<pkgrel>-<arch>.pkg.tar.<ext>`, split from the right. A
    pkgver containing a dash would defeat this, which is why pacman discourages
    them.
    """
    stem = filename
    for marker in (".pkg.tar.zst", ".pkg.tar.xz", ".pkg.tar.gz", ".pkg.tar"):
        if stem.endswith(marker):
            stem = stem[: -len(marker)]
            break
    else:
        if not stem.endswith(".part"):
            return None
        stem = stem[: -len(".part")]
    parts = stem.rsplit("-", 3)
    if len(parts) != 4:
        return None
    name, version, _rel, _arch = parts
    return (name, f"{version}-{_rel}") if name else None


@dataclass
class CacheUsage:
    total: int = 0
    reclaimable: int = 0
    files: int = 0
    reclaimable_files: int = 0
    keep: int = 3

    @property
    def has_reclaimable(self) -> bool:
        return self.reclaimable > 0


def incomplete_downloads(cache_dir: Path = CACHE_DIR) -> Leftovers:
    """Count pacman's abandoned `download-*` scratch files.

    Never raises: an unreadable cache directory is a normal state.
    """
    try:
        entries = list(cache_dir.iterdir())
    except OSError:
        return Leftovers()

    result = Leftovers()
    for entry in entries:
        if not entry.name.startswith("download-"):
            continue
        try:
            info = entry.stat()
        except OSError:
            continue
        result.count += 1
        result.bytes += info.st_size
        if result.oldest == 0.0 or info.st_mtime < result.oldest:
            result.oldest = info.st_mtime
        result.newest = max(result.newest, info.st_mtime)
    return result


def cache_usage(
    cache_dir: Path = CACHE_DIR, *, keep: int = 3, installed: dict[str, str] | None = None
) -> CacheUsage:
    """How much of the package cache could be pruned.

    Keeps the newest `keep` versions of every package plus whatever is currently
    installed, which is the same rule `paccache -rk3` uses. The installed version
    is kept even if it is older than the newest: downgrading is a legitimate
    thing to want to do.
    """
    usage = CacheUsage(keep=keep)
    try:
        entries = list(os.scandir(cache_dir))
    except OSError:
        return usage

    by_name: dict[str, list[tuple[float, int, str]]] = {}
    for entry in entries:
        parsed = parse_cache_name(entry.name)
        if parsed is None:
            continue
        name, version = parsed
        try:
            stat = entry.stat()
        except OSError:
            continue
        usage.total += stat.st_size
        usage.files += 1
        by_name.setdefault(name, []).append((stat.st_mtime, stat.st_size, version))

    for name, versions in by_name.items():
        versions.sort(key=lambda row: row[0], reverse=True)
        keep_versions = {v for _m, _s, v in versions[:keep]}
        if installed and name in installed:
            keep_versions.add(installed[name])
        for _mtime, size, version in versions:
            if version not in keep_versions:
                usage.reclaimable += size
                usage.reclaimable_files += 1
    return usage


@dataclass
class Drift:
    pacnew: list[Path] = field(default_factory=list)
    pacsave: list[Path] = field(default_factory=list)
    pacorig: list[Path] = field(default_factory=list)

    @property
    def total(self) -> int:
        return len(self.pacnew) + len(self.pacsave) + len(self.pacorig)

    @property
    def actionable(self) -> list[Path]:
        """`.pacsave` and `.pacorig` are informational; `.pacnew` needs merging."""
        return list(self.pacnew)


def config_drift(root: Path = ETC, *, max_depth: int = 6) -> Drift:
    """Unmerged configuration files left behind by package upgrades."""
    drift = Drift()
    root = root.resolve()
    base_depth = len(root.parts)
    for dirpath, dirnames, filenames in os.walk(root, onerror=lambda _e: None):
        current = Path(dirpath)
        if len(current.parts) - base_depth >= max_depth:
            dirnames[:] = []
            continue
        # don't descend into the places pacnew files are never left
        dirnames[:] = [
            d for d in dirnames if d not in ("proc", "sys", "dev", ".git", "node_modules")
        ]
        for filename in filenames:
            if filename.endswith(".pacnew"):
                drift.pacnew.append(current / filename)
            elif filename.endswith(".pacsave"):
                drift.pacsave.append(current / filename)
            elif filename.endswith(".pacorig"):
                drift.pacorig.append(current / filename)
    for bucket in (drift.pacnew, drift.pacsave, drift.pacorig):
        bucket.sort()
    return drift


def running_kernel() -> str:
    return platform.release()


def installed_kernels(modules: Path = MODULES) -> list[str]:
    try:
        return sorted(p.name for p in modules.iterdir() if p.is_dir())
    except OSError:
        return []


def reboot_reason(updated: set[str] | None = None, modules: Path = MODULES) -> str | None:
    """Why a reboot is needed, or None.

    The authoritative signal is that the running kernel has no module tree on
    disk any more, which means its modules were replaced underneath it.
    """
    running = running_kernel()
    kernels = installed_kernels(modules)
    if kernels and running not in kernels:
        return f"running kernel {running} is no longer installed"
    if updated:
        touched = sorted(n for n in updated if n.startswith(REBOOT_PREFIXES))
        if touched:
            return f"updated {', '.join(touched[:3])}"
    return None


#: the kernel's marker for an executable whose file has been unlinked
DELETED_SUFFIX = " (deleted)"


@dataclass(frozen=True)
class StaleProcess:
    pid: int
    name: str
    path: str


def stale_processes(*, limit: int = 400) -> list[StaleProcess]:
    """Processes still running a binary that has been replaced on disk.

    The kernel marks an executable whose file was unlinked with a ` (deleted)`
    suffix on `/proc/<pid>/exe`. That is the *usual* result of a package
    upgrade, because the package manager replaces the file rather than editing
    it -- and this check used to skip exactly those, so it reported almost
    nothing. A deleted executable is the strongest possible signal that the
    process is out of date, not a reason to look away.

    Where the file still exists, the inode the process is running is compared
    with the inode at that path: a mismatch means the same thing.

    Only readable processes are inspected, so this needs no privileges and simply
    sees less when run as a normal user.
    """
    out: list[StaleProcess] = []
    seen: set[tuple[int, str]] = set()
    checked = 0
    proc = Path("/proc")
    try:
        entries = [e for e in os.listdir(proc) if e.isdigit()]
    except OSError:
        return out

    for entry in entries:
        if checked >= limit:
            break
        pid = int(entry)
        try:
            exe = os.readlink(proc / entry / "exe")
        except OSError:
            continue
        if not exe.startswith("/"):
            continue

        deleted = exe.endswith(DELETED_SUFFIX)
        if deleted:
            exe = exe[: -len(DELETED_SUFFIX)]

        checked += 1
        if not deleted:
            try:
                running_inode = os.stat(proc / entry / "exe").st_ino
                disk_inode = os.stat(exe).st_ino
            except OSError:
                # the path is gone or unreadable, so the process is running
                # something that is no longer there
                pass
            else:
                if running_inode == disk_inode:
                    continue

        key = (pid, exe)
        if key in seen:
            continue
        seen.add(key)
        try:
            name = (proc / entry / "comm").read_text().strip()
        except OSError:
            name = os.path.basename(exe)
        out.append(StaleProcess(pid=pid, name=name, path=exe))
    out.sort(key=lambda s: s.name)
    return out


@dataclass
class Leftovers:
    """Incomplete downloads pacman abandoned in the cache.

    pacman downloads to `download-XXXXXX` and renames on success, so anything
    left behind is a transfer that failed. They are usually empty, which is why
    this reports a count rather than a size -- the useful signal is that
    downloads keep failing, not that any disk can be recovered.
    """

    count: int = 0
    bytes: int = 0
    oldest: float = 0.0
    newest: float = 0.0

    @property
    def worth_mentioning(self) -> bool:
        # a handful of failures is normal; a pile of them is a symptom
        return self.count >= 10

    @property
    def all_empty(self) -> bool:
        """Every leftover is zero bytes, which points at a write failure.

        A transfer that ran out of network leaves partial data; one that could
        not create or rename its file leaves nothing at all.
        """
        return self.count > 0 and self.bytes == 0

    def explain(self, cache_dir) -> list[str]:
        """What these are, why they are probably there, and how to clear them."""

        def age(stamp: float) -> str:
            if not stamp:
                return "?"
            seconds = max(0.0, time.time() - stamp)
            if seconds < 5400:
                return f"{int(seconds // 60)}m"
            if seconds < 172800:
                return f"{seconds / 3600:.0f}h"
            return f"{seconds / 86400:.0f}d"

        lines = [
            f"{self.count} file(s), {self.bytes:,} bytes total",
            f"oldest {age(self.oldest)} old, newest {age(self.newest)} old",
            "",
            "pacman downloads each package to a scratch name and renames it",
            "on success, so anything still called download-* is a transfer",
            "that never finished.",
        ]
        if self.all_empty:
            lines += [
                "",
                "They are all zero bytes, which is not what running out of",
                "network looks like: a partial download has partial data.",
                "Empty ones mean pacman could not create or rename the file,",
                "which is the same failure the databases check reports above.",
            ]
        lines += [
            "",
            "clear them with:",
            f"  sudo find {cache_dir} -maxdepth 1 -name 'download-*' -delete",
        ]
        return lines


@dataclass
class Hygiene:
    cache: CacheUsage = field(default_factory=CacheUsage)
    leftovers: Leftovers = field(default_factory=Leftovers)
    drift: Drift = field(default_factory=Drift)
    orphans: list[str] = field(default_factory=list)
    reboot: str | None = None
    reboot_after: set[str] = field(default_factory=set)
    stale: list[StaleProcess] = field(default_factory=list)
    removed_from_aur: list[str] = field(default_factory=list)
    orphaned_aur: list[str] = field(default_factory=list)
    stale_aur: list[tuple[str, float]] = field(default_factory=list)
    stale_scan_seconds: float = 0.0

    @property
    def has_anything(self) -> bool:
        return bool(
            self.cache.has_reclaimable
            or self.drift.total
            or self.orphans
            or self.reboot
            or self.stale
            or self.removed_from_aur
            or self.orphaned_aur
            or self.leftovers.worth_mentioning
        )


def collect(
    *,
    cache_dir: Path = CACHE_DIR,
    etc: Path = ETC,
    modules: Path = MODULES,
    installed: dict[str, str] | None = None,
    orphans: list[str] | None = None,
    updated: set[str] | None = None,
    include_processes: bool = True,
) -> Hygiene:
    """Gather everything that does not need a network or a password."""
    report = Hygiene(
        cache=cache_usage(cache_dir, installed=installed),
        leftovers=incomplete_downloads(cache_dir),
        drift=config_drift(etc),
        orphans=sorted(orphans or []),
    )
    report.reboot_after = set(updated or ())
    report.reboot = reboot_reason(report.reboot_after, modules=modules)
    if include_processes:
        started = time.monotonic()
        report.stale = stale_processes()
        report.stale_scan_seconds = time.monotonic() - started
    return report