"""Lizenzverwaltung auf Client-Seite (TESM) und fuer die eigene Lizenz des Servers.

Zustaendig fuer: Lizenzdatei laden und verifizieren, Aussteller pinnen,
Aktivierung/Deaktivierung/Heartbeat (online und offline), Modul-Freischaltung
und die Anzeige in der Topbar.

Wesentliche Aenderungen gegenueber dem Vorgaenger:

* **Vertrauensanker.** Die Signatur wird gegen gepinnte Aussteller-Schluessel
  geprueft (Tabelle ``trusted_issuer_keys``). Beim ersten Import wird der
  Schluessel per Trust-on-First-Use uebernommen und dabei protokolliert; jede
  spaetere Lizenz eines *anderen* Ausstellers wird abgelehnt, bis ein
  Administrator den neuen Schluessel ausdruecklich freigibt. Der Vorgaenger
  pruefte ausschliesslich gegen den Schluessel *in derselben Datei* -- damit war
  jede selbst signierte Datei "gueltig".
* **Trennung von Dokument und Client-Schluessel** (siehe ``tesm_licensing``):
  ``license.json`` (0640) und ``license_key.json`` (0600).
* **Ein Gate, eine Wahrheit.** Der Vorgaenger hatte eine Docstring, die mehr
  behauptete als der Code durchsetzte (LDAP/NGINX waren nur im Template
  gegated). Hier gibt es genau eine Funktion ``require_license()`` und eine
  Tabelle in ``LICENSED_FEATURES`` -- was dort steht, wird serverseitig
  durchgesetzt, ohne Ausnahme.
"""

from __future__ import annotations

import json
import logging
import sqlite3
import threading
import time
import urllib.error
import urllib.request
from dataclasses import dataclass, field
from datetime import datetime, timezone
from pathlib import Path
from typing import Any, Callable

import tesm_licensing as lic

from . import audit
from .db import Database
from .keystore import write_private_file

logger = logging.getLogger(__name__)

SETTING_STATUS = "license_activation_status"
SETTING_PENDING = "license_pending_request"
SETTING_LAST_HEARTBEAT = "license_last_heartbeat_at"
SETTING_HEARTBEAT_ERROR = "license_heartbeat_error"
SETTING_EXPORT_GRACE = "license_export_grace"

STATUS_INACTIVE = "inactive"
STATUS_PENDING = "pending"
STATUS_ACTIVE = "active"

API_PATHS = {
    "activate": "/api/v1/licenses/activate",
    "deactivate": "/api/v1/licenses/deactivate",
    "heartbeat": "/api/v1/licenses/heartbeat",
}

#: Funktionen, die eine gueltige, aktivierte Lizenz brauchen. Diese Tabelle ist
#: die einzige Wahrheit -- Templates lesen sie, Routen erzwingen sie.
LICENSED_FEATURES: dict[str, str] = {
    "directory": "Verzeichnisanmeldung (LDAP/AD) konfigurieren",
    "nginx": "Webserver-Einstellungen aendern",
    "export": "Daten exportieren",
    "trash_restore": "Papierkorb wiederherstellen und endgueltig loeschen",
    "monitor_interval": "Pruefintervall der Ueberwachung aendern",
    "auto_poe_restart": "Automatischer PoE-Neustart bei Ausfall",
}


def describe_transport_error(exc: BaseException, endpoint: str) -> str:
    """Uebersetzt einen Netzfehler in etwas, mit dem man arbeiten kann.

    "War nicht erreichbar" ist bei einem Zertifikatsfehler schlicht falsch --
    der Server war sehr wohl erreichbar, nur nicht vertrauenswuerdig. Wer das
    nicht weiss, sucht am falschen Ende.
    """
    text = str(exc)
    if "CERTIFICATE_VERIFY_FAILED" in text or "certificate verify failed" in text:
        detail = "Das Zertifikat konnte nicht geprueft werden"
        if "self-signed" in text or "self signed" in text:
            detail = "Der Lizenzserver benutzt ein selbstsigniertes Zertifikat"
        return (
            f"{detail} ({endpoint}). Hinterlegen Sie sein Zertifikat unter "
            "Einstellungen -> Lizenz, oder geben Sie dem Lizenzserver ein Zertifikat "
            "einer anerkannten Stelle."
        )
    if "Name or service not known" in text or "nodename nor servname" in text:
        return f"Der Name im Endpunkt ({endpoint}) laesst sich nicht aufloesen."
    if "Connection refused" in text:
        return f"{endpoint} nimmt keine Verbindung an."
    if "timed out" in text or "timeout" in text.lower():
        return f"{endpoint} hat nicht rechtzeitig geantwortet."
    return f"Der Lizenzserver ({endpoint}) war nicht erreichbar: {text[:160]}"


