tesm-license/apps/tesm-license/src/tesm_license/services/issuance.py

"""Kunden, Tickets und die Ausstellung von Lizenzen."""

from __future__ import annotations

import json
import re
import secrets
import sqlite3
from dataclasses import dataclass
from datetime import datetime, timezone
from typing import Any, Iterable, Sequence

import tesm_licensing as lic
import tesm_licensing_server as licsrv
from tesm_core.db import Database

TICKET_TYPES = lic.LICENSE_TYPES
TICKET_STATES = ("open", "blocked")
LICENSE_STATES = ("issued", "active", "deactivated", "revoked")

#: Ampel-Schwellen der Anzeige. Bewusst getrennt von der Kulanzlogik in
#: ``tesm_licensing`` -- gleiche Zahl (30), anderes Konzept.
GREEN_ABOVE_DAYS = 90
RED_BELOW_DAYS = 30

EMAIL_RE = re.compile(r"^[^@\s]+@[^@\s]+\.[^@\s]+$")


class IssuanceError(Exception):
    """Eingabe- oder Regelverstoss mit Klartextmeldung."""


def _now() -> str:
    return datetime.now(timezone.utc).isoformat(timespec="seconds")


# --------------------------------------------------------------------------- #
# Kunden
# --------------------------------------------------------------------------- #


def validate_email(value: str, *, required: bool) -> str:
    text = (value or "").strip()
    if not text:
        if required:
            raise IssuanceError("Eine E-Mail-Adresse ist erforderlich.")
        # Leer bleibt erlaubt: Altkunden von vor der Pflichtfeld-Einfuehrung
        # sollen weiter bearbeitbar bleiben, statt ausgesperrt zu werden.
        return ""
    if not EMAIL_RE.match(text):
        raise IssuanceError(f"{text!r} ist keine gueltige E-Mail-Adresse.")
    return text


def list_customers(conn: sqlite3.Connection) -> list[dict[str, Any]]:
    return Database.all(
        conn,
        """
        SELECT c.*,
               (SELECT COUNT(*) FROM license_tickets t WHERE t.customer_id = c.id) AS ticket_count,
               (SELECT COUNT(*) FROM licenses l
                 WHERE l.customer_id = c.id AND l.status IN ('issued','active')) AS open_licenses,
               (SELECT l.license_id FROM licenses l
                 WHERE l.customer_id = c.id AND l.status='active' AND l.role='license_server'
                 LIMIT 1) AS license_server_id
          FROM license_customers c
         ORDER BY c.name COLLATE NOCASE
        """,
    )


def get_customer(conn: sqlite3.Connection, customer_id: int) -> dict[str, Any] | None:
    return Database.one(conn, "SELECT * FROM license_customers WHERE id=?", (customer_id,))


def create_customer(conn: sqlite3.Connection, *, actor: str, **fields: Any) -> int:
    name = str(fields.get("name") or "").strip()
    if not name:
        raise IssuanceError("Der Firmenname darf nicht leer sein.")
    email = validate_email(str(fields.get("contact_email") or ""), required=True)
    cursor = conn.execute(
        "INSERT INTO license_customers (name, contact_email, contact_phone, contact_person, "
        "street, house_number, postal_code, city, notes, created_at, updated_at, created_by) "
        "VALUES (?,?,?,?,?,?,?,?,?,?,?,?)",
        (
            name,
            email,
            str(fields.get("contact_phone") or "").strip(),
            str(fields.get("contact_person") or "").strip(),
            str(fields.get("street") or "").strip(),
            str(fields.get("house_number") or "").strip(),
            str(fields.get("postal_code") or "").strip(),
            str(fields.get("city") or "").strip(),
            str(fields.get("notes") or "").strip(),
            _now(),
            _now(),
            actor,
        ),
    )
    return int(cursor.lastrowid or 0)


