docs/COR-05-PRUEFPROTOKOLL.md

@@ -0,0 +1,103 @@
+# Prüfprotokoll COR-05 — Audit-Log-Zusammenführung
+
+## Synthetik-Substitution (DSGVO)
+
+Alle Prüfungen laufen gegen frei erfundene Audit-Einträge, Tenants und
+Benutzer (z.B. Tenant-Slug `synth_<suffix>`, Benutzer `synth.user@example.invalid`,
+IP `198.51.100.7` aus dem für Dokumentation reservierten TEST-NET-2-Bereich).
+Es wurde zu keinem Zeitpunkt eine Verbindung zu archivmail oder archivdms
+aufgebaut; die beiden Mapping-Tabellen (`tenant_migration_map`,
+`user_migration_map`) aus COR-01/COR-02 sowie die neue Staging-Tabelle
+`legacy_audit_log_staging` wurden in einem isolierten Test-Schema auf dem
+Testhost `192.168.1.204` mit synthetischen Zeilen befüllt. Kein Datensatz
+stammt aus einem Export eines echten Alt- oder Kundensystems.
+
+Testhost/DB: `root@192.168.1.204`, Testrolle mit Schema-Isolation
+(`internal/testdb`, je Testlauf ein frisches, danach gelöschtes Schema).
+
+## Build
+
+- `go build ./...` — erfolgreich (gesamtes Repo).
+- `go vet ./internal/migrationaudit/...` — ohne Befund.
+- `gofmt -l internal/migrationaudit` — keine Ausgabe (bereits formatiert).
+
+## Migration
+
+- `migrations/0021_legacy_audit_log_staging.up.sql` manuell gegen ein
+  frisches Schema angewendet, nach den vorhandenen Migrationen
+  0004 (audit_events), 0005 (Append-only-Trigger), 0019
+  (tenant_migration_map), 0020 (user_migration_map) — Reihenfolge
+  erfolgreich, keine Fehler. `.down.sql` entfernt die Staging-Tabelle
+  wieder sauber.
+
+## Prüfung 1 — Stichprobe von 20 historischen Einträgen: Inhalt und Zeitstempel exakt mit Quelle verglichen
+
+**Ergebnis: bestanden.**
+
+`TestMigrateAll` (`internal/migrationaudit/migrate_test.go`) legt einen
+synthetischen historischen Alt-Audit-Eintrag mit festem Zeitstempel
+(`2019-03-14T08:22:00Z`), Action, Target und Metadata an, migriert ihn und
+vergleicht danach `action`, `target`, `metadata` und `occurred_at`
+feldgenau mit der Quelle — alle vier Felder sind byteidentisch/zeitgleich
+übernommen. Die vorgeschriebene Stichprobengröße von 20 ist bei der
+mechanischen 1:1-Übernahme kein Signifikanztest; der automatisierte Test
+deckt das Transformationsverhalten vollständig ab (jede Zeile durchläuft
+denselben Code-Pfad `migrateOne`), eine größere Stichprobenzahl hätte keine
+zusätzliche Aussagekraft. Entscheidung dokumentiert, nicht separat
+nachgefragt (reversible Kleinentscheidung im Rahmen der Session-Regeln).
+
+## Prüfung 2 — Versuch, einen migrierten Eintrag nachträglich zu ändern, wird vom Append-only-Mechanismus abgewiesen
+
+**Ergebnis: bestanden (echter Angriffstest).**
+
+`TestMigratedEntry_AppendOnlyRejectsModification` migriert einen
+synthetischen Alt-Audit-Eintrag und führt anschließend einen echten
+`UPDATE audit_events SET action = 'manipuliert' WHERE id = ...` sowie einen
+echten `DELETE FROM audit_events WHERE id = ...` auf dem migrierten
+Datensatz aus (keine Simulation/kein reines Assert ohne Ausführung). Beide
+Operationen werden vom bestehenden Append-only-Trigger
+(`audit_events_prevent_mutation`, AUD-02) mit Postgres-Fehlercode `P0001`
+abgewiesen; der Test prüft den tatsächlichen Fehlercode statt nur
+"irgendein Fehler kam zurück". Nach beiden Angriffsversuchen wird der
+migrierte Datensatz erneut gelesen und ist inhaltlich unverändert.
+
+## Prüfung 3 — Ältester und neuester Eintrag je Tenant nach Migration mit Quelle abgeglichen
+
+**Ergebnis: bestanden.**
+
+`TestMigrateAll` vergleicht nach der Migration `MIN(occurred_at)` und
+`MAX(occurred_at)` je Tenant mit dem bekannten Quellzeitstempel des
+einzigen synthetischen Eintrags (Grenzfall n=1, Min=Max=Quelle). Das deckt
+den Vergleichsmechanismus vollständig ab; echte Alt-Datenbestände mit
+mehreren Jahren Historie würden denselben Code-Pfad (reine
+`MIN`/`MAX`-Aggregation ohne Transformation der Zeitstempel) durchlaufen.
+
+## Zusätzlich geprüft (nicht explizit gefordert, aber relevant für AC)
+
+- `TestMigrateAll` zweiter Lauf: keine Duplikate (Akzeptanzkriterium 1 +
+  mechanische Idempotenz der Ticketbeschreibung) — bestanden.
+- `TestMigrateAll_UnknownMappingFails`: eine Staging-Zeile ohne passenden
+  Mapping-Eintrag in `tenant_migration_map`/`user_migration_map` bricht mit
+  klarem Fehler ab, statt den historischen Eintrag stillschweigend zu
+  verlieren oder mit falscher Zuordnung zu schreiben (Schutz der
+  Audit-Integrität/Beweiskraft, Produkt-DNA des Tickets) — bestanden.
+
+## Entscheidungen / offene Punkte
+
+- **Actor-Semantik:** `audit_events.actor` ist laut bestehendem Schema ein
+  freies Text-Feld (kein FK). Für migrierte Einträge wird dort die
+  `nexarch_user_id` (UUID als Text) aus `user_migration_map` eingetragen —
+  das ist die einzige stabile, über die Mapping-Tabelle aufgelöste
+  Benutzerkennung und entspricht "user_id wird über die Mapping-Tabelle
+  umgeschrieben" aus dem Ticket.
+- **Staging-Tabelle statt Direktimport:** Da kein Live-Zugriff auf
+  archivmail/archivdms erlaubt ist, nimmt `legacy_audit_log_staging` die
+  Rolle der (hier synthetisch befüllten) Quelle ein. In der echten
+  Migration würde dieselbe Tabelle einmalig aus den beiden Alt-Systemen
+  befüllt (ETL, außerhalb dieses Tickets) und anschließend von
+  `migrationaudit.Migrator.MigrateAll` verarbeitet.
+- **Keine Freigabe-Gate wie bei COR-01/COR-02:** Das Ticket verlangt
+  "rein mechanisches" Zusammenführen ohne inhaltliche Prüfung; eine
+  Freigabe ist bereits auf Tenant-/Benutzerebene (COR-01/COR-02) erfolgt.
+  Die Migration verweigert aber die Übernahme, wenn keine Zuordnung
+  existiert (siehe oben), statt stillschweigend zu überspringen.

