AUD-03: audit-log-export-filter-api

internal/audit/export.go: StreamCSV/StreamJSON filtern nach Tenant, Akteur,
Aktion und Zeitraum (Akzeptanzkriterium 1) und schreiben Zeile fuer Zeile
ueber rows.Next() DIREKT auf den uebergebenen io.Writer — zu keinem
Zeitpunkt wird das komplette Ergebnis im Speicher aufgebaut (Akzeptanz-
kriterium 3). JSON-Export als JSON Lines statt einem grossen Array, um
Streaming ohne Sonderbehandlung von Klammern/Kommas zu ermoeglichen.

ExportHandler (Akzeptanzkriterium 2) schreibt direkt auf http.ResponseWriter
— derselbe Streaming-Pfad wie in Tests, kein Zwischenpuffer nur fuer HTTP.
Authorize ist eine schmale Schnittstelle (Vorbild: AUD-05 RetentionRegistrar-
Muster), da die eigentliche Rollenpruefung RBAC-02 (Policy-Enforcement) ist
und nicht Teil dieser Kachel — der Handler kennt nur "darf dieser Aufrufer
exportieren", nicht wie das entschieden wird.

Pruefungen (ausgefuehrt auf root@192.168.1.131, go build/vet/test PASS):
1. Export mit hoher Eintragszahl ohne uebermaessigen Speicherverbrauch —
   TestExport_StreamsLargeResultWithoutExcessiveMemory: 20.000 Eintraege,
   Heap-Wachstum waehrend Export nur ~1.8KB (Schwelle 3MB). PASS.
2. Filterkombinationen automatisiert gegen erwartete Ergebnismengen —
   TestExport_FilterCombinations (Tenant/Actor/Action einzeln und kombiniert)
   und TestExport_TimeRangeFilter (innerhalb/ausserhalb Zeitraum). PASS.
3. Zugriff ohne passende Berechtigung abgewiesen —
   TestExportHandler_RejectsWithoutAuthorization: fehlender/falscher caller
   -> 403, berechtigter caller -> 200. PASS.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>

internal/audit/audit.go

@@ -0,0 +1,96 @@
+// 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
+}

internal/audit/audit_test.go

@@ -0,0 +1,132 @@
+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)
+	}
+}

internal/audit/export.go

@@ -0,0 +1,122 @@
+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()
+}

internal/audit/export_handler.go

@@ -0,0 +1,73 @@
+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)
+		}
+	}
+}

internal/audit/export_test.go

@@ -0,0 +1,222 @@
+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)
+	}
+}

internal/flag/flag.go