def update_customer(conn: sqlite3.Connection, customer_id: int, **fields: Any) -> list[str]:
    row = get_customer(conn, customer_id)
    if row is None:
        raise IssuanceError("Dieser Kunde existiert nicht.")
    name = str(fields.get("name") or "").strip()
    if not name:
        raise IssuanceError("Der Firmenname darf nicht leer sein.")
    email = validate_email(str(fields.get("contact_email") or ""), required=False)

    columns = {
        "name": name,
        "contact_email": email,
        "contact_phone": str(fields.get("contact_phone") or "").strip(),
        "contact_person": str(fields.get("contact_person") or "").strip(),
        "street": str(fields.get("street") or "").strip(),
        "house_number": str(fields.get("house_number") or "").strip(),
        "postal_code": str(fields.get("postal_code") or "").strip(),
        "city": str(fields.get("city") or "").strip(),
        "notes": str(fields.get("notes") or "").strip(),
    }
    changes = [key for key, value in columns.items() if str(row.get(key) or "") != value]
    if not changes:
        return []
    assignments = ", ".join(f"{key}=?" for key in columns)
    conn.execute(
        f"UPDATE license_customers SET {assignments}, updated_at=? WHERE id=?",
        (*columns.values(), _now(), customer_id),
    )
    return changes


def delete_customer(conn: sqlite3.Connection, customer_id: int) -> None:
    tickets = Database.value(
        conn, "SELECT COUNT(*) FROM license_tickets WHERE customer_id=?", (customer_id,), default=0
    )
    if int(tickets or 0):
        raise IssuanceError(
            f"An diesem Kunden haengen noch {tickets} Ticket(s). Bitte zuerst entfernen."
        )
    conn.execute("DELETE FROM license_customers WHERE id=?", (customer_id,))


def find_customer_by_email(conn: sqlite3.Connection, email: str) -> dict[str, Any] | None:
    return Database.one(
        conn,
        "SELECT * FROM license_customers WHERE contact_email <> '' "
        "AND lower(contact_email)=lower(?)",
        ((email or "").strip(),),
    )


# --------------------------------------------------------------------------- #
# Tickets
# --------------------------------------------------------------------------- #


def list_tickets(
    conn: sqlite3.Connection, *, customer_id: int | None = None
) -> list[dict[str, Any]]:
    clause = "WHERE t.customer_id = ?" if customer_id else ""
    params = (customer_id,) if customer_id else ()
    return Database.all(
        conn,
        f"""
        SELECT t.*, c.name AS customer_name, c.contact_email,
               (SELECT COUNT(*) FROM licenses l WHERE l.ticket_id = t.id) AS license_count,
               (SELECT l.license_id FROM licenses l
                 WHERE l.ticket_id = t.id AND l.status IN ('issued','active') LIMIT 1) AS open_license_id
          FROM license_tickets t
          JOIN license_customers c ON c.id = t.customer_id
          {clause}
         ORDER BY t.created_at DESC
        """,
        params,
    )


def get_ticket(conn: sqlite3.Connection, ticket_id: str) -> dict[str, Any] | None:
    return Database.one(
        conn,
        "SELECT t.*, c.name AS customer_name, c.contact_email, c.contact_person "
        "FROM license_tickets t JOIN license_customers c ON c.id = t.customer_id WHERE t.id=?",
        (ticket_id,),
    )


def normalize_modules(license_type: str, role: str, modules: Iterable[str]) -> list[str]:
    """Bestimmt den Modulsatz eines Tickets.

    Regeln (bewusst hier zentral, nicht in der Oberflaeche):
    * ``role != "customer"``: immer leer -- das TESM-Modulkonzept ist fuer einen
      Lizenzserver bedeutungslos.
    * ``trial``/``standard``: leer (Trial bekommt den vollen Satz erst bei der
      Ausstellung als Trial-Override).
    * ``enterprise``: alle Module.
    * ``custom``: die uebergebene Auswahl.
    """
    if role != "customer":
        return []
    if license_type == "enterprise":
        return sorted(lic.ALL_MODULES)
    if license_type == "custom":
        return sorted(set(modules) & set(lic.ALL_MODULES))
    return []


