RET-07: fristablauf-benachrichtigungen

- archive/internal/notifyclient: HTTP-Client fuer Core CFG-05 (gleiches
  Muster wie rbacclient/RET-08 fuer RBAC-06)
- archive/internal/retentionnotify.Run: ermittelt faellige Objekte ueber
  dieselbe Funktion wie RET-02-Job/RET-06-API-Preview, filtert je
  Klasse nach Vorlauf+Ein-Aus-Schalter, Postgres-persistente Dedupe
  (retention_notifications), Fehlschlag wird protokolliert statt
  verworfen (kein Eintrag -> Retry beim naechsten Durchlauf)
- archive/cmd/retention-notify-job: systemd-Timer-CLI, analog scrub-cli
- Abweichung vom urspruenglichen Ticket-Text dokumentiert: CFG-05 statt
  direktem CFG-02-Import (Modul-Trennung), konfigurierte zustaendige
  Rolle statt Objekt-Owner (RET-01 fuehrt keinen)
- real deployed auf 131 (timer taeglich 07:00 UTC), end-zu-ende
  bewiesen: echte notification_jobs-Zeile in Core-DB, zweiter
  Dienststart ohne Doppelversand

Pruefungen siehe archive/docs/RET-07-PRUEFPROTOKOLL.md

archive/internal/retentionnotify/retentionnotify.go

