feat(dsgvo): Löschkonzept/Aufbewahrungsfristen (Auto-Purge)

Neuer retention_service.py: Lohn-/zeitrelevante Daten (time_entries,
hours_payouts) werden nach konfigurierbarer Frist gelöscht
(company.settings.retention_lohn_years, Default 10 Jahre). Technische
Tabellen mit fester Frist: audit_logs (3 Jahre), abgelaufene
sessions/password_resets (sofort).

Täglicher Scheduler-Job (03:00 Uhr, Redis-Tageslock analog Reminder-Jobs)
plus manuelle Trigger: POST /companies/me/run-retention-purge
(COMPANY_ADMIN/HR, nur eigene Firma) und POST /admin/run-retention-purge
(SUPER_ADMIN, global inkl. technischer Tabellen).

Letzter offener Punkt aus dem DSGVO-Löschkonzept (Art. 15/17 bereits erledigt).

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01Gis16MnuwkYcivLrSxK1pD

backend/app/services/retention_service.py

@@ -0,0 +1,111 @@
+"""DSGVO-Löschkonzept: Aufbewahrungsfristen / Auto-Purge.
+
+Zwei Kategorien:
+- Lohn-/zeitrelevante Daten (time_entries, hours_payouts): gesetzliche
+  Aufbewahrungspflicht, Frist pro Firma konfigurierbar über
+  company.settings["retention_lohn_years"] (Default RETENTION_LOHN_DEFAULT_YEARS).
+  Nach Ablauf werden die Zeilen hart gelöscht (keine Anonymisierung nötig,
+  da nach Fristablauf keine Aufbewahrungspflicht mehr besteht und die Daten
+  für den ursprünglichen Zweck nicht mehr benötigt werden).
+- Rein technische Daten (audit_logs, sessions, password_resets): feste,
+  nicht pro Firma konfigurierbare Fristen. AuditLog RETENTION_AUDITLOG_YEARS,
+  Sessions/Password-Resets werden gelöscht sobald abgelaufen (keine
+  Aufbewahrungspflicht für Alt-Sessions).
+
+Läuft wie die Reminder-Jobs (scheduler_service.py) mit eigener DB-Session und
+RLS-Bypass, da firmenübergreifend bzw. ohne Tenant-Kontext gearbeitet wird.
+"""
+from __future__ import annotations
+
+import logging
+from datetime import date, datetime, timedelta, timezone
+
+from sqlalchemy import delete, select
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from app.models.company import Company
+from app.models.audit_log import AuditLog
+from app.models.hours_payout import HoursPayout
+from app.models.password_reset import PasswordReset
+from app.models.session import Session
+from app.models.time_entry import TimeEntry
+from app.models.user import User
+
+log = logging.getLogger(__name__)
+
+RETENTION_LOHN_DEFAULT_YEARS = 10
+RETENTION_AUDITLOG_YEARS = 3
+
+
+def _lohn_retention_years(company: Company) -> int:
+    s = company.settings or {}
+    years = s.get("retention_lohn_years")
+    return int(years) if years else RETENTION_LOHN_DEFAULT_YEARS
+
+
+async def _active_companies(db: AsyncSession, company_id=None) -> list[Company]:
+    q = select(Company).where(Company.is_active.is_(True))
+    if company_id:
+        q = q.where(Company.id == company_id)
+    return list((await db.scalars(q)).all())
+
+
+async def purge_company_lohn_data(db: AsyncSession, company: Company) -> dict:
+    """Löscht time_entries + hours_payouts der Firma, die älter als die
+    konfigurierte Aufbewahrungsfrist sind."""
+    years = _lohn_retention_years(company)
+    cutoff_date = date.today() - timedelta(days=years * 365)
+    cutoff_dt = datetime.now(timezone.utc) - timedelta(days=years * 365)
+
+    user_ids_q = select(User.id).where(User.company_id == company.id)
+
+    te_result = await db.execute(
+        delete(TimeEntry)
+        .where(TimeEntry.user_id.in_(user_ids_q), TimeEntry.date < cutoff_date)
+    )
+    hp_result = await db.execute(
+        delete(HoursPayout)
+        .where(HoursPayout.company_id == company.id, HoursPayout.created_at < cutoff_dt)
+    )
+    return {"time_entries": te_result.rowcount or 0, "hours_payouts": hp_result.rowcount or 0}
+
+
+async def purge_audit_logs(db: AsyncSession) -> int:
+    cutoff = datetime.now(timezone.utc) - timedelta(days=RETENTION_AUDITLOG_YEARS * 365)
+    result = await db.execute(delete(AuditLog).where(AuditLog.created_at < cutoff))
+    return result.rowcount or 0
+
+
+async def purge_expired_sessions(db: AsyncSession) -> int:
+    now = datetime.now(timezone.utc)
+    result = await db.execute(delete(Session).where(Session.expires_at < now))
+    return result.rowcount or 0
+
+
+async def purge_expired_password_resets(db: AsyncSession) -> int:
+    now = datetime.now(timezone.utc)
+    result = await db.execute(
+        delete(PasswordReset).where(
+            (PasswordReset.expires_at < now) | (PasswordReset.used_at.isnot(None))
+        )
+    )
+    return result.rowcount or 0
+
+
+async def run_retention_purge(db: AsyncSession, company_id=None) -> dict:
+    """Orchestriert den vollständigen Purge-Lauf. Liefert Zähler je Kategorie."""
+    totals = {"time_entries": 0, "hours_payouts": 0}
+    for company in await _active_companies(db, company_id):
+        counts = await purge_company_lohn_data(db, company)
+        totals["time_entries"] += counts["time_entries"]
+        totals["hours_payouts"] += counts["hours_payouts"]
+
+    # Technische Tabellen sind firmenübergreifend – nur beim globalen Lauf mitziehen,
+    # nicht wenn ein einzelner Company-Admin gezielt nur seine eigene Firma anstößt.
+    if company_id is None:
+        totals["audit_logs"] = await purge_audit_logs(db)
+        totals["sessions"] = await purge_expired_sessions(db)
+        totals["password_resets"] = await purge_expired_password_resets(db)
+
+    log.info("Retention-Purge abgeschlossen: %s", totals)
+    return totals

