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/routers/companies.py

@@ -52,6 +52,21 @@ async def run_reminders_now(
     return {"sent": result}
 
 
+@router.post("/me/run-retention-purge")
+async def run_retention_purge_now(
+    current_user: User = require_role(*_admin_roles),
+    db: AsyncSession = Depends(get_db),
+):
+    """Löscht Zeiterfassung/Auszahlungen der eigenen Firma, die älter als die
+    konfigurierte Aufbewahrungsfrist (company.settings.retention_lohn_years,
+    Default 10 Jahre) sind. Technische Tabellen (audit_logs/sessions) laufen
+    nur im globalen Scheduler-Job, nicht firmenscoped."""
+    from app.services.retention_service import run_retention_purge
+    result = await run_retention_purge(db, company_id=current_user.company_id)
+    await db.commit()
+    return {"deleted": result}
+
+
 @router.patch("/me", response_model=CompanyOut)
 async def update_my_company(
     data: CompanyUpdate,

backend/app/routers/tenants.py

@@ -217,3 +217,18 @@ async def update_reseller(
         last_name=reseller.last_name, is_active=reseller.is_active,
         created_at=reseller.created_at, company_count=counts.get(reseller.id, 0),
     )
+
+
+@router.post("/run-retention-purge")
+async def run_global_retention_purge(
+    current_user: User = require_role(*_sa),
+    db: AsyncSession = Depends(get_db),
+):
+    """Globaler DSGVO-Purge über alle Firmen: Zeiterfassung/Auszahlungen je Firma
+    (Frist konfigurierbar) plus firmenübergreifende technische Tabellen
+    (audit_logs > 3 Jahre, abgelaufene sessions/password_resets). Gleiche Logik
+    wie der tägliche Scheduler-Job."""
+    from app.services.retention_service import run_retention_purge
+    result = await run_retention_purge(db, company_id=None)
+    await db.commit()
+    return {"deleted": result}

backend/app/schemas/company.py

@@ -13,12 +13,14 @@ class CompanySettingsUpdate(BaseModel):
 
     Nur bekannte Top-Level-Keys sind erlaubt (extra="forbid").
     Derzeit genutzte Keys: carryover_expires_month, carryover_expires_day
-    (gelesen in absences.py für Resturlaub-Verfall-Berechnung).
+    (gelesen in absences.py für Resturlaub-Verfall-Berechnung), retention_lohn_years
+    (gelesen in retention_service.py für Aufbewahrungsfrist Zeiterfassung/Auszahlungen).
     """
     model_config = ConfigDict(extra="forbid")
 
     carryover_expires_month: int | None = Field(None, ge=1, le=12)
     carryover_expires_day: int | None = Field(None, ge=1, le=31)
+    retention_lohn_years: int | None = Field(None, ge=1, le=30)
 
 
 class CompanyOut(BaseModel):

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)
 

backend/tests/test_retention.py

@@ -0,0 +1,108 @@
+"""Tests für DSGVO-Löschkonzept (retention_service.py)."""
+from datetime import date, datetime, timedelta, timezone
+
+import pytest
+import pytest_asyncio
+from httpx import AsyncClient
+from sqlalchemy import text
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from app.services.retention_service import run_retention_purge
+
+
+@pytest_asyncio.fixture(scope="session", loop_scope="session")
+async def retention_headers(client: AsyncClient):
+    resp = await client.post("/api/v1/auth/register", json={
+        "company_name": "Retention GmbH",
+        "first_name": "Rita",
+        "last_name": "Tention",
+        "email": "admin@retentiongmbh.de",
+        "password": "Secret123",
+    })
+    assert resp.status_code == 201, resp.text
+    data = resp.json()
+    return {"Authorization": f"Bearer {data['access_token']}"}, data["user"]["company_id"]
+
+
+@pytest.mark.asyncio(loop_scope="session")
+async def test_purge_deletes_old_time_entries_respects_retention(
+    client: AsyncClient, db_session: AsyncSession, retention_headers
+):
+    headers, company_id = retention_headers
+    me = await client.get("/api/v1/users/me", headers=headers)
+    user_id = me.json()["id"]
+
+    old_date = date.today() - timedelta(days=11 * 365)  # älter als Default 10 Jahre
+    recent_date = date.today() - timedelta(days=30)
+
+    await db_session.execute(text(
+        "INSERT INTO time_entries (id, user_id, date, start_time, end_time, status, source) "
+        "VALUES (gen_random_uuid(), :uid, :d, '08:00', '16:00', 'approved', 'web')"
+    ), {"uid": user_id, "d": old_date})
+    await db_session.execute(text(
+        "INSERT INTO time_entries (id, user_id, date, start_time, end_time, status, source) "
+        "VALUES (gen_random_uuid(), :uid, :d, '08:00', '16:00', 'approved', 'web')"
+    ), {"uid": user_id, "d": recent_date})
+    await db_session.commit()
+
+    result = await run_retention_purge(db_session, company_id=company_id)
+    await db_session.commit()
+    assert result["time_entries"] == 1
+
+    remaining = await db_session.execute(text(
+        "SELECT date FROM time_entries WHERE user_id = :uid"
+    ), {"uid": user_id})
+    dates = [r[0] for r in remaining]
+    assert old_date not in dates
+    assert recent_date in dates
+
+
+@pytest.mark.asyncio(loop_scope="session")
+async def test_purge_configurable_retention_years(
+    client: AsyncClient, db_session: AsyncSession, retention_headers
+):
+    headers, company_id = retention_headers
+    up = await client.patch("/api/v1/companies/me",
+                            json={"settings": {"retention_lohn_years": 2}}, headers=headers)
+    assert up.status_code == 200, up.text
+
+    me = await client.get("/api/v1/users/me", headers=headers)
+    user_id = me.json()["id"]
+    three_years_ago = date.today() - timedelta(days=3 * 365)
+
+    await db_session.execute(text(
+        "INSERT INTO time_entries (id, user_id, date, start_time, end_time, status, source) "
+        "VALUES (gen_random_uuid(), :uid, :d, '08:00', '16:00', 'approved', 'web')"
+    ), {"uid": user_id, "d": three_years_ago})
+    await db_session.commit()
+
+    result = await run_retention_purge(db_session, company_id=company_id)
+    await db_session.commit()
+    assert result["time_entries"] >= 1
+
+
+@pytest.mark.asyncio(loop_scope="session")
+async def test_purge_expired_sessions_and_audit_logs_global(
+    client: AsyncClient, db_session: AsyncSession, retention_headers
+):
+    headers, company_id = retention_headers
+    me = await client.get("/api/v1/users/me", headers=headers)
+    user_id = me.json()["id"]
+
+    expired = datetime.now(timezone.utc) - timedelta(days=1)
+    old_audit = datetime.now(timezone.utc) - timedelta(days=4 * 365)
+
+    await db_session.execute(text(
+        "INSERT INTO sessions (id, user_id, refresh_token_hash, expires_at) "
+        "VALUES (gen_random_uuid(), :uid, :h, :exp)"
+    ), {"uid": user_id, "h": "expired-hash-retention-test", "exp": expired})
+    await db_session.execute(text(
+        "INSERT INTO audit_logs (id, company_id, user_id, action, created_at) "
+        "VALUES (gen_random_uuid(), :cid, :uid, 'test_old_action', :ts)"
+    ), {"cid": company_id, "uid": user_id, "ts": old_audit})
+    await db_session.commit()
+
+    result = await run_retention_purge(db_session, company_id=None)
+    await db_session.commit()
+    assert result["sessions"] >= 1
+    assert result["audit_logs"] >= 1