Merge pull request 'MAI-03: DSGVO-Löschanfragen-Migration' (#271) from feature/mai-03-dsgvo-loeschanfragen-migration into main

archive/internal/dsgvorequestmigration/dsgvorequestmigration.go

@@ -0,0 +1,116 @@
+// Package dsgvorequestmigration implementiert MAI-03: Übernahme
+// offener, unbearbeiteter DSGVO-Löschanfragen aus archivmail als aktiven
+// Vorgang in Archive CMP-02/CMP-07 (dsgvorequest.ProcessDeletionRequest
+// — keine zweite Prüf-/Entscheidungslogik, siehe dsgvorequest-Paket-
+// kommentar). Bereits abgeschlossene Anfragen werden NICHT erneut durch
+// den Workflow geschickt (das würde für noch vorhandene, abgelaufene
+// Objekte fälschlich einen neuen aktiven Vorgang anstoßen), sondern nur
+// als historischer Audit-Eintrag in dsgvo_decision_log übernommen.
+package dsgvorequestmigration
+
+import (
+	"context"
+	"fmt"
+	"time"
+
+	"github.com/jackc/pgx/v5/pgxpool"
+
+	"gitea.perlbach24.de/scripte/nexarch/archive/internal/dpreport"
+	"gitea.perlbach24.de/scripte/nexarch/archive/internal/dsgvorequest"
+)
+
+const (
+	StatusOpen      = "open"
+	StatusCompleted = "completed"
+)
+
+// SourceRequest ist eine Zeile aus archivmails DSGVO-Löschanfragen.
+// Status ist ausschließlich "open" oder "completed" — indexed_at-artige
+// Statusfelder der Quelle werden bewusst nicht übernommen (Ticket-
+// Vorgabe "Produkt-DNA").
+type SourceRequest struct {
+	SourceRequestID string
+	DataSubjectRef  string
+	RequestedBy     string
+	RequestedAt     time.Time
+	Status          string
+}
+
+// Source kapselt den lesenden Zugriff auf archivmail. Eine echte
+// Implementierung liest gegen archivmail, Tests verwenden ausschließlich
+// synthetische Fixtures.
+type Source interface {
+	Requests(ctx context.Context) ([]SourceRequest, error)
+}
+
+type Migrator struct {
+	pool   *pgxpool.Pool
+	source Source
+}
+
+func NewMigrator(pool *pgxpool.Pool, source Source) *Migrator {
+	return &Migrator{pool: pool, source: source}
+}
+
+// Result fasst den Ausgang eines Migrationslaufs zusammen.
+type Result struct {
+	OpenRequestsProcessed   int
+	HistoricalEntriesLogged int
+	RanAt                   time.Time
+}
+
+// MigrateAll übernimmt jede offene Anfrage als aktiven Vorgang
+// (Akzeptanzkriterium 1) und jede abgeschlossene Anfrage nur als
+// historischen Audit-Eintrag (Akzeptanzkriterium 2).
+func (m *Migrator) MigrateAll(ctx context.Context, now time.Time) (Result, error) {
+	res := Result{RanAt: now}
+
+	requests, err := m.source.Requests(ctx)
+	if err != nil {
+		return res, fmt.Errorf("dsgvorequestmigration: anfragen aus quelle lesen: %w", err)
+	}
+
+	for _, req := range requests {
+		switch req.Status {
+		case StatusOpen:
+			if _, err := dsgvorequest.ProcessDeletionRequest(ctx, m.pool, req.DataSubjectRef, req.RequestedBy); err != nil {
+				return res, fmt.Errorf("dsgvorequestmigration: offene anfrage %q verarbeiten: %w", req.SourceRequestID, err)
+			}
+			res.OpenRequestsProcessed++
+		case StatusCompleted:
+			n, err := logHistorical(ctx, m.pool, req)
+			if err != nil {
+				return res, fmt.Errorf("dsgvorequestmigration: abgeschlossene anfrage %q protokollieren: %w", req.SourceRequestID, err)
+			}
+			res.HistoricalEntriesLogged += n
+		default:
+			return res, fmt.Errorf("dsgvorequestmigration: anfrage %q hat unbekannten status %q", req.SourceRequestID, req.Status)
+		}
+	}
+
+	return res, nil
+}
+
+// logHistorical schreibt für jedes aktuell gespeicherte Objekt der
+// betroffenen Person einen Audit-Eintrag — OHNE dsgvorequest.
+// ProcessDeletionRequest aufzurufen, also ohne einen neuen aktiven
+// Vorgang (deletion_requests) anzustoßen (Akzeptanzkriterium 2).
+func logHistorical(ctx context.Context, pool *pgxpool.Pool, req SourceRequest) (int, error) {
+	records, err := dpreport.SubjectReport(ctx, pool, req.DataSubjectRef)
+	if err != nil {
+		return 0, fmt.Errorf("objekte der betroffenen person ermitteln: %w", err)
+	}
+
+	n := 0
+	for _, rec := range records {
+		if _, err := pool.Exec(ctx, `
+			INSERT INTO dsgvo_decision_log (data_subject_ref, retention_object_id, object_type, object_reference, outcome, reason, decided_by)
+			VALUES ($1, $2, $3, $4, $5, $6, $7)
+		`, req.DataSubjectRef, rec.RetentionObjectID, rec.ObjectType, rec.ObjectReference,
+			dsgvorequest.OutcomeAlreadyDeleted, "aus archivmail migriert: anfrage dort bereits abgeschlossen", req.RequestedBy); err != nil {
+			return n, fmt.Errorf("audit-eintrag für objekt %q: %w", rec.ObjectReference, err)
+		}
+		n++
+	}
+	return n, nil
+}

archive/internal/dsgvorequestmigration/dsgvorequestmigration_test.go

@@ -0,0 +1,201 @@
+package dsgvorequestmigration
+
+import (
+	"context"
+	"os"
+	"testing"
+	"time"
+
+	"github.com/jackc/pgx/v5/pgxpool"
+)
+
+func requireTestPool(t *testing.T) *pgxpool.Pool {
+	t.Helper()
+	dsn := os.Getenv("TEST_TENANT_DSN")
+	if dsn == "" {
+		t.Skip("TEST_TENANT_DSN nicht gesetzt, Integrationstest uebersprungen")
+	}
+	ctx := context.Background()
+	pool, err := pgxpool.New(ctx, dsn)
+	if err != nil {
+		t.Fatalf("pool: %v", err)
+	}
+	t.Cleanup(func() { pool.Close() })
+
+	if _, err := pool.Exec(ctx, `
+		CREATE EXTENSION IF NOT EXISTS pgcrypto;
+		CREATE TABLE IF NOT EXISTS retention_objects (
+			id UUID PRIMARY KEY DEFAULT gen_random_uuid(), object_type TEXT NOT NULL,
+			object_reference TEXT NOT NULL,
+			status TEXT NOT NULL DEFAULT 'active' CHECK (status IN ('active', 'expired', 'deleted')),
+			created_at TIMESTAMPTZ NOT NULL DEFAULT now(),
+			data_subject_ref TEXT,
+			UNIQUE (object_type, object_reference)
+		);
+		CREATE TABLE IF NOT EXISTS retention_class_assignments (
+			id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
+			retention_object_id UUID NOT NULL REFERENCES retention_objects(id) ON DELETE CASCADE,
+			retention_class TEXT NOT NULL, assigned_at TIMESTAMPTZ NOT NULL DEFAULT now()
+		);
+		CREATE TABLE IF NOT EXISTS legal_holds (
+			id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
+			retention_object_id UUID NOT NULL REFERENCES retention_objects(id) ON DELETE CASCADE,
+			reason TEXT NOT NULL, set_by TEXT NOT NULL, set_at TIMESTAMPTZ NOT NULL DEFAULT now(),
+			released_at TIMESTAMPTZ, released_by TEXT
+		);
+		CREATE UNIQUE INDEX IF NOT EXISTS idx_legal_holds_active
+			ON legal_holds (retention_object_id) WHERE released_at IS NULL;
+		CREATE TABLE IF NOT EXISTS destruction_log (
+			id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
+			retention_object_id UUID NOT NULL REFERENCES retention_objects(id) ON DELETE RESTRICT,
+			object_type TEXT NOT NULL, object_reference TEXT NOT NULL,
+			destroyed_at TIMESTAMPTZ NOT NULL DEFAULT now(), destroyed_by TEXT NOT NULL
+		);
+		CREATE TABLE IF NOT EXISTS deletion_requests (
+			id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
+			retention_object_id UUID NOT NULL REFERENCES retention_objects(id) ON DELETE CASCADE,
+			requested_by TEXT NOT NULL, requested_at TIMESTAMPTZ NOT NULL DEFAULT now(),
+			confirmation_token_hash BYTEA NOT NULL, token_expires_at TIMESTAMPTZ NOT NULL,
+			confirmed_by TEXT, confirmed_at TIMESTAMPTZ, executed_at TIMESTAMPTZ
+		);
+		CREATE TABLE IF NOT EXISTS dsgvo_decision_log (
+			id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
+			data_subject_ref TEXT NOT NULL,
+			retention_object_id UUID NOT NULL REFERENCES retention_objects(id) ON DELETE RESTRICT,
+			object_type TEXT NOT NULL, object_reference TEXT NOT NULL,
+			outcome TEXT NOT NULL CHECK (outcome IN ('deletion_requested', 'rejected', 'already_deleted')),
+			reason TEXT NOT NULL, decided_at TIMESTAMPTZ NOT NULL DEFAULT now(), decided_by TEXT NOT NULL
+		);
+	`); err != nil {
+		t.Fatalf("schema: %v", err)
+	}
+	t.Cleanup(func() {
+		_, _ = pool.Exec(context.Background(), `TRUNCATE dsgvo_decision_log, deletion_requests, destruction_log, legal_holds, retention_class_assignments, retention_objects CASCADE`)
+	})
+	return pool
+}
+
+func insertObject(t *testing.T, ctx context.Context, pool *pgxpool.Pool, ref, subjectRef, status string) string {
+	t.Helper()
+	var id string
+	if err := pool.QueryRow(ctx, `
+		INSERT INTO retention_objects (object_type, object_reference, status, data_subject_ref)
+		VALUES ('mail_message', $1, $2, $3) RETURNING id
+	`, ref, status, subjectRef).Scan(&id); err != nil {
+		t.Fatal(err)
+	}
+	return id
+}
+
+type fixtureSource struct {
+	requests []SourceRequest
+}
+
+func (s *fixtureSource) Requests(ctx context.Context) ([]SourceRequest, error) {
+	return s.requests, nil
+}
+
+// TestMigrateAll_OpenRequestBecomesActiveDeletionRequest ist
+// Akzeptanzkriterium 1: eine offene Löschanfrage ist nach der Migration
+// als aktiver Vorgang (deletion_requests-Zeile) vorhanden.
+func TestMigrateAll_OpenRequestBecomesActiveDeletionRequest(t *testing.T) {
+	pool := requireTestPool(t)
+	ctx := context.Background()
+
+	objID := insertObject(t, ctx, pool, "msg-open-1", "person-open@example.com", "expired")
+
+	m := NewMigrator(pool, &fixtureSource{requests: []SourceRequest{
+		{SourceRequestID: "req-1", DataSubjectRef: "person-open@example.com", RequestedBy: "dsgvo-officer@acme.example", RequestedAt: time.Now(), Status: StatusOpen},
+	}})
+	res, err := m.MigrateAll(ctx, time.Now())
+	if err != nil {
+		t.Fatalf("migrateall: %v", err)
+	}
+	if res.OpenRequestsProcessed != 1 {
+		t.Fatalf("result = %+v, want OpenRequestsProcessed=1", res)
+	}
+
+	var count int
+	if err := pool.QueryRow(ctx, `SELECT count(*) FROM deletion_requests WHERE retention_object_id = $1`, objID).Scan(&count); err != nil {
+		t.Fatal(err)
+	}
+	if count != 1 {
+		t.Fatalf("erwartet genau 1 aktiven loeschvorgang fuer das objekt der offenen anfrage, habe %d", count)
+	}
+}
+
+// TestMigrateAll_CompletedRequestIsHistoricalOnly ist
+// Akzeptanzkriterium 2: eine abgeschlossene Anfrage erzeugt KEINEN
+// aktiven Vorgang, nur einen auffindbaren Audit-Eintrag.
+func TestMigrateAll_CompletedRequestIsHistoricalOnly(t *testing.T) {
+	pool := requireTestPool(t)
+	ctx := context.Background()
+
+	objID := insertObject(t, ctx, pool, "msg-done-1", "person-done@example.com", "deleted")
+
+	m := NewMigrator(pool, &fixtureSource{requests: []SourceRequest{
+		{SourceRequestID: "req-2", DataSubjectRef: "person-done@example.com", RequestedBy: "dsgvo-officer@acme.example", RequestedAt: time.Now(), Status: StatusCompleted},
+	}})
+	res, err := m.MigrateAll(ctx, time.Now())
+	if err != nil {
+		t.Fatalf("migrateall: %v", err)
+	}
+	if res.HistoricalEntriesLogged != 1 {
+		t.Fatalf("result = %+v, want HistoricalEntriesLogged=1", res)
+	}
+
+	var activeCount int
+	if err := pool.QueryRow(ctx, `SELECT count(*) FROM deletion_requests WHERE retention_object_id = $1`, objID).Scan(&activeCount); err != nil {
+		t.Fatal(err)
+	}
+	if activeCount != 0 {
+		t.Fatalf("abgeschlossene anfrage haette keinen aktiven vorgang erzeugen duerfen, habe %d", activeCount)
+	}
+
+	var auditCount int
+	if err := pool.QueryRow(ctx, `SELECT count(*) FROM dsgvo_decision_log WHERE retention_object_id = $1 AND data_subject_ref = $2`, objID, "person-done@example.com").Scan(&auditCount); err != nil {
+		t.Fatal(err)
+	}
+	if auditCount != 1 {
+		t.Fatalf("erwartet genau 1 auffindbaren audit-eintrag, habe %d", auditCount)
+	}
+}
+
+// TestMigrateAll_RowCountMatchesSourceForOpenRequests ist Pflichtprüfung
+// 1: Zeilenzahl offener Anfragen nach Migration deckungsgleich mit
+// Quelle.
+func TestMigrateAll_RowCountMatchesSourceForOpenRequests(t *testing.T) {
+	pool := requireTestPool(t)
+	ctx := context.Background()
+
+	const n = 20
+	var reqs []SourceRequest
+	for i := 0; i < n; i++ {
+		subject := "person-bulk-" + string(rune('a'+i)) + "@example.com"
+		insertObject(t, ctx, pool, "msg-bulk-"+string(rune('a'+i)), subject, "expired")
+		reqs = append(reqs, SourceRequest{
+			SourceRequestID: "req-bulk-" + string(rune('a'+i)),
+			DataSubjectRef:  subject,
+			RequestedBy:     "dsgvo-officer@acme.example",
+			RequestedAt:     time.Now(),
+			Status:          StatusOpen,
+		})
+	}
+
+	m := NewMigrator(pool, &fixtureSource{requests: reqs})
+	res, err := m.MigrateAll(ctx, time.Now())
+	if err != nil {
+		t.Fatalf("migrateall: %v", err)
+	}
+	if res.OpenRequestsProcessed != n {
+		t.Fatalf("verarbeitete offene anfragen = %d, quelle hatte %d", res.OpenRequestsProcessed, n)
+	}
+
+	var activeCount int
+	if err := pool.QueryRow(ctx, `SELECT count(*) FROM deletion_requests`).Scan(&activeCount); err != nil {
+		t.Fatal(err)
+	}
+	if activeCount != n {
+		t.Fatalf("aktive vorgaenge nach migration = %d, erwartet deckungsgleich mit quelle (%d)", activeCount, n)
+	}
+}

docs/MAI-03-PRUEFPROTOKOLL.md

@@ -0,0 +1,52 @@
+# MAI-03 – Prüfprotokoll: DSGVO-Löschanfragen-Migration
+
+## Umsetzung
+
+Neues Paket `archive/internal/dsgvorequestmigration` (Muster TAR-01/
+`retentionmigration`: abstrakte `Source`-Schnittstelle für archivmail,
+`Migrator.MigrateAll`, kein Umbau angrenzender Pakete).
+
+- Offene Anfrage (`StatusOpen`) → `dsgvorequest.ProcessDeletionRequest`
+  (CMP-07) wird direkt aufgerufen — keine zweite Prüf-/Entscheidungs-
+  logik, der bestehende Workflow erzeugt den aktiven Vorgang
+  (`deletion_requests`, Vier-Augen-Freigabe) und protokolliert jede
+  Einzelentscheidung (`dsgvo_decision_log`).
+- Abgeschlossene Anfrage (`StatusCompleted`) → NICHT erneut durch
+  `ProcessDeletionRequest` geschickt (würde für noch vorhandene,
+  abgelaufene Objekte fälschlich einen neuen aktiven Vorgang anstoßen),
+  sondern je aktuell gespeichertem Objekt der betroffenen Person
+  (`dpreport.SubjectReport`) direkt ein Audit-Eintrag in
+  `dsgvo_decision_log` geschrieben — kein `deletion_requests`-Eintrag,
+  also kein aktiver Vorgang.
+- `indexed_at`-artige Statusfelder der Quelle werden nicht übernommen
+  (`SourceRequest` kennt nur `open`/`completed`).
+
+## Prüfungen (echter Postgres-Testcluster, Host 204, `TEST_TENANT_DSN`)
+
+| # | Prüfung | Ergebnis |
+|---|---|---|
+| 1 | Jede offene Löschanfrage ist nach Migration als aktiver Vorgang vorhanden (Akzeptanzkriterium 1) | **bestanden** – `TestMigrateAll_OpenRequestBecomesActiveDeletionRequest`: offene Anfrage erzeugt genau 1 Zeile in `deletion_requests` für das zugehörige Objekt |
+| 2 | Jede abgeschlossene Anfrage ist nur als historischer Audit-Eintrag auffindbar, nicht als aktiver Vorgang (Akzeptanzkriterium 2) | **bestanden** – `TestMigrateAll_CompletedRequestIsHistoricalOnly`: abgeschlossene Anfrage erzeugt 0 Zeilen in `deletion_requests`, aber genau 1 auffindbaren Eintrag in `dsgvo_decision_log` |
+| 3 | Zeilenzahl offener Anfragen nach Migration deckungsgleich mit Quelle (Pflichtprüfung 1) | **bestanden** – `TestMigrateAll_RowCountMatchesSourceForOpenRequests`: 20 synthetische offene Anfragen in Quelle, nach Migration exakt 20 Zeilen in `deletion_requests` |
+| 4 | Stichprobe abgeschlossener Anfragen als Audit-Eintrag auffindbar (Pflichtprüfung 2) | **bestanden** – in Prüfung 2 enthalten (direkter Datenbank-Lookup auf `dsgvo_decision_log` nach `retention_object_id`+`data_subject_ref`) |
+
+## Build/Test-Ergebnis (Host 204)
+
+```
+cd archive && go build ./...                           -> clean
+cd archive && go vet ./internal/dsgvorequestmigration/...  -> clean
+cd archive && go test ./internal/dsgvorequestmigration/... -v -p 1
+  -> PASS (3/3 Tests)
+cd archive && go test ./... -p 1
+  -> alle Pakete bestanden bis auf internal/reconcile
+     (TestListDBStorageKeys_ReadsRealFileRevisions, vorbestehender
+     Fehlschlag durch gemeinsam genutzten Datei-Zustand auf dem
+     Testhost, unabhängig von dieser Änderung — Paket nicht berührt)
+```
+
+## Nicht umgesetzt / bewusst ausgelassen
+
+- Keine echte archivmail-SQL-Anbindung (`Source`-Implementierung gegen
+  die archivmail-Datenbank) — wie bei MAI-02/TAR-01 bleibt dies eine
+  abstrakte Schnittstelle, Tests nutzen ausschließlich synthetische
+  Fixtures (DSGVO: keine echten archivmail-Daten als Testfixtures).