#!/usr/bin/python3
"""
Copyright (C) 2026 Shawn W Dunn

    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/>.

txnupd-notifier — Desktop notification daemon for transactional update systems.

Listens on the D-Bus system bus for update completion signals from:
  org.opensuse.tukit.Updated  — transactional-update / tukit
  org.rpm.dnf.v0.Automatic    — dnf5-automatic

Delivers desktop notifications via org.freedesktop.Notifications on the
session bus.

Set TRANSACTIONAL_UPDATE_NOTIFY_URGENT=1 to escalate urgency to critical
(workaround for desktop environments that silently drop normal notifications,
bsc#1219525).
"""

import configparser
import dataclasses
import enum
import logging
import os
import re
import shutil
import sys
from datetime import datetime, timedelta, timezone
from pathlib import Path

import dbus
import dbus.mainloop.glib
import dbus.service
import gi
gi.require_version("Notify", "0.7")
from gi.repository import GLib, Notify


LOG = logging.getLogger("txnupd-notifier")
APP_NAME = "txnupd-notifier"

CHECK_INTERVAL_SECONDS = 86400  # check once per day

@dataclasses.dataclass
class _Config:
    stale_threshold: timedelta = timedelta(days=14)
    notify_interval: timedelta = timedelta(days=7)

_cfg = _Config()

# /usr/share/txnupd-maintenance-tools is substituted by CMake (configure_file @ONLY).
# If the literal placeholder is still present, the script is running directly
# from the source tree without a build step; fall back to the Fedora default.
_raw_defaultconfdir = "/usr/share/txnupd-maintenance-tools"
_VENDOR_CONF = (
    Path("/usr/share/txnupd-notifier")
    if _raw_defaultconfdir.startswith("@")
    else Path(_raw_defaultconfdir)
) / "txnupd-notifier.conf"
_ADMIN_CONF = Path("/etc/txnupd-notifier.conf")
_USER_CONF = (
    Path(os.environ["XDG_CONFIG_HOME"])
    if "XDG_CONFIG_HOME" in os.environ
    else Path.home() / ".config"
) / "txnupd-notifier.conf"

STATE_DIR = (
    Path(os.environ["STATE_DIRECTORY"])
    if "STATE_DIRECTORY" in os.environ
    else Path(os.environ.get("XDG_STATE_HOME", Path.home() / ".local/state")) / APP_NAME
)
STATE_FILE = STATE_DIR / "last-update"
STALE_NOTIFY_FILE = STATE_DIR / "last-stale-notify"

TUKIT_LOG = Path("/var/log/transactional-update.log")
_TUKIT_LOG_RE = re.compile(r"^(\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2})\s+New default snapshot")

DNF5_LOGS = [Path("/var/log/dnf5.log")] + [Path(f"/var/log/dnf5.log.{i}") for i in range(1, 5)]
# Matches both "2026-06-05T14:25:12+0000" and "2026-06-05 14:25:12" prefixes
_DNF5_LOG_RE = re.compile(
    r"^(\d{4}-\d{2}-\d{2}[T ]\d{2}:\d{2}:\d{2}(?:[+-]\d{2}:?\d{2})?)"
    r".*txnupd plugin\s*:\s*New default snapshot"
)

class Backend(enum.Enum):
    TUKIT      = "tukit"
    DNF5_TXNUPD = "dnf5-txnupd"
    UNKNOWN    = "unknown"


def _detect_backend() -> Backend:
    if shutil.which("transactional-update") is not None:
        return Backend.TUKIT
    if Path("/etc/dnf/plugins/txnupd.conf").exists():
        return Backend.DNF5_TXNUPD
    return Backend.UNKNOWN


def _drop_ins(base: Path) -> list[Path]:
    d = base.parent / (base.name + ".d")
    return sorted(d.glob("*.conf")) if d.is_dir() else []


def load_config() -> None:
    parser = configparser.ConfigParser()
    parser.read([
        _VENDOR_CONF, *_drop_ins(_VENDOR_CONF),
        _ADMIN_CONF,  *_drop_ins(_ADMIN_CONF),
        _USER_CONF,   *_drop_ins(_USER_CONF),
    ])

    def _getint(section: str, key: str) -> int | None:
        try:
            val = parser.getint(section, key)
            if val < 1:
                LOG.warning("Config %s.%s must be >= 1, ignoring", section, key)
                return None
            return val
        except (configparser.NoSectionError, configparser.NoOptionError):
            return None
        except (ValueError, configparser.Error) as exc:
            LOG.warning("Config %s.%s invalid: %s", section, key, exc)
            return None

    if (days := _getint("staleness", "stale_threshold_days")) is not None:
        _cfg.stale_threshold = timedelta(days=days)
    if (days := _getint("staleness", "notify_interval_days")) is not None:
        _cfg.notify_interval = timedelta(days=days)

    LOG.info(
        "Config: stale_threshold=%d days, notify_interval=%d days",
        _cfg.stale_threshold.days,
        _cfg.notify_interval.days,
    )


