Compare commits

..
Author SHA1 Message Date
sysops c344dea218 TEN-08: tenant-loeschung-unter-retention-vorbehalt-gobd (RetentionChecker-Schnittstelle gegen Archive RET-03/CMP-06, ProcessDueDeletions haelt gesperrte Tenants zurueck) 2026-08-28 22:51:05 +02:00
sysopsandClaude Sonnet 5 45bc10719a TEN-04: tenant-lifecycle-suspendieren-reaktivieren-loeschen
Zustandsautomat active/suspended/pending_deletion/deleted als First-Class-
Konzept (previous_status + deletion_scheduled_at in der Registry). Alle
Uebergaenge in Registry.transition als atomarer Check-and-Set (UPDATE ...
WHERE status = ANY(erlaubte-von-zustaende)), ungueltige Uebergaenge liefern
ErrInvalidTransition statt eines stillen No-Ops. ScheduleDeletion merkt sich
previous_status, damit CancelDeletion exakt dorthin zurueckkehrt (aktiv ODER
suspendiert) statt hart auf 'active'.

Lifecycle.ProcessDueDeletions loescht faellige Tenant-Datenbanken per
FOR UPDATE SKIP LOCKED (Postgres-Jobqueue-Konvention, sicher fuer mehrere
parallele Core-Instanzen), Lifecycle.RunSweeper triggert das periodisch per
In-Prozess-Goroutine. Lifecycle.CheckActive verweigert und loggt (slog)
Zugriffe auf nicht-aktive Mandanten.

Bugfix nebenbei: Registry.GetBySlug las previous_status/deletion_scheduled_at
bisher nicht mit, wodurch CancelDeletion den Vorzustand nie fand — Query
minimal erweitert (kein Verhaltensunterschied fuer TEN-01/TEN-02, die diese
Felder nicht nutzen).

Neu: scripts/reset-test-env.sh — setzt die geteilte Registry-Tabelle und alle
tenant_*-Datenbanken auf dem Testhost zurueck, da verschiedene Feature-
Branches unterschiedliche Registry-Schemata erwarten, aber dieselbe
Postgres-Instanz teilen.

Pruefungen (ausgefuehrt auf root@192.168.1.131, go build/vet/test PASS):
1. Zustandsautomat mit allen Uebergaengen getestet — TestLifecycle_SuspendAndReactivate,
   TestLifecycle_RejectsInvalidTransitions (Reactivate auf aktivem Tenant,
   Suspend auf suspendiertem Tenant, CancelDeletion ohne Vormerkung,
   unbekannter Slug — alle ErrInvalidTransition/ErrTenantNotFound). PASS.
2. Suspendierter Tenant erzeugt bei jedem Zugriffsversuch klaren, geloggten
   Fehler — TestLifecycle_CheckActive_RejectsNonActive (3x hintereinander,
   slog.Warn nachweislich pro Aufruf). PASS.
3. Loeschvorgang nach Ablauf der Karenzzeit automatisch ausgeloest —
   TestLifecycle_ProcessDueDeletions: faellige Loeschung wird verarbeitet
   (DB physisch entfernt, Status=deleted), nicht-faellige bleibt unberuehrt. PASS.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