def create_ticket(
    conn: sqlite3.Connection,
    *,
    customer_id: int,
    license_type: str,
    modules: Sequence[str],
    valid_days: int,
    lifetime: bool,
    role: str,
    notes: str = "",
    actor: str = "",
) -> str:
    if get_customer(conn, customer_id) is None:
        raise IssuanceError("Dieser Kunde existiert nicht.")
    if license_type not in TICKET_TYPES:
        raise IssuanceError("Unbekannter Lizenztyp.")
    if role not in lic.LICENSE_ROLES or role == "master":
        # "master" entsteht ausschliesslich lokal per Bootstrap-Skript.
        raise IssuanceError("Unzulaessige Rolle.")

    # Trial + unbefristet ist ein sinnloser Zustand: eine permanente Lizenz ohne
    # Module. Diese Regel wird an drei Stellen geprueft (hier, beim Bearbeiten
    # und unmittelbar vor dem Signieren).
    if lifetime and license_type == "trial":
        raise IssuanceError("Eine Testlizenz kann nicht unbefristet sein.")
    if not lifetime and int(valid_days) <= 0:
        raise IssuanceError("Die Laufzeit muss positiv sein.")

    ticket_id = secrets.token_urlsafe(24)
    conn.execute(
        "INSERT INTO license_tickets (id, customer_id, type, modules, valid_days, lifetime, "
        "status, role, notes, created_by, created_at, updated_at) "
        "VALUES (?,?,?,?,?,?,'open',?,?,?,?,?)",
        (
            ticket_id,
            customer_id,
            license_type,
            json.dumps(normalize_modules(license_type, role, modules)),
            int(valid_days),
            int(bool(lifetime)),
            role,
            (notes or "").strip(),
            actor,
            _now(),
            _now(),
        ),
    )
    return ticket_id


def update_ticket(
    conn: sqlite3.Connection,
    ticket_id: str,
    *,
    license_type: str,
    modules: Sequence[str],
    valid_days: int,
    lifetime: bool,
    notes: str = "",
) -> list[str]:
    """Aendert die Parameter eines Tickets.

    Wirkt ausschliesslich auf die **naechste** Ausstellung. Bereits ausgestellte
    Lizenzen behalten die Werte, mit denen sie signiert wurden -- eine
    nachtraegliche Aenderung waere sonst eine Signaturaenderung ohne neue Datei.

    Kunde und Rolle sind nach der Anlage unveraenderlich: das Self-Service-Ticket
    ist an genau diesen Kunden gebunden, und ein Rollenwechsel wuerde eine
    Vertrauensgrenze verschieben.
    """
    ticket = get_ticket(conn, ticket_id)
    if ticket is None:
        raise IssuanceError("Dieses Ticket existiert nicht.")
    if license_type not in TICKET_TYPES:
        raise IssuanceError("Unbekannter Lizenztyp.")
    if lifetime and license_type == "trial":
        raise IssuanceError("Eine Testlizenz kann nicht unbefristet sein.")

    role = str(ticket["role"])
    new_modules = json.dumps(normalize_modules(license_type, role, modules))
    changes: list[str] = []
    if license_type != str(ticket["type"]):
        changes.append(f"Typ: {ticket['type']} -> {license_type}")
    if new_modules != str(ticket["modules"]):
        changes.append("Module geaendert")
    if int(valid_days) != int(ticket["valid_days"]):
        changes.append(f"Laufzeit: {ticket['valid_days']} -> {valid_days} Tage")
    if bool(lifetime) != bool(ticket["lifetime"]):
        changes.append(f"Unbefristet: {bool(ticket['lifetime'])} -> {bool(lifetime)}")

    conn.execute(
        "UPDATE license_tickets SET type=?, modules=?, valid_days=?, lifetime=?, notes=?, "
        "updated_at=? WHERE id=?",
        (
            license_type,
            new_modules,
            int(valid_days),
            int(bool(lifetime)),
            (notes or "").strip(),
            _now(),
            ticket_id,
        ),
    )
    return changes