backend/app/services/scheduler_service.py

@@ -23,6 +23,7 @@ from app.models.user import User, UserRole
 from app.models.vacation_balance import VacationBalance
 from app.services.absence_service import absence_service
 from app.services.email_service import email_service
+from app.services.retention_service import run_retention_purge
 
 log = logging.getLogger(__name__)
 
@@ -188,6 +189,22 @@ async def _run_job(name: str, fn) -> None:
         log.exception("Reminder-Job %s fehlgeschlagen: %s", name, exc)
 
 
+async def _run_purge_job(name: str, fn) -> None:
+    if not _acquire_daily_lock(name):
+        log.info("Purge-Job %s bereits von anderem Prozess übernommen.", name)
+        return
+    from sqlalchemy import text
+    from app.core.database import AsyncSessionLocal
+    try:
+        async with AsyncSessionLocal() as db:
+            await db.execute(text("SET LOCAL app.bypass_rls = 'on'"))
+            result = await fn(db)
+            await db.commit()
+        log.info("Purge-Job %s: %s", name, result)
+    except Exception as exc:
+        log.exception("Purge-Job %s fehlgeschlagen: %s", name, exc)
+
+
 def start() -> None:
     global _scheduler
     if _scheduler is not None:
@@ -207,6 +224,8 @@ def start() -> None:
                        args=["carryover_expiry", run_carryover_expiry], id="carryover_expiry")
     _scheduler.add_job(_run_job, CronTrigger(hour=hour, minute=20),
                        args=["certificate_overdue", run_certificate_overdue], id="certificate_overdue")
+    _scheduler.add_job(_run_purge_job, CronTrigger(hour=3, minute=0),
+                       args=["retention_purge", run_retention_purge], id="retention_purge")
     _scheduler.start()
     log.info("Reminder-Scheduler gestartet (täglich ab %02d:00 Uhr).", hour)