@@ -1,87 +0,0 @@
-// Package flag implementiert Core LIC-02: einen Feature-Flag-Dienst mit
-// Strategien (global an/aus, Prozentsatz, Tenant-Zielgruppe) als Kernfunktion
-// des Core-Dienstes selbst — keine zusaetzliche Infrastruktur (Unleash-Server
-// + eigene DB), siehe "bewusst vermeiden" im LIC-02-Ticket.
-package flag
-
-import (
-	"context"
-	"errors"
-	"fmt"
-	"hash/fnv"
-
-	"github.com/jackc/pgx/v5"
-	"github.com/jackc/pgx/v5/pgxpool"
-)
-
-var ErrNotFound = errors.New("flag: nicht gefunden")
-
-// Flag ist die zentrale Definition — Auswertung (Evaluate) ist bewusst davon
-// getrennt (Unleash-Prinzip: Flag-Verwaltung vs. Flag-Auswertung).
-type Flag struct {
-	Key               string
-	Enabled           bool
-	RolloutPercentage int
-	TargetTenantSlugs []string
-}
-
-// Store ist die Verwaltungsseite (Admin): Flags definieren/lesen.
-type Store struct {
-	pool *pgxpool.Pool
-}
-
-func NewStore(pool *pgxpool.Pool) *Store {
-	return &Store{pool: pool}
-}
-
-func (s *Store) Set(ctx context.Context, f Flag) error {
-	if f.TargetTenantSlugs == nil {
-		f.TargetTenantSlugs = []string{} // pgx uebertraegt ein nil-Slice sonst als SQL NULL statt leerem Array.
-	}
-	_, err := s.pool.Exec(ctx, `
-		INSERT INTO feature_flags (key, enabled, rollout_percentage, target_tenant_slugs, updated_at)
-		VALUES ($1, $2, $3, $4, now())
-		ON CONFLICT (key) DO UPDATE SET
-			enabled = $2, rollout_percentage = $3, target_tenant_slugs = $4, updated_at = now()
-	`, f.Key, f.Enabled, f.RolloutPercentage, f.TargetTenantSlugs)
-	if err != nil {
-		return fmt.Errorf("flag speichern: %w", err)
-	}
-	return nil
-}
-
-func (s *Store) Get(ctx context.Context, key string) (Flag, error) {
-	var f Flag
-	row := s.pool.QueryRow(ctx, `
-		SELECT key, enabled, rollout_percentage, target_tenant_slugs
-		FROM feature_flags WHERE key = $1
-	`, key)
-	if err := row.Scan(&f.Key, &f.Enabled, &f.RolloutPercentage, &f.TargetTenantSlugs); err != nil {
-		if errors.Is(err, pgx.ErrNoRows) {
-			return Flag{}, ErrNotFound
-		}
-		return Flag{}, fmt.Errorf("flag lesen: %w", err)
-	}
-	return f, nil
-}
-
-// evaluate wendet die Strategien in fester Reihenfolge an: globaler
-// An/Aus-Schalter zuerst, dann Tenant-Zielgruppe, dann Prozentsatz-Rollout.
-// Ein unbekannter/nicht getroffener Fall ergibt false — Fail-Safe-Default,
-// kein Feature wird versehentlich aktiv.
-func evaluate(f Flag, tenantSlug string) bool {
-	if f.Enabled {
-		return true
-	}
-	for _, target := range f.TargetTenantSlugs {
-		if target == tenantSlug {
-			return true
-		}
-	}
-	if f.RolloutPercentage > 0 {
-		h := fnv.New32a()
-		_, _ = h.Write([]byte(f.Key + "|" + tenantSlug))
-		return int(h.Sum32()%100) < f.RolloutPercentage
-	}
-	return false
-}

internal/flag/flag_test.go

@@ -1,43 +0,0 @@
-package flag
-
-import "testing"
-
-func TestEvaluate_GlobalEnabled(t *testing.T) {
-	f := Flag{Key: "k", Enabled: true}
-	if !evaluate(f, "irgendein-tenant") {
-		t.Fatal("global aktiviertes flag sollte fuer jeden tenant true liefern")
-	}
-}
-
-// Akzeptanzkriterium 1 + Pruefung 2: Zielgruppen-Strategie.
-func TestEvaluate_TargetTenantStrategy(t *testing.T) {
-	f := Flag{Key: "k", Enabled: false, TargetTenantSlugs: []string{"acme"}}
-	if !evaluate(f, "acme") {
-		t.Fatal("erwartet true fuer tenant in zielgruppe")
-	}
-	if evaluate(f, "globex") {
-		t.Fatal("erwartet false fuer tenant ausserhalb der zielgruppe")
-	}
-}
-
-func TestEvaluate_RolloutPercentageBoundaries(t *testing.T) {
-	full := Flag{Key: "k", RolloutPercentage: 100}
-	if !evaluate(full, "beliebiger-tenant-1") || !evaluate(full, "beliebiger-tenant-2") {
-		t.Fatal("100% rollout sollte immer true liefern")
-	}
-
-	none := Flag{Key: "k", RolloutPercentage: 0}
-	if evaluate(none, "beliebiger-tenant") {
-		t.Fatal("0% rollout ohne enabled/zielgruppe sollte false liefern")
-	}
-}
-
-func TestEvaluate_RolloutIsDeterministicPerTenant(t *testing.T) {
-	f := Flag{Key: "k", RolloutPercentage: 50}
-	first := evaluate(f, "stabiler-tenant")
-	for i := 0; i < 5; i++ {
-		if evaluate(f, "stabiler-tenant") != first {
-			t.Fatal("rollout-auswertung sollte fuer denselben tenant/key stabil sein")
-		}
-	}
-}

internal/flag/service.go