@@ -0,0 +1,158 @@
+// Package retentionnotify implementiert RET-07: Benachrichtigung einer
+// konfigurierten zuständigen Rolle (Tenant-Admin — RET-01 führt bewusst
+// keine Objekt-Owner-Beziehung) vor Ablauf einer Aufbewahrungsfrist,
+// konfigurierbarer Vorlauf je Aufbewahrungsklasse. Erzeugt NUR das
+// Ereignis über Core CFG-05 (archive/internal/notifyclient) — versendet
+// selbst keine E-Mail.
+package retentionnotify
+
+import (
+	"context"
+	"fmt"
+	"time"
+
+	"github.com/jackc/pgx/v5/pgxpool"
+
+	"gitea.perlbach24.de/scripte/nexarch/archive/internal/notifyclient"
+	"gitea.perlbach24.de/scripte/nexarch/archive/internal/retentionengine"
+)
+
+// EventType ist der an CFG-05 übergebene Ereignistyp — muss mit dem im
+// Frontend/Core bekannten Namen übereinstimmen (siehe CFG-05-Tests).
+const EventType = "retention_due_soon"
+
+// Recipient benennt Tenant-Slug, User-ID und E-Mail-Adresse der
+// konfigurierten zuständigen Rolle (Tenant-Admin), an die alle
+// Fristablauf-Benachrichtigungen dieses Tenants gehen.
+type Recipient struct {
+	TenantSlug string
+	UserID     string
+	Email      string
+}
+
+// Result ist das Ergebnis EINES benachrichtigten (oder fehlgeschlagenen)
+// Objekts — der Aufrufer (cmd/retention-notify-job) protokolliert Err
+// explizit, kein stillschweigendes Verwerfen (Pflichtprüfung 3).
+type Result struct {
+	RetentionObjectID string
+	RetentionClass    string
+	JobID             string
+	Skipped           bool
+	Err               error
+}
+
+type classRuleLead struct {
+	leadDays int
+	enabled  bool
+}
+
+// Run führt EINEN Durchlauf des Benachrichtigungs-Jobs aus: ermittelt je
+// aktiver, benachrichtigungs-aktivierter Aufbewahrungsklasse die Objekte,
+// deren Stichtag innerhalb des konfigurierten Vorlaufs liegt, überspringt
+// bereits benachrichtigte Objekte (Akzeptanzkriterium 2, Postgres-
+// persistent — übersteht einen Job-Neustart) und löst für den Rest je ein
+// Ereignis über CFG-05 aus.
+func Run(ctx context.Context, pool *pgxpool.Pool, client *notifyclient.Client, now time.Time, recipient Recipient) ([]Result, error) {
+	// notify_lead_days/notify_enabled sind nicht Teil von
+	// retentionengine.ClassRule (RET-02/RET-06 kennen sie nicht) — direkt
+	// gelesen, um retentionengine nicht um RET-07-eigene Felder zu
+	// erweitern (kein Umbau angrenzender Bereiche).
+	leadByClass := make(map[string]classRuleLead)
+	maxLeadDays := 0
+	leadRows, err := pool.Query(ctx, `SELECT retention_class, notify_lead_days, notify_enabled FROM retention_class_rules WHERE active`)
+	if err != nil {
+		return nil, fmt.Errorf("retentionnotify: benachrichtigungs-konfiguration laden: %w", err)
+	}
+	for leadRows.Next() {
+		var class string
+		var lead classRuleLead
+		if err := leadRows.Scan(&class, &lead.leadDays, &lead.enabled); err != nil {
+			leadRows.Close()
+			return nil, fmt.Errorf("retentionnotify: konfigurationszeile lesen: %w", err)
+		}
+		leadByClass[class] = lead
+		if lead.leadDays > maxLeadDays {
+			maxLeadDays = lead.leadDays
+		}
+	}
+	if err := leadRows.Err(); err != nil {
+		return nil, fmt.Errorf("retentionnotify: benachrichtigungs-konfiguration lesen: %w", err)
+	}
+	leadRows.Close()
+
+	if maxLeadDays == 0 {
+		return nil, nil
+	}
+
+	// Nutzt DIESELBE Funktion wie der RET-02-Job/RET-06-API-Preview
+	// (kein zweiter Ermittlungspfad) — asOf auf den größten konfigurierten
+	// Vorlauf gesetzt, je Klasse wird unten mit deren EIGENEM Vorlauf
+	// gefiltert.
+	candidates, err := retentionengine.ListExpiringObjects(ctx, pool, now.AddDate(0, 0, maxLeadDays))
+	if err != nil {
+		return nil, fmt.Errorf("retentionnotify: ablaufende objekte ermitteln: %w", err)
+	}
+
+	alreadyNotified, err := loadAlreadyNotified(ctx, pool)
+	if err != nil {
+		return nil, err
+	}
+
+	var results []Result
+	for _, obj := range candidates {
+		lead, known := leadByClass[obj.RetentionClass]
+		if !known || !lead.enabled {
+			continue
+		}
+		if !obj.DueDate.Before(now.AddDate(0, 0, lead.leadDays+1)) {
+			// Ausserhalb des klassen-eigenen Vorlaufs (nur mit dem
+			// globalen maxLeadDays vorselektiert).
+			continue
+		}
+		if alreadyNotified[obj.RetentionObjectID] {
+			continue
+		}
+
+		res := Result{RetentionObjectID: obj.RetentionObjectID, RetentionClass: obj.RetentionClass}
+		enq, err := client.Enqueue(ctx, recipient.TenantSlug, recipient.UserID, EventType, "email", recipient.Email, map[string]any{
+			"object_type":      obj.ObjectType,
+			"object_reference": obj.ObjectReference,
+			"retention_class":  obj.RetentionClass,
+			"due_date":         obj.DueDate.Format(time.RFC3339),
+		})
+		if err != nil {
+			res.Err = err
+			results = append(results, res)
+			// Kein INSERT in retention_notifications bei Fehler — das
+			// Objekt wird beim naechsten Durchlauf erneut versucht,
+			// statt stillschweigend als erledigt zu gelten.
+			continue
+		}
+		res.JobID = enq.JobID
+		res.Skipped = enq.Skipped
+
+		if _, err := pool.Exec(ctx, `INSERT INTO retention_notifications (retention_object_id) VALUES ($1)`, obj.RetentionObjectID); err != nil {
+			res.Err = fmt.Errorf("retentionnotify: benachrichtigung als versendet markieren: %w", err)
+		}
+		results = append(results, res)
+	}
+	return results, nil
+}
+
+func loadAlreadyNotified(ctx context.Context, pool *pgxpool.Pool) (map[string]bool, error) {
+	rows, err := pool.Query(ctx, `SELECT retention_object_id FROM retention_notifications`)
+	if err != nil {
+		return nil, fmt.Errorf("retentionnotify: bereits benachrichtigte objekte laden: %w", err)
+	}
+	defer rows.Close()
+
+	out := make(map[string]bool)
+	for rows.Next() {
+		var id string
+		if err := rows.Scan(&id); err != nil {
+			return nil, fmt.Errorf("retentionnotify: zeile lesen: %w", err)
+		}
+		out[id] = true
+	}
+	return out, rows.Err()
+}

archive/internal/retentionnotify/retentionnotify_test.go