def set_ticket_status(conn: sqlite3.Connection, ticket_id: str, status: str) -> None:
    if status not in TICKET_STATES:
        raise IssuanceError("Unbekannter Ticketzustand.")
    conn.execute(
        "UPDATE license_tickets SET status=?, updated_at=? WHERE id=?",
        (status, _now(), ticket_id),
    )


def ticket_has_open_license(conn: sqlite3.Connection, ticket_id: str) -> bool:
    return (
        Database.value(
            conn,
            "SELECT COUNT(*) FROM licenses WHERE ticket_id=? AND status IN ('issued','active')",
            (ticket_id,),
            default=0,
        )
        or 0
    ) > 0


def delete_ticket(conn: sqlite3.Connection, ticket_id: str) -> None:
    if ticket_has_open_license(conn, ticket_id):
        raise IssuanceError(
            "Zu diesem Ticket existiert noch eine offene Lizenz. Bitte zuerst widerrufen."
        )
    # Historische Lizenzzeilen haengen per ON DELETE CASCADE mit -- sonst blieben
    # verwaiste Zeilen mit toter ticket_id zurueck.
    conn.execute("DELETE FROM license_tickets WHERE id=?", (ticket_id,))


# --------------------------------------------------------------------------- #
# Lizenzen
# --------------------------------------------------------------------------- #


def list_licenses(
    conn: sqlite3.Connection,
    *,
    customer_id: int | None = None,
    ticket_id: str | None = None,
    status: str | None = None,
) -> list[dict[str, Any]]:
    clauses: list[str] = []
    params: list[Any] = []
    if customer_id:
        clauses.append("l.customer_id = ?")
        params.append(customer_id)
    if ticket_id:
        clauses.append("l.ticket_id = ?")
        params.append(ticket_id)
    if status:
        clauses.append("l.status = ?")
        params.append(status)
    where = f"WHERE {' AND '.join(clauses)}" if clauses else ""
    return Database.all(
        conn,
        f"""
        SELECT l.*, c.name AS customer_name, c.contact_email
          FROM licenses l JOIN license_customers c ON c.id = l.customer_id
          {where}
         ORDER BY l.created_at DESC
        """,
        tuple(params),
    )


def get_license(conn: sqlite3.Connection, license_id: str) -> dict[str, Any] | None:
    return Database.one(
        conn,
        "SELECT l.*, c.name AS customer_name, c.contact_email, c.contact_person "
        "FROM licenses l JOIN license_customers c ON c.id = l.customer_id WHERE l.license_id=?",
        (license_id,),
    )


@dataclass(frozen=True, slots=True)
class IssuedLicense:
    license_id: str
    bundle: dict[str, Any]
    document: dict[str, Any]
    already_existed: bool = False


def client_key_aad(license_id: str) -> str:
    return f"licenses.client_key:{license_id}"