@@ -1,87 +0,0 @@
-package flag
-
-import (
-	"context"
-	"log/slog"
-	"sync"
-	"time"
-)
-
-// DefaultCacheTTL ist die dokumentierte Cache-Invalidierungszeit
-// (Akzeptanzkriterium 2/3): eine Aenderung wirkt spaetestens nach dieser
-// Zeit auf allen Core-Instanzen, ohne dass ein Dienst neu gestartet werden
-// muss (Akzeptanzkriterium 3).
-const DefaultCacheTTL = 5 * time.Second
-
-type cacheEntry struct {
-	flag      Flag
-	expiresAt time.Time
-}
-
-// Service ist die Auswertungsseite (SDK/Client-Analogon zu Unleash) mit
-// lokalem TTL-Cache. Bewusst getrennt von Store (Verwaltung).
-type Service struct {
-	store *Store
-	ttl   time.Duration
-
-	mu    sync.RWMutex
-	cache map[string]cacheEntry
-}
-
-func NewService(store *Store, ttl time.Duration) *Service {
-	if ttl <= 0 {
-		ttl = DefaultCacheTTL
-	}
-	return &Service{store: store, ttl: ttl, cache: make(map[string]cacheEntry)}
-}
-
-// IsEnabled wertet ein Flag fuer einen Tenant aus. Liefert IMMER einen
-// bool ohne Fehlerwert — ein nicht erreichbarer Flag-Dienst darf abhaengige
-// Aufrufer nicht zum Absturz bringen oder zu Fehlerbehandlungscode zwingen,
-// der leicht vergessen wird (Akzeptanzkriterium 3 / Pruefung 3: dokumentiertes
-// Fallback-Verhalten = false, ggf. aus dem zuletzt bekannten Zwischenspeicher).
-func (s *Service) IsEnabled(ctx context.Context, tenantSlug, key string) bool {
-	f, ok := s.resolve(ctx, key)
-	if !ok {
-		return false
-	}
-	return evaluate(f, tenantSlug)
-}
-
-func (s *Service) resolve(ctx context.Context, key string) (Flag, bool) {
-	s.mu.RLock()
-	entry, exists := s.cache[key]
-	fresh := exists && time.Now().Before(entry.expiresAt)
-	s.mu.RUnlock()
-	if fresh {
-		return entry.flag, true
-	}
-
-	f, err := s.store.Get(ctx, key)
-	if err != nil {
-		if exists {
-			slog.Warn("feature-flag-dienst nicht erreichbar, nutze zwischengespeicherten stand",
-				"flag_key", key, "error", err)
-			return entry.flag, true
-		}
-		slog.Warn("feature-flag-dienst nicht erreichbar, kein zwischengespeicherter stand vorhanden, fallback: deaktiviert",
-			"flag_key", key, "error", err)
-		return Flag{}, false
-	}
-
-	s.mu.Lock()
-	s.cache[key] = cacheEntry{flag: f, expiresAt: time.Now().Add(s.ttl)}
-	s.mu.Unlock()
-	return f, true
-}
-
-// Invalidate erzwingt beim naechsten IsEnabled-Aufruf ein sofortiges Neuladen
-// aus der Datenbank statt auf den TTL-Ablauf zu warten — wird nach Store.Set
-// auf derselben Instanz aufgerufen, damit der Schreiber die eigene Aenderung
-// ohne Wartezeit sieht. Andere Core-Instanzen sehen sie spaetestens nach
-// DefaultCacheTTL (siehe Akzeptanzkriterium 3).
-func (s *Service) Invalidate(key string) {
-	s.mu.Lock()
-	delete(s.cache, key)
-	s.mu.Unlock()
-}

internal/flag/service_test.go