@@ -0,0 +1,223 @@
+package retentionnotify
+
+import (
+	"context"
+	"encoding/json"
+	"net/http"
+	"net/http/httptest"
+	"os"
+	"sync/atomic"
+	"testing"
+	"time"
+
+	"github.com/jackc/pgx/v5/pgxpool"
+
+	"gitea.perlbach24.de/scripte/nexarch/archive/internal/notifyclient"
+	"gitea.perlbach24.de/scripte/nexarch/archive/internal/retentionengine"
+)
+
+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(),
+			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 retention_class_rules (
+			retention_class TEXT PRIMARY KEY, duration INTERVAL NOT NULL,
+			active BOOLEAN NOT NULL DEFAULT true
+		);
+		ALTER TABLE retention_class_rules ADD COLUMN IF NOT EXISTS notify_lead_days INT NOT NULL DEFAULT 7;
+		ALTER TABLE retention_class_rules ADD COLUMN IF NOT EXISTS notify_enabled BOOLEAN NOT NULL DEFAULT true;
+		CREATE TABLE IF NOT EXISTS retention_notifications (
+			retention_object_id UUID PRIMARY KEY REFERENCES retention_objects(id) ON DELETE CASCADE,
+			notified_at TIMESTAMPTZ NOT NULL DEFAULT now()
+		);
+	`); err != nil {
+		t.Fatalf("schema: %v", err)
+	}
+	t.Cleanup(func() {
+		_, _ = pool.Exec(context.Background(), `TRUNCATE retention_notifications, retention_class_assignments, retention_objects CASCADE; TRUNCATE retention_class_rules`)
+	})
+	return pool
+}
+
+// fakeCFG05Server zaehlt Aufrufe und liefert eine feste Antwort - simuliert
+// den echten CFG-05-Endpunkt, ohne das Core-Modul einzubinden (Archive kann
+// es nicht direkt importieren, siehe notifyclient).
+func fakeCFG05Server(t *testing.T, fail bool) (*notifyclient.Client, *int32) {
+	t.Helper()
+	var calls int32
+	mux := http.NewServeMux()
+	mux.HandleFunc("POST /notify/enqueue", func(w http.ResponseWriter, r *http.Request) {
+		atomic.AddInt32(&calls, 1)
+		if fail {
+			http.Error(w, "simulierter zustellfehler", http.StatusInternalServerError)
+			return
+		}
+		w.Header().Set("Content-Type", "application/json")
+		_ = json.NewEncoder(w).Encode(map[string]any{"job_id": "fake-job-id", "skipped": false})
+	})
+	server := httptest.NewServer(mux)
+	t.Cleanup(server.Close)
+	return notifyclient.New(server.URL, "test-token"), &calls
+}
+
+// insertDueObject legt ein Objekt an, dessen Stichtag (assigned_at +
+// Klassen-Frist) genau dueIn ab jetzt liegt — die Subtraktion der Frist
+// erfolgt IN Postgres (dieselbe INTERVAL-Arithmetik wie ComputeDueDate),
+// keine eigene Kalenderrechnung in Go.
+func insertDueObject(t *testing.T, ctx context.Context, pool *pgxpool.Pool, objectRef, class string, dueIn time.Duration) string {
+	t.Helper()
+	var objID string
+	if err := pool.QueryRow(ctx, `INSERT INTO retention_objects (object_type, object_reference) VALUES ('dms_document', $1) RETURNING id`, objectRef).Scan(&objID); err != nil {
+		t.Fatalf("objekt anlegen: %v", err)
+	}
+	desiredDue := time.Now().UTC().Add(dueIn)
+	if _, err := pool.Exec(ctx, `
+		INSERT INTO retention_class_assignments (retention_object_id, retention_class, assigned_at)
+		SELECT $1, $2, $3::timestamptz - r.duration FROM retention_class_rules r WHERE r.retention_class = $2
+	`, objID, class, desiredDue); err != nil {
+		t.Fatalf("zuordnung anlegen: %v", err)
+	}
+	return objID
+}
+
+var testRecipient = Recipient{TenantSlug: "acme", UserID: "tenant-admin", Email: "admin@acme.example"}
+
+// TestRun_ShortLeadTimeTriggersExactlyOneNotification ist die geforderte
+// Pflichtpruefung 1.
+func TestRun_ShortLeadTimeTriggersExactlyOneNotification(t *testing.T) {
+	pool := requireTestPool(t)
+	ctx := context.Background()
+
+	if err := retentionengine.ConfigureClassRule(ctx, pool, "klasse-kurz", "1 day"); err != nil {
+		t.Fatal(err)
+	}
+	if _, err := pool.Exec(ctx, `UPDATE retention_class_rules SET notify_lead_days = 1, notify_enabled = true WHERE retention_class = 'klasse-kurz'`); err != nil {
+		t.Fatal(err)
+	}
+	insertDueObject(t, ctx, pool, "kurz-doc", "klasse-kurz", 12*time.Hour)
+
+	client, calls := fakeCFG05Server(t, false)
+	results, err := Run(ctx, pool, client, time.Now().UTC(), testRecipient)
+	if err != nil {
+		t.Fatalf("run: %v", err)
+	}
+	if len(results) != 1 || results[0].Err != nil {
+		t.Fatalf("erwartet genau ein ergebnis ohne fehler, habe: %+v", results)
+	}
+	if atomic.LoadInt32(calls) != 1 {
+		t.Fatalf("erwartet genau einen cfg-05-aufruf, habe %d", *calls)
+	}
+}
+
+// TestRun_DisabledNotificationSendsNothing ist die geforderte
+// Pflichtpruefung 2.
+func TestRun_DisabledNotificationSendsNothing(t *testing.T) {
+	pool := requireTestPool(t)
+	ctx := context.Background()
+
+	if err := retentionengine.ConfigureClassRule(ctx, pool, "klasse-deaktiviert", "1 day"); err != nil {
+		t.Fatal(err)
+	}
+	if _, err := pool.Exec(ctx, `UPDATE retention_class_rules SET notify_lead_days = 1, notify_enabled = false WHERE retention_class = 'klasse-deaktiviert'`); err != nil {
+		t.Fatal(err)
+	}
+	insertDueObject(t, ctx, pool, "deaktiviert-doc", "klasse-deaktiviert", 12*time.Hour)
+
+	client, calls := fakeCFG05Server(t, false)
+	results, err := Run(ctx, pool, client, time.Now().UTC(), testRecipient)
+	if err != nil {
+		t.Fatalf("run: %v", err)
+	}
+	if len(results) != 0 {
+		t.Fatalf("erwartet keine benachrichtigung bei deaktivierter klasse, habe: %+v", results)
+	}
+	if atomic.LoadInt32(calls) != 0 {
+		t.Fatalf("erwartet keinen cfg-05-aufruf, habe %d", *calls)
+	}
+}
+
+// TestRun_FailedDeliveryIsReportedNotSwallowed ist die geforderte
+// Pflichtpruefung 3.
+func TestRun_FailedDeliveryIsReportedNotSwallowed(t *testing.T) {
+	pool := requireTestPool(t)
+	ctx := context.Background()
+
+	if err := retentionengine.ConfigureClassRule(ctx, pool, "klasse-fehler", "1 day"); err != nil {
+		t.Fatal(err)
+	}
+	if _, err := pool.Exec(ctx, `UPDATE retention_class_rules SET notify_lead_days = 1, notify_enabled = true WHERE retention_class = 'klasse-fehler'`); err != nil {
+		t.Fatal(err)
+	}
+	objID := insertDueObject(t, ctx, pool, "fehler-doc", "klasse-fehler", 12*time.Hour)
+
+	client, _ := fakeCFG05Server(t, true)
+	results, err := Run(ctx, pool, client, time.Now().UTC(), testRecipient)
+	if err != nil {
+		t.Fatalf("run: %v", err)
+	}
+	if len(results) != 1 || results[0].Err == nil {
+		t.Fatalf("erwartet ein ergebnis MIT protokolliertem fehler, habe: %+v", results)
+	}
+
+	var count int
+	if err := pool.QueryRow(ctx, `SELECT count(*) FROM retention_notifications WHERE retention_object_id = $1`, objID).Scan(&count); err != nil {
+		t.Fatal(err)
+	}
+	if count != 0 {
+		t.Fatalf("fehlgeschlagener versand darf NICHT als benachrichtigt markiert werden (sonst kein retry)")
+	}
+}
+
+// TestRun_PreventsDoubleNotificationAcrossRuns ist Akzeptanzkriterium 2 -
+// Mehrfachversand wird verhindert, auch nach einem simulierten Neustart
+// (zweiter Run() mit frischem Aufruf, kein In-Memory-Zustand zwischen den
+// Durchlaeufen).
+func TestRun_PreventsDoubleNotificationAcrossRuns(t *testing.T) {
+	pool := requireTestPool(t)
+	ctx := context.Background()
+
+	if err := retentionengine.ConfigureClassRule(ctx, pool, "klasse-doppelt", "1 day"); err != nil {
+		t.Fatal(err)
+	}
+	if _, err := pool.Exec(ctx, `UPDATE retention_class_rules SET notify_lead_days = 1, notify_enabled = true WHERE retention_class = 'klasse-doppelt'`); err != nil {
+		t.Fatal(err)
+	}
+	insertDueObject(t, ctx, pool, "doppelt-doc", "klasse-doppelt", 12*time.Hour)
+
+	client, calls := fakeCFG05Server(t, false)
+	now := time.Now().UTC()
+
+	if _, err := Run(ctx, pool, client, now, testRecipient); err != nil {
+		t.Fatalf("erster run: %v", err)
+	}
+	if _, err := Run(ctx, pool, client, now, testRecipient); err != nil {
+		t.Fatalf("zweiter run (simulierter neustart): %v", err)
+	}
+
+	if atomic.LoadInt32(calls) != 1 {
+		t.Fatalf("erwartet genau einen cfg-05-aufruf ueber beide durchlaeufe, habe %d", *calls)
+	}
+}