def issue_from_ticket(
    conn: sqlite3.Connection,
    ticket_id: str,
    *,
    master_private_key: str,
    master_public_key: str,
    master_endpoint: str,
    vendor: dict[str, Any],
    actor: str,
    keystore: Any,
    as_trial: bool = False,
) -> IssuedLicense:
    """Stellt aus einem Ticket eine Lizenz aus.

    ``as_trial`` ist ein reiner **Ausstellungs-Override**: 30 Tage, alle Module,
    Typ ``trial``. Das Ticket selbst bleibt dabei unveraendert -- nur diese eine
    Datei traegt Trial-Konditionen. Fuer Tickets mit einer anderen Rolle als
    ``customer`` ist der Override sinnlos und wird abgewiesen.
    """
    ticket = get_ticket(conn, ticket_id)
    if ticket is None:
        raise IssuanceError("Dieses Ticket existiert nicht.")
    if str(ticket["status"]) != "open":
        raise IssuanceError("Dieses Ticket ist gesperrt.")

    role = str(ticket["role"])
    if as_trial and role != "customer":
        raise IssuanceError(
            "Eine Testlizenz gibt es nur fuer Kundeninstanzen, nicht fuer Lizenzserver."
        )

    if as_trial:
        license_type = "trial"
        modules = sorted(lic.ALL_MODULES)
        valid_days = lic.TRIAL_DEFAULT_DAYS
        lifetime = False
    else:
        license_type = str(ticket["type"])
        modules = json.loads(str(ticket["modules"] or "[]"))
        valid_days = int(ticket["valid_days"])
        lifetime = bool(ticket["lifetime"])

    # Dritte Absicherung derselben Regel, unmittelbar vor dem Signieren.
    if lifetime and license_type == "trial":
        raise IssuanceError("Eine Testlizenz kann nicht unbefristet sein.")

    # Der Endpunkt wird mitsigniert. Ohne ihn entsteht eine Lizenz, die sich
    # nie online aktivieren laesst -- und das laesst sich nachtraeglich nur
    # durch eine neue Signatur beheben, nicht durch eine Einstellung. Deshalb
    # hier, an der einzigen Stelle, durch die jede Ausstellung laeuft.
    if not str(master_endpoint or "").strip():
        raise IssuanceError(
            "Es ist kein oeffentlicher Endpunkt hinterlegt (Einstellungen -> Anbieter). "
            "Ohne ihn koennte sich diese Lizenz nie online aktivieren -- der Endpunkt "
            "wird in die Lizenz signiert und laesst sich spaeter nicht nachtragen."
        )

    customer = get_customer(conn, int(ticket["customer_id"]))
    assert customer is not None
    bundle, license_pub = licsrv.issue_license(
        customer={
            "name": str(customer["name"]),
            "contact_email": str(customer["contact_email"] or ""),
            "customer_ref": str(customer["id"]),
        },
        license_type=license_type,
        modules=modules,
        valid_days=valid_days,
        master_private_key_b64=master_private_key,
        master_public_key_b64=master_public_key,
        master_endpoint=master_endpoint,
        vendor=vendor,
        role=role,
        lifetime=lifetime,
    )
    document = bundle["license"]

    try:
        conn.execute(
            "INSERT INTO licenses (license_id, ticket_id, customer_id, type, role, modules, "
            "lifetime, seq, issued_at, expires_at, license_pubkey, license_document, "
            "client_key_encrypted, status, created_by, created_at) "
            "VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,'issued',?,?)",
            (
                document["license_id"],
                ticket_id,
                int(ticket["customer_id"]),
                license_type,
                role,
                json.dumps(modules),
                int(bool(lifetime)),
                int(document["seq"]),
                document["issued_at"],
                document["expires_at"],
                license_pub,
                json.dumps(document),
                keystore.encrypt(
                    str(bundle["client_key"]), aad=client_key_aad(str(document["license_id"]))
                ),
                actor,
                _now(),
            ),
        )
    except sqlite3.IntegrityError:
        # Der partielle Unique-Index hat zugeschlagen: ein zweiter Klick oder ein
        # zweiter Tab war schneller. Wir geben die tatsaechlich erzeugte Lizenz
        # zurueck statt einen Fehler zu zeigen.
        existing = Database.one(
            conn,
            "SELECT license_id, license_document FROM licenses "
            "WHERE ticket_id=? AND status IN ('issued','active')",
            (ticket_id,),
        )
        if existing is None:
            raise
        return IssuedLicense(
            license_id=str(existing["license_id"]),
            bundle={},
            document=json.loads(str(existing["license_document"])),
            already_existed=True,
        )

    return IssuedLicense(
        license_id=str(document["license_id"]), bundle=bundle, document=document
    )


def delivery_bundle(conn: sqlite3.Connection, license_id: str, *, keystore: Any) -> dict[str, Any]:
    """Baut das Auslieferungs-Bundle inklusive Client-Schluessel neu auf.

    Der Schluessel liegt verschluesselt in der Datenbank; hier wird er genau
    fuer die Auslieferung entschluesselt. Fehlt er (Lizenz aus einer aelteren
    Version), enthaelt das Bundle nur das Dokument -- der Kunde braucht dann eine
    neu ausgestellte Lizenz.
    """
    row = get_license(conn, license_id)
    if row is None:
        raise IssuanceError("Diese Lizenz existiert nicht.")
    document = json.loads(str(row["license_document"]))
    bundle: dict[str, Any] = {
        "kind": "tesm-license-bundle",
        "fmt": lic.FORMAT_VERSION,
        "license": document,
    }
    encrypted = str(row["client_key_encrypted"] or "")
    if encrypted:
        try:
            bundle["client_key"] = keystore.decrypt(encrypted, aad=client_key_aad(license_id))
        except Exception as exc:  # noqa: BLE001
            raise IssuanceError(
                "Der Client-Schluessel dieser Lizenz konnte nicht entschluesselt werden. "
                "Bitte eine neue Lizenz ausstellen."
            ) from exc
    return bundle


def revoke_license(conn: sqlite3.Connection, license_id: str, *, reason: str = "") -> None:
    """Widerruft eine Lizenz.

    Das Ticket bleibt bewusst offen: ein Widerruf ist Routine (Systemwechsel,
    Neuinstallation), keine Sperre. Wer den Kunden aussperren will, sperrt das
    Ticket.
    """
    row = get_license(conn, license_id)
    if row is None:
        raise IssuanceError("Diese Lizenz existiert nicht.")
    if str(row["status"]) == "revoked":
        raise IssuanceError("Diese Lizenz ist bereits widerrufen.")
    conn.execute(
        "UPDATE licenses SET status='revoked', revoked_at=?, revoke_reason=? WHERE license_id=?",
        (_now(), (reason or "").strip()[:400], license_id),
    )


def delete_license(conn: sqlite3.Connection, license_id: str) -> None:
    row = get_license(conn, license_id)
    if row is None:
        raise IssuanceError("Diese Lizenz existiert nicht.")
    if str(row["status"]) not in {"revoked", "issued"}:
        raise IssuanceError(
            "Eine aktive Lizenz kann nicht geloescht werden. Bitte zuerst widerrufen."
        )
    conn.execute("DELETE FROM licenses WHERE license_id=?", (license_id,))


def resign_license(
    conn: sqlite3.Connection,
    license_id: str,
    *,
    master_private_key: str,
    changes: dict[str, Any],
) -> dict[str, Any]:
    """Signiert eine neue Revision -- z. B. nach einer Modulerweiterung.

    Der Client erhaelt sie beim naechsten Heartbeat. ``seq`` steigt dabei; eine
    aeltere Revision kann ihm damit nicht mehr eingespielt werden.
    """
    row = get_license(conn, license_id)
    if row is None:
        raise IssuanceError("Diese Lizenz existiert nicht.")
    document = json.loads(str(row["license_document"]))
    updated = licsrv.resign_license(
        document, master_private_key_b64=master_private_key, changes=changes
    )
    conn.execute(
        "UPDATE licenses SET license_document=?, seq=?, type=?, modules=?, lifetime=?, "
        "expires_at=?, revision_at=? WHERE license_id=?",
        (
            json.dumps(updated),
            int(updated["seq"]),
            str(updated["type"]),
            json.dumps(updated["modules"]),
            int(bool(updated["lifetime"])),
            str(updated["expires_at"]),
            _now(),
            license_id,
        ),
    )
    return updated


def record_delivery(
    conn: sqlite3.Connection,
    *,
    license_id: str,
    channel: str,
    recipient: str,
    actor: str,
    detail: str = "",
) -> None:
    conn.execute(
        "INSERT INTO license_deliveries (license_id, channel, recipient, delivered_at, "
        "delivered_by, detail) VALUES (?,?,?,?,?,?)",
        (license_id, channel, recipient, _now(), actor, detail[:400]),
    )


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


