#!/usr/bin/python3

# sd-debian: Software-defined Debian
# Copyright (C) 2024-2026 Daniel Baumann <daniel@debian.org>
# Copyright (C) 2024-2026 Patrik Dufresne <patrik@ikus-soft.com>
#
# SPDX-License-Identifier: GPL-3.0+
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program.  If not, see <https://www.gnu.org/licenses/>.

import fcntl
import getpass
import logging
import os
import re
import subprocess
import sys
from pathlib import Path

VARDIR = Path("/var/lib/sd-debian")
RUNDIR = Path("/run/sd-debian")
LOCKFILE = RUNDIR / "dpkg_status-logger.lock"
LOGDIR = Path("/var/log/sd-debian")
LOGFILE = LOGDIR / "sd-debian.log"
DPKG_QUERY_BIN = Path("/usr/bin/dpkg-query")

PKG_NAME_RE = re.compile(r"^[a-zA-Z0-9][a-zA-Z0-9+.-]*$")

LOGUSER = getpass.getuser()

logger = logging.getLogger("dpkg_status-logger")


def setup_logger(path: Path):
    """
    Configure logging system.
    """
    try:
        path.parent.mkdir(parents=True, exist_ok=True)
    except OSError as e:
        print(
            f"[WARNING]: could not create log directory '{path.parent}': {e}, aborting\n",
            file=sys.stderr,
        )
        return

    writable = (path.exists() and os.access(path, os.W_OK)) or (
        not path.exists() and os.access(path.parent, os.W_OK)
    )
    if not writable:
        print(
            f"[WARNING]: log file at '{path}' is not writeable and we cannot create it, aborting\n",
            file=sys.stderr,
        )
        return

    if path.is_dir():
        print(
            f"[WARNING]: log path can not be a directory '{path}', aborting",
            file=sys.stderr,
        )
        return

    logging.basicConfig(
        filename=path,
        level=logging.DEBUG,
        format=f"%(asctime)s p=%(process)d u={LOGUSER} n=%(name)s %(levelname)s| %(message)s",
    )


def get_source_package(binary_package: str) -> str:
    """Get source package of a binary package using dpkg-query.

    Falls back to the binary package name if the source package cannot
    be determined (e.g. package no longer in dpkg status, dpkg-query
    missing, or invalid/unexpected output).
    """
    source_package = ""
    try:
        result = subprocess.run(
            [DPKG_QUERY_BIN, "-W", "-f=${source:Package}\n", binary_package],
            capture_output=True,
            text=True,
            check=False,
        )
        if result.returncode != 0:
            logger.debug(
                f"dpkg-query failed for '{binary_package}' "
                f"(rc={result.returncode}): {result.stderr.strip()}"
            )
        else:
            output = result.stdout.strip()
            if output and not PKG_NAME_RE.match(output):
                logger.warning(
                    f"invalid package name returned by dpkg-query: {output!r}"
                )
            else:
                source_package = output
    except FileNotFoundError:
        # When dpkg-query is not found.
        logger.warning(f"'{DPKG_QUERY_BIN}' not found, cannot resolve source package")

    return source_package if source_package else binary_package


def process_line(line: str, state: dict) -> None:
    """Process a single line of dpkg hook output.

    state is a dict holding 'pending_action' and 'pending_binary' keys
    to track state across lines (equivalent to shell script globals).

    Collect a list of packages getting installed, removed or purged:

        processing: install: <pkgname>
        status: <pkgname>: installed --> confirmed install

        processing: configure: <pkgname>
        status: <pkgname>: installed --> confirmed install

        processing: remove: <pkgname>
        status: <pkgname>: config-files --> confirm remove

        processing: purge: <pkgname>
        status: <pkgname>: not-installed --> confirm purge

        Note: this happen when package doesn't have conffiles

        processing: remove: <pkgname>
        status: <pkgname>: not-installed --> confirm purge
    """
    strip_line = line.rstrip("\n")
    fields = strip_line.split(maxsplit=3)
    if not fields or len(fields) < 3:
        return
    tag, p0, p1 = fields[0:3]
    logger.debug(f"{strip_line}")

    if tag == "processing:":
        action = p0.rsplit(":", 1)[0]

        if action in ("install", "configure", "remove", "purge"):
            # Strip architecture from binary package name (p1)
            binary_package = p1.rsplit(":", 1)[0]

            # Validate pkg name
            if not PKG_NAME_RE.match(binary_package):
                logger.warning(f"skipping invalid package name: {binary_package}")
                return

            state["pending_action"] = action
            state["pending_binary"] = binary_package

    elif tag == "status:":
        status_pkg = p0.rstrip(":")
        status_state = p1

        pending_action = state.get("pending_action")
        pending_binary = state.get("pending_binary")

        if pending_action and status_pkg == pending_binary:
            confirmed_action = None

            key = f"{pending_action}:{status_state}"
            if key in ("install:installed", "configure:installed"):
                confirmed_action = "install"
            elif key == "remove:config-files":
                confirmed_action = "remove"
            elif key in ("remove:not-installed", "purge:not-installed"):
                confirmed_action = "purge"

            if confirmed_action:
                source_package = get_source_package(pending_binary)

                out_file = VARDIR / confirmed_action
                with open(out_file, "a") as f:
                    f.write(f"{source_package}/{pending_binary}\n")

                logger.debug(
                    f"confirmed: {confirmed_action} {source_package}/{pending_binary} "
                    f"(state={status_state})"
                )


def main() -> int:
    VARDIR.mkdir(parents=True, exist_ok=True)
    RUNDIR.mkdir(parents=True, exist_ok=True)
    setup_logger(path=LOGFILE)

    lock_fd = open(LOCKFILE, "w")
    try:
        fcntl.flock(lock_fd, fcntl.LOCK_EX)  # blocks until acquired

        state: dict = {}
        for line in sys.stdin:
            process_line(line, state)
    finally:
        fcntl.flock(lock_fd, fcntl.LOCK_UN)
        lock_fd.close()

    return 0


if __name__ == "__main__":
    sys.exit(main())