2026-08-27 18:42:24 +02:00
19 changed files with 734 additions and 691 deletions
-96
View File
@@ -1,96 +0,0 @@
// Package audit implementiert Core AUD-01: das zentrale, vom allgemeinen
// Anwendungs-Log getrennte Audit-Datenmodell fuer sicherheits- und
// compliancerelevante Ereignisse (wer, was, wann, an welchem Tenant).
// Unveraenderlichkeit (Append-only) ist AUD-02, Export/Filter-API ist AUD-03
// — dieses Paket liefert nur das Datenmodell und den EINEN zentralen
// Schreibpfad (Akzeptanzkriterium 3).
package audit
import (
"context"
"encoding/json"
"errors"
"fmt"
"time"
"github.com/jackc/pgx/v5/pgxpool"
)
// SystemTenant ist der reservierte Tenant-Bezug fuer mandantenuebergreifende
// Ereignisse (z.B. Superadmin-Aktionen) — es gibt bewusst KEINEN Weg, ein
// Ereignis ganz ohne Tenant-Bezug zu schreiben (Akzeptanzkriterium 2).
const SystemTenant = "system"
var ErrMissingTenant = errors.New("audit: tenant_slug darf nicht leer sein")
var ErrMissingActor = errors.New("audit: actor darf nicht leer sein")
var ErrMissingAction = errors.New("audit: action darf nicht leer sein")
// Event ist ein strukturiertes Audit-Ereignis (Akzeptanzkriterium 1: Akteur,
// Aktion, Zielobjekt, Zeitpunkt, Tenant).
type Event struct {
TenantSlug string
Actor string
Action string
Target string
Metadata map[string]any
OccurredAt time.Time
}
// Log ist der EINE zentrale Schreibpfad fuer Audit-Ereignisse — es gibt
// bewusst keine zweite Schreibmoeglichkeit, damit kein Handler versehentlich
// direkt in audit_events schreibt und dabei die Validierung umgeht
// (Akzeptanzkriterium 3).
type Log struct {
pool *pgxpool.Pool
}
func NewLog(pool *pgxpool.Pool) *Log {
return &Log{pool: pool}
}
// Record persistiert genau einen Audit-Eintrag. Fehlender Tenant-Bezug wird
// bereits hier abgewiesen (klarer Fehler statt Constraint-Verletzung im
// Normalfall) — die Datenbank-CHECK-Constraint aus der Migration ist die
// zweite, unumgehbare Verteidigungslinie (Akzeptanzkriterium 2 / Pruefung 2).
func (l *Log) Record(ctx context.Context, e Event) error {
if e.TenantSlug == "" {
return ErrMissingTenant
}
if e.Actor == "" {
return ErrMissingActor
}
if e.Action == "" {
return ErrMissingAction
}
if e.Metadata == nil {
e.Metadata = map[string]any{}
}
metadataJSON, err := json.Marshal(e.Metadata)
if err != nil {
return fmt.Errorf("metadaten serialisieren: %w", err)
}
if e.OccurredAt.IsZero() {
e.OccurredAt = time.Now()
}
_, err = l.pool.Exec(ctx, `
INSERT INTO audit_events (occurred_at, tenant_slug, actor, action, target, metadata)
VALUES ($1, $2, $3, $4, $5, $6)
`, e.OccurredAt, e.TenantSlug, e.Actor, e.Action, e.Target, metadataJSON)
if err != nil {
return fmt.Errorf("audit-ereignis schreiben: %w", err)
}
return nil
}
// CountByTenant ist eine schlanke Lesehilfe fuer Tests/Diagnose — die
// eigentliche Filter-/Export-API ist AUD-03, hier bewusst nicht vorgezogen.
func (l *Log) CountByTenant(ctx context.Context, tenantSlug string) (int, error) {
var n int
if err := l.pool.QueryRow(ctx, `
SELECT count(*) FROM audit_events WHERE tenant_slug = $1
`, tenantSlug).Scan(&n); err != nil {
return 0, fmt.Errorf("audit-ereignisse zaehlen: %w", err)
}
return n, nil
}
-132
View File
@@ -1,132 +0,0 @@
package audit
import (
"context"
"errors"
"os"
"testing"
"github.com/jackc/pgx/v5/pgxpool"
)
func setupAuditTest(t *testing.T) (*Log, *pgxpool.Pool, func()) {
t.Helper()
adminDSN := os.Getenv("TEST_ADMIN_DSN")
if adminDSN == "" {
t.Skip("TEST_ADMIN_DSN nicht gesetzt, Integrationstest uebersprungen")
}
ctx := context.Background()
pool, err := pgxpool.New(ctx, adminDSN)
if err != nil {
t.Fatalf("pool: %v", err)
}
if _, err := pool.Exec(ctx, `
CREATE TABLE IF NOT EXISTS audit_events (
id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
occurred_at TIMESTAMPTZ NOT NULL DEFAULT now(),
tenant_slug TEXT NOT NULL CHECK (tenant_slug <> ''),
actor TEXT NOT NULL CHECK (actor <> ''),
action TEXT NOT NULL CHECK (action <> ''),
target TEXT NOT NULL,
metadata JSONB NOT NULL DEFAULT '{}'::jsonb
)`); err != nil {
t.Fatalf("schema: %v", err)
}
cleanup := func() {
_, _ = pool.Exec(ctx, `DELETE FROM audit_events WHERE tenant_slug LIKE 'test\_%' ESCAPE '\' OR tenant_slug = $1`, SystemTenant)
pool.Close()
}
return NewLog(pool), pool, cleanup
}
// Akzeptanzkriterium 1 + Pruefung 1: ein sicherheitsrelevanter Vorgang
// (hier: fehlgeschlagener Login) erzeugt zuverlaessig genau einen Eintrag.
func TestRecord_PersistsExactlyOneEventPerSecurityIncident(t *testing.T) {
log, pool, cleanup := setupAuditTest(t)
defer cleanup()
ctx := context.Background()
err := log.Record(ctx, Event{
TenantSlug: "test_acme",
Actor: "alice@example.com",
Action: "iam.login_failed",
Target: "user:alice@example.com",
Metadata: map[string]any{"reason": "falsches passwort"},
})
if err != nil {
t.Fatalf("record: %v", err)
}
count, err := log.CountByTenant(ctx, "test_acme")
if err != nil {
t.Fatalf("count: %v", err)
}
if count != 1 {
t.Fatalf("erwartet genau 1 audit-eintrag, habe %d", count)
}
var actor, action, target string
if err := pool.QueryRow(ctx, `
SELECT actor, action, target FROM audit_events WHERE tenant_slug = 'test_acme'
`).Scan(&actor, &action, &target); err != nil {
t.Fatalf("eintrag lesen: %v", err)
}
if actor != "alice@example.com" || action != "iam.login_failed" || target != "user:alice@example.com" {
t.Fatalf("eintrag unerwartet: actor=%q action=%q target=%q", actor, action, target)
}
}
// Akzeptanzkriterium 2 + Pruefung 2 (App-Ebene): fehlender Tenant-Bezug wird
// bereits vom zentralen Schreibpfad abgewiesen.
func TestRecord_RejectsMissingTenant(t *testing.T) {
log, _, cleanup := setupAuditTest(t)
defer cleanup()
ctx := context.Background()
err := log.Record(ctx, Event{TenantSlug: "", Actor: "alice", Action: "irgendwas"})
if !errors.Is(err, ErrMissingTenant) {
t.Fatalf("erwartet ErrMissingTenant, habe %v", err)
}
}
// Akzeptanzkriterium 2 + Pruefung 2 (DB-Ebene): selbst ein direkter INSERT,
// der Log.Record umgeht, wird durch die CHECK-Constraint verhindert — der
// Schutz haengt nicht allein von der Go-Validierung ab.
func TestConstraint_RejectsMissingTenantAtDatabaseLevel(t *testing.T) {
_, pool, cleanup := setupAuditTest(t)
defer cleanup()
ctx := context.Background()
_, err := pool.Exec(ctx, `
INSERT INTO audit_events (tenant_slug, actor, action, target)
VALUES ('', 'alice', 'irgendwas', 'ziel')
`)
if err == nil {
t.Fatal("erwartet fehler durch CHECK-constraint bei leerem tenant_slug, habe nil")
}
}
func TestRecord_RejectsMissingActorAndAction(t *testing.T) {
log, _, cleanup := setupAuditTest(t)
defer cleanup()
ctx := context.Background()
if err := log.Record(ctx, Event{TenantSlug: "test_acme", Actor: "", Action: "x"}); !errors.Is(err, ErrMissingActor) {
t.Fatalf("erwartet ErrMissingActor, habe %v", err)
}
if err := log.Record(ctx, Event{TenantSlug: "test_acme", Actor: "alice", Action: ""}); !errors.Is(err, ErrMissingAction) {
t.Fatalf("erwartet ErrMissingAction, habe %v", err)
}
}
func TestRecord_SystemTenantForCrossTenantEvents(t *testing.T) {
log, _, cleanup := setupAuditTest(t)
defer cleanup()
ctx := context.Background()
if err := log.Record(ctx, Event{TenantSlug: SystemTenant, Actor: "superadmin", Action: "tenant.provisioned", Target: "tenant:acme"}); err != nil {
t.Fatalf("record mit SystemTenant: %v", err)
}
}
-122
View File
@@ -1,122 +0,0 @@
package audit
import (
"context"
"encoding/csv"
"encoding/json"
"fmt"
"io"
"time"
)
// Filter grenzt einen Export ein (Akzeptanzkriterium 1). Leere/Nil-Felder
// bedeuten "kein Filter auf diesem Feld".
type Filter struct {
TenantSlug string
Actor string
Action string
From *time.Time
To *time.Time
}
func buildFilterQuery(f Filter) (string, []any) {
query := `SELECT occurred_at, tenant_slug, actor, action, target, metadata FROM audit_events WHERE 1=1`
var args []any
if f.TenantSlug != "" {
args = append(args, f.TenantSlug)
query += fmt.Sprintf(" AND tenant_slug = $%d", len(args))
}
if f.Actor != "" {
args = append(args, f.Actor)
query += fmt.Sprintf(" AND actor = $%d", len(args))
}
if f.Action != "" {
args = append(args, f.Action)
query += fmt.Sprintf(" AND action = $%d", len(args))
}
if f.From != nil {
args = append(args, *f.From)
query += fmt.Sprintf(" AND occurred_at >= $%d", len(args))
}
if f.To != nil {
args = append(args, *f.To)
query += fmt.Sprintf(" AND occurred_at <= $%d", len(args))
}
query += " ORDER BY occurred_at"
return query, args
}
// StreamCSV schreibt gefilterte Audit-Eintraege direkt als CSV in w, Zeile
// fuer Zeile ueber rows.Next() — es wird zu keinem Zeitpunkt das gesamte
// Ergebnis im Speicher aufgebaut (Akzeptanzkriterium 3 / Pruefung 1).
func (l *Log) StreamCSV(ctx context.Context, filter Filter, w io.Writer) error {
query, args := buildFilterQuery(filter)
rows, err := l.pool.Query(ctx, query, args...)
if err != nil {
return fmt.Errorf("export abfragen: %w", err)
}
defer rows.Close()
cw := csv.NewWriter(w)
if err := cw.Write([]string{"occurred_at", "tenant_slug", "actor", "action", "target", "metadata"}); err != nil {
return fmt.Errorf("csv-header schreiben: %w", err)
}
for rows.Next() {
var occurredAt time.Time
var tenantSlug, actor, action, target string
var metadataJSON []byte
if err := rows.Scan(&occurredAt, &tenantSlug, &actor, &action, &target, &metadataJSON); err != nil {
return fmt.Errorf("zeile lesen: %w", err)
}
if err := cw.Write([]string{
occurredAt.Format(time.RFC3339), tenantSlug, actor, action, target, string(metadataJSON),
}); err != nil {
return fmt.Errorf("csv-zeile schreiben: %w", err)
}
}
cw.Flush()
if err := rows.Err(); err != nil {
return fmt.Errorf("export lesen: %w", err)
}
return cw.Error()
}
// exportRecord ist die JSON-Repraesentation einer exportierten Zeile.
type exportRecord struct {
OccurredAt time.Time `json:"occurred_at"`
TenantSlug string `json:"tenant_slug"`
Actor string `json:"actor"`
Action string `json:"action"`
Target string `json:"target"`
Metadata json.RawMessage `json:"metadata"`
}
// StreamJSON schreibt gefilterte Audit-Eintraege als JSON Lines (ein
// JSON-Objekt pro Zeile) — bewusst kein einzelnes grosses JSON-Array, da
// dessen korrektes Streaming (Kommas/Klammern ohne Zwischenpufferung)
// unnoetige Komplexitaet fuer denselben Zweck waere. Wie StreamCSV
// zeilenweise ueber rows.Next(), kein Aufbau im Speicher.
func (l *Log) StreamJSON(ctx context.Context, filter Filter, w io.Writer) error {
query, args := buildFilterQuery(filter)
rows, err := l.pool.Query(ctx, query, args...)
if err != nil {
return fmt.Errorf("export abfragen: %w", err)
}
defer rows.Close()
enc := json.NewEncoder(w)
for rows.Next() {
var rec exportRecord
var metadataJSON []byte
if err := rows.Scan(&rec.OccurredAt, &rec.TenantSlug, &rec.Actor, &rec.Action, &rec.Target, &metadataJSON); err != nil {
return fmt.Errorf("zeile lesen: %w", err)
}
rec.Metadata = metadataJSON
if err := enc.Encode(rec); err != nil {
return fmt.Errorf("json-zeile schreiben: %w", err)
}
}
return rows.Err()
}
-73
View File
@@ -1,73 +0,0 @@
package audit
import (
"context"
"net/http"
"time"
)
// Authorize entscheidet, ob caller den Export ausfuehren darf. Die
// eigentliche Rollen-/Rechtepruefung (RBAC-02 Policy-Enforcement) ist nicht
// Teil dieser Kachel — ExportHandler kennt nur diese schmale Schnittstelle,
// analog zum RetentionRegistrar-Muster aus AUD-05.
type Authorize func(ctx context.Context, caller string) bool
// ExportHandler stellt den Export als HTTP-Endpunkt bereit
// (Akzeptanzkriterium 2: fuer berechtigte Rollen verfuegbar).
type ExportHandler struct {
log *Log
authorize Authorize
}
func NewExportHandler(log *Log, authorize Authorize) *ExportHandler {
return &ExportHandler{log: log, authorize: authorize}
}
// Export liest Filter-Query-Parameter (tenant, actor, action, from, to,
// format) und schreibt DIREKT auf den ResponseWriter (io.Writer) — dieselbe
// Streaming-Funktion wie in export.go, kein zusaetzlicher Pufferungsschritt.
func (h *ExportHandler) Export(w http.ResponseWriter, r *http.Request) {
// "caller" identifiziert die anfragende Person fuer die Berechtigungs-
// pruefung — bewusst getrennt vom Filterfeld "actor" (das den
// AUDIT-Akteur meint, ueber den gefiltert wird).
caller := r.URL.Query().Get("caller")
if caller == "" || !h.authorize(r.Context(), caller) {
http.Error(w, "keine berechtigung fuer audit-log-export", http.StatusForbidden)
return
}
filter := Filter{
TenantSlug: r.URL.Query().Get("tenant"),
Actor: r.URL.Query().Get("actor"),
Action: r.URL.Query().Get("action"),
}
if from := r.URL.Query().Get("from"); from != "" {
t, err := time.Parse(time.RFC3339, from)
if err != nil {
http.Error(w, "ungueltiges from-datum, erwartet RFC3339", http.StatusBadRequest)
return
}
filter.From = &t
}
if to := r.URL.Query().Get("to"); to != "" {
t, err := time.Parse(time.RFC3339, to)
if err != nil {
http.Error(w, "ungueltiges to-datum, erwartet RFC3339", http.StatusBadRequest)
return
}
filter.To = &t
}
switch r.URL.Query().Get("format") {
case "json":
w.Header().Set("Content-Type", "application/x-ndjson")
if err := h.log.StreamJSON(r.Context(), filter, w); err != nil {
http.Error(w, "export fehlgeschlagen", http.StatusInternalServerError)
}
default:
w.Header().Set("Content-Type", "text/csv")
if err := h.log.StreamCSV(r.Context(), filter, w); err != nil {
http.Error(w, "export fehlgeschlagen", http.StatusInternalServerError)
}
}
}
-222
View File
@@ -1,222 +0,0 @@
package audit
import (
"bytes"
"context"
"encoding/csv"
"fmt"
"net/http"
"net/http/httptest"
"os"
"runtime"
"strings"
"testing"
"time"
"github.com/jackc/pgx/v5/pgxpool"
)
func setupExportTest(t *testing.T) (*Log, func()) {
t.Helper()
adminDSN := os.Getenv("TEST_ADMIN_DSN")
if adminDSN == "" {
t.Skip("TEST_ADMIN_DSN nicht gesetzt, Integrationstest uebersprungen")
}
ctx := context.Background()
pool, err := pgxpool.New(ctx, adminDSN)
if err != nil {
t.Fatalf("pool: %v", err)
}
if _, err := pool.Exec(ctx, `
CREATE TABLE IF NOT EXISTS audit_events (
id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
occurred_at TIMESTAMPTZ NOT NULL DEFAULT now(),
tenant_slug TEXT NOT NULL CHECK (tenant_slug <> ''),
actor TEXT NOT NULL CHECK (actor <> ''),
action TEXT NOT NULL CHECK (action <> ''),
target TEXT NOT NULL,
metadata JSONB NOT NULL DEFAULT '{}'::jsonb
)`); err != nil {
t.Fatalf("schema: %v", err)
}
cleanup := func() { pool.Close() }
return NewLog(pool), cleanup
}
// Akzeptanzkriterium 1 + Pruefung 2: Filterkombinationen liefern korrekte
// Teilmengen.
func TestExport_FilterCombinations(t *testing.T) {
log, cleanup := setupExportTest(t)
defer cleanup()
ctx := context.Background()
run := fmt.Sprintf("_%d", time.Now().UnixNano())
tenantA, tenantB := "test_fa"+run, "test_fb"+run
alice, bob := "alice"+run, "bob"+run
events := []Event{
{TenantSlug: tenantA, Actor: alice, Action: "login", Target: "x"},
{TenantSlug: tenantA, Actor: bob, Action: "login", Target: "x"},
{TenantSlug: tenantA, Actor: alice, Action: "logout", Target: "x"},
{TenantSlug: tenantB, Actor: alice, Action: "login", Target: "x"},
}
for _, e := range events {
if err := log.Record(ctx, e); err != nil {
t.Fatalf("record: %v", err)
}
}
cases := []struct {
name string
filter Filter
wantLen int
}{
{"nach tenant", Filter{TenantSlug: tenantA}, 3},
{"nach tenant+actor", Filter{TenantSlug: tenantA, Actor: alice}, 2},
{"nach tenant+actor+action", Filter{TenantSlug: tenantA, Actor: alice, Action: "login"}, 1},
{"nach actor ueber beide tenants", Filter{Actor: alice, Action: "login"}, 2},
}
for _, c := range cases {
t.Run(c.name, func(t *testing.T) {
var buf bytes.Buffer
if err := log.StreamCSV(ctx, c.filter, &buf); err != nil {
t.Fatalf("stream: %v", err)
}
rows, err := csv.NewReader(&buf).ReadAll()
if err != nil {
t.Fatalf("csv parsen: %v", err)
}
got := len(rows) - 1 // Header abziehen
if got != c.wantLen {
t.Fatalf("erwartet %d zeilen, habe %d", c.wantLen, got)
}
})
}
}
// Akzeptanzkriterium 1 + Pruefung 2: Zeitraum-Filter.
func TestExport_TimeRangeFilter(t *testing.T) {
log, cleanup := setupExportTest(t)
defer cleanup()
ctx := context.Background()
tenant := fmt.Sprintf("test_tr_%d", time.Now().UnixNano())
past := time.Now().Add(-48 * time.Hour)
future := time.Now().Add(48 * time.Hour)
if err := log.Record(ctx, Event{TenantSlug: tenant, Actor: "a", Action: "x", Target: "t", OccurredAt: time.Now()}); err != nil {
t.Fatalf("record: %v", err)
}
var buf bytes.Buffer
if err := log.StreamCSV(ctx, Filter{TenantSlug: tenant, From: &past, To: &future}, &buf); err != nil {
t.Fatalf("stream (innerhalb range): %v", err)
}
if got := countLines(buf.String()) - 1; got != 1 {
t.Fatalf("erwartet 1 eintrag innerhalb des zeitraums, habe %d", got)
}
farPast := time.Now().Add(-96 * time.Hour)
buf.Reset()
if err := log.StreamCSV(ctx, Filter{TenantSlug: tenant, From: &farPast, To: &past}, &buf); err != nil {
t.Fatalf("stream (ausserhalb range): %v", err)
}
if got := countLines(buf.String()) - 1; got != 0 {
t.Fatalf("erwartet 0 eintraege ausserhalb des zeitraums, habe %d", got)
}
}
func countLines(s string) int {
s = strings.TrimRight(s, "\n")
if s == "" {
return 0
}
return len(strings.Split(s, "\n"))
}
// Akzeptanzkriterium 3 + Pruefung 1: Export mit hoher Eintragszahl ohne
// uebermaessigen Speicherverbrauch — Stichprobe per runtime.MemStats.
func TestExport_StreamsLargeResultWithoutExcessiveMemory(t *testing.T) {
log, cleanup := setupExportTest(t)
defer cleanup()
ctx := context.Background()
tenant := fmt.Sprintf("test_large_%d", time.Now().UnixNano())
const n = 20000
for i := 0; i < n; i++ {
if err := log.Record(ctx, Event{TenantSlug: tenant, Actor: "bulk", Action: "test.bulk", Target: fmt.Sprintf("obj-%d", i)}); err != nil {
t.Fatalf("record %d: %v", i, err)
}
}
runtime.GC()
var before runtime.MemStats
runtime.ReadMemStats(&before)
lineCount := 0
cw := &countingWriter{onWrite: func(p []byte) { lineCount += strings.Count(string(p), "\n") }}
if err := log.StreamCSV(ctx, Filter{TenantSlug: tenant}, cw); err != nil {
t.Fatalf("stream: %v", err)
}
runtime.GC()
var after runtime.MemStats
runtime.ReadMemStats(&after)
if lineCount != n+1 { // +1 Header
t.Fatalf("erwartet %d zeilen (inkl. header), habe %d", n+1, lineCount)
}
// Grobe Stichprobe: ein NICHT streamender Export haette hier locker
// mehrere MB an einmal gehaltenen Zeilen/Strings erzeugt. Grosszuegige
// Schwelle, da Go-Heap-Messungen naturgemaess rauschen.
const maxAcceptableGrowth = 3 * 1024 * 1024 // 3 MB
growth := int64(after.HeapAlloc) - int64(before.HeapAlloc)
t.Logf("heap-wachstum waehrend export von %d zeilen: %d bytes (schwelle: %d)", n, growth, maxAcceptableGrowth)
if growth > maxAcceptableGrowth {
t.Fatalf("heap ist um %d bytes gewachsen, erwartet unter %d (hinweis auf vollstaendige pufferung statt streaming)", growth, maxAcceptableGrowth)
}
}
type countingWriter struct {
onWrite func(p []byte)
}
func (w *countingWriter) Write(p []byte) (int, error) {
w.onWrite(p)
return len(p), nil
}
// Akzeptanzkriterium 2 + Pruefung 3: Zugriff ohne passende Berechtigung wird abgewiesen.
func TestExportHandler_RejectsWithoutAuthorization(t *testing.T) {
log, cleanup := setupExportTest(t)
defer cleanup()
handler := NewExportHandler(log, func(ctx context.Context, caller string) bool {
return caller == "berechtigte-person@example.com"
})
req := httptest.NewRequest(http.MethodGet, "/audit/export?caller=unberechtigt@example.com", nil)
rec := httptest.NewRecorder()
handler.Export(rec, req)
if rec.Code != http.StatusForbidden {
t.Fatalf("unberechtigt: status = %d, want 403", rec.Code)
}
reqNoCaller := httptest.NewRequest(http.MethodGet, "/audit/export", nil)
recNoCaller := httptest.NewRecorder()
handler.Export(recNoCaller, reqNoCaller)
if recNoCaller.Code != http.StatusForbidden {
t.Fatalf("ohne caller: status = %d, want 403", recNoCaller.Code)
}
reqOK := httptest.NewRequest(http.MethodGet, "/audit/export?caller=berechtigte-person@example.com", nil)
recOK := httptest.NewRecorder()
handler.Export(recOK, reqOK)
if recOK.Code != http.StatusOK {
t.Fatalf("berechtigt: status = %d, want 200", recOK.Code)
}
}
+241
View File
@@ -0,0 +1,241 @@
package tenant
import (
"context"
"errors"
"fmt"
"log/slog"
"time"
"github.com/jackc/pgx/v5"
"github.com/jackc/pgx/v5/pgxpool"
)
var (
ErrTenantNotFound = errors.New("tenant: nicht gefunden")
ErrInvalidTransition = errors.New("tenant: ungueltiger zustandsuebergang")
// ErrTenantNotActive wird von Lifecycle.CheckActive verwendet — bewusst
// EIN Fehler fuer suspendiert/zur-Loeschung-vorgemerkt/geloescht, da der
// Aufrufer (z.B. Login) nur wissen muss "kein Zugriff", nicht welcher der
// Nicht-aktiv-Zustaende genau vorliegt.
ErrTenantNotActive = errors.New("tenant: nicht aktiv")
)
func scanTenantWithLifecycle(row pgx.Row) (Tenant, error) {
var t Tenant
if err := row.Scan(&t.ID, &t.Slug, &t.Name, &t.DBName, &t.DBDSN, &t.Status,
&t.CreatedAt, &t.PreviousStatus, &t.DeletionScheduledAt); err != nil {
return Tenant{}, err
}
return t, nil
}
// transition fuehrt einen bewachten Zustandsuebergang aus: das UPDATE greift
// nur, wenn der aktuelle Status einer von allowedFrom ist (atomarer
// Check-and-Set, kein Race zwischen Lesen und Schreiben). Greift es nicht,
// wird zwischen "Tenant existiert nicht" und "Uebergang nicht erlaubt"
// unterschieden, damit AC1 ("ungueltige Uebergaenge werden abgewiesen") einen
// sprechenden Fehler liefert statt eines stillen No-Ops.
func (r *Registry) transition(ctx context.Context, slug string, allowedFrom []Status, to Status, previousStatus *string, deletionAt *time.Time) (Tenant, error) {
from := make([]string, len(allowedFrom))
for i, s := range allowedFrom {
from[i] = string(s)
}
row := r.pool.QueryRow(ctx, `
UPDATE tenants
SET status = $2, previous_status = $3, deletion_scheduled_at = $4
WHERE slug = $1 AND status = ANY($5)
RETURNING id, slug, name, db_name, db_dsn, status, created_at, previous_status, deletion_scheduled_at
`, slug, string(to), previousStatus, deletionAt, from)
t, err := scanTenantWithLifecycle(row)
if err == nil {
return t, nil
}
if !errors.Is(err, pgx.ErrNoRows) {
return Tenant{}, fmt.Errorf("zustandsuebergang: %w", err)
}
existing, getErr := r.GetBySlug(ctx, slug)
if getErr != nil {
return Tenant{}, ErrTenantNotFound
}
return Tenant{}, fmt.Errorf("%w: von %q nach %q (aktuell: %q)", ErrInvalidTransition, allowedFrom, to, existing.Status)
}
// Suspend haelt die Daten des Mandanten unveraendert, sperrt aber den Zugriff
// (Akzeptanzkriterium 1) — es findet keine Loeschung/Migration statt.
func (r *Registry) Suspend(ctx context.Context, slug string) (Tenant, error) {
return r.transition(ctx, slug, []Status{StatusActive}, StatusSuspended, nil, nil)
}
// Reactivate stellt den Zustand vor der Suspendierung vollstaendig wieder her
// (Akzeptanzkriterium 2) — da Suspend keine weiteren Daten veraendert, genuegt
// die Rueckkehr nach StatusActive.
func (r *Registry) Reactivate(ctx context.Context, slug string) (Tenant, error) {
return r.transition(ctx, slug, []Status{StatusSuspended}, StatusActive, nil, nil)
}
// ScheduleDeletion merkt den Mandanten zur Loeschung vor und startet die
// Karenzzeit (Akzeptanzkriterium 3). previous_status wird festgehalten, damit
// CancelDeletion exakt dorthin zurueckkehren kann (aktiv ODER suspendiert).
func (r *Registry) ScheduleDeletion(ctx context.Context, slug string, grace time.Duration) (Tenant, error) {
existing, err := r.GetBySlug(ctx, slug)
if err != nil {
return Tenant{}, ErrTenantNotFound
}
prev := string(existing.Status)
deletionAt := time.Now().Add(grace)
return r.transition(ctx, slug, []Status{StatusActive, StatusSuspended}, StatusPendingDeletion, &prev, &deletionAt)
}
// CancelDeletion widerruft eine Loeschvormerkung innerhalb der Karenzzeit und
// stellt exakt den zuvor gesicherten Zustand wieder her.
func (r *Registry) CancelDeletion(ctx context.Context, slug string) (Tenant, error) {
existing, err := r.GetBySlug(ctx, slug)
if err != nil {
return Tenant{}, ErrTenantNotFound
}
if existing.Status != StatusPendingDeletion || existing.PreviousStatus == nil {
return Tenant{}, fmt.Errorf("%w: von %q nach aktiv/suspendiert (aktuell: %q)", ErrInvalidTransition, StatusPendingDeletion, existing.Status)
}
restoreTo := Status(*existing.PreviousStatus)
return r.transition(ctx, slug, []Status{StatusPendingDeletion}, restoreTo, nil, nil)
}
// Lifecycle fuehrt die tatsaechliche, physische Loeschung nach Ablauf der
// Karenzzeit aus (Datenbank-Drop) und stellt die Zugriffsschutz-Pruefung
// bereit. Getrennt von Registry, weil hierfuer zusaetzlich der adminPool
// (fuer DROP DATABASE) noetig ist, siehe internal/tenant.Provisioner.
type Lifecycle struct {
registry *Registry
adminPool *pgxpool.Pool
// retention ist die Pruef-Schnittstelle gegen Archive RET-03/CMP-06 (TEN-08).
// Default NoRetentionCheck{}, bis Archive angebunden ist — siehe retention.go.
retention RetentionChecker
}
func NewLifecycle(registry *Registry, adminPool *pgxpool.Pool) *Lifecycle {
return &Lifecycle{registry: registry, adminPool: adminPool, retention: NoRetentionCheck{}}
}
// WithRetentionChecker ersetzt den Retention-Checker (z.B. im Test durch einen
// Fake, oder in Produktion durch den echten Archive-RET-03-Client). Gibt
// dasselbe *Lifecycle zurueck, um Verkettung beim Aufbau zu erlauben.
func (l *Lifecycle) WithRetentionChecker(checker RetentionChecker) *Lifecycle {
l.retention = checker
return l
}
// CheckActive verweigert Zugriff fuer jeden Nicht-aktiv-Zustand und loggt den
// Vorgang strukturiert (Akzeptanzkriterium 1 / Pruefung 2).
func (l *Lifecycle) CheckActive(ctx context.Context, slug string) error {
t, err := l.registry.GetBySlug(ctx, slug)
if err != nil {
return ErrTenantNotFound
}
if t.Status != StatusActive {
slog.Warn("zugriff auf nicht-aktiven mandanten verweigert",
"tenant_slug", slug, "tenant_status", t.Status)
return ErrTenantNotActive
}
return nil
}
// ProcessDueDeletions loescht alle Mandanten-Datenbanken, deren Karenzzeit
// abgelaufen ist (Akzeptanzkriterium 3 / Pruefung 3). FOR UPDATE SKIP LOCKED
// folgt der projektweiten Postgres-Jobqueue-Konvention (siehe
// SKALIERUNGSKONZEPT.md) und macht die Funktion sicher fuer mehrere parallel
// laufende Core-Instanzen.
func (l *Lifecycle) ProcessDueDeletions(ctx context.Context) (int, error) {
tx, err := l.registry.pool.Begin(ctx)
if err != nil {
return 0, fmt.Errorf("sweep-transaktion starten: %w", err)
}
defer func() { _ = tx.Rollback(ctx) }()
rows, err := tx.Query(ctx, `
SELECT id, slug, db_name FROM tenants
WHERE status = $1 AND deletion_scheduled_at <= now()
FOR UPDATE SKIP LOCKED
`, string(StatusPendingDeletion))
if err != nil {
return 0, fmt.Errorf("faellige loeschungen abfragen: %w", err)
}
type due struct{ id, slug, dbName string }
var candidates []due
for rows.Next() {
var d due
if err := rows.Scan(&d.id, &d.slug, &d.dbName); err != nil {
rows.Close()
return 0, fmt.Errorf("faellige loeschung lesen: %w", err)
}
candidates = append(candidates, d)
}
rows.Close()
if err := rows.Err(); err != nil {
return 0, err
}
processed := 0
for _, c := range candidates {
// TEN-08: vor der physischen Loeschung gegen Archive RET-03/CMP-06 pruefen.
// Solange eine Sperre besteht, bleibt der Tenant in pending_deletion
// ("zur Loeschung vorgemerkt, aber gesperrt") — der Grund wird
// festgehalten (Akzeptanzkriterium 2), die naechste Sweeper-Runde
// prueft automatisch erneut (Akzeptanzkriterium 3), ohne dass ein
// manueller Re-Trigger noetig waere.
result, err := l.retention.CheckTenantRetention(ctx, c.id)
if err != nil {
return processed, fmt.Errorf("retention-pruefung fuer tenant %q: %w", c.id, err)
}
if result.Blocked {
slog.Warn("tenant-loeschung wegen aufbewahrungspflicht/legal-hold zurueckgehalten",
"tenant_slug", c.slug, "reason", result.Reason)
if _, err := tx.Exec(ctx, `
UPDATE tenants SET retention_block_reason = $2, retention_checked_at = now()
WHERE id = $1
`, c.id, result.Reason); err != nil {
return processed, fmt.Errorf("retention-sperrgrund fuer tenant %q speichern: %w", c.id, err)
}
continue
}
if _, err := l.adminPool.Exec(ctx, fmt.Sprintf(`DROP DATABASE IF EXISTS %q`, c.dbName)); err != nil {
return processed, fmt.Errorf("tenant-datenbank %q loeschen: %w", c.dbName, err)
}
if _, err := tx.Exec(ctx, `
UPDATE tenants
SET status = $2, previous_status = NULL, deletion_scheduled_at = NULL,
retention_block_reason = NULL, retention_checked_at = now()
WHERE id = $1
`, c.id, string(StatusDeleted)); err != nil {
return processed, fmt.Errorf("tenant %q als geloescht markieren: %w", c.id, err)
}
processed++
}
if err := tx.Commit(ctx); err != nil {
return 0, fmt.Errorf("sweep-transaktion committen: %w", err)
}
return processed, nil
}
// RunSweeper triggert ProcessDueDeletions periodisch, bis ctx beendet wird —
// die "In-Prozess-Worker-Goroutine" aus der projektweiten Jobqueue-Konvention.
func (l *Lifecycle) RunSweeper(ctx context.Context, interval time.Duration) {
ticker := time.NewTicker(interval)
defer ticker.Stop()
for {
select {
case <-ctx.Done():
return
case <-ticker.C:
if _, err := l.ProcessDueDeletions(ctx); err != nil {
slog.Error("tenant-loeschung-sweep fehlgeschlagen", "error", err)
}
}
}
}
+255
View File
@@ -0,0 +1,255 @@
package tenant
import (
"context"
"errors"
"fmt"
"os"
"strings"
"testing"
"time"
"github.com/jackc/pgx/v5/pgxpool"
)
func newLifecycleTestSetup(t *testing.T) (*Registry, *Lifecycle, *pgxpool.Pool, func()) {
t.Helper()
adminDSN := os.Getenv("TEST_ADMIN_DSN")
if adminDSN == "" {
t.Skip("TEST_ADMIN_DSN nicht gesetzt, Integrationstest uebersprungen")
}
ctx := context.Background()
adminPool, err := pgxpool.New(ctx, adminDSN)
if err != nil {
t.Fatalf("admin pool: %v", err)
}
registryPool, err := pgxpool.New(ctx, adminDSN)
if err != nil {
t.Fatalf("registry pool: %v", err)
}
if _, err := registryPool.Exec(ctx, `
CREATE TABLE IF NOT EXISTS tenants (
id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
slug TEXT NOT NULL UNIQUE,
name TEXT NOT NULL,
db_name TEXT NOT NULL UNIQUE,
db_dsn TEXT NOT NULL,
status TEXT NOT NULL DEFAULT 'active',
created_at TIMESTAMPTZ NOT NULL DEFAULT now(),
previous_status TEXT,
deletion_scheduled_at TIMESTAMPTZ,
retention_block_reason TEXT,
retention_checked_at TIMESTAMPTZ
)`); err != nil {
t.Fatalf("registry-schema: %v", err)
}
registry := NewRegistry(registryPool)
dsnTemplate := strings.Replace(adminDSN, "/postgres?", "/%s?", 1)
provisioner := NewProvisioner(adminPool, registry, dsnTemplate)
lifecycle := NewLifecycle(registry, adminPool)
cleanup := func() {
registryPool.Close()
adminPool.Close()
}
_ = provisioner
return registry, lifecycle, adminPool, cleanup
}
func provisionTestTenant(t *testing.T, registry *Registry, adminPool *pgxpool.Pool, slug string) {
t.Helper()
dsnTemplate := strings.Replace(os.Getenv("TEST_ADMIN_DSN"), "/postgres?", "/%s?", 1)
provisioner := NewProvisioner(adminPool, registry, dsnTemplate)
if _, err := provisioner.Provision(context.Background(), slug, slug); err != nil {
t.Fatalf("provision %s: %v", slug, err)
}
t.Cleanup(func() {
ctx := context.Background()
_, _ = adminPool.Exec(ctx, fmt.Sprintf(`DROP DATABASE IF EXISTS %q`, dbNameForSlug(slug)))
_, _ = registry.pool.Exec(ctx, `DELETE FROM tenants WHERE slug = $1`, slug)
})
}
// Akzeptanzkriterium 1 (Suspend) + 2 (Reactivate) + Pruefung 1 (Uebergaenge).
func TestLifecycle_SuspendAndReactivate(t *testing.T) {
registry, _, adminPool, cleanup := newLifecycleTestSetup(t)
defer cleanup()
provisionTestTenant(t, registry, adminPool, "lc_suspend")
ctx := context.Background()
suspended, err := registry.Suspend(ctx, "lc_suspend")
if err != nil {
t.Fatalf("suspend: %v", err)
}
if suspended.Status != StatusSuspended {
t.Fatalf("status = %q, want suspended", suspended.Status)
}
reactivated, err := registry.Reactivate(ctx, "lc_suspend")
if err != nil {
t.Fatalf("reactivate: %v", err)
}
if reactivated.Status != StatusActive {
t.Fatalf("status = %q, want active", reactivated.Status)
}
}
// Pruefung 1: ungueltige Uebergaenge werden abgewiesen.
func TestLifecycle_RejectsInvalidTransitions(t *testing.T) {
registry, _, adminPool, cleanup := newLifecycleTestSetup(t)
defer cleanup()
provisionTestTenant(t, registry, adminPool, "lc_invalid")
ctx := context.Background()
// Reactivate auf einem bereits aktiven Tenant ist kein gueltiger Uebergang.
if _, err := registry.Reactivate(ctx, "lc_invalid"); !errors.Is(err, ErrInvalidTransition) {
t.Fatalf("erwartet ErrInvalidTransition, habe %v", err)
}
if _, err := registry.Suspend(ctx, "lc_invalid"); err != nil {
t.Fatalf("suspend: %v", err)
}
// Suspend auf einem bereits suspendierten Tenant ist ebenfalls ungueltig.
if _, err := registry.Suspend(ctx, "lc_invalid"); !errors.Is(err, ErrInvalidTransition) {
t.Fatalf("erwartet ErrInvalidTransition, habe %v", err)
}
// CancelDeletion ohne vorherige Loeschvormerkung ist ungueltig.
if _, err := registry.CancelDeletion(ctx, "lc_invalid"); !errors.Is(err, ErrInvalidTransition) {
t.Fatalf("erwartet ErrInvalidTransition, habe %v", err)
}
if _, err := registry.Suspend(ctx, "unbekannter-slug-xyz"); !errors.Is(err, ErrTenantNotFound) {
t.Fatalf("erwartet ErrTenantNotFound, habe %v", err)
}
}
// Akzeptanzkriterium 3: Loeschung zweistufig mit Karenzzeit, innerhalb der
// Frist widerrufbar — sowohl aus 'active' als auch aus 'suspended' heraus,
// mit exakter Wiederherstellung des jeweiligen Vorzustands.
func TestLifecycle_ScheduleAndCancelDeletion_RestoresExactPreviousState(t *testing.T) {
registry, _, adminPool, cleanup := newLifecycleTestSetup(t)
defer cleanup()
provisionTestTenant(t, registry, adminPool, "lc_cancel_active")
provisionTestTenant(t, registry, adminPool, "lc_cancel_suspended")
ctx := context.Background()
// Fall 1: aus 'active' heraus vorgemerkt und widerrufen.
scheduled, err := registry.ScheduleDeletion(ctx, "lc_cancel_active", time.Hour)
if err != nil {
t.Fatalf("schedule deletion: %v", err)
}
if scheduled.Status != StatusPendingDeletion {
t.Fatalf("status = %q, want pending_deletion", scheduled.Status)
}
if scheduled.DeletionScheduledAt == nil {
t.Fatal("erwartet gesetzte deletion_scheduled_at")
}
restored, err := registry.CancelDeletion(ctx, "lc_cancel_active")
if err != nil {
t.Fatalf("cancel deletion: %v", err)
}
if restored.Status != StatusActive {
t.Fatalf("status = %q, want active (vorheriger zustand)", restored.Status)
}
// Fall 2: aus 'suspended' heraus vorgemerkt und widerrufen — muss zu
// 'suspended' zurueckkehren, NICHT zu 'active'.
if _, err := registry.Suspend(ctx, "lc_cancel_suspended"); err != nil {
t.Fatalf("suspend: %v", err)
}
if _, err := registry.ScheduleDeletion(ctx, "lc_cancel_suspended", time.Hour); err != nil {
t.Fatalf("schedule deletion: %v", err)
}
restoredSuspended, err := registry.CancelDeletion(ctx, "lc_cancel_suspended")
if err != nil {
t.Fatalf("cancel deletion: %v", err)
}
if restoredSuspended.Status != StatusSuspended {
t.Fatalf("status = %q, want suspended (vorheriger zustand)", restoredSuspended.Status)
}
}
// Akzeptanzkriterium 1 + Pruefung 2: suspendierter Tenant erzeugt bei jedem
// Zugriffsversuch einen klaren Fehler.
func TestLifecycle_CheckActive_RejectsNonActive(t *testing.T) {
registry, lifecycle, adminPool, cleanup := newLifecycleTestSetup(t)
defer cleanup()
provisionTestTenant(t, registry, adminPool, "lc_checkactive")
ctx := context.Background()
if err := lifecycle.CheckActive(ctx, "lc_checkactive"); err != nil {
t.Fatalf("aktiver tenant sollte durchgehen, habe %v", err)
}
if _, err := registry.Suspend(ctx, "lc_checkactive"); err != nil {
t.Fatalf("suspend: %v", err)
}
for i := 0; i < 3; i++ {
if err := lifecycle.CheckActive(ctx, "lc_checkactive"); !errors.Is(err, ErrTenantNotActive) {
t.Fatalf("versuch %d: erwartet ErrTenantNotActive, habe %v", i, err)
}
}
if err := lifecycle.CheckActive(ctx, "nie-registriert"); !errors.Is(err, ErrTenantNotFound) {
t.Fatalf("erwartet ErrTenantNotFound, habe %v", err)
}
}
// Akzeptanzkriterium 3 + Pruefung 3: Loeschvorgang nach Ablauf der Karenzzeit
// automatisch ausgeloest (hier durch direkten Aufruf von ProcessDueDeletions,
// das RunSweeper periodisch aufruft).
func TestLifecycle_ProcessDueDeletions(t *testing.T) {
registry, lifecycle, adminPool, cleanup := newLifecycleTestSetup(t)
defer cleanup()
provisionTestTenant(t, registry, adminPool, "lc_due")
provisionTestTenant(t, registry, adminPool, "lc_not_due")
ctx := context.Background()
// lc_due: Karenzzeit liegt bereits in der Vergangenheit -> faellig.
if _, err := registry.ScheduleDeletion(ctx, "lc_due", -time.Minute); err != nil {
t.Fatalf("schedule deletion (due): %v", err)
}
// lc_not_due: Karenzzeit liegt weit in der Zukunft -> nicht faellig.
if _, err := registry.ScheduleDeletion(ctx, "lc_not_due", time.Hour); err != nil {
t.Fatalf("schedule deletion (not due): %v", err)
}
processed, err := lifecycle.ProcessDueDeletions(ctx)
if err != nil {
t.Fatalf("process due deletions: %v", err)
}
if processed != 1 {
t.Fatalf("erwartet genau 1 verarbeitete loeschung, habe %d", processed)
}
due, err := registry.GetBySlug(ctx, "lc_due")
if err != nil {
t.Fatalf("get lc_due: %v", err)
}
if due.Status != StatusDeleted {
t.Fatalf("lc_due status = %q, want deleted", due.Status)
}
notDue, err := registry.GetBySlug(ctx, "lc_not_due")
if err != nil {
t.Fatalf("get lc_not_due: %v", err)
}
if notDue.Status != StatusPendingDeletion {
t.Fatalf("lc_not_due status = %q, want pending_deletion (noch nicht faellig)", notDue.Status)
}
// Datenbank von lc_due wurde tatsaechlich physisch entfernt.
var exists bool
if err := adminPool.QueryRow(ctx, `SELECT EXISTS(SELECT 1 FROM pg_database WHERE datname = $1)`,
dbNameForSlug("lc_due")).Scan(&exists); err != nil {
t.Fatalf("pg_database pruefen: %v", err)
}
if exists {
t.Fatal("erwartet, dass die tenant-datenbank von lc_due geloescht wurde")
}
}
+9 -2
View File
@@ -35,13 +35,20 @@ func (r *Registry) insertTx(ctx context.Context, tx pgx.Tx, t Tenant) (Tenant, e
}
func (r *Registry) GetBySlug(ctx context.Context, slug string) (Tenant, error) {
// previous_status/deletion_scheduled_at werden mitgelesen, damit TEN-04
// (internal/tenant/lifecycle.go) den vollstaendigen Lebenszyklus-Zustand
// ueber GetBySlug ansehen kann, statt eine eigene Abfrage zu duplizieren.
// retention_block_reason/retention_checked_at (TEN-08) aus demselben Grund
// fuer die Admin-Einsehbarkeit des Sperrgrunds (Akzeptanzkriterium 2).
var t Tenant
row := r.pool.QueryRow(ctx, `
SELECT id, slug, name, db_name, db_dsn, status, created_at
SELECT id, slug, name, db_name, db_dsn, status, created_at, previous_status, deletion_scheduled_at,
retention_block_reason, retention_checked_at
FROM tenants WHERE slug = $1
`, slug)
if err := row.Scan(&t.ID, &t.Slug, &t.Name, &t.DBName, &t.DBDSN, &t.Status, &t.CreatedAt); err != nil {
if err := row.Scan(&t.ID, &t.Slug, &t.Name, &t.DBName, &t.DBDSN, &t.Status, &t.CreatedAt,
&t.PreviousStatus, &t.DeletionScheduledAt, &t.RetentionBlockReason, &t.RetentionCheckedAt); err != nil {
return Tenant{}, fmt.Errorf("tenant laden: %w", err)
}
return t, nil
+34
View File
@@ -0,0 +1,34 @@
package tenant
import "context"
// RetentionResult ist das Ergebnis einer Pruefung gegen Archive RET-03/CMP-06
// vor einer endgueltigen Tenant-Loeschung (TEN-08).
type RetentionResult struct {
// Blocked ist true, solange GoBD-relevante Daten des Tenants unter
// Aufbewahrungspflicht oder Legal Hold stehen (Akzeptanzkriterium 1).
Blocked bool
// Reason beschreibt Aufbewahrungsklasse/Frist oder Legal-Hold-Grund,
// fuer Admins einsehbar (Akzeptanzkriterium 2). Nur aussagekraeftig, wenn Blocked true ist.
Reason string
}
// RetentionChecker ist die Schnittstelle zu Archive RET-03 (Loeschworkflow &
// Aufbewahrungssperre) / CMP-06 (Vier-Augen-Freigabe fuer Loeschungen).
// Core kennt bewusst keine Retention-Logik selbst — diese Kachel ruft nur auf,
// siehe TEN-08 "Nicht Bestandteil dieser Kachel". Solange Archive RET-03 noch
// nicht implementiert ist, wird ein no-op-Checker verwendet (siehe
// NoRetentionCheck), der niemals blockiert — Core faellt damit auf das
// TEN-04-Verhalten vor diesem Ticket zurueck, statt fehlzuschlagen.
type RetentionChecker interface {
CheckTenantRetention(ctx context.Context, tenantID string) (RetentionResult, error)
}
// NoRetentionCheck ist der Platzhalter-Checker, solange Archive RET-03 noch
// nicht angebunden ist — blockiert nie. Wird in Produktion durch den echten
// HTTP-Client gegen Archive ersetzt, sobald RET-03 existiert.
type NoRetentionCheck struct{}
func (NoRetentionCheck) CheckTenantRetention(context.Context, string) (RetentionResult, error) {
return RetentionResult{Blocked: false}, nil
}
+164
View File
@@ -0,0 +1,164 @@
package tenant
import (
"context"
"testing"
"time"
)
// fakeRetentionChecker simuliert Archive RET-03/CMP-06 in Tests — echte
// Anbindung existiert noch nicht (siehe retention.go), diese Kachel ruft nur auf.
type fakeRetentionChecker struct {
blocked map[string]string // tenantID -> Grund
}
func (f fakeRetentionChecker) CheckTenantRetention(_ context.Context, tenantID string) (RetentionResult, error) {
if reason, ok := f.blocked[tenantID]; ok {
return RetentionResult{Blocked: true, Reason: reason}, nil
}
return RetentionResult{Blocked: false}, nil
}
// Akzeptanzkriterium 1 + Pruefung 1: Loeschung eines Tenants mit aktiver
// GoBD-Aufbewahrungspflicht wird abgewiesen, Grund wird protokolliert
// (Akzeptanzkriterium 2).
func TestLifecycle_ProcessDueDeletions_BlockedByRetention(t *testing.T) {
registry, lifecycle, adminPool, cleanup := newLifecycleTestSetup(t)
defer cleanup()
provisionTestTenant(t, registry, adminPool, "lc_retention_blocked")
ctx := context.Background()
tenantBeforeSchedule, err := registry.GetBySlug(ctx, "lc_retention_blocked")
if err != nil {
t.Fatalf("get tenant: %v", err)
}
if _, err := registry.ScheduleDeletion(ctx, "lc_retention_blocked", -time.Minute); err != nil {
t.Fatalf("schedule deletion: %v", err)
}
lifecycle.WithRetentionChecker(fakeRetentionChecker{
blocked: map[string]string{
tenantBeforeSchedule.ID: "GoBD-Aufbewahrungsfrist bis 2034-01-01 (Buchungsbeleg-Klasse)",
},
})
processed, err := lifecycle.ProcessDueDeletions(ctx)
if err != nil {
t.Fatalf("process due deletions: %v", err)
}
if processed != 0 {
t.Fatalf("erwartet 0 tatsaechlich verarbeitete loeschungen, habe %d", processed)
}
after, err := registry.GetBySlug(ctx, "lc_retention_blocked")
if err != nil {
t.Fatalf("get tenant nach sweep: %v", err)
}
if after.Status != StatusPendingDeletion {
t.Fatalf("status = %q, want pending_deletion (gesperrt, nicht geloescht)", after.Status)
}
if after.RetentionBlockReason == nil || *after.RetentionBlockReason == "" {
t.Fatal("erwartet gesetzten retention_block_reason (Akzeptanzkriterium 2)")
}
if after.RetentionCheckedAt == nil {
t.Fatal("erwartet gesetzten retention_checked_at")
}
var exists bool
if err := adminPool.QueryRow(ctx, `SELECT EXISTS(SELECT 1 FROM pg_database WHERE datname = $1)`,
dbNameForSlug("lc_retention_blocked")).Scan(&exists); err != nil {
t.Fatalf("pg_database pruefen: %v", err)
}
if !exists {
t.Fatal("tenant-datenbank haette NICHT geloescht werden duerfen (retention-sperre)")
}
}
// Akzeptanzkriterium 1 + Pruefung 2: Loeschung eines Tenants mit Legal Hold
// wird ebenfalls abgewiesen — derselbe Mechanismus wie GoBD-Frist, nur anderer Grund.
func TestLifecycle_ProcessDueDeletions_BlockedByLegalHold(t *testing.T) {
registry, lifecycle, adminPool, cleanup := newLifecycleTestSetup(t)
defer cleanup()
provisionTestTenant(t, registry, adminPool, "lc_legal_hold")
ctx := context.Background()
tenant, err := registry.GetBySlug(ctx, "lc_legal_hold")
if err != nil {
t.Fatalf("get tenant: %v", err)
}
if _, err := registry.ScheduleDeletion(ctx, "lc_legal_hold", -time.Minute); err != nil {
t.Fatalf("schedule deletion: %v", err)
}
lifecycle.WithRetentionChecker(fakeRetentionChecker{
blocked: map[string]string{
tenant.ID: "Legal Hold: laufendes Gerichtsverfahren, Aktenzeichen XY-2026-042",
},
})
if _, err := lifecycle.ProcessDueDeletions(ctx); err != nil {
t.Fatalf("process due deletions: %v", err)
}
after, err := registry.GetBySlug(ctx, "lc_legal_hold")
if err != nil {
t.Fatalf("get tenant nach sweep: %v", err)
}
if after.Status != StatusPendingDeletion {
t.Fatalf("status = %q, want pending_deletion", after.Status)
}
if after.RetentionBlockReason == nil || *after.RetentionBlockReason == "" {
t.Fatal("erwartet gesetzten retention_block_reason")
}
}
// Akzeptanzkriterium 3 + Pruefung 3: nach Aufhebung aller Sperren wird die
// Loeschung bei der naechsten Sweep-Runde automatisch ausgefuehrt — kein
// manueller Re-Trigger noetig, derselbe Sweeper-Aufruf greift erneut.
func TestLifecycle_ProcessDueDeletions_ExecutesAfterRetentionCleared(t *testing.T) {
registry, lifecycle, adminPool, cleanup := newLifecycleTestSetup(t)
defer cleanup()
provisionTestTenant(t, registry, adminPool, "lc_retention_cleared")
ctx := context.Background()
tenant, err := registry.GetBySlug(ctx, "lc_retention_cleared")
if err != nil {
t.Fatalf("get tenant: %v", err)
}
if _, err := registry.ScheduleDeletion(ctx, "lc_retention_cleared", -time.Minute); err != nil {
t.Fatalf("schedule deletion: %v", err)
}
blockingChecker := fakeRetentionChecker{blocked: map[string]string{tenant.ID: "Aufbewahrungsfrist laeuft noch"}}
lifecycle.WithRetentionChecker(blockingChecker)
if _, err := lifecycle.ProcessDueDeletions(ctx); err != nil {
t.Fatalf("erster sweep (blockiert): %v", err)
}
blockedState, err := registry.GetBySlug(ctx, "lc_retention_cleared")
if err != nil {
t.Fatalf("get tenant nach erstem sweep: %v", err)
}
if blockedState.Status != StatusPendingDeletion {
t.Fatalf("status nach erstem sweep = %q, want pending_deletion", blockedState.Status)
}
// Sperre aufgehoben: naechster Checker blockiert nicht mehr (fakeRetentionChecker.blocked leer).
lifecycle.WithRetentionChecker(fakeRetentionChecker{})
processed, err := lifecycle.ProcessDueDeletions(ctx)
if err != nil {
t.Fatalf("zweiter sweep (unblockiert): %v", err)
}
if processed != 1 {
t.Fatalf("erwartet genau 1 verarbeitete loeschung im zweiten sweep, habe %d", processed)
}
final, err := registry.GetBySlug(ctx, "lc_retention_cleared")
if err != nil {
t.Fatalf("get tenant nach zweitem sweep: %v", err)
}
if final.Status != StatusDeleted {
t.Fatalf("status = %q, want deleted", final.Status)
}
}
+14
View File
@@ -12,6 +12,10 @@ type Status string
const (
StatusActive Status = "active"
// Lebenszyklus-Zustaende aus TEN-04 (siehe internal/tenant/lifecycle.go).
StatusSuspended Status = "suspended"
StatusPendingDeletion Status = "pending_deletion"
StatusDeleted Status = "deleted"
)
type Tenant struct {
@@ -22,6 +26,16 @@ type Tenant struct {
DBDSN string
Status Status
CreatedAt time.Time
// PreviousStatus und DeletionScheduledAt sind nur waehrend
// StatusPendingDeletion gesetzt (TEN-04) — sie halten fest, in welchen
// Zustand CancelDeletion zurueckkehrt und wann die Karenzzeit ablaeuft.
PreviousStatus *string
DeletionScheduledAt *time.Time
// RetentionBlockReason ist nur gesetzt, wenn eine faellige Loeschung wegen
// GoBD-Aufbewahrungspflicht oder Legal Hold zurueckgehalten wurde (TEN-08,
// siehe internal/tenant/retention.go) — fuer Admins einsehbar (Akzeptanzkriterium 2).
RetentionBlockReason *string
RetentionCheckedAt *time.Time
}
// slugPattern erzwingt sichere, als SQL-Identifier verwendbare Slugs, damit
@@ -0,0 +1,2 @@
ALTER TABLE tenants DROP COLUMN previous_status;
ALTER TABLE tenants DROP COLUMN deletion_scheduled_at;
+6
View File
@@ -0,0 +1,6 @@
-- Lebenszyklus-Zustaende fuer Mandanten (TEN-04, siehe core-kanban/tickets/TEN-04.md).
-- previous_status haelt den Zustand VOR einer Loeschvormerkung, damit
-- CancelDeletion "den vorherigen Zustand vollstaendig wiederherstellt"
-- (aktiv ODER suspendiert), statt hart auf 'active' zurueckzusetzen.
ALTER TABLE tenants ADD COLUMN previous_status TEXT;
ALTER TABLE tenants ADD COLUMN deletion_scheduled_at TIMESTAMPTZ;
-1
View File
@@ -1 +0,0 @@
DROP TABLE IF EXISTS audit_events;
-17
View File
@@ -1,17 +0,0 @@
-- Zentrales Audit-Log-Modell (AUD-01, siehe core-kanban/tickets/AUD-01.md).
-- Getrennt vom allgemeinen Anwendungs-Log (Akzeptanzkriterium 2): eigene
-- Tabelle, eigenes Paket (internal/audit), kein Log-Framework.
-- tenant_slug ist NOT NULL + darf nicht leer sein (Akzeptanzkriterium 2 /
-- Pruefung 2) — mandantenuebergreifende Ereignisse nutzen den reservierten
-- Wert 'system', niemals NULL oder leeren String.
CREATE TABLE audit_events (
id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
occurred_at TIMESTAMPTZ NOT NULL DEFAULT now(),
tenant_slug TEXT NOT NULL CHECK (tenant_slug <> ''),
actor TEXT NOT NULL CHECK (actor <> ''),
action TEXT NOT NULL CHECK (action <> ''),
target TEXT NOT NULL,
metadata JSONB NOT NULL DEFAULT '{}'::jsonb
);
CREATE INDEX audit_events_tenant_slug_idx ON audit_events (tenant_slug, occurred_at);
@@ -0,0 +1,2 @@
ALTER TABLE tenants DROP COLUMN retention_block_reason;
ALTER TABLE tenants DROP COLUMN retention_checked_at;
@@ -0,0 +1,6 @@
-- TEN-08: Haelt fest, warum eine faellige Tenant-Loeschung zurueckgehalten wurde
-- (GoBD-Aufbewahrungspflicht oder Legal Hold aus Archive RET-03), damit Admins
-- den Grund einsehen koennen (Akzeptanzkriterium 2), ohne dass die Registry
-- selbst modulspezifische Retention-Logik kennen muss — nur den Grund-Text.
ALTER TABLE tenants ADD COLUMN retention_block_reason TEXT;
ALTER TABLE tenants ADD COLUMN retention_checked_at TIMESTAMPTZ;
+1 -2
View File
@@ -13,8 +13,7 @@ ROLE="nexarch_test"
export PGPASSWORD="$PASS"
psql -h localhost -U "$ROLE" -d postgres -v ON_ERROR_STOP=1 -c "DROP TABLE IF EXISTS tenants CASCADE;"
psql -h localhost -U "$ROLE" -d postgres -v ON_ERROR_STOP=1 -c "DROP TABLE IF EXISTS audit_events CASCADE;"
psql -h localhost -U "$ROLE" -d postgres -v ON_ERROR_STOP=1 -c "DROP TABLE IF EXISTS tenants;"
dbs=$(psql -h localhost -U "$ROLE" -d postgres -tAc "SELECT datname FROM pg_database WHERE datname LIKE 'tenant\_%' ESCAPE '\'")
for db in $dbs; do
-24
View File
@@ -1,24 +0,0 @@
#!/usr/bin/env bash
# Ein-Kommando-Pruefung fuer den aktuellen Code-Stand auf dem Testhost:
# Registry+Tenant-DBs zuruecksetzen, dann build/vet/test in einem Rutsch.
# -p 1 ist Pflicht, da mehrere Pakete dieselbe physische Registry-Tabelle auf
# dem Testhost teilen (siehe [[project-nexarch-test-infra]]).
#
# Aufruf: NEXARCH_TEST_DB_PASSWORD=... ./scripts/run-checks.sh
set -euo pipefail
PASS="${NEXARCH_TEST_DB_PASSWORD:?Setze NEXARCH_TEST_DB_PASSWORD vor dem Aufruf}"
cd "$(dirname "$0")/.."
NEXARCH_TEST_DB_PASSWORD="$PASS" bash scripts/reset-test-env.sh
export TEST_ADMIN_DSN="postgresql://nexarch_test:${PASS}@localhost:5432/postgres?sslmode=disable"
echo "== go build =="
go build ./...
echo "== go vet =="
go vet ./...
echo "== go test (-p 1) =="
go test ./... -p 1 -count=1