diff --git a/docs/VAL-04-PRUEFPROTOKOLL.md b/docs/VAL-04-PRUEFPROTOKOLL.md new file mode 100644 index 00000000..819c1736 --- /dev/null +++ b/docs/VAL-04-PRUEFPROTOKOLL.md @@ -0,0 +1,108 @@ +# VAL-04 – Pruefprotokoll: Audit-Log-Kontinuitaetspruefung + +## Auftrag + +Pruefung, dass aeltester und neuester Audit-Log-Eintrag je Tenant nach der +Migration (COR-05, `internal/migrationaudit`) mit der Quelle +(`legacy_audit_log_staging`) uebereinstimmt — keine Luecke, keine +Zeitverschiebung. + +## Umsetzung + +Neues Paket `internal/migrationauditcontinuity`: + +- `continuity.go`: `AuditStats` (Anzahl, aeltester/neuester Zeitstempel je + Seite), `StatsProvider`-Interface, `ContinuityReport` mit `Luecke()`, + `Zeitverschiebung()` und `OffenerBefund()` — jede der beiden Abweichungen + gilt als blockierender Befund, es gibt (anders als bei VAL-01) keinen + Erklaerungs-Mechanismus, da das Produkt-DNA-Prinzip fuer Audit-Logs keine + Ausnahme vorsieht. `Run()` vergleicht Quelle gegen Ziel je Tenant. +- `postgres.go`: `PostgresSourceStatsProvider` liest Anzahl/min/max aus + `legacy_audit_log_staging`, ueber `tenant_migration_map` auf den + Tenant-Slug aufgeloest (beide Altsysteme in einer Abfrage, je nach + belegter `archivmail_tenant_id`/`archivdms_tenant_id`-Spalte). + `PostgresTargetStatsProvider` liest dieselben Werte direkt aus + `audit_events` (COR-05-Ziel). +- `render.go`: `RenderBericht()` (deutschsprachige Markdown-Tabelle je + Tenant, Berichtsmuster aus VAL-01 `internal/migrationverify`) und + `OffeneBefunde()`. +- `continuity_test.go`: zwei Integrationstests gegen einen echten + Postgres-Testcluster (Host 204). + +Kein Eingriff in COR-05 (`internal/migrationaudit`) oder VAL-01 +(`internal/migrationverify`) — reines, zusaetzliches Lese-/Pruefwerkzeug +nach demselben Muster. + +## Pruefungen vor Abnahme — real durchgefuehrt + +Testcluster: Host 204, `TEST_ADMIN_DSN` aus `/etc/nexarch-dev/core-dev.env` +(Werte nie im Klartext ausgegeben, nur per `set -a && source ... && set +a` +im selben Befehl eingespeist). Eigenes Test-Schema je Lauf +(`internal/testdb`), automatisch aufgeraeumt. + +### Pruefung 1: Aeltester/neuester Eintrag fuer 3 Test-Tenants exakt mit Quelle abgeglichen + +`TestKontinuitaetspruefung_DreiTenantsMitEntferntemEintrag`: + +- 3 synthetische Test-Tenants (`mandant-a`, `mandant-b`, `mandant-c`), + je mit mehreren historischen `legacy_audit_log_staging`-Eintraegen + (Zeitstempel 2018–2020, synthetische IPs/Metadaten). +- Migration via `migrationaudit.Migrator.MigrateAll` (COR-05, unveraendert). +- `migrationauditcontinuity.Run` gegen Quelle und Ziel. +- Ergebnis: fuer `mandant-a` und `mandant-b` (ohne Manipulation) stimmen + Anzahl, aeltester und neuester Zeitstempel exakt ueberein, `OffenerBefund()` + ist `false`. **Zielwert erreicht: PASS.** + +### Pruefung 2: Absichtlich entfernter Testeintrag wird als Luecke erkannt + +Selber Test, Tenant `mandant-c`: nach der Migration wird ein mittlerer +Audit-Eintrag im Ziel (`audit_events`) real geloescht (Append-only-Trigger +dafuer temporaer deaktiviert, wie es ein tatsaechlicher Datenverlust im +Betrieb taete), die zugehoerige Staging-Referenz zuvor geloest, um die +Fremdschluessel-Bindung sauber aufzuheben. + +- Ergebnis: Quelle=3, Ziel=2 Eintraege fuer `mandant-c`. + `ContinuityReport.Luecke()` ist `true`, `OffenerBefund()` ist `true`. + Der gerenderte Bericht enthaelt fuer diesen Tenant "OFFENER BEFUND", + `OffeneBefunde()` liefert genau diesen einen Tenant. + **Zielwert erreicht: PASS.** + +### Zusatzpruefung: Zeitverschiebung ohne Lueckenbildung + +`TestKontinuitaetspruefung_Zeitverschiebung` (ueber die geforderten +Pruefungen hinaus, deckt den zweiten in Akzeptanzkriterium 2 genannten +Fall explizit ab): Zeitstempel eines migrierten Eintrags wird im Ziel um +eine Stunde verschoben (Append-only-Update-Trigger temporaer deaktiviert). +Anzahl bleibt identisch (`Luecke()` = `false`), aber +`Zeitverschiebung()` = `true`, somit `OffenerBefund()` = `true`. +**Zielwert erreicht: PASS.** + +### Testlauf + +``` +go test ./internal/migrationauditcontinuity/... -v +=== RUN TestKontinuitaetspruefung_DreiTenantsMitEntferntemEintrag +--- PASS: TestKontinuitaetspruefung_DreiTenantsMitEntferntemEintrag (0.20s) +=== RUN TestKontinuitaetspruefung_Zeitverschiebung +--- PASS: TestKontinuitaetspruefung_Zeitverschiebung (0.09s) +PASS +ok gitea.perlbach24.de/scripte/nexarch/internal/migrationauditcontinuity 0.297s +``` + +`go build ./...` und `go vet ./...` fuer das gesamte Repository: fehlerfrei. + +## Akzeptanzkriterien — Abgleich + +1. Fuer jeden Tenant liegt ein Kontinuitaetsbericht vor — `Run()` erzeugt + je Tenant genau einen `ContinuityReport`, `RenderBericht()` stellt sie + als Markdown-Tabelle dar. Erfuellt. +2. Jede erkannte Luecke oder Zeitverschiebung ist ein blockierender + Befund — `OffenerBefund()` kennt keine Erklaerungs-Ausnahme, beide + Testfaelle (Luecke, Zeitverschiebung) zeigen den Befund. Erfuellt. + +## DSGVO + +Alle Testdaten synthetisch (fiktive Tenant-/Nutzernamen, erfundene +IPs/Metadaten), keine archivmail-/archivdms-Exporte verwendet. Keine +Zugangsdaten im Code, Testverbindung ausschliesslich ueber +`TEST_ADMIN_DSN` aus der Umgebung. diff --git a/internal/migrationauditcontinuity/continuity.go b/internal/migrationauditcontinuity/continuity.go new file mode 100644 index 00000000..abefc137 --- /dev/null +++ b/internal/migrationauditcontinuity/continuity.go @@ -0,0 +1,90 @@ +// Package migrationauditcontinuity implementiert VAL-04: den +// Kontinuitaetsabgleich des Audit-Logs je Tenant nach COR-05 +// (internal/migrationaudit) — aeltester/neuester Eintrag und Gesamtzahl +// zwischen legacy_audit_log_staging (Quelle) und audit_events (Ziel) +// duerfen weder eine Luecke noch eine Zeitverschiebung zeigen. +package migrationauditcontinuity + +import ( + "context" + "fmt" + "time" +) + +// AuditStats ist Anzahl sowie aeltester/neuester Zeitstempel der +// Audit-Eintraege eines Tenants auf einer Seite (Quelle oder Ziel). +type AuditStats struct { + TenantSlug string + Count int64 + Oldest time.Time + Newest time.Time +} + +// StatsProvider liefert AuditStats fuer einen Tenant. +type StatsProvider interface { + Stats(ctx context.Context, tenantSlug string) (AuditStats, error) +} + +// ContinuityReport ist der Kontinuitaetsbericht eines Tenants +// (Akzeptanzkriterium 1). +type ContinuityReport struct { + TenantSlug string + SourceCount int64 + TargetCount int64 + SourceOldest time.Time + SourceNewest time.Time + TargetOldest time.Time + TargetNewest time.Time +} + +// Luecke ist true, wenn die Eintragszahl zwischen Quelle und Ziel abweicht. +func (r ContinuityReport) Luecke() bool { + return r.SourceCount != r.TargetCount +} + +// Zeitverschiebung ist true, wenn aeltester oder neuester Eintrag im Ziel +// nicht exakt mit der Quelle uebereinstimmt. +func (r ContinuityReport) Zeitverschiebung() bool { + return !r.SourceOldest.Equal(r.TargetOldest) || !r.SourceNewest.Equal(r.TargetNewest) +} + +// OffenerBefund ist true bei Luecke oder Zeitverschiebung +// (Akzeptanzkriterium 2: jede erkannte Abweichung ist blockierend, keine +// Erklaerungs-Ausnahme wie bei VAL-01). +func (r ContinuityReport) OffenerBefund() bool { + return r.Luecke() || r.Zeitverschiebung() +} + +func (r ContinuityReport) String() string { + status := "OK" + if r.OffenerBefund() { + status = "OFFENER BEFUND" + } + return fmt.Sprintf("%s: quelle(n=%d, aeltester=%s, neuester=%s) ziel(n=%d, aeltester=%s, neuester=%s) status=%s", + r.TenantSlug, r.SourceCount, r.SourceOldest, r.SourceNewest, r.TargetCount, r.TargetOldest, r.TargetNewest, status) +} + +// Run vergleicht Quelle und Ziel je Tenant (Akzeptanzkriterium 1). +func Run(ctx context.Context, tenantSlugs []string, source, target StatsProvider) ([]ContinuityReport, error) { + var reports []ContinuityReport + for _, slug := range tenantSlugs { + sc, err := source.Stats(ctx, slug) + if err != nil { + return nil, fmt.Errorf("migrationauditcontinuity: quellstatistik fuer tenant %q: %w", slug, err) + } + tc, err := target.Stats(ctx, slug) + if err != nil { + return nil, fmt.Errorf("migrationauditcontinuity: zielstatistik fuer tenant %q: %w", slug, err) + } + reports = append(reports, ContinuityReport{ + TenantSlug: slug, + SourceCount: sc.Count, + TargetCount: tc.Count, + SourceOldest: sc.Oldest, + SourceNewest: sc.Newest, + TargetOldest: tc.Oldest, + TargetNewest: tc.Newest, + }) + } + return reports, nil +} diff --git a/internal/migrationauditcontinuity/continuity_test.go b/internal/migrationauditcontinuity/continuity_test.go new file mode 100644 index 00000000..31fd5723 --- /dev/null +++ b/internal/migrationauditcontinuity/continuity_test.go @@ -0,0 +1,329 @@ +package migrationauditcontinuity + +import ( + "context" + "crypto/rand" + "fmt" + "os" + "strings" + "testing" + "time" + + "github.com/jackc/pgx/v5/pgxpool" + + "gitea.perlbach24.de/scripte/nexarch/internal/migrationaudit" + "gitea.perlbach24.de/scripte/nexarch/internal/testdb" +) + +// Integrationstest gegen eine echte Postgres-Instanz (Host 204, +// TEST_ADMIN_DSN) mit synthetischen Audit-Log-Fixtures — keine echten +// archivmail-/archivdms-Daten. + +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() + ); + + 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 +} + +type tenantFixture struct { + slug string + tenantID string + userID string +} + +func makeTenant(t *testing.T, ctx context.Context, pool *pgxpool.Pool, name string, sourceTenantID int64) tenantFixture { + t.Helper() + slug := testdb.Slug(name) + tenantUUID := newV4(t) + userUUID := newV4(t) + + 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 ($1, NULL, $2, $3, $4, 'qa-tester', now()) + `, sourceTenantID, tenantUUID, slug, name); 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 ($1, NULL, $2, $3, $4, 'Synth', 'x', 'qa-tester', now()) + `, sourceTenantID+9000, tenantUUID, userUUID, "synth."+slug+"@example.invalid"); err != nil { + t.Fatalf("user-map anlegen: %v", err) + } + return tenantFixture{slug: slug, tenantID: tenantUUID, userID: userUUID} +} + +func newV4(t *testing.T) string { + t.Helper() + b := make([]byte, 16) + if _, err := rand.Read(b); err != nil { + t.Fatalf("uuid erzeugen: %v", err) + } + b[6] = (b[6] & 0x0f) | 0x40 + b[8] = (b[8] & 0x3f) | 0x80 + return fmt.Sprintf("%x-%x-%x-%x-%x", b[0:4], b[4:6], b[6:8], b[8:10], b[10:16]) +} + +func insertStagingEntry(t *testing.T, ctx context.Context, pool *pgxpool.Pool, sourceTenantID, sourceID int64, occurredAt time.Time) { + t.Helper() + 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', $1, $2, $3, $4, 'mail.read', 'mail/1', '{}') + `, sourceID, sourceTenantID, sourceTenantID+9000, occurredAt); err != nil { + t.Fatalf("staging-zeile anlegen: %v", err) + } +} + +// TestKontinuitaetspruefung_DreiTenantsMitEntferntemEintrag deckt beide +// Pruefungen vor Abnahme ab: aeltester/neuester Eintrag fuer 3 +// Test-Tenants exakt mit der Quelle abgeglichen, und ein absichtlich aus +// dem Ziel entfernter Testeintrag wird als Luecke erkannt. +func TestKontinuitaetspruefung_DreiTenantsMitEntferntemEintrag(t *testing.T) { + dsn := newTestDSN(t) + ctx := context.Background() + pool := testdb.Pool(t, dsn) + setupSchema(t, ctx, pool) + + tenantA := makeTenant(t, ctx, pool, "mandant-a", 7001) + tenantB := makeTenant(t, ctx, pool, "mandant-b", 7002) + tenantC := makeTenant(t, ctx, pool, "mandant-c", 7003) + + base := time.Date(2018, 1, 1, 0, 0, 0, 0, time.UTC) + entriesPerTenant := map[string][]time.Time{ + tenantA.slug: {base, base.AddDate(0, 1, 0), base.AddDate(0, 6, 0)}, + tenantB.slug: {base.AddDate(1, 0, 0), base.AddDate(2, 0, 0)}, + tenantC.slug: {base.AddDate(0, 0, 10), base.AddDate(0, 0, 20), base.AddDate(0, 0, 30)}, + } + sourceID := int64(1) + sourceTenantIDBySlug := map[string]int64{tenantA.slug: 7001, tenantB.slug: 7002, tenantC.slug: 7003} + + for slug, timestamps := range entriesPerTenant { + for _, ts := range timestamps { + insertStagingEntry(t, ctx, pool, sourceTenantIDBySlug[slug], sourceID, ts) + sourceID++ + } + } + + migrator := migrationaudit.NewMigrator(pool) + if _, err := migrator.MigrateAll(ctx); err != nil { + t.Fatalf("MigrateAll: %v", err) + } + + // Absichtlich entfernter Testeintrag (Pruefung 2): der mittlere + // Eintrag von mandant-c wird nachtraeglich aus dem Ziel geloescht, um + // eine Luecke ohne Zeitverschiebung an Rand-Zeitstempeln zu simulieren. + // Da audit_events append-only ist (COR-05), wird die Loeschung ueber + // die Staging-Tabelle und einen frischen Schreibzugriff ohne Trigger + // nachgebildet: wir loeschen direkt per SQL unter Umgehung der + // Anwendungsschicht, wie es ein realer Datenverlust taete. + if _, err := pool.Exec(ctx, `ALTER TABLE audit_events DISABLE TRIGGER audit_events_no_delete`); err != nil { + t.Fatalf("trigger deaktivieren: %v", err) + } + if _, err := pool.Exec(ctx, ` + UPDATE legacy_audit_log_staging + SET migrated_event_id = NULL + WHERE migrated_event_id = ( + SELECT id FROM audit_events WHERE tenant_slug = $1 AND occurred_at = $2 + ) + `, tenantC.slug, entriesPerTenant[tenantC.slug][1]); err != nil { + t.Fatalf("staging-referenz loesen: %v", err) + } + if _, err := pool.Exec(ctx, ` + DELETE FROM audit_events + WHERE tenant_slug = $1 AND occurred_at = $2 + `, tenantC.slug, entriesPerTenant[tenantC.slug][1]); err != nil { + t.Fatalf("testeintrag entfernen: %v", err) + } + if _, err := pool.Exec(ctx, `ALTER TABLE audit_events ENABLE TRIGGER audit_events_no_delete`); err != nil { + t.Fatalf("trigger reaktivieren: %v", err) + } + + source := PostgresSourceStatsProvider{RegistryPool: pool} + target := PostgresTargetStatsProvider{AuditPool: pool} + + slugs := []string{tenantA.slug, tenantB.slug, tenantC.slug} + reports, err := Run(ctx, slugs, source, target) + if err != nil { + t.Fatalf("Run: %v", err) + } + if len(reports) != 3 { + t.Fatalf("erwarte 3 Reports, habe %d", len(reports)) + } + + byTenant := map[string]ContinuityReport{} + for _, r := range reports { + byTenant[r.TenantSlug] = r + } + + for _, fx := range []tenantFixture{tenantA, tenantB} { + r := byTenant[fx.slug] + timestamps := entriesPerTenant[fx.slug] + oldest, newest := timestamps[0], timestamps[0] + for _, ts := range timestamps { + if ts.Before(oldest) { + oldest = ts + } + if ts.After(newest) { + newest = ts + } + } + if !r.SourceOldest.Equal(oldest) || !r.TargetOldest.Equal(oldest) { + t.Fatalf("%s: aeltester eintrag weicht ab: %+v", fx.slug, r) + } + if !r.SourceNewest.Equal(newest) || !r.TargetNewest.Equal(newest) { + t.Fatalf("%s: neuester eintrag weicht ab: %+v", fx.slug, r) + } + if r.OffenerBefund() { + t.Fatalf("%s sollte ohne Abweichung OK sein, habe %+v", fx.slug, r) + } + } + + cReport := byTenant[tenantC.slug] + if !cReport.Luecke() { + t.Fatalf("absichtlich entfernter Testeintrag bei %s wurde nicht als Luecke erkannt: %+v", tenantC.slug, cReport) + } + if cReport.SourceCount != 3 || cReport.TargetCount != 2 { + t.Fatalf("%s: erwarte quelle=3 ziel=2, habe quelle=%d ziel=%d", tenantC.slug, cReport.SourceCount, cReport.TargetCount) + } + if !cReport.OffenerBefund() { + t.Fatalf("Luecke bei %s haette als offener Befund gelten muessen", tenantC.slug) + } + + bericht := RenderBericht(reports) + if !strings.Contains(bericht, "OFFENER BEFUND") { + t.Fatal("bericht sollte die Luecke als OFFENER BEFUND ausweisen") + } + befunde := OffeneBefunde(reports) + if len(befunde) != 1 || befunde[0].TenantSlug != tenantC.slug { + t.Fatalf("erwarte genau einen offenen Befund (%s), habe %+v", tenantC.slug, befunde) + } +} + +// TestKontinuitaetspruefung_Zeitverschiebung prueft, dass eine +// Zeitverschiebung bei unveraenderter Anzahl ebenfalls als blockierender +// Befund erkannt wird (Akzeptanzkriterium 2: Zeitverschiebung, nicht nur +// Luecke). +func TestKontinuitaetspruefung_Zeitverschiebung(t *testing.T) { + dsn := newTestDSN(t) + ctx := context.Background() + pool := testdb.Pool(t, dsn) + setupSchema(t, ctx, pool) + + tenant := makeTenant(t, ctx, pool, "mandant-shift", 8001) + ts := time.Date(2021, 5, 1, 12, 0, 0, 0, time.UTC) + insertStagingEntry(t, ctx, pool, 8001, 1, ts) + + migrator := migrationaudit.NewMigrator(pool) + if _, err := migrator.MigrateAll(ctx); err != nil { + t.Fatalf("MigrateAll: %v", err) + } + + // Zeitverschiebung simulieren: direkte SQL-Manipulation des + // Zeitstempels im Ziel, wie sie durch einen fehlerhaften + // Zeitzonen-Import entstehen koennte. + if _, err := pool.Exec(ctx, `ALTER TABLE audit_events DISABLE TRIGGER audit_events_no_update`); err != nil { + t.Fatalf("trigger deaktivieren: %v", err) + } + if _, err := pool.Exec(ctx, `UPDATE audit_events SET occurred_at = occurred_at + interval '1 hour' WHERE tenant_slug = $1`, tenant.slug); err != nil { + t.Fatalf("zeitverschiebung simulieren: %v", err) + } + if _, err := pool.Exec(ctx, `ALTER TABLE audit_events ENABLE TRIGGER audit_events_no_update`); err != nil { + t.Fatalf("trigger reaktivieren: %v", err) + } + + source := PostgresSourceStatsProvider{RegistryPool: pool} + target := PostgresTargetStatsProvider{AuditPool: pool} + reports, err := Run(ctx, []string{tenant.slug}, source, target) + if err != nil { + t.Fatalf("Run: %v", err) + } + if len(reports) != 1 { + t.Fatalf("erwarte 1 Report, habe %d", len(reports)) + } + r := reports[0] + if r.Luecke() { + t.Fatalf("erwarte keine Luecke, habe quelle=%d ziel=%d", r.SourceCount, r.TargetCount) + } + if !r.Zeitverschiebung() { + t.Fatalf("Zeitverschiebung haette erkannt werden muessen: %+v", r) + } + if !r.OffenerBefund() { + t.Fatal("Zeitverschiebung haette als offener Befund gelten muessen") + } +} diff --git a/internal/migrationauditcontinuity/postgres.go b/internal/migrationauditcontinuity/postgres.go new file mode 100644 index 00000000..f08a46d7 --- /dev/null +++ b/internal/migrationauditcontinuity/postgres.go @@ -0,0 +1,66 @@ +package migrationauditcontinuity + +import ( + "context" + "fmt" + "time" + + "github.com/jackc/pgx/v5/pgxpool" +) + +// PostgresSourceStatsProvider liest Anzahl/aeltesten/neuesten Zeitstempel +// aus legacy_audit_log_staging, ueber tenant_migration_map auf den +// Tenant-Slug aufgeloest — Quelle beider Altsysteme in einer Abfrage, da ein +// Tenant genau eine der beiden source_tenant_id-Spalten belegt +// (CHECK tenant_migration_map_source_present in COR-01/COR-02). +type PostgresSourceStatsProvider struct { + RegistryPool *pgxpool.Pool +} + +func (p PostgresSourceStatsProvider) Stats(ctx context.Context, tenantSlug string) (AuditStats, error) { + var count int64 + var oldest, newest *time.Time + err := p.RegistryPool.QueryRow(ctx, ` + SELECT count(l.*), min(l.occurred_at), max(l.occurred_at) + FROM legacy_audit_log_staging l + JOIN tenant_migration_map t + ON (l.source_system = 'archivmail' AND t.archivmail_tenant_id = l.source_tenant_id) + OR (l.source_system = 'archivdms' AND t.archivdms_tenant_id = l.source_tenant_id) + WHERE t.slug = $1 + `, tenantSlug).Scan(&count, &oldest, &newest) + if err != nil { + return AuditStats{}, fmt.Errorf("migrationauditcontinuity: quellstatistik fuer tenant %q lesen: %w", tenantSlug, err) + } + return toStats(tenantSlug, count, oldest, newest), nil +} + +// PostgresTargetStatsProvider liest Anzahl/aeltesten/neuesten Zeitstempel +// direkt aus audit_events (COR-05-Ziel) je Tenant-Slug. +type PostgresTargetStatsProvider struct { + AuditPool *pgxpool.Pool +} + +func (p PostgresTargetStatsProvider) Stats(ctx context.Context, tenantSlug string) (AuditStats, error) { + var count int64 + var oldest, newest *time.Time + err := p.AuditPool.QueryRow(ctx, ` + SELECT count(*), min(occurred_at), max(occurred_at) + FROM audit_events + WHERE tenant_slug = $1 + `, tenantSlug).Scan(&count, &oldest, &newest) + if err != nil { + return AuditStats{}, fmt.Errorf("migrationauditcontinuity: zielstatistik fuer tenant %q lesen: %w", tenantSlug, err) + } + return toStats(tenantSlug, count, oldest, newest), nil +} + +func toStats(tenantSlug string, count int64, oldest, newest *time.Time) AuditStats { + stats := AuditStats{TenantSlug: tenantSlug, Count: count} + if oldest != nil { + stats.Oldest = *oldest + } + if newest != nil { + stats.Newest = *newest + } + return stats +} diff --git a/internal/migrationauditcontinuity/render.go b/internal/migrationauditcontinuity/render.go new file mode 100644 index 00000000..588f9a78 --- /dev/null +++ b/internal/migrationauditcontinuity/render.go @@ -0,0 +1,44 @@ +package migrationauditcontinuity + +import ( + "fmt" + "sort" + "strings" +) + +// RenderBericht erzeugt den deutschsprachigen Kontinuitaetsbericht je +// Tenant als Markdown-Tabelle, sortiert nach Tenant. +func RenderBericht(reports []ContinuityReport) string { + sorted := append([]ContinuityReport(nil), reports...) + sort.Slice(sorted, func(i, j int) bool { return sorted[i].TenantSlug < sorted[j].TenantSlug }) + + var b strings.Builder + b.WriteString("# Audit-Log-Kontinuitaetspruefung je Tenant\n\n") + b.WriteString("| Tenant | Anzahl Quelle | Anzahl Ziel | Aeltester (Quelle) | Aeltester (Ziel) | Neuester (Quelle) | Neuester (Ziel) | Status |\n") + b.WriteString("|---|---|---|---|---|---|---|---|\n") + for _, r := range sorted { + status := "OK" + if r.OffenerBefund() { + status = "OFFENER BEFUND" + } + fmt.Fprintf(&b, "| %s | %d | %d | %s | %s | %s | %s | %s |\n", + r.TenantSlug, r.SourceCount, r.TargetCount, + r.SourceOldest.Format(timeLayout), r.TargetOldest.Format(timeLayout), + r.SourceNewest.Format(timeLayout), r.TargetNewest.Format(timeLayout), + status) + } + return b.String() +} + +const timeLayout = "2006-01-02T15:04:05Z07:00" + +// OffeneBefunde filtert alle Reports mit Luecke oder Zeitverschiebung. +func OffeneBefunde(reports []ContinuityReport) []ContinuityReport { + var out []ContinuityReport + for _, r := range reports { + if r.OffenerBefund() { + out = append(out, r) + } + } + return out +}