"""pacman's alpm hooks: the work that happens around a transaction.

A hook is an INI-ish file with one or more `[Trigger]` sections describing when
it fires and an `[Action]` section describing what it runs. They are how a
kernel upgrade rebuilds its initramfs, how systemd notices a unit changed, and
how snapper takes its snapshots -- the things a plan is supposed to tell you
about and currently does not.

Parsed, never executed.
"""

from __future__ import annotations

import fnmatch
from dataclasses import dataclass, field
from pathlib import Path

#: system hooks ship here; local ones override by filename in /etc
HOOK_DIRS = (Path("/usr/share/libalpm/hooks"), Path("/etc/pacman.d/hooks"))

#: `snap-pac` names its snapshots, so a snapshot stage can be recognised
SNAPSHOT_MARKER = "snap-pac"


@dataclass(frozen=True)
class Trigger:
    kind: str  # "Package" or "Path"
    operations: tuple[str, ...]  # Upgrade, Install, Remove
    targets: tuple[str, ...]  # glob patterns

    def matches_name(self, name: str, operation: str = "Upgrade") -> bool:
        if self.kind != "Package":
            return False
        if self.operations and operation not in self.operations:
            return False
        return any(fnmatch.fnmatchcase(name, pat) for pat in self.targets)

    def matches_path(self, path: str, operation: str = "Upgrade") -> bool:
        if self.kind != "Path":
            return False
        if self.operations and operation not in self.operations:
            return False
        # hook paths are relative to the filesystem root, without a leading /
        cleaned = str(path).lstrip("/")
        return any(fnmatch.fnmatchcase(cleaned, pat) for pat in self.targets)


@dataclass
class Hook:
    path: Path
    description: str = ""
    when: str = "PostTransaction"
    exec_: str = ""
    depends: tuple[str, ...] = ()
    triggers: list[Trigger] = field(default_factory=list)
    needs_targets: bool = False
    abort_on_fail: bool = False

    @property
    def is_snapshot(self) -> bool:
        return SNAPSHOT_MARKER in self.exec_.lower() or "snapper" in self.exec_.lower()

    @property
    def label(self) -> str:
        """What to show a person. pacman's own words where it gave any."""
        if self.description:
            return self.description.rstrip(".").strip()
        return self.exec_.split()[0].rsplit("/", 1)[-1] if self.exec_ else self.path.name

    def fires_for(
        self, names: set[str], paths_by_package: dict[str, list[str]], operation: str = "Upgrade"
    ) -> bool:
        """Whether this hook will run for any of the given packages.

        Package triggers are exact. Path triggers are matched against the file
        list of the *installed* version, which is an approximation: a package
        that ships a systemd unit today almost certainly still will, but it is
        the old file list and not the new one.
        """
        for trigger in self.triggers:
            if trigger.kind == "Package":
                if any(trigger.matches_name(n, operation) for n in names):
                    return True
            elif trigger.kind == "Path":
                for name in names:
                    for path in paths_by_package.get(name, ()):
                        if trigger.matches_path(path, operation):
                            return True
        return False


def _sections(path: Path) -> list[tuple[str, dict[str, list[str]]]]:
    """Every section in a hook file, in order, with repeated keys kept.

    Hand-written rather than ConfigParser, for three reasons: a hook may declare
    several `[Trigger]` sections and ConfigParser keeps only one; the flags
    `NeedsTargets` and `AbortOnFail` are bare words with no `=` at all, which
    ConfigParser treats as a parse error and drops the whole file; and keys
    repeat -- a trigger lists `Operation = Upgrade`, `Install` and `Remove` on
    three lines, and keeping only the last one silently changed which
    transactions a hook fires for.
    """
    try:
        text = path.read_text(encoding="utf-8", errors="replace")
    except OSError:
        return []
    out: list[tuple[str, dict[str, list[str]]]] = []
    current: dict[str, list[str]] | None = None
    for raw in text.splitlines():
        line = raw.split("#", 1)[0].strip()
        if not line:
            continue
        if line.startswith("[") and line.endswith("]"):
            current = {}
            out.append((line[1:-1].strip().lower(), current))
            continue
        if current is None:
            continue
        if "=" in line:
            key, _, value = line.partition("=")
            current.setdefault(key.strip().lower(), []).append(value.strip())
        else:
            current.setdefault(line.lower(), []).append("")
    return out


def _parse(path: Path) -> Hook | None:
    """One hook file, or None if it declares no usable trigger."""
    sections = _sections(path)
    action: dict[str, list[str]] = {}
    for name, body in sections:
        if name == "action":
            action.update(body)
    if not action:
        return None

    hook = Hook(
        path=path,
        description=_first(action, "description") or "",
        when=_first(action, "when") or "PostTransaction",
        exec_=_first(action, "exec") or "",
        depends=tuple(_split(_first(action, "depends"))),
        needs_targets="needstargets" in action,
        abort_on_fail="abortonfail" in action,
    )

    for name, block in sections:
        if name != "trigger":
            continue
        kind = _first(block, "type")
        targets = tuple(_split(_first(block, "target")))
        operations = tuple(_split_all(block.get("operation"))) or ("Upgrade",)
        if kind and targets:
            hook.triggers.append(Trigger(kind=kind, operations=operations, targets=targets))
    if not hook.triggers:
        return None
    return hook


def _first(section: dict[str, list[str]], key: str) -> str | None:
    """The first value for a key. Keys are already lowercased by the reader."""
    values = section.get(key.lower())
    return values[0] if values else None


def _split_all(values: list[str] | None) -> list[str]:
    out: list[str] = []
    for value in values or ():
        out.extend(_split(value))
    return out


def _split(value: str | None) -> list[str]:
    if not value:
        return []
    return [part for part in value.replace(",", " ").split() if part]


def load(dirs: tuple[Path, ...] = HOOK_DIRS) -> list[Hook]:
    """Every readable hook, later directories overriding earlier ones by name.

    That is what pacman does: a hook in /etc/pacman.d/hooks with the same
    filename replaces the one the package shipped.
    """
    seen: dict[str, Hook] = {}
    order: list[str] = []
    for directory in dirs:
        try:
            entries = sorted(directory.glob("*.hook"))
        except OSError:
            continue
        for path in entries:
            hook = _parse(path)
            if hook is None:
                continue
            if path.name not in seen:
                order.append(path.name)
            seen[path.name] = hook
    return [seen[name] for name in order]


def scheduled(
    hooks: list[Hook],
    names: set[str],
    paths_by_package: dict[str, list[str]] | None = None,
    *,
    operation: str = "Upgrade",
) -> tuple[list[Hook], list[Hook]]:
    """Split the hooks that will run into (before the transaction, after it)."""
    paths = paths_by_package or {}
    firing = [h for h in hooks if h.fires_for(names, paths, operation)]
    before = [h for h in firing if h.when.lower().startswith("pre")]
    after = [h for h in firing if not h.when.lower().startswith("pre")]
    return before, after