TUKIT_INTERFACE = "org.opensuse.tukit.Updated"
TUKIT_PATH      = "/org/opensuse/tukit/Updated"
TUKIT_MEMBER    = "Notify"

DNF_AUTO_INTERFACE = "org.rpm.dnf.v0.Automatic"
DNF_AUTO_PATH      = "/org/rpm/dnf/Automatic"
DNF_AUTO_MEMBER    = "Finished"   # Finished(success: bool, message: string)


def current_urgency() -> Notify.Urgency:
    if os.environ.get("TRANSACTIONAL_UPDATE_NOTIFY_URGENT") == "1":
        return Notify.Urgency.CRITICAL
    return Notify.Urgency.NORMAL


def timestamp() -> str:
    return datetime.now(tz=timezone.utc).strftime("%a, %d %b %Y %H:%M:%S %Z")


def _write_state(path: Path, dt: datetime) -> None:
    try:
        STATE_DIR.mkdir(parents=True, exist_ok=True)
        path.write_text(dt.isoformat())
    except OSError:
        LOG.warning("Could not write state to %s", path)


def write_last_update() -> None:
    _write_state(STATE_FILE, datetime.now(tz=timezone.utc))


def _bootstrap_from_tukit_log() -> datetime | None:
    if not TUKIT_LOG.exists():
        return None
    last = None
    try:
        for line in TUKIT_LOG.read_text(errors="replace").splitlines():
            m = _TUKIT_LOG_RE.match(line)
            if m:
                try:
                    last = datetime.strptime(m.group(1), "%Y-%m-%d %H:%M:%S").replace(
                        tzinfo=timezone.utc
                    )
                except ValueError:
                    pass
    except OSError:
        pass
    return last


def _parse_dnf5_timestamp(ts_str: str) -> datetime | None:
    for fmt in ("%Y-%m-%dT%H:%M:%S%z", "%Y-%m-%d %H:%M:%S"):
        try:
            dt = datetime.strptime(ts_str, fmt)
            return dt if dt.tzinfo else dt.replace(tzinfo=timezone.utc)
        except ValueError:
            pass
    return None


def _bootstrap_from_dnf5_log() -> datetime | None:
    last = None
    for log_path in DNF5_LOGS:
        if not log_path.exists():
            continue
        try:
            for line in log_path.read_text(errors="replace").splitlines():
                m = _DNF5_LOG_RE.match(line)
                if not m:
                    continue
                dt = _parse_dnf5_timestamp(m.group(1))
                if dt is not None and (last is None or dt > last):
                    last = dt
        except OSError:
            pass
    return last


def read_last_update() -> datetime | None:
    if STATE_FILE.exists():
        try:
            return datetime.fromisoformat(STATE_FILE.read_text().strip())
        except (ValueError, OSError):
            pass
    backend = _detect_backend()
    if backend is Backend.TUKIT:
        ts = _bootstrap_from_tukit_log()
    elif backend is Backend.DNF5_TXNUPD:
        ts = _bootstrap_from_dnf5_log()
    else:
        ts = _bootstrap_from_tukit_log() or _bootstrap_from_dnf5_log()
    if ts is not None:
        _write_state(STATE_FILE, ts)
    return ts


def check_staleness() -> bool:
    last = read_last_update()
    if last is None:
        return GLib.SOURCE_CONTINUE

    now = datetime.now(tz=timezone.utc)
    if now - last < _cfg.stale_threshold:
        return GLib.SOURCE_CONTINUE

    if STALE_NOTIFY_FILE.exists():
        try:
            last_notified = datetime.fromisoformat(STALE_NOTIFY_FILE.read_text().strip())
            if now - last_notified < _cfg.notify_interval:
                return GLib.SOURCE_CONTINUE
        except (ValueError, OSError):
            pass

    age_days = (now - last).days
    send_notification(
        summary="System update overdue",
        body=(
            f"Your system has not been updated in {age_days} days. "
            "Consider running a system update."
        ),
        icon="software-update-available",
    )
    _write_state(STALE_NOTIFY_FILE, now)
    return GLib.SOURCE_CONTINUE