def revision_pending(row: Any) -> bool:
    """Wurde seit der letzten Revision noch kein Kontakt vom Client gesehen?

    Widerrufene Lizenzen zaehlen nicht -- dort ist nichts mehr abzuholen.
    """
    def field(name: str) -> Any:
        # Die Zeile kommt aus verschiedenen Abfragen; nicht jede waehlt alles.
        # Eine fehlende Spalte darf keine Ausnahme werfen -- die Funktion sitzt
        # in Listen, Detailseiten und im Kundenportal.
        try:
            return row[name]
        except (KeyError, IndexError, TypeError):
            return None

    revision_at = field("revision_at")
    if not revision_at or str(field("status") or "") == "revoked":
        return False
    last = field("last_heartbeat_at") or field("activated_at")
    if not last:
        return True
    # Beide Zeitstempel stammen aus ``_now()`` und sind damit gleich formatiert.
    return str(last) < str(revision_at)


def display_state(row: dict[str, Any], *, now: datetime | None = None) -> dict[str, Any]:
    """Ampelfarbe und Resttage einer Lizenzzeile.

    Genau eine Stelle im Code, von jeder Oberflaeche benutzt (Dashboard,
    Listen, Detailseiten, Self-Service, Kundenportal, gemeldete Bestaende der
    Sub-Lizenzserver).

    Grenzen: mehr als 90 Tage = gruen, 30 bis 90 Tage = orange, unter 30 Tage
    (auch abgelaufen) = rot. Unbefristete Lizenzen sind immer gruen.
    """
    moment = now or datetime.now(timezone.utc)
    result = {
        "color": "green",
        "days_left": None,
        "is_expired": False,
        "lifetime": bool(row.get("lifetime")),
        # Kein eigener Status: eine Lizenz kann aktiv sein *und* eine noch
        # nicht abgeholte Revision haben.
        "revision_pending": revision_pending(row),
    }
    if row.get("lifetime"):
        return result
    raw = row.get("expires_at")
    if not raw:
        return {**result, "color": "red"}
    try:
        expires = lic.parse_iso(str(raw))
    except ValueError:
        # Fail-safe: unlesbares Datum gilt als kritisch, nicht als in Ordnung.
        return {**result, "color": "red"}

    delta = expires - moment
    days_left = int(delta.total_seconds() // 86400)
    expired = delta.total_seconds() <= 0
    if expired:
        color = "red"
    elif days_left < RED_BELOW_DAYS:
        color = "red"
    elif days_left <= GREEN_ABOVE_DAYS:
        color = "orange"
    else:
        color = "green"
    return {
        **result,
        "color": color,
        "days_left": max(0, days_left),
        "is_expired": expired,
        "lifetime": False,
    }


def dashboard_summary(conn: sqlite3.Connection) -> dict[str, Any]:
    rows = list_licenses(conn)
    buckets: dict[str, list[dict[str, Any]]] = {"red": [], "orange": [], "green": []}
    for row in rows:
        if str(row["status"]) in {"revoked", "deactivated"}:
            continue
        state = display_state(row)
        entry = {**row, **state}
        buckets[state["color"]].append(entry)
    return {
        "buckets": buckets,
        "counts": {
            "customers": int(
                Database.value(conn, "SELECT COUNT(*) FROM license_customers", default=0) or 0
            ),
            "tickets_open": int(
                Database.value(
                    conn, "SELECT COUNT(*) FROM license_tickets WHERE status='open'", default=0
                )
                or 0
            ),
            "tickets_blocked": int(
                Database.value(
                    conn, "SELECT COUNT(*) FROM license_tickets WHERE status='blocked'", default=0
                )
                or 0
            ),
            "licenses_active": int(
                Database.value(
                    conn, "SELECT COUNT(*) FROM licenses WHERE status='active'", default=0
                )
                or 0
            ),
            "licenses_issued": int(
                Database.value(
                    conn, "SELECT COUNT(*) FROM licenses WHERE status='issued'", default=0
                )
                or 0
            ),
            "licenses_revoked": int(
                Database.value(
                    conn, "SELECT COUNT(*) FROM licenses WHERE status='revoked'", default=0
                )
                or 0
            ),
        },
    }