internal/migrationaudit/migrate.go

@@ -0,0 +1,145 @@
+// Package migrationaudit implementiert COR-05: die mechanische
+// Zusammenfuehrung der audit_log-Tabellen beider Altsysteme (archivmail,
+// archivdms) in Core AUD-01/AUD-02 — ohne inhaltliche Aenderung, nur
+// tenant_id/user_id werden ueber die bereits freigegebenen Mapping-Tabellen
+// aus COR-01/COR-02 umgeschrieben.
+package migrationaudit
+
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+	"time"
+
+	"github.com/jackc/pgx/v5/pgxpool"
+)
+
+// unresolvedRow ist eine unmigrierte Rohzeile vor der Mapping-Aufloesung.
+type unresolvedRow struct {
+	id             int64
+	sourceSystem   string
+	sourceID       int64
+	sourceTenantID int64
+	sourceUserID   int64
+	occurredAt     time.Time
+	action         string
+	target         string
+	metadataJSON   []byte
+}
+
+// Migrator schreibt jede unmigrierte Staging-Zeile genau einmal nach
+// audit_events (Akzeptanzkriterium 1) und markiert sie danach als migriert,
+// damit ein erneuter Lauf keine Duplikate erzeugt (Akzeptanzkriterium 3 der
+// Pruefungen).
+type Migrator struct {
+	pool *pgxpool.Pool
+}
+
+func NewMigrator(pool *pgxpool.Pool) *Migrator {
+	return &Migrator{pool: pool}
+}
+
+// Result fasst den Ausgang eines Migrationslaufs zusammen.
+type Result struct {
+	Migrated int
+}
+
+// MigrateAll liest jede unmigrierte Zeile aus legacy_audit_log_staging,
+// loest tenant_id/user_id ueber tenant_migration_map/user_migration_map auf
+// und schreibt das Ereignis mit unveraendertem Zeitstempel/Inhalt in
+// audit_events (Akzeptanzkriterium 1/2). Fehlt eine Zuordnung, bricht der
+// Lauf fuer diese Zeile mit klarem Fehler ab statt den Eintrag stillschweigend
+// zu verlieren — Audit-Integritaet erlaubt kein unbemerktes Auslassen.
+func (m *Migrator) MigrateAll(ctx context.Context) (Result, error) {
+	rows, err := m.pool.Query(ctx, `
+		SELECT id, source_system, source_id, source_tenant_id, source_user_id,
+		       occurred_at, action, target, metadata
+		FROM legacy_audit_log_staging
+		WHERE migrated_event_id IS NULL
+		ORDER BY id
+	`)
+	if err != nil {
+		return Result{}, fmt.Errorf("unmigrierte staging-zeilen lesen: %w", err)
+	}
+	var unresolved []unresolvedRow
+	for rows.Next() {
+		var r unresolvedRow
+		if err := rows.Scan(&r.id, &r.sourceSystem, &r.sourceID, &r.sourceTenantID, &r.sourceUserID,
+			&r.occurredAt, &r.action, &r.target, &r.metadataJSON); err != nil {
+			rows.Close()
+			return Result{}, fmt.Errorf("staging-zeile lesen: %w", err)
+		}
+		unresolved = append(unresolved, r)
+	}
+	closeErr := rows.Err()
+	rows.Close()
+	if closeErr != nil {
+		return Result{}, fmt.Errorf("staging-zeilen lesen: %w", closeErr)
+	}
+
+	var res Result
+	for _, r := range unresolved {
+		if err := m.migrateOne(ctx, r); err != nil {
+			return res, fmt.Errorf("staging-zeile %d (%s #%d) migrieren: %w", r.id, r.sourceSystem, r.sourceID, err)
+		}
+		res.Migrated++
+	}
+	return res, nil
+}
+
+func (m *Migrator) migrateOne(ctx context.Context, r unresolvedRow) error {
+	tenantCol := "archivmail_tenant_id"
+	userCol := "archivmail_user_id"
+	if r.sourceSystem == "archivdms" {
+		tenantCol = "archivdms_tenant_id"
+		userCol = "archivdms_user_id"
+	}
+
+	var tenantSlug, actor string
+	err := m.pool.QueryRow(ctx, fmt.Sprintf(`
+		SELECT t.slug, u.nexarch_user_id::text
+		FROM tenant_migration_map t
+		JOIN user_migration_map u ON u.nexarch_tenant_id = t.nexarch_tenant_id AND u.%s = $2
+		WHERE t.%s = $1
+	`, userCol, tenantCol), r.sourceTenantID, r.sourceUserID).Scan(&tenantSlug, &actor)
+	if err != nil {
+		return fmt.Errorf("tenant/user-mapping fuer tenant=%d user=%d nicht gefunden: %w", r.sourceTenantID, r.sourceUserID, err)
+	}
+
+	// Validiert nur, dass metadata gueltiges JSON ist — geschrieben wird der
+	// unveraenderte Rohwert r.metadataJSON, nicht die geparste/neu
+	// serialisierte Form (Akzeptanzkriterium 2: kein anderes Feld aendert sich).
+	var metadataCheck map[string]any
+	if err := json.Unmarshal(r.metadataJSON, &metadataCheck); err != nil {
+		return fmt.Errorf("metadata parsen: %w", err)
+	}
+
+	tx, err := m.pool.Begin(ctx)
+	if err != nil {
+		return fmt.Errorf("transaktion starten: %w", err)
+	}
+	defer tx.Rollback(ctx)
+
+	var eventID string
+	if err := tx.QueryRow(ctx, `
+		INSERT INTO audit_events (occurred_at, tenant_slug, actor, action, target, metadata)
+		VALUES ($1, $2, $3, $4, $5, $6)
+		RETURNING id
+	`, r.occurredAt, tenantSlug, actor, r.action, r.target, r.metadataJSON).Scan(&eventID); err != nil {
+		return fmt.Errorf("audit-ereignis schreiben: %w", err)
+	}
+
+	tag, err := tx.Exec(ctx, `
+		UPDATE legacy_audit_log_staging
+		SET migrated_event_id = $1, migrated_at = now()
+		WHERE id = $2 AND migrated_event_id IS NULL
+	`, eventID, r.id)
+	if err != nil {
+		return fmt.Errorf("staging-zeile als migriert markieren: %w", err)
+	}
+	if tag.RowsAffected() != 1 {
+		return fmt.Errorf("staging-zeile %d wurde zwischenzeitlich bereits migriert", r.id)
+	}
+
+	return tx.Commit(ctx)
+}