def _startup_staleness_check() -> bool:
    check_staleness()
    return GLib.SOURCE_REMOVE


def send_notification(summary: str, body: str, icon: str) -> None:
    n = Notify.Notification.new(summary, body, icon)
    n.set_urgency(current_urgency())
    try:
        n.show()
    except GLib.Error:
        LOG.exception("Failed to send notification")


def on_tukit_notify(status: str) -> None:
    ts = timestamp()
    if str(status) == "success":
        write_last_update()
        send_notification(
            summary="Updates successfully installed",
            body=f"System has been upgraded on {ts}. Please reboot to take effect.",
            icon="appointment-soon",
        )
    else:
        send_notification(
            summary="Update process failed",
            body=f"An error was encountered while upgrading on {ts}.",
            icon="appointment-missed",
        )


def on_dnf_automatic_finished(success: bool, message: str = "") -> None:
    ts = timestamp()
    if bool(success):
        write_last_update()
        send_notification(
            summary="Automatic updates installed",
            body=f"Automatic updates were applied on {ts}. A reboot may be required.",
            icon="appointment-soon",
        )
    else:
        send_notification(
            summary="Automatic update failed",
            body=f"An error was encountered during automatic updates on {ts}.",
            icon="appointment-missed",
        )


def run_daemon() -> None:
    load_config()
    Notify.init(APP_NAME)
    dbus.mainloop.glib.DBusGMainLoop(set_as_default=True)

    try:
        system_bus = dbus.SystemBus()
    except dbus.DBusException:
        LOG.exception("Cannot connect to D-Bus system bus")
        sys.exit(1)

    system_bus.add_signal_receiver(
        handler_function=on_tukit_notify,
        signal_name=TUKIT_MEMBER,
        dbus_interface=TUKIT_INTERFACE,
        bus_name=None,
        path=TUKIT_PATH,
    )

    system_bus.add_signal_receiver(
        handler_function=on_dnf_automatic_finished,
        signal_name=DNF_AUTO_MEMBER,
        dbus_interface=DNF_AUTO_INTERFACE,
        bus_name=None,
        path=DNF_AUTO_PATH,
    )

    LOG.info("Detected backend: %s", _detect_backend().value)
    LOG.info("Listening on %s and %s", TUKIT_INTERFACE, DNF_AUTO_INTERFACE)

    GLib.timeout_add_seconds(60, _startup_staleness_check)
    GLib.timeout_add_seconds(CHECK_INTERVAL_SECONDS, check_staleness)

    loop = GLib.MainLoop()
    try:
        loop.run()
    except KeyboardInterrupt:
        pass


def run_client(status: str) -> None:
    """Acquire org.opensuse.tukit.Updated and emit Notify on the system bus.

    Must run as root — the D-Bus policy only grants ownership to root.
    Called by transactional-update when REBOOT_METHOD=notify.
    """
    dbus.mainloop.glib.DBusGMainLoop(set_as_default=True)

    try:
        system_bus = dbus.SystemBus()
        bus_name = dbus.service.BusName(TUKIT_INTERFACE, system_bus)
    except dbus.DBusException:
        LOG.exception("Cannot acquire %s — must run as root", TUKIT_INTERFACE)
        sys.exit(1)

    class _Notifier(dbus.service.Object):
        @dbus.service.signal(dbus_interface=TUKIT_INTERFACE, signature="s")
        def Notify(self, status: str) -> None:
            pass

    notifier = _Notifier(bus_name, TUKIT_PATH)
    loop = GLib.MainLoop()

    def _emit_and_quit() -> bool:
        notifier.Notify(status)
        LOG.info("Emitted %s.%s: %s", TUKIT_INTERFACE, TUKIT_MEMBER, status)
        loop.quit()
        return False

    GLib.idle_add(_emit_and_quit)
    loop.run()


def main() -> None:
    logging.basicConfig(
        level=logging.INFO,
        format="%(name)s[%(process)d]: %(levelname)s %(message)s",
    )

    if len(sys.argv) < 2 or sys.argv[1] not in ("daemon", "client"):
        print(f"Usage: {sys.argv[0]} {{daemon|client [success|failure]}}", file=sys.stderr)
        sys.exit(1)

    match sys.argv[1]:
        case "daemon":
            run_daemon()

        case "client":
            status = sys.argv[2] if len(sys.argv) > 2 else "success"
            if status not in ("success", "failure"):
                print("client: status must be 'success' or 'failure'", file=sys.stderr)
                sys.exit(1)
            run_client(status)


if __name__ == "__main__":
    main()