class LicenseError(Exception):
    """Fehler beim Import oder bei der Aktivierung -- Text ist fuer Anwender gedacht."""


@dataclass(slots=True)
class LicenseState:
    """Zwischengespeicherter Lizenzzustand. Wird bei jeder Aenderung neu gebaut."""

    present: bool = False
    valid: bool = False
    issuer_pinned: bool = False
    activated: bool = False
    document: dict[str, Any] = field(default_factory=dict)
    status: lic.LicenseStatus | None = None
    error: str = ""
    warnings: tuple[str, ...] = ()
    last_heartbeat_at: str = ""
    heartbeat_error: str = ""
    pending_action: str = ""

    @property
    def role(self) -> str:
        return self.status.role if self.status else "customer"

    @property
    def modules(self) -> frozenset[str]:
        return self.status.modules if self.status else frozenset()

    @property
    def customer_name(self) -> str:
        return str((self.status.customer if self.status else {}).get("name") or "")

    @property
    def vendor(self) -> dict[str, Any]:
        return dict(self.status.vendor) if self.status else {}


class LicenseManager:
    """Haelt den Lizenzzustand und fuehrt die Handshakes durch."""

    def __init__(
        self,
        *,
        database: Database,
        settings: Any,
        license_path: Path,
        key_path: Path,
        app_key: str,
        built_in_trusted_keys: tuple[str, ...] = (),
    ) -> None:
        self.database = database
        self.settings = settings
        self.license_path = Path(license_path)
        self.key_path = Path(key_path)
        self.app_key = app_key
        self.built_in_trusted_keys = tuple(built_in_trusted_keys)
        self._state = LicenseState()
        self._lock = threading.RLock()
        self._heartbeat_thread: threading.Thread | None = None
        self._report_builder: Callable[[sqlite3.Connection], dict[str, Any]] | None = None

    # ------------------------------------------------------------------ #
    # Zustand
    # ------------------------------------------------------------------ #

    @property
    def state(self) -> LicenseState:
        return self._state

    def trusted_keys(self, conn: sqlite3.Connection) -> list[str]:
        rows = Database.all(conn, "SELECT public_key FROM trusted_issuer_keys")
        return [*self.built_in_trusted_keys, *(str(row["public_key"]) for row in rows)]

    def reload(self, conn: sqlite3.Connection) -> LicenseState:
        """Liest Lizenzdatei und Status neu ein."""
        with self._lock:
            state = LicenseState()
            state.last_heartbeat_at = str(self.settings.get(conn, SETTING_LAST_HEARTBEAT) or "")
            state.heartbeat_error = str(self.settings.get(conn, SETTING_HEARTBEAT_ERROR) or "")
            pending = self._pending(conn)
            state.pending_action = str(pending.get("action") or "") if pending else ""

            if not self.license_path.is_file():
                state.error = "Es ist keine Lizenzdatei installiert."
                self._state = state
                return state

            try:
                document = json.loads(self.license_path.read_text(encoding="utf-8"))
            except (OSError, ValueError) as exc:
                state.error = f"Die Lizenzdatei ist unlesbar: {exc}"
                self._state = state
                return state

            if isinstance(document, dict) and "license" in document:
                # Jemand hat das Auslieferungs-Bundle direkt hierher kopiert.
                document, _ = lic.split_bundle(document)

            state.present = True
            state.document = document
            trusted = self.trusted_keys(conn)
            try:
                lic.verify_license(document, trusted_master_keys=trusted or None)
                state.valid = True
                state.issuer_pinned = bool(trusted)
            except lic.UntrustedIssuer as exc:
                state.error = str(exc)
            except lic.LicensingError as exc:
                state.error = str(exc)

            if state.valid:
                state.status = lic.license_status(
                    document, issuer_pinned=state.issuer_pinned
                )
                state.warnings = state.status.warnings

            state.activated = (
                str(self.settings.get(conn, SETTING_STATUS) or STATUS_INACTIVE) == STATUS_ACTIVE
            )
            self._state = state
            return state

    # ------------------------------------------------------------------ #
    # Gates
    # ------------------------------------------------------------------ #

    def active(self) -> bool:
        """Grundvoraussetzung fuer alles Lizenzpflichtige."""
        state = self._state
        return bool(
            state.valid
            and state.activated
            and state.status is not None
            and state.status.modules_active
        )

    def module_active(self, name: str) -> bool:
        """Ist ein Vollmodul (dhcp/fileshare/maintenance) freigeschaltet?"""
        return self.active() and name in self._state.modules

    def feature_allowed(self, feature: str) -> bool:
        """Lizenzpflichtige Einzelfunktion.

        Ausnahme ``export``: bleibt erlaubt, wenn der lizenzlose Zustand durch
        eine vom Kunden selbst ausgeloeste Deaktivierung entstanden ist
        (Systemwechsel) -- sonst waeren die eigenen Daten eingeschlossen.
        """
        if feature == "export" and not self.active():
            with self.database.session() as conn:
                return str(self.settings.get(conn, SETTING_EXPORT_GRACE) or "") == "1"
        return self.active()

    def require(self, feature: str) -> None:
        if not self.feature_allowed(feature):
            raise LicenseError(
                f"{LICENSED_FEATURES.get(feature, feature)} ist ohne gueltige, "
                "aktivierte Lizenz nicht moeglich."
            )

    # ------------------------------------------------------------------ #
    # Import
    # ------------------------------------------------------------------ #

    def import_bundle(
        self,
        conn: sqlite3.Connection,
        raw: bytes | str,
        *,
        actor: str,
        allow_new_issuer: bool = False,
    ) -> LicenseState:
        """Nimmt eine hochgeladene Lizenzdatei an.

        ``allow_new_issuer`` ist nur fuer den ausdruecklichen Fall gedacht, dass
        ein Administrator einen Aussteller-Wechsel bestaetigt (z. B. Wechsel des
        Lizenzservers). Ohne dieses Kennzeichen wird eine Datei mit unbekanntem
        Aussteller abgelehnt, sobald schon einer gepinnt ist.
        """
        text = raw.decode("utf-8") if isinstance(raw, bytes) else raw
        try:
            payload = json.loads(text)
        except ValueError as exc:
            raise LicenseError("Die Datei ist kein gueltiges JSON.") from exc

        try:
            document, client_key = lic.split_bundle(payload)
        except lic.LicensingError as exc:
            raise LicenseError(str(exc)) from exc

        try:
            lic.verify_license(document)  # innere Konsistenz
        except lic.LicensingError as exc:
            raise LicenseError(f"Die Lizenzdatei ist ungueltig: {exc}") from exc

        master_pubkey = str(document.get("master_pubkey") or "")
        known = self.trusted_keys(conn)
        if known and master_pubkey not in set(known) and not allow_new_issuer:
            raise LicenseError(
                "Diese Lizenz wurde von einem anderen Lizenzserver ausgestellt "
                f"(Schluessel {lic.key_id(master_pubkey)}). Ein Administrator muss den "
                "Wechsel des Ausstellers ausdruecklich bestaetigen."
            )

        if not client_key and not self.key_path.is_file():
            raise LicenseError(
                "Der Datei fehlt der Client-Schluessel. Bitte das vollstaendige "
                "Lizenz-Bundle des Lizenzservers verwenden."
            )

        # Reihenfolge: erst Schluessel pinnen, dann Dateien schreiben, dann
        # Aktivierungsstatus zuruecksetzen -- ein Abbruch dazwischen hinterlaesst
        # keinen halb gueltigen Zustand.
        if master_pubkey not in set(known):
            conn.execute(
                "INSERT INTO trusted_issuer_keys (key_id, public_key, label, source, added_at, added_by) "
                "VALUES (?,?,?,?,?,?) ON CONFLICT(key_id) DO NOTHING",
                (
                    lic.key_id(master_pubkey),
                    master_pubkey,
                    str((document.get("vendor") or {}).get("name") or "Lizenzserver"),
                    "tofu" if not known else "admin-confirmed",
                    datetime.now(timezone.utc).isoformat(timespec="seconds"),
                    actor,
                ),
            )
            audit.log(
                conn,
                actor=actor,
                action="license.issuer_pinned",
                target=lic.key_id(master_pubkey),
                detail="Trust on first use" if not known else "Aussteller-Wechsel bestaetigt",
                severity="notice",
            )

        self.license_path.parent.mkdir(parents=True, exist_ok=True)
        self.license_path.write_text(
            json.dumps(document, indent=2, sort_keys=True, ensure_ascii=False) + "\n",
            encoding="utf-8",
        )
        try:
            self.license_path.chmod(0o640)
        except OSError:
            pass
        if client_key:
            write_private_file(
                self.key_path,
                json.dumps({"license_id": document.get("license_id"), "client_key": client_key})
                + "\n",
            )

        self.settings.set(conn, SETTING_STATUS, STATUS_INACTIVE, actor=actor)
        self.settings.set(conn, SETTING_PENDING, "", actor=actor)
        self.settings.set(conn, SETTING_EXPORT_GRACE, "", actor=actor)
        audit.log(
            conn,
            actor=actor,
            action="license.imported",
            target=str(document.get("license_id") or ""),
            detail=f"Typ {document.get('type')}, Rolle {document.get('role')}",
            severity="notice",
        )
        return self.reload(conn)

    def client_key(self) -> str:
        if not self.key_path.is_file():
            return ""
        try:
            data = json.loads(self.key_path.read_text(encoding="utf-8"))
        except (OSError, ValueError):
            return ""
        return str(data.get("client_key") or "")

    # ------------------------------------------------------------------ #
    # Handshake
    # ------------------------------------------------------------------ #

    def _pending(self, conn: sqlite3.Connection) -> dict[str, Any]:
        raw = self.settings.get(conn, SETTING_PENDING) or ""
        if not raw:
            return {}
        try:
            return json.loads(raw) if isinstance(raw, str) else dict(raw)
        except ValueError:
            return {}

    def _endpoint(self, action: str) -> str:
        base = str(self._state.document.get("master_endpoint") or "").rstrip("/")
        if not base:
            return ""
        return base + API_PATHS[action]

    def build_request(
        self, action: str, *, hostname: str = "", report: dict[str, Any] | None = None
    ) -> dict[str, Any]:
        if not self._state.valid:
            raise LicenseError("Es ist keine gueltige Lizenz installiert.")
        key = self.client_key()
        if not key:
            raise LicenseError(
                "Der Client-Schluessel dieser Lizenz fehlt. Bitte die Lizenzdatei erneut importieren."
            )
        return lic.build_client_request(
            action, self._state.document, key, hostname=hostname, report=report
        )

    def handshake(
        self,
        conn: sqlite3.Connection,
        action: str,
        *,
        actor: str,
        hostname: str = "",
        report: dict[str, Any] | None = None,
        timeout: int = 12,
    ) -> dict[str, Any]:
        """Fuehrt eine Aktion durch. Rueckgabe beschreibt das Ergebnis fuer die UI.

        ``{"ok": bool, "transport": "online"|"offline", "code": str, "message": str}``
        """
        request_payload = self.build_request(action, hostname=hostname, report=report)
        endpoint = self._endpoint(action)

        reason = ""
        if endpoint:
            try:
                response = self._post_json(
                    endpoint,
                    request_payload,
                    timeout=timeout,
                    ca_pem=str(self.settings.get(conn, "license_server_ca") or ""),
                )
            except Exception as exc:  # noqa: BLE001 - jeder Netzfehler fuehrt offline weiter
                reason = describe_transport_error(exc, endpoint)
                logger.info("Online-Handshake fehlgeschlagen (%s): %s", action, exc)
            else:
                return self._apply_response(
                    conn, action, request_payload, response, actor=actor, transport="online"
                )
        else:
            reason = (
                "In dieser Lizenz ist kein Lizenzserver hinterlegt -- sie kann sich nur "
                "ueber den Offline-Weg aktivieren."
            )

        # Offline-Weg: Anfrage merken -- **mit** dem Code. Zwischen Anfrage und
        # Antwort liegen in der Praxis eine E-Mail und ein Arbeitstag; der Code
        # muss ein Neuladen der Seite ueberstehen.
        code = lic.encode_code(request_payload)
        self.settings.set(
            conn,
            SETTING_PENDING,
            json.dumps(
                {
                    "action": action,
                    "nonce": request_payload["nonce"],
                    "fingerprint": request_payload["fingerprint"],
                    "license_id": request_payload["license_id"],
                    "created_at": datetime.now(timezone.utc).isoformat(timespec="seconds"),
                    "code": code,
                    "reason": reason,
                }
            ),
            actor=actor,
        )
        if action == "activate":
            self.settings.set(conn, SETTING_STATUS, STATUS_PENDING, actor=actor)
        self.reload(conn)
        return {
            "ok": False,
            "transport": "offline",
            "code": code,
            "message": (
                (reason + " " if reason else "")
                + "Der Offline-Weg steht offen: uebermitteln Sie den angezeigten Code an "
                "Ihren Ansprechpartner und tragen Sie die Antwort hier ein."
            ),
        }

    def submit_offline_response(
        self, conn: sqlite3.Connection, code: str, *, actor: str
    ) -> dict[str, Any]:
        """Nimmt den Antwortcode des Lizenzservers an (Offline-Weg)."""
        pending = self._pending(conn)
        if not pending:
            raise LicenseError("Es liegt keine offene Anfrage vor.")
        try:
            response = lic.decode_code(code)
        except ValueError as exc:
            raise LicenseError(f"Der Code ist unlesbar: {exc}") from exc
        return self._apply_response(
            conn,
            str(pending.get("action") or ""),
            {
                "license_id": str(pending.get("license_id") or ""),
                "fingerprint": str(pending.get("fingerprint") or ""),
                "nonce": str(pending.get("nonce") or ""),
            },
            response,
            actor=actor,
            transport="offline",
        )

    def _apply_response(
        self,
        conn: sqlite3.Connection,
        action: str,
        request_payload: dict[str, Any],
        response: dict[str, Any],
        *,
        actor: str,
        transport: str,
    ) -> dict[str, Any]:
        master_pubkey = str(self._state.document.get("master_pubkey") or "")
        current_seq = int(self._state.document.get("seq") or 1)
        try:
            lic.verify_master_response(
                response,
                master_pubkey,
                expected_license_id=str(request_payload["license_id"]),
                expected_fingerprint=str(request_payload["fingerprint"]),
                expected_action=action,
                expected_nonce=str(request_payload["nonce"]),
                current_seq=current_seq,
            )
        except lic.LicensingError as exc:
            raise LicenseError(f"Die Antwort des Lizenzservers ist nicht gueltig: {exc}") from exc

        status = str(response.get("status") or "")
        detail = str(response.get("detail") or "")
        update = response.get("license_update")

        if isinstance(update, dict):
            try:
                lic.verify_license(update, trusted_master_keys=self.trusted_keys(conn) or None)
            except lic.LicensingError as exc:
                logger.warning("Mitgeliefertes Lizenz-Update verworfen: %s", exc)
            else:
                self.license_path.write_text(
                    json.dumps(update, indent=2, sort_keys=True, ensure_ascii=False) + "\n",
                    encoding="utf-8",
                )
                audit.log(
                    conn,
                    actor=actor,
                    action="license.updated_by_master",
                    target=str(update.get("license_id") or ""),
                    detail=f"Revision {update.get('seq')}",
                    severity="notice",
                )

        if status == "revoked":
            self.settings.set(conn, SETTING_STATUS, STATUS_INACTIVE, actor=actor)
            self.settings.set(conn, SETTING_PENDING, "", actor=actor)
            audit.log(
                conn,
                actor=actor,
                action="license.revoked_by_master",
                target=str(request_payload["license_id"]),
                severity="warning",
            )
            self.reload(conn)
            return {
                "ok": False,
                "transport": transport,
                "code": "",
                "message": "Diese Lizenz wurde vom Lizenzserver widerrufen.",
            }

        if status != "ok":
            self.reload(conn)
            return {
                "ok": False,
                "transport": transport,
                "code": "",
                "message": detail or f"Der Lizenzserver hat die Anfrage abgelehnt ({status}).",
            }

        if action == "activate":
            self.settings.set(conn, SETTING_STATUS, STATUS_ACTIVE, actor=actor)
            self.settings.set(conn, SETTING_EXPORT_GRACE, "", actor=actor)
            message = "Die Lizenz ist jetzt aktiv."
        elif action == "deactivate":
            self.settings.set(conn, SETTING_STATUS, STATUS_INACTIVE, actor=actor)
            # Der Kunde hat selbst deaktiviert (Systemwechsel): Export bleibt
            # erlaubt, damit er seine Daten mitnehmen kann.
            self.settings.set(conn, SETTING_EXPORT_GRACE, "1", actor=actor)
            message = "Die Lizenz wurde deaktiviert und kann auf einem anderen System aktiviert werden."
        else:
            self.settings.set(
                conn,
                SETTING_LAST_HEARTBEAT,
                datetime.now(timezone.utc).isoformat(timespec="seconds"),
                actor=actor,
            )
            self.settings.set(conn, SETTING_HEARTBEAT_ERROR, "", actor=actor)
            message = "Verbindung zum Lizenzserver bestaetigt."

        self.settings.set(conn, SETTING_PENDING, "", actor=actor)
        audit.log(
            conn,
            actor=actor,
            action=f"license.{action}",
            target=str(request_payload["license_id"]),
            detail=f"Transport: {transport}",
            severity="notice",
        )
        self.reload(conn)
        return {"ok": True, "transport": transport, "code": "", "message": message}

    @staticmethod
    def _tls_context(ca_pem: str) -> Any:
        """Prueft gegen ein hinterlegtes Zertifikat -- oder gegen die Systemanker.

        Nie ohne Pruefung: das Lizenzprotokoll ist zwar selbst signiert und
        gegen Wiedereinspielung geschuetzt, aber das ist kein Grund, den
        Transport offen zu lassen.
        """
        import ssl

        context = ssl.create_default_context()
        pem = (ca_pem or "").strip()
        if not pem:
            return context
        try:
            # cadata nimmt PEM direkt entgegen -- keine temporaere Datei, die
            # liegen bleiben oder mitgelesen werden koennte.
            context.load_verify_locations(cadata=pem)
        except (ssl.SSLError, ValueError) as exc:
            raise ValueError(
                "Das hinterlegte Zertifikat des Lizenzservers ist unlesbar: " + str(exc)
            ) from exc
        return context

    @staticmethod
    def _post_json(
        url: str, payload: dict[str, Any], *, timeout: int, ca_pem: str = ""
    ) -> dict[str, Any]:
        if not url.startswith(("https://", "http://")):
            raise ValueError("Ungueltiger Endpunkt")
        data = json.dumps(payload).encode("utf-8")
        request = urllib.request.Request(  # noqa: S310 - Schema oben geprueft
            url,
            data=data,
            headers={"Content-Type": "application/json", "Accept": "application/json"},
            method="POST",
        )
        context = LicenseManager._tls_context(ca_pem) if url.startswith("https://") else None
        try:
            with urllib.request.urlopen(  # noqa: S310
                request, timeout=timeout, context=context
            ) as response:
                body = response.read().decode("utf-8")
        except urllib.error.HTTPError as exc:
            body = exc.read().decode("utf-8", errors="replace")
            try:
                parsed = json.loads(body)
            except ValueError:
                raise
            if isinstance(parsed, dict) and "signature" in parsed:
                return parsed  # Auch abweisende Antworten sind signiert.
            raise
        return json.loads(body)

    # ------------------------------------------------------------------ #
    # Heartbeat im Hintergrund
    # ------------------------------------------------------------------ #

    def set_report_builder(self, builder: Callable[[sqlite3.Connection], dict[str, Any]]) -> None:
        """Nur der Lizenzserver setzt das: Bericht an den eigenen Master."""
        self._report_builder = builder

    def start_heartbeat(self, *, hostname: str = "") -> None:
        if self._heartbeat_thread is not None:
            return

        def loop() -> None:
            # Erster Lauf verzoegert: der Dienst soll erst sauber hochkommen.
            time.sleep(90)
            while True:
                try:
                    self.run_heartbeat(hostname=hostname)
                except Exception as exc:  # noqa: BLE001
                    logger.warning("Heartbeat fehlgeschlagen: %s", exc)
                time.sleep(lic.HEARTBEAT_INTERVAL_SECONDS)

        thread = threading.Thread(target=loop, name="license-heartbeat", daemon=True)
        thread.start()
        self._heartbeat_thread = thread
        logger.info("Lizenz-Heartbeat gestartet (Intervall %ss)", lic.HEARTBEAT_INTERVAL_SECONDS)

    def run_heartbeat(self, *, hostname: str = "") -> dict[str, Any]:
        with self.database.session() as conn:
            self.reload(conn)
            if not self._state.valid or not self._state.activated:
                return {"ok": False, "message": "Keine aktive Lizenz."}
            report = None
            if self._report_builder is not None and self._state.role == "license_server":
                report = self._report_builder(conn)
            try:
                with self.database.transaction(conn):
                    return self.handshake(
                        conn, "heartbeat", actor="system", hostname=hostname, report=report
                    )
            except LicenseError as exc:
                with self.database.transaction(conn):
                    self.settings.set(conn, SETTING_HEARTBEAT_ERROR, str(exc), actor="system")
                return {"ok": False, "message": str(exc)}

    # ------------------------------------------------------------------ #
    # Anzeige
    # ------------------------------------------------------------------ #

    def topbar_pill(self) -> dict[str, str] | None:
        """Zustandsanzeige fuer die Topbar. Reihenfolge = Dringlichkeit."""
        state = self._state
        if not state.present:
            return {
                "label": "Keine Lizenz",
                "css_class": "status-pill--danger",
                "dot": "1",
                "dot_class": "status-pill__dot--pulse",
                "title": "Es ist keine Lizenzdatei installiert.",
            }
        if not state.valid:
            return {
                "label": "Lizenz ungueltig",
                "css_class": "status-pill--danger",
                "dot": "1",
                "dot_class": "status-pill__dot--pulse",
                "title": state.error,
            }
        if not state.activated:
            return {
                "label": "Lizenz nicht aktiviert",
                "css_class": "status-pill--warning",
                "dot": "1",
                "dot_class": "status-pill__dot--pulse",
                "title": "Die Lizenz muss noch aktiviert werden.",
            }
        status = state.status
        if status is None:
            return None
        if status.expired:
            days = status.days_since_expiry or 0
            return {
                "label": f"Lizenz abgelaufen (seit {days} Tagen)",
                "css_class": "status-pill--danger",
                "dot": "1",
                "dot_class": "status-pill__dot--pulse",
                "title": (
                    "Die Kulanzfrist laeuft noch, Funktionen bleiben vorerst verfuegbar."
                    if status.grace_active
                    else "Lizenzpflichtige Funktionen sind abgeschaltet."
                ),
            }
        if status.expiring_soon:
            return {
                "label": f"Lizenz laeuft in {status.days_left} Tagen ab",
                "css_class": "status-pill--warning",
                "dot": "1",
                "dot_class": "status-pill__dot--pulse",
                "title": "Bitte rechtzeitig verlaengern.",
            }
        if state.last_heartbeat_at:
            try:
                last = lic.parse_iso(state.last_heartbeat_at)
                days = (lic.utcnow() - last).days
            except ValueError:
                days = 0
            if days >= lic.HEARTBEAT_WARNING_DAYS:
                return {
                    "label": f"Lizenzserver seit {days} Tagen nicht erreichbar",
                    "css_class": "status-pill--warning",
                    "dot": "1",
                    "dot_class": "",
                    "title": "Rein informativ -- es wird nichts abgeschaltet.",
                }
        return None