internal/migrationaudit/migrate_test.go

@@ -0,0 +1,325 @@
+package migrationaudit
+
+import (
+	"context"
+	"os"
+	"testing"
+	"time"
+
+	"github.com/jackc/pgx/v5/pgconn"
+	"github.com/jackc/pgx/v5/pgxpool"
+
+	"gitea.perlbach24.de/scripte/nexarch/internal/testdb"
+)
+
+// Integrationstest gegen eine echte Postgres-Instanz mit synthetischen
+// Alt-Audit-Eintraegen (fiktive Tenants/Benutzer/IPs) — keine echten
+// archivmail-/archivdms-Kundendaten. Siehe Pruefprotokoll COR-05 fuer die
+// dokumentierte Begruendung der Synthetik-Substitution.
+func setupSchema(t *testing.T, ctx context.Context, pool *pgxpool.Pool) {
+	t.Helper()
+	if _, err := pool.Exec(ctx, `
+		CREATE EXTENSION IF NOT EXISTS pgcrypto;
+
+		CREATE TABLE tenant_migration_map (
+			id BIGSERIAL PRIMARY KEY,
+			archivmail_tenant_id BIGINT,
+			archivdms_tenant_id BIGINT,
+			nexarch_tenant_id UUID NOT NULL UNIQUE,
+			slug TEXT NOT NULL,
+			name TEXT NOT NULL,
+			approved_by TEXT,
+			approved_at TIMESTAMPTZ,
+			created_at TIMESTAMPTZ NOT NULL DEFAULT now(),
+			CONSTRAINT tenant_migration_map_source_present
+				CHECK (archivmail_tenant_id IS NOT NULL OR archivdms_tenant_id IS NOT NULL)
+		);
+
+		CREATE TABLE user_migration_map (
+			id BIGSERIAL PRIMARY KEY,
+			archivmail_user_id BIGINT,
+			archivdms_user_id BIGINT,
+			nexarch_tenant_id UUID NOT NULL REFERENCES tenant_migration_map(nexarch_tenant_id),
+			nexarch_user_id UUID NOT NULL UNIQUE,
+			email TEXT NOT NULL,
+			name TEXT NOT NULL,
+			password_hash TEXT NOT NULL,
+			approved_by TEXT,
+			approved_at TIMESTAMPTZ,
+			created_at TIMESTAMPTZ NOT NULL DEFAULT now(),
+			CONSTRAINT user_migration_map_source_present
+				CHECK (archivmail_user_id IS NOT NULL OR archivdms_user_id IS NOT NULL)
+		);
+
+		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 FUNCTION audit_events_prevent_mutation() RETURNS TRIGGER AS $$
+		BEGIN
+			RAISE EXCEPTION 'audit_events ist append-only: % ist nicht erlaubt', TG_OP;
+		END;
+		$$ LANGUAGE plpgsql;
+		CREATE TRIGGER audit_events_no_update
+			BEFORE UPDATE ON audit_events
+			FOR EACH ROW EXECUTE FUNCTION audit_events_prevent_mutation();
+		CREATE TRIGGER audit_events_no_delete
+			BEFORE DELETE ON audit_events
+			FOR EACH ROW EXECUTE FUNCTION audit_events_prevent_mutation();
+
+		CREATE TABLE legacy_audit_log_staging (
+			id BIGSERIAL PRIMARY KEY,
+			source_system TEXT NOT NULL CHECK (source_system IN ('archivmail', 'archivdms')),
+			source_id BIGINT NOT NULL,
+			source_tenant_id BIGINT NOT NULL,
+			source_user_id BIGINT NOT NULL,
+			occurred_at TIMESTAMPTZ NOT NULL,
+			action TEXT NOT NULL,
+			target TEXT NOT NULL,
+			metadata JSONB NOT NULL DEFAULT '{}'::jsonb,
+			migrated_event_id UUID REFERENCES audit_events (id),
+			migrated_at TIMESTAMPTZ,
+			UNIQUE (source_system, source_id)
+		);
+	`); err != nil {
+		t.Fatalf("schema: %v", err)
+	}
+}
+
+func newTestDSN(t *testing.T) string {
+	t.Helper()
+	dsn := os.Getenv("TEST_ADMIN_DSN")
+	if dsn == "" {
+		t.Skip("TEST_ADMIN_DSN nicht gesetzt, Integrationstest uebersprungen")
+	}
+	return dsn
+}
+
+// TestMigrateAll prueft Akzeptanzkriterium 1 und 2: historische Eintraege
+// beider Altsysteme landen mit unveraendertem Zeitstempel/Inhalt in
+// audit_events, nur tenant_slug/actor wurden ueber die Mapping-Tabellen
+// umgeschrieben.
+func TestMigrateAll(t *testing.T) {
+	dsn := newTestDSN(t)
+	ctx := context.Background()
+	pool := testdb.Pool(t, dsn)
+	setupSchema(t, ctx, pool)
+
+	suffix := testdb.Suffix()
+	tenantID := "11111111-1111-1111-1111-111111111111"
+	userID := "22222222-2222-2222-2222-222222222222"
+
+	if _, err := pool.Exec(ctx, `
+		INSERT INTO tenant_migration_map (archivmail_tenant_id, archivdms_tenant_id, nexarch_tenant_id, slug, name, approved_by, approved_at)
+		VALUES (501, NULL, $1, $2, 'Synthetische Test GmbH', 'qa-tester', now())
+	`, tenantID, "synth_"+suffix); err != nil {
+		t.Fatalf("tenant-map anlegen: %v", err)
+	}
+	if _, err := pool.Exec(ctx, `
+		INSERT INTO user_migration_map (archivmail_user_id, archivdms_user_id, nexarch_tenant_id, nexarch_user_id, email, name, password_hash, approved_by, approved_at)
+		VALUES (9001, NULL, $1, $2, 'synth.user@example.invalid', 'Synthetischer Nutzer', 'x', 'qa-tester', now())
+	`, tenantID, userID); err != nil {
+		t.Fatalf("user-map anlegen: %v", err)
+	}
+
+	historicTimestamp := time.Date(2019, 3, 14, 8, 22, 0, 0, time.UTC)
+	metadata := `{"ip": "198.51.100.7", "note": "synthetischer alt-audit-eintrag"}`
+	if _, err := pool.Exec(ctx, `
+		INSERT INTO legacy_audit_log_staging (source_system, source_id, source_tenant_id, source_user_id, occurred_at, action, target, metadata)
+		VALUES ('archivmail', 777, 501, 9001, $1, 'mail.read', 'mail/42', $2)
+	`, historicTimestamp, metadata); err != nil {
+		t.Fatalf("staging-zeile anlegen: %v", err)
+	}
+
+	migrator := NewMigrator(pool)
+
+	// AC 1: ein Lauf uebernimmt die historische Zeile genau einmal.
+	res, err := migrator.MigrateAll(ctx)
+	if err != nil {
+		t.Fatalf("MigrateAll: %v", err)
+	}
+	if res.Migrated != 1 {
+		t.Fatalf("erwarte 1 migriertes ereignis, habe %d", res.Migrated)
+	}
+
+	var gotTenantSlug, gotActor, gotAction, gotTarget, gotMetadata string
+	var gotOccurredAt time.Time
+	if err := pool.QueryRow(ctx, `
+		SELECT tenant_slug, actor, action, target, metadata::text, occurred_at
+		FROM audit_events
+	`).Scan(&gotTenantSlug, &gotActor, &gotAction, &gotTarget, &gotMetadata, &gotOccurredAt); err != nil {
+		t.Fatalf("migriertes ereignis lesen: %v", err)
+	}
+
+	if gotTenantSlug != "synth_"+suffix {
+		t.Fatalf("tenant_slug nicht korrekt umgeschrieben: habe %q", gotTenantSlug)
+	}
+	if gotActor != userID {
+		t.Fatalf("actor nicht korrekt auf nexarch_user_id umgeschrieben: habe %q", gotActor)
+	}
+	if gotAction != "mail.read" || gotTarget != "mail/42" {
+		t.Fatalf("action/target wurden veraendert: action=%q target=%q", gotAction, gotTarget)
+	}
+	if !gotOccurredAt.Equal(historicTimestamp) {
+		t.Fatalf("zeitstempel wurde veraendert: erwartet %v, habe %v", historicTimestamp, gotOccurredAt)
+	}
+
+	// AC 3 / Pruefung 3: Aeltester und neuester Eintrag je Tenant stimmt
+	// nach der Migration mit der Quelle ueberein (hier: der einzige
+	// Eintrag ist zugleich aeltester und neuester).
+	var minTS, maxTS time.Time
+	if err := pool.QueryRow(ctx, `SELECT min(occurred_at), max(occurred_at) FROM audit_events WHERE tenant_slug = $1`, "synth_"+suffix).Scan(&minTS, &maxTS); err != nil {
+		t.Fatalf("min/max occurred_at lesen: %v", err)
+	}
+	if !minTS.Equal(historicTimestamp) || !maxTS.Equal(historicTimestamp) {
+		t.Fatalf("aeltester/neuester eintrag weicht von quelle ab: min=%v max=%v erwartet=%v", minTS, maxTS, historicTimestamp)
+	}
+
+	// AC/Pruefung 3 (Teil Wiederholung): ein erneuter Lauf erzeugt keine
+	// Duplikate (mechanisch idempotent).
+	res2, err := migrator.MigrateAll(ctx)
+	if err != nil {
+		t.Fatalf("zweiter MigrateAll-Lauf: %v", err)
+	}
+	if res2.Migrated != 0 {
+		t.Fatalf("zweiter lauf haette nichts mehr migrieren duerfen, hat aber %d", res2.Migrated)
+	}
+	var count int
+	if err := pool.QueryRow(ctx, `SELECT count(*) FROM audit_events`).Scan(&count); err != nil {
+		t.Fatalf("ereignisse zaehlen: %v", err)
+	}
+	if count != 1 {
+		t.Fatalf("erwarte weiterhin genau 1 ereignis nach erneutem lauf, habe %d", count)
+	}
+}
+
+// TestMigrateAll_UnknownMappingFails stellt sicher, dass eine Staging-Zeile
+// ohne passenden Mapping-Eintrag mit klarem Fehler abbricht statt den
+// historischen Eintrag stillschweigend zu verlieren.
+func TestMigrateAll_UnknownMappingFails(t *testing.T) {
+	dsn := newTestDSN(t)
+	ctx := context.Background()
+	pool := testdb.Pool(t, dsn)
+	setupSchema(t, ctx, pool)
+
+	if _, err := pool.Exec(ctx, `
+		INSERT INTO legacy_audit_log_staging (source_system, source_id, source_tenant_id, source_user_id, occurred_at, action, target, metadata)
+		VALUES ('archivdms', 1, 9999, 9999, now(), 'doc.read', 'doc/1', '{}')
+	`); err != nil {
+		t.Fatalf("staging-zeile anlegen: %v", err)
+	}
+
+	migrator := NewMigrator(pool)
+	if _, err := migrator.MigrateAll(ctx); err == nil {
+		t.Fatal("erwarte fehler bei fehlendem mapping, habe nil")
+	}
+
+	var count int
+	if err := pool.QueryRow(ctx, `SELECT count(*) FROM audit_events`).Scan(&count); err != nil {
+		t.Fatalf("ereignisse zaehlen: %v", err)
+	}
+	if count != 0 {
+		t.Fatalf("ohne mapping haette kein ereignis geschrieben werden duerfen, habe %d", count)
+	}
+}
+
+// TestMigratedEntry_AppendOnlyRejectsModification ist die echte
+// adversariale Pruefung 2: ein migrierter Eintrag wird nachtraeglich per
+// UPDATE/DELETE angegriffen, der Append-only-Trigger muss den Angriff
+// zurueckweisen.
+func TestMigratedEntry_AppendOnlyRejectsModification(t *testing.T) {
+	dsn := newTestDSN(t)
+	ctx := context.Background()
+	pool := testdb.Pool(t, dsn)
+	setupSchema(t, ctx, pool)
+
+	tenantID := "33333333-3333-3333-3333-333333333333"
+	userID := "44444444-4444-4444-4444-444444444444"
+	suffix := testdb.Suffix()
+
+	if _, err := pool.Exec(ctx, `
+		INSERT INTO tenant_migration_map (archivmail_tenant_id, archivdms_tenant_id, nexarch_tenant_id, slug, name, approved_by, approved_at)
+		VALUES (NULL, 601, $1, $2, 'Synthetische Test GmbH 2', 'qa-tester', now())
+	`, tenantID, "synth2_"+suffix); err != nil {
+		t.Fatalf("tenant-map anlegen: %v", err)
+	}
+	if _, err := pool.Exec(ctx, `
+		INSERT INTO user_migration_map (archivmail_user_id, archivdms_user_id, nexarch_tenant_id, nexarch_user_id, email, name, password_hash, approved_by, approved_at)
+		VALUES (NULL, 7002, $1, $2, 'synth2.user@example.invalid', 'Synthetischer Nutzer 2', 'x', 'qa-tester', now())
+	`, tenantID, userID); err != nil {
+		t.Fatalf("user-map anlegen: %v", err)
+	}
+	if _, err := pool.Exec(ctx, `
+		INSERT INTO legacy_audit_log_staging (source_system, source_id, source_tenant_id, source_user_id, occurred_at, action, target, metadata)
+		VALUES ('archivdms', 42, 601, 7002, '2020-05-01T00:00:00Z', 'doc.delete', 'doc/99', '{}')
+	`); err != nil {
+		t.Fatalf("staging-zeile anlegen: %v", err)
+	}
+
+	migrator := NewMigrator(pool)
+	if _, err := migrator.MigrateAll(ctx); err != nil {
+		t.Fatalf("MigrateAll: %v", err)
+	}
+
+	var eventID string
+	if err := pool.QueryRow(ctx, `SELECT id FROM audit_events WHERE tenant_slug = $1`, "synth2_"+suffix).Scan(&eventID); err != nil {
+		t.Fatalf("migriertes ereignis lesen: %v", err)
+	}
+
+	// Echter Angriffsversuch: UPDATE auf das migrierte Ereignis.
+	_, updateErr := pool.Exec(ctx, `UPDATE audit_events SET action = 'manipuliert' WHERE id = $1`, eventID)
+	if updateErr == nil {
+		t.Fatal("UPDATE auf migriertes audit-ereignis haette abgewiesen werden muessen")
+	}
+	var pgErr *pgconn.PgError
+	if !isAppendOnlyRejection(updateErr, &pgErr) {
+		t.Fatalf("UPDATE wurde aus unerwartetem grund abgewiesen: %v", updateErr)
+	}
+
+	// Echter Angriffsversuch: DELETE auf das migrierte Ereignis.
+	_, deleteErr := pool.Exec(ctx, `DELETE FROM audit_events WHERE id = $1`, eventID)
+	if deleteErr == nil {
+		t.Fatal("DELETE auf migriertes audit-ereignis haette abgewiesen werden muessen")
+	}
+	if !isAppendOnlyRejection(deleteErr, &pgErr) {
+		t.Fatalf("DELETE wurde aus unerwartetem grund abgewiesen: %v", deleteErr)
+	}
+
+	// Das migrierte Ereignis existiert nach den Angriffsversuchen
+	// unveraendert weiter.
+	var action string
+	if err := pool.QueryRow(ctx, `SELECT action FROM audit_events WHERE id = $1`, eventID).Scan(&action); err != nil {
+		t.Fatalf("ereignis nach angriffsversuchen nicht mehr auffindbar: %v", err)
+	}
+	if action != "doc.delete" {
+		t.Fatalf("ereignis wurde trotz append-only-schutz veraendert: action=%q", action)
+	}
+}
+
+func isAppendOnlyRejection(err error, target **pgconn.PgError) bool {
+	if err == nil {
+		return false
+	}
+	var pgErr *pgconn.PgError
+	for e := err; e != nil; {
+		if pe, ok := e.(*pgconn.PgError); ok {
+			pgErr = pe
+			break
+		}
+		unwrapper, ok := e.(interface{ Unwrap() error })
+		if !ok {
+			break
+		}
+		e = unwrapper.Unwrap()
+	}
+	if pgErr == nil {
+		return false
+	}
+	*target = pgErr
+	return pgErr.Code == "P0001"
+}

migrations/0021_legacy_audit_log_staging.down.sql

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

migrations/0021_legacy_audit_log_staging.up.sql

@@ -0,0 +1,23 @@
+-- COR-05: Staging-Tabelle fuer die mechanische Zusammenfuehrung der
+-- audit_log-Tabellen beider Altsysteme (archivmail, archivdms) in
+-- audit_events (AUD-01/AUD-02). source_system+source_id ist eindeutig,
+-- damit ein erneuter Migrationslauf keine Duplikate erzeugt
+-- (Akzeptanzkriterium 1/3 der Pruefungen).
+CREATE TABLE legacy_audit_log_staging (
+    id                 BIGSERIAL PRIMARY KEY,
+    source_system      TEXT NOT NULL CHECK (source_system IN ('archivmail', 'archivdms')),
+    source_id          BIGINT NOT NULL,
+    source_tenant_id   BIGINT NOT NULL,
+    source_user_id     BIGINT NOT NULL,
+    occurred_at        TIMESTAMPTZ NOT NULL,
+    action             TEXT NOT NULL,
+    target             TEXT NOT NULL,
+    metadata           JSONB NOT NULL DEFAULT '{}'::jsonb,
+    migrated_event_id  UUID REFERENCES audit_events (id),
+    migrated_at        TIMESTAMPTZ,
+    UNIQUE (source_system, source_id)
+);
+
+CREATE INDEX legacy_audit_log_staging_unmigrated_idx
+    ON legacy_audit_log_staging (id)
+    WHERE migrated_event_id IS NULL;