@@ -1,179 +0,0 @@
-package flag
-
-import (
-	"context"
-	"os"
-	"testing"
-	"time"
-
-	"github.com/jackc/pgx/v5/pgxpool"
-)
-
-func setupFlagStoreTest(t *testing.T) (*Store, 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 feature_flags (
-			key TEXT PRIMARY KEY,
-			enabled BOOLEAN NOT NULL DEFAULT false,
-			rollout_percentage INT NOT NULL DEFAULT 0,
-			target_tenant_slugs TEXT[] NOT NULL DEFAULT '{}',
-			updated_at TIMESTAMPTZ NOT NULL DEFAULT now()
-		)`); err != nil {
-		t.Fatalf("schema: %v", err)
-	}
-
-	cleanup := func() {
-		_, _ = pool.Exec(ctx, `DELETE FROM feature_flags WHERE key LIKE 'test\_%' ESCAPE '\'`)
-		pool.Close()
-	}
-	return NewStore(pool), cleanup
-}
-
-// Akzeptanzkriterium 1 + Pruefung 2: Zielgruppen-Strategie liefert im Test
-// die erwartete Auswertung.
-func TestService_TargetTenantStrategy(t *testing.T) {
-	store, cleanup := setupFlagStoreTest(t)
-	defer cleanup()
-	ctx := context.Background()
-
-	if err := store.Set(ctx, Flag{Key: "test_target_flag", TargetTenantSlugs: []string{"acme"}}); err != nil {
-		t.Fatalf("set: %v", err)
-	}
-	svc := NewService(store, time.Hour)
-
-	if !svc.IsEnabled(ctx, "acme", "test_target_flag") {
-		t.Fatal("erwartet true fuer tenant in zielgruppe")
-	}
-	if svc.IsEnabled(ctx, "globex", "test_target_flag") {
-		t.Fatal("erwartet false fuer tenant ausserhalb der zielgruppe")
-	}
-}
-
-// Akzeptanzkriterium 2 + 3 + Pruefung 1: Flag-Aenderung wirkt innerhalb der
-// dokumentierten Cache-Invalidierungszeit, automatisiert gemessen.
-func TestService_CacheInvalidationTiming(t *testing.T) {
-	store, cleanup := setupFlagStoreTest(t)
-	defer cleanup()
-	ctx := context.Background()
-
-	const ttl = 150 * time.Millisecond
-	if err := store.Set(ctx, Flag{Key: "test_ttl_flag", Enabled: false}); err != nil {
-		t.Fatalf("set: %v", err)
-	}
-	svc := NewService(store, ttl)
-
-	if svc.IsEnabled(ctx, "acme", "test_ttl_flag") {
-		t.Fatal("erwartet false vor der aenderung")
-	}
-
-	// Aenderung "auf einer anderen instanz" simulieren: direkt ueber den
-	// Store, ohne svc.Invalidate aufzurufen.
-	changedAt := time.Now()
-	if err := store.Set(ctx, Flag{Key: "test_ttl_flag", Enabled: true}); err != nil {
-		t.Fatalf("set: %v", err)
-	}
-
-	// Sofort danach sollte der Cache noch den alten Stand liefern.
-	if svc.IsEnabled(ctx, "acme", "test_ttl_flag") {
-		t.Fatal("cache haette den alten (false) stand liefern sollen, direkt nach der aenderung")
-	}
-
-	deadline := changedAt.Add(ttl + 100*time.Millisecond)
-	for time.Now().Before(deadline) {
-		if svc.IsEnabled(ctx, "acme", "test_ttl_flag") {
-			elapsed := time.Since(changedAt)
-			t.Logf("aenderung wurde nach %s wirksam (ziel: innerhalb %s + toleranz)", elapsed, ttl)
-			return
-		}
-		time.Sleep(10 * time.Millisecond)
-	}
-	t.Fatalf("aenderung wurde nicht innerhalb von %s wirksam", deadline.Sub(changedAt))
-}
-
-func TestService_InvalidateForcesImmediateRefresh(t *testing.T) {
-	store, cleanup := setupFlagStoreTest(t)
-	defer cleanup()
-	ctx := context.Background()
-
-	if err := store.Set(ctx, Flag{Key: "test_invalidate_flag", Enabled: false}); err != nil {
-		t.Fatalf("set: %v", err)
-	}
-	svc := NewService(store, time.Hour) // lange TTL, damit Invalidate den unterschied macht
-	_ = svc.IsEnabled(ctx, "acme", "test_invalidate_flag")
-
-	if err := store.Set(ctx, Flag{Key: "test_invalidate_flag", Enabled: true}); err != nil {
-		t.Fatalf("set: %v", err)
-	}
-	svc.Invalidate("test_invalidate_flag")
-
-	if !svc.IsEnabled(ctx, "acme", "test_invalidate_flag") {
-		t.Fatal("erwartet sofort sichtbaren neuen stand nach Invalidate")
-	}
-}
-
-// Akzeptanzkriterium 3 + Pruefung 3: Ausfall des Flag-Dienstes fuehrt zu
-// dokumentiertem Fallback-Verhalten, nicht zum Absturz.
-func TestService_FallsBackOnStoreFailure(t *testing.T) {
-	store, cleanup := setupFlagStoreTest(t)
-	defer cleanup()
-	ctx := context.Background()
-
-	if err := store.Set(ctx, Flag{Key: "test_fallback_flag", Enabled: true}); err != nil {
-		t.Fatalf("set: %v", err)
-	}
-	svc := NewService(store, time.Hour)
-
-	// Cache vorwaermen, waehrend die DB noch erreichbar ist.
-	if !svc.IsEnabled(ctx, "acme", "test_fallback_flag") {
-		t.Fatal("erwartet true bei funktionierender db")
-	}
-
-	brokenPool, err := pgxpool.New(ctx, "postgresql://nonexistent-host-fuer-test:5432/x?connect_timeout=1")
-	if err != nil {
-		t.Fatalf("broken pool erstellen (sollte nicht sofort verbinden): %v", err)
-	}
-	brokenStore := NewStore(brokenPool)
-
-	svcWithCache := NewService(brokenStore, time.Nanosecond) // TTL sofort abgelaufen, erzwingt reload-versuch
-	svcWithCache.mu.Lock()
-	svcWithCache.cache["test_fallback_flag"] = cacheEntry{
-		flag:      Flag{Key: "test_fallback_flag", Enabled: true},
-		expiresAt: time.Now().Add(-time.Hour), // bereits abgelaufen
-	}
-	svcWithCache.mu.Unlock()
-
-	func() {
-		defer func() {
-			if r := recover(); r != nil {
-				t.Fatalf("IsEnabled hat gepanict statt einen fallback zu liefern: %v", r)
-			}
-		}()
-		if !svcWithCache.IsEnabled(ctx, "acme", "test_fallback_flag") {
-			t.Fatal("erwartet fallback auf zwischengespeicherten (true) stand bei db-ausfall")
-		}
-	}()
-
-	// Voellig frischer Dienst ohne jeglichen cache + kaputte db -> sicherer
-	// default false, kein absturz.
-	freshSvc := NewService(brokenStore, time.Hour)
-	func() {
-		defer func() {
-			if r := recover(); r != nil {
-				t.Fatalf("IsEnabled hat gepanict: %v", r)
-			}
-		}()
-		if freshSvc.IsEnabled(ctx, "acme", "test_fallback_flag") {
-			t.Fatal("erwartet fail-safe false ohne cache und mit kaputter db")
-		}
-	}()
-}

migrations/0004_audit_events.down.sql

@@ -0,0 +1 @@
+DROP TABLE IF EXISTS audit_events;

migrations/0004_audit_events.up.sql

@@ -0,0 +1,17 @@
+-- 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);

migrations/0004_feature_flags.down.sql

@@ -1 +0,0 @@
-DROP TABLE IF EXISTS feature_flags;

migrations/0004_feature_flags.up.sql

@@ -1,10 +0,0 @@
--- Feature-Flags zentral je Mandant/Zielgruppe (LIC-02, siehe core-kanban/tickets/LIC-02.md).
--- Lebt in der Registry-DB, nicht pro Tenant-Datenbank — Flags sind eine
--- Core-weite Konfiguration, keine Mandanten-Geschaeftsdaten.
-CREATE TABLE feature_flags (
-    key                 TEXT PRIMARY KEY,
-    enabled             BOOLEAN NOT NULL DEFAULT false,
-    rollout_percentage  INT NOT NULL DEFAULT 0 CHECK (rollout_percentage BETWEEN 0 AND 100),
-    target_tenant_slugs TEXT[] NOT NULL DEFAULT '{}',
-    updated_at          TIMESTAMPTZ NOT NULL DEFAULT now()
-);

scripts/reset-test-env.sh

@@ -14,6 +14,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;"
 
 dbs=$(psql -h localhost -U "$ROLE" -d postgres -tAc "SELECT datname FROM pg_database WHERE datname LIKE 'tenant\_%' ESCAPE '\'")
 for db in $dbs; do