imap-mail-filter-service/app/services/maintenance.py

"""Datenbank-Wartung: Log-Rotation, verwaiste Marker entfernen und bei Bedarf
VACUUM, damit die SQLite-Datei nicht unbegrenzt wächst.

SQLite gibt durch DELETE frei gewordenen Platz nicht ans Dateisystem zurück —
die Datei schrumpft erst durch VACUUM. Das kopiert die Datenbank einmal neu
und braucht dafür kurzzeitig freien Platz in Größe der verbleibenden Daten."""
import logging
import os
import shutil
import threading
import time
from datetime import datetime, timedelta

from sqlalchemy import text

from app.config import settings
from app.database import engine

logger = logging.getLogger(__name__)

# VACUUM nur, wenn sich das lohnt: mindestens so viel Anteil bzw. Größe frei
VACUUM_MIN_FREE_RATIO = 0.25
VACUUM_MIN_FREE_BYTES = 50 * 1024 * 1024

_lock = threading.Lock()
last_result: dict | None = None


def _db_path() -> str | None:
    url = settings.database_url
    if not url.startswith("sqlite:///"):
        return None
    return url.removeprefix("sqlite:///")


def _file_size(path: str | None) -> int:
    if not path:
        return 0
    total = 0
    for suffix in ("", "-wal"):
        try:
            total += os.path.getsize(path + suffix)
        except OSError:
            pass
    return total


def db_status() -> dict:
    path = _db_path()
    with engine.connect() as conn:
        page_size = conn.exec_driver_sql("PRAGMA page_size").scalar()
        page_count = conn.exec_driver_sql("PRAGMA page_count").scalar()
        free_pages = conn.exec_driver_sql("PRAGMA freelist_count").scalar()
        log_rows = conn.execute(text("SELECT COUNT(*) FROM filter_logs")).scalar()
        processed_rows = conn.execute(text("SELECT COUNT(*) FROM processed_mails")).scalar()
    disk_free = shutil.disk_usage(os.path.dirname(os.path.abspath(path))).free if path else None
    return {
        "file_bytes": _file_size(path),
        "used_bytes": (page_count - free_pages) * page_size,
        "free_bytes": free_pages * page_size,
        "log_rows": log_rows,
        "processed_rows": processed_rows,
        "disk_free_bytes": disk_free,
        "retention_days": settings.log_retention_days,
        "max_entries": settings.log_max_entries,
        "last_run": last_result,
    }


def run_maintenance(force_vacuum: bool = False) -> dict:
    """Läuft synchron (aus Thread/Executor aufrufen). Parallel-Läufe werden
    übersprungen."""
    global last_result
    if not _lock.acquire(blocking=False):
        return {"skipped": "Wartung läuft bereits"}
    started = time.monotonic()
    try:
        path = _db_path()
        size_before = _file_size(path)
        result: dict = {"started_at": datetime.utcnow().isoformat(), "size_before": size_before}

        with engine.begin() as conn:
            cutoff = datetime.utcnow() - timedelta(days=settings.log_retention_days)
            result["logs_deleted_age"] = conn.execute(
                text("DELETE FROM filter_logs WHERE created_at < :cutoff"), {"cutoff": cutoff}
            ).rowcount

            # Obergrenze: nur die neuesten N behalten
            boundary = conn.execute(
                text("SELECT id FROM filter_logs ORDER BY id DESC LIMIT 1 OFFSET :n"),
                {"n": settings.log_max_entries},
            ).scalar()
            result["logs_deleted_cap"] = (
                conn.execute(text("DELETE FROM filter_logs WHERE id <= :b"), {"b": boundary}).rowcount
                if boundary is not None else 0
            )

            # Marker gelöschter Regeln/Konten (SQLite erzwingt die Fremdschlüssel nicht)
            result["processed_orphans_deleted"] = conn.execute(text(
                "DELETE FROM processed_mails "
                "WHERE rule_id NOT IN (SELECT id FROM filter_rules) "
                "OR account_id NOT IN (SELECT id FROM accounts)"
            )).rowcount

        with engine.connect() as conn:
            conn.exec_driver_sql("PRAGMA wal_checkpoint(TRUNCATE)")
            page_size = conn.exec_driver_sql("PRAGMA page_size").scalar()
            page_count = conn.exec_driver_sql("PRAGMA page_count").scalar()
            free_pages = conn.exec_driver_sql("PRAGMA freelist_count").scalar()
        free_bytes = free_pages * page_size
        used_bytes = (page_count - free_pages) * page_size
        worthwhile = free_bytes >= VACUUM_MIN_FREE_BYTES and free_bytes >= VACUUM_MIN_FREE_RATIO * page_count * page_size

        result["vacuum"] = "nicht nötig"
        if force_vacuum or worthwhile:
            disk_free = shutil.disk_usage(os.path.dirname(os.path.abspath(path))).free if path else 0
            # VACUUM schreibt eine Kopie der verbleibenden Daten (plus WAL)
            if path and disk_free < used_bytes * 1.2:
                result["vacuum"] = (
                    f"übersprungen: zu wenig Plattenplatz ({disk_free // 2**20} MB frei, "
                    f"~{int(used_bytes * 1.2) // 2**20} MB nötig)"
                )
            else:
                t = time.monotonic()
                # VACUUM darf nicht in einer Transaktion laufen
                with engine.connect().execution_options(isolation_level="AUTOCOMMIT") as conn:
                    conn.exec_driver_sql("VACUUM")
                    conn.exec_driver_sql("PRAGMA wal_checkpoint(TRUNCATE)")
                result["vacuum"] = f"ausgeführt ({time.monotonic() - t:.0f}s)"

        result["size_after"] = _file_size(path)
        result["duration_s"] = round(time.monotonic() - started, 1)
        logger.info("DB-Wartung: %s", result)
        last_result = result
        return result
    except Exception as e:
        logger.exception("DB-Wartung fehlgeschlagen")
        last_result = {"error": str(e), "started_at": datetime.utcnow().isoformat()}
        raise
    finally:
        _lock.release()