From a445bfff562771664c165ec00e5f7936ceac7f3e Mon Sep 17 00:00:00 2001 From: patrick Date: Wed, 14 Oct 2026 15:12:29 +0000 Subject: [PATCH] WFL-01: workflow-historie-migration Uebernahme abgeschlossener Workflow-/Automatisierungsvorgaenge aus archivdms als Historie nach ENG-01 (workflow_instances/ workflow_instance_history). Laufende Instanzen werden nicht migriert, das ist der sicherheitskritische Teil der Kachel und per Test abgedeckt. Source/UserResolver-Interfaces nach dem Muster aus COR-02/ DOK-01, deterministische Instanz-IDs fuer Idempotenz, fail-closed bei unbekannten Akteuren. Co-Authored-By: Claude Sonnet 5 --- workflow/docs/WFL-01-PRUEFPROTOKOLL.md | 92 +++++++ .../internal/migrationworkflow/migrate.go | 196 +++++++++++++++ .../migrationworkflow/migrate_test.go | 234 ++++++++++++++++++ 3 files changed, 522 insertions(+) create mode 100644 workflow/docs/WFL-01-PRUEFPROTOKOLL.md create mode 100644 workflow/internal/migrationworkflow/migrate.go create mode 100644 workflow/internal/migrationworkflow/migrate_test.go diff --git a/workflow/docs/WFL-01-PRUEFPROTOKOLL.md b/workflow/docs/WFL-01-PRUEFPROTOKOLL.md new file mode 100644 index 00000000..1a5d7330 --- /dev/null +++ b/workflow/docs/WFL-01-PRUEFPROTOKOLL.md @@ -0,0 +1,92 @@ +# WFL-01 – Prüfprotokoll: Workflow-Historie-Migration + +## Umfang + +Übernahme abgeschlossener Workflow-/Automatisierungsvorgänge aus archivdms als +Historie nach ENG-01 (`workflow_instances`/`workflow_instance_history`). +Laufende Instanzen werden nicht migriert. Implementiert in +`workflow/internal/migrationworkflow` (Paket `migrationworkflow`). + +Quelle und Benutzerauflösung sind über Interfaces (`Source`, `UserResolver`) +abstrahiert, analog zum Muster aus COR-02 (`internal/migrationuser`) und +DOK-01 (`dms/internal/migrationdocument`): Tests laufen ausschließlich gegen +synthetische In-Memory-Fixtures, keine echten archivdms-Daten. + +## Geprüft auf Host 204 (Wegwerf-Cluster `coredev`, Port 5433, `TEST_TENANT_DSN` +über lokalen Unix-Socket mit Peer-Auth als Postgres-Systembenutzer, keine +Zugangsdaten im Code oder in Dateien hinterlegt) + +- `go build ./...`: sauber (BUILD_OK) +- `go vet ./...`: sauber (VET_OK) +- `gofmt -l internal/migrationworkflow/`: keine Ausgabe (sauber formatiert) +- `go test -count=1 ./...` (gesamtes Workflow-Modul, MIT `TEST_TENANT_DSN` + gesetzt, keine stillen Skips): alle Pakete `ok`, inklusive + `internal/migrationworkflow`. + +## Tests je Akzeptanzkriterium + +Paket: `workflow/internal/migrationworkflow`, Fixtures: 3 synthetische +Vorgänge (`archivdms-case-1` abgeschlossen/completed, `archivdms-case-2` +abgebrochen/aborted, `archivdms-case-3` laufend/running). + +1. **Jeder abgeschlossene Workflow-Vorgang ist nach der Migration als + Historieneintrag auffindbar** + `TestMigrateCompleted_AkzeptanzkriteriumAuffindbar` — PASS. + Ergebnis: 2 migrierte Instanzen, 5 Historieneinträge insgesamt; + Instanz 1 (`archivdms-case-1`) mit state=`closed`, version=4 (3 + Übergänge + 1), 3 Historieneinträge in `workflow_instance_history` + nachweisbar. + +2. **Keine laufende Workflow-Instanz wird in einem halb migrierten Zustand + übernommen** (sicherheitskritischer Teil der Kachel) + `TestMigrateCompleted_AkzeptanzkriteriumKeineLaufendeInstanz` — PASS. + Ergebnis: `archivdms-case-3` (Status running) erzeugt weder eine Zeile + in `workflow_instances` noch in `workflow_instance_history` — 0 Zeilen + in beiden Tabellen für diese Instanz, `InstancesSkippedOpen=1`. + +3. **Liste offener Instanzen zum Stichtag ist vor dem Cutover vollständig + abgearbeitet oder bewusst abgebrochen** + `TestOpenCases_AkzeptanzkriteriumListeOffenerInstanzen` — PASS. + `Migrator.OpenCases` liefert genau `archivdms-case-3` als einzige offene + Instanz — Grundlage für die operative Prüfung vor dem Stichtag. + +Zusätzliche Robustheitstests: + +- `TestMigrateCompleted_AkzeptanzkriteriumWiederholbar` — PASS. Ein + zweiter Lauf gegen dieselbe Quelle migriert 0 neue Instanzen + (`InstancesSkippedExisting=2`), keine Duplikate in beiden Tabellen + (deterministische Instanz-ID aus Quell-ID, `ON CONFLICT (id) DO + NOTHING`). +- `TestMigrateCompleted_FailClosedBeiUnbekanntemAkteur` — PASS. Ein nicht + auflösbarer Akteur in einem Historieneintrag bricht die Migration dieses + Vorgangs mit `ErrUnresolvedActor` ab statt ihn mit einem Platzhalter zu + übernehmen. + +## Prüfungen vor der Abnahme (Ticketvorgabe) + +1. **Zeilenzahl abgeschlossener Vorgänge nach Migration deckungsgleich mit + Quelle**: durch `TestMigrateCompleted_AkzeptanzkriteriumAuffindbar` und + `TestMigrateCompleted_AkzeptanzkriteriumWiederholbar` abgedeckt + (2 abgeschlossene Quell-Vorgänge → 2 Zeilen in `workflow_instances`, + stabil über mehrere Läufe). +2. **Zum geplanten Stichtag-Zeitpunkt sind keine offenen Instanzen in der + Processing-Queue des Altsystems mehr vorhanden**: operative Prüfung + gegen das echte Altsystem zum Stichtag selbst, nicht automatisierbar + gegen synthetische Fixtures. `Migrator.OpenCases` liefert das + Werkzeug dafür (siehe Akzeptanzkriterium 3); der tatsächliche Leerstand + der Altsystem-Queue ist zum Zeitpunkt des Cutovers operativ zu + bestätigen und hier nachzutragen. + +## Hinweise zu Scope-Grenzen + +- Benutzer-/Akteursauflösung nutzt `UserResolver` nach dem Muster aus + COR-02 (`user_migration_map`); die konkrete Anbindung an die Map ist + nicht Teil dieser Kachel, Tests verwenden einen statischen Resolver. +- Dokumentreferenzen in Workflow-Historieneinträgen sind laut Ticket + konditional ("falls... enthalten"); das bestehende ENG-01-Schema + (`workflow_instance_history`) trägt keine Dokumentreferenz-Spalte, ein + Schema-Umbau dafür wäre ein Eingriff außerhalb des Tickettextes ("kein + Umbau angrenzender Bereiche") und wurde nicht vorgenommen. +- Keine neue SQL-Migrationsdatei nötig: Die Migration schreibt direkt in + das bestehende ENG-01-Schema (`0001_workflow_instances.up.sql`), keine + Kollision mit der Migrationsnummern-Vergabe in `workflow/migrations/`. diff --git a/workflow/internal/migrationworkflow/migrate.go b/workflow/internal/migrationworkflow/migrate.go new file mode 100644 index 00000000..0523c993 --- /dev/null +++ b/workflow/internal/migrationworkflow/migrate.go @@ -0,0 +1,196 @@ +// Package migrationworkflow implementiert WFL-01: Uebernahme abgeschlossener +// Workflow-/Automatisierungsvorgaenge aus archivdms als Historie nach +// ENG-01 (workflow_instances/workflow_instance_history). Laufende Instanzen +// werden NICHT migriert, sie muessen vor dem Stichtag im Altsystem zu Ende +// gefuehrt oder abgebrochen werden. +package migrationworkflow + +import ( + "context" + "crypto/sha256" + "errors" + "fmt" + + "github.com/jackc/pgx/v5" + "github.com/jackc/pgx/v5/pgxpool" +) + +// Status ist der Abschlusszustand eines Vorgangs in der Quelle. +type Status string + +const ( + StatusCompleted Status = "completed" + StatusAborted Status = "aborted" + StatusRunning Status = "running" +) + +// done listet die Status-Werte, die migriert werden duerfen (Whitelist statt +// Blacklist: ein unbekannter/neuer Status-Wert wird NICHT migriert, siehe +// migrateDone). +var done = map[Status]bool{ + StatusCompleted: true, + StatusAborted: true, +} + +// SourceCase ist ein Vorgang (Workflow-/Automatisierungsinstanz) aus +// archivdms. +type SourceCase struct { + SourceCaseID string + ProcessName string + FinalState string + Status Status +} + +// SourceHistoryEntry ist ein Historieneintrag eines Vorgangs aus archivdms, +// in Reihenfolge ihres Auftretens. +type SourceHistoryEntry struct { + FromState string + ToState string + SourceActor string + Reason string + OccurredAtRFC3339 string +} + +// Source kapselt den lesenden Zugriff auf archivdms (SQL-Export). Eine +// echte Implementierung liest gegen archivdms, Tests verwenden +// ausschliesslich synthetische Fixtures. +type Source interface { + Cases(ctx context.Context) ([]SourceCase, error) + History(ctx context.Context, sourceCaseID string) ([]SourceHistoryEntry, error) +} + +// UserResolver bildet eine archivdms-Akteurs-ID auf die bereits migrierte +// NEXARCH-Benutzer-ID ab (siehe COR-02, user_migration_map). Benutzer-/ +// ACL-Migration selbst ist nicht Bestandteil von WFL-01. +type UserResolver interface { + ResolveUserID(ctx context.Context, sourceActorID string) (string, error) +} + +// ErrUnresolvedActor wird zurueckgegeben, wenn ein Akteur in einem +// Historieneintrag nicht auf eine NEXARCH-Benutzer-ID abgebildet werden kann +// (fail-closed statt stillem Platzhalter). +var ErrUnresolvedActor = errors.New("migrationworkflow: akteur nicht aufloesbar") + +// Migrator uebertraegt abgeschlossene Vorgaenge einer Quelle als Historie +// in Workflow ENG-01. +type Migrator struct { + pool *pgxpool.Pool + source Source + users UserResolver +} + +func NewMigrator(pool *pgxpool.Pool, source Source, users UserResolver) *Migrator { + return &Migrator{pool: pool, source: source, users: users} +} + +// Result fasst den Ausgang eines Migrationslaufs zusammen. +type Result struct { + InstancesMigrated int + InstancesSkippedExisting int + InstancesSkippedOpen int + HistoryEntriesMigrated int +} + +// instanceID leitet eine deterministische Ziel-ID aus der Quell-ID ab, damit +// ein erneuter Lauf gegen dieselbe Quelle dieselbe Zeile trifft (keine +// Duplikate, Akzeptanzkriterium 2/Pruefung 1). +func instanceID(sourceCaseID string) string { + h := sha256.Sum256([]byte("workflow_instance\x00" + sourceCaseID)) + h[6] = (h[6] & 0x0f) | 0x50 + h[8] = (h[8] & 0x3f) | 0x80 + return fmt.Sprintf("%x-%x-%x-%x-%x", h[0:4], h[4:6], h[6:8], h[8:10], h[10:16]) +} + +// MigrateCompleted uebertraegt jeden abgeschlossenen Vorgang (Status +// completed/aborted) samt vollstaendiger Historie (Akzeptanzkriterium 1). Ein +// laufender Vorgang (oder ein unbekannter, nicht in der Whitelist gefuehrter +// Status) wird NICHT angefasst (Akzeptanzkriterium 2, sicherheitskritischer +// Teil dieser Kachel). +func (m *Migrator) MigrateCompleted(ctx context.Context) (Result, error) { + var res Result + cases, err := m.source.Cases(ctx) + if err != nil { + return res, fmt.Errorf("migrationworkflow: vorgaenge aus quelle lesen: %w", err) + } + + for _, c := range cases { + if !done[c.Status] { + res.InstancesSkippedOpen++ + continue + } + if err := m.migrateCase(ctx, c, &res); err != nil { + return res, err + } + } + return res, nil +} + +func (m *Migrator) migrateCase(ctx context.Context, c SourceCase, res *Result) error { + id := instanceID(c.SourceCaseID) + + var alreadyMigrated bool + if err := m.pool.QueryRow(ctx, `SELECT true FROM workflow_instances WHERE id = $1`, id).Scan(&alreadyMigrated); err != nil && !errors.Is(err, pgx.ErrNoRows) { + return fmt.Errorf("migrationworkflow: bestand von instanz %q pruefen: %w", c.SourceCaseID, err) + } + if alreadyMigrated { + res.InstancesSkippedExisting++ + return nil + } + + history, err := m.source.History(ctx, c.SourceCaseID) + if err != nil { + return fmt.Errorf("migrationworkflow: historie fuer vorgang %q lesen: %w", c.SourceCaseID, err) + } + + resolved := make([]struct { + entry SourceHistoryEntry + actor string + }, len(history)) + for i, h := range history { + actorID, err := m.users.ResolveUserID(ctx, h.SourceActor) + if err != nil { + return fmt.Errorf("%w: vorgang %q, akteur %q: %v", ErrUnresolvedActor, c.SourceCaseID, h.SourceActor, err) + } + resolved[i].entry = h + resolved[i].actor = actorID + } + + version := len(history) + 1 + if _, err := m.pool.Exec(ctx, ` + INSERT INTO workflow_instances (id, process_name, state, version) + VALUES ($1, $2, $3, $4) + ON CONFLICT (id) DO NOTHING + `, id, c.ProcessName, c.FinalState, version); err != nil { + return fmt.Errorf("migrationworkflow: instanz %q anlegen: %w", c.SourceCaseID, err) + } + res.InstancesMigrated++ + + for _, r := range resolved { + if _, err := m.pool.Exec(ctx, ` + INSERT INTO workflow_instance_history (instance_id, from_state, to_state, actor, reason, occurred_at) + VALUES ($1, $2, $3, $4, $5, $6) + `, id, r.entry.FromState, r.entry.ToState, r.actor, r.entry.Reason, r.entry.OccurredAtRFC3339); err != nil { + return fmt.Errorf("migrationworkflow: historieneintrag fuer instanz %q schreiben: %w", c.SourceCaseID, err) + } + res.HistoryEntriesMigrated++ + } + return nil +} + +// OpenCases liefert alle Vorgaenge der Quelle, die NICHT abgeschlossen sind +// (Status ausserhalb der Whitelist, insbesondere laufend). Grundlage fuer +// die Pruefung vor dem Cutover (Akzeptanzkriterium 3/Pruefung 2): diese +// Liste muss vor dem Stichtag leer sein oder bewusst abgearbeitet werden. +func (m *Migrator) OpenCases(ctx context.Context) ([]SourceCase, error) { + cases, err := m.source.Cases(ctx) + if err != nil { + return nil, fmt.Errorf("migrationworkflow: vorgaenge aus quelle lesen: %w", err) + } + var open []SourceCase + for _, c := range cases { + if !done[c.Status] { + open = append(open, c) + } + } + return open, nil +} diff --git a/workflow/internal/migrationworkflow/migrate_test.go b/workflow/internal/migrationworkflow/migrate_test.go new file mode 100644 index 00000000..4ebba1f5 --- /dev/null +++ b/workflow/internal/migrationworkflow/migrate_test.go @@ -0,0 +1,234 @@ +package migrationworkflow + +import ( + "context" + "errors" + "fmt" + "os" + "testing" + + "github.com/jackc/pgx/v5/pgxpool" + + "gitea.perlbach24.de/scripte/nexarch/workflow/internal/testdb" +) + +// memorySource ist eine In-Memory-Testquelle: simuliert den SQL-Export von +// archivdms rein synthetisch (DSGVO: keine echten Altsystem-Daten). +type memorySource struct { + cases []SourceCase + history map[string][]SourceHistoryEntry +} + +func (s *memorySource) Cases(ctx context.Context) ([]SourceCase, error) { + return s.cases, nil +} + +func (s *memorySource) History(ctx context.Context, sourceCaseID string) ([]SourceHistoryEntry, error) { + return s.history[sourceCaseID], nil +} + +// staticUserResolver bildet archivdms-Akteurs-IDs 1:1 auf eine feste +// Ziel-Benutzer-ID ab (Benutzermigration ist COR-02, nicht Teil von WFL-01). +type staticUserResolver struct{ known map[string]string } + +func (r staticUserResolver) ResolveUserID(ctx context.Context, sourceActorID string) (string, error) { + id, ok := r.known[sourceActorID] + if !ok { + return "", fmt.Errorf("unbekannter testakteur %q", sourceActorID) + } + return id, nil +} + +func newSyntheticSource() *memorySource { + return &memorySource{ + cases: []SourceCase{ + {SourceCaseID: "archivdms-case-1", ProcessName: "case", FinalState: "closed", Status: StatusCompleted}, + {SourceCaseID: "archivdms-case-2", ProcessName: "case", FinalState: "closed", Status: StatusAborted}, + {SourceCaseID: "archivdms-case-3", ProcessName: "case", FinalState: "in_progress", Status: StatusRunning}, + }, + history: map[string][]SourceHistoryEntry{ + "archivdms-case-1": { + {FromState: "", ToState: "open", SourceActor: "archivdms-user-1", Reason: "gestartet", OccurredAtRFC3339: "2025-01-10T09:00:00Z"}, + {FromState: "open", ToState: "in_progress", SourceActor: "archivdms-user-1", Reason: "bearbeitung begonnen", OccurredAtRFC3339: "2025-01-11T09:00:00Z"}, + {FromState: "in_progress", ToState: "closed", SourceActor: "archivdms-user-2", Reason: "abgeschlossen", OccurredAtRFC3339: "2025-01-12T09:00:00Z"}, + }, + "archivdms-case-2": { + {FromState: "", ToState: "open", SourceActor: "archivdms-user-1", Reason: "gestartet", OccurredAtRFC3339: "2025-02-01T09:00:00Z"}, + {FromState: "open", ToState: "closed", SourceActor: "archivdms-user-1", Reason: "abgebrochen", OccurredAtRFC3339: "2025-02-02T09:00:00Z"}, + }, + "archivdms-case-3": { + {FromState: "", ToState: "open", SourceActor: "archivdms-user-1", Reason: "gestartet", OccurredAtRFC3339: "2025-03-01T09:00:00Z"}, + {FromState: "open", ToState: "in_progress", SourceActor: "archivdms-user-1", Reason: "bearbeitung begonnen", OccurredAtRFC3339: "2025-03-02T09:00:00Z"}, + }, + }, + } +} + +func setupMigratorTest(t *testing.T) (*Migrator, *pgxpool.Pool) { + t.Helper() + dsn := os.Getenv("TEST_TENANT_DSN") + if dsn == "" { + t.Skip("TEST_TENANT_DSN nicht gesetzt, Integrationstest uebersprungen") + } + ctx := context.Background() + + pool := testdb.Pool(t, dsn) + + if _, err := pool.Exec(ctx, ` + CREATE EXTENSION IF NOT EXISTS pgcrypto; + CREATE TABLE IF NOT EXISTS workflow_instances ( + id UUID PRIMARY KEY DEFAULT gen_random_uuid(), + process_name TEXT NOT NULL, + state TEXT NOT NULL, + version INT NOT NULL DEFAULT 1, + created_at TIMESTAMPTZ NOT NULL DEFAULT now(), + updated_at TIMESTAMPTZ NOT NULL DEFAULT now() + ); + CREATE TABLE IF NOT EXISTS workflow_instance_history ( + id BIGSERIAL PRIMARY KEY, + instance_id UUID NOT NULL REFERENCES workflow_instances(id) ON DELETE CASCADE, + from_state TEXT NOT NULL, + to_state TEXT NOT NULL, + actor TEXT NOT NULL, + reason TEXT NOT NULL DEFAULT '', + occurred_at TIMESTAMPTZ NOT NULL DEFAULT now() + ); + `); err != nil { + t.Fatalf("schema: %v", err) + } + + resolver := staticUserResolver{known: map[string]string{ + "archivdms-user-1": "11111111-1111-1111-1111-111111111111", + "archivdms-user-2": "22222222-2222-2222-2222-222222222222", + }} + migrator := NewMigrator(pool, newSyntheticSource(), resolver) + return migrator, pool +} + +// TestMigrateCompleted_AkzeptanzkriteriumAuffindbar prueft Akzeptanzkriterium +// 1: jeder abgeschlossene Vorgang ist nach der Migration als Historieneintrag +// auffindbar, inklusive vollstaendiger Uebergangshistorie. +func TestMigrateCompleted_AkzeptanzkriteriumAuffindbar(t *testing.T) { + migrator, pool := setupMigratorTest(t) + ctx := context.Background() + + res, err := migrator.MigrateCompleted(ctx) + if err != nil { + t.Fatalf("MigrateCompleted: %v", err) + } + if res.InstancesMigrated != 2 || res.HistoryEntriesMigrated != 5 { + t.Fatalf("erwartet 2 instanzen/5 historieneintraege, habe %+v", res) + } + + id1 := instanceID("archivdms-case-1") + var state string + var version int + if err := pool.QueryRow(ctx, `SELECT state, version FROM workflow_instances WHERE id = $1`, id1).Scan(&state, &version); err != nil { + t.Fatalf("instanz 1 lesen: %v", err) + } + if state != "closed" || version != 4 { + t.Fatalf("erwartet state=closed version=4, habe state=%q version=%d", state, version) + } + + var historyCount int + if err := pool.QueryRow(ctx, `SELECT count(*) FROM workflow_instance_history WHERE instance_id = $1`, id1).Scan(&historyCount); err != nil { + t.Fatalf("historie zaehlen: %v", err) + } + if historyCount != 3 { + t.Fatalf("erwartet 3 historieneintraege fuer instanz 1, habe %d", historyCount) + } +} + +// TestMigrateCompleted_AkzeptanzkriteriumKeineLaufendeInstanz prueft den +// sicherheitskritischen Teil der Kachel (Akzeptanzkriterium 2): eine NICHT +// abgeschlossene/laufende Workflow-Instanz in der Quelle wird NICHT +// migriert, auch nicht halb (keine Zeile in workflow_instances). +func TestMigrateCompleted_AkzeptanzkriteriumKeineLaufendeInstanz(t *testing.T) { + migrator, pool := setupMigratorTest(t) + ctx := context.Background() + + res, err := migrator.MigrateCompleted(ctx) + if err != nil { + t.Fatalf("MigrateCompleted: %v", err) + } + if res.InstancesSkippedOpen != 1 { + t.Fatalf("erwartet 1 uebersprungene laufende instanz, habe %+v", res) + } + + id3 := instanceID("archivdms-case-3") + var exists bool + err = pool.QueryRow(ctx, `SELECT true FROM workflow_instances WHERE id = $1`, id3).Scan(&exists) + if err == nil { + t.Fatalf("laufende instanz archivdms-case-3 wurde migriert, darf nicht sein") + } + + var historyCount int + if err := pool.QueryRow(ctx, `SELECT count(*) FROM workflow_instance_history WHERE instance_id = $1`, id3).Scan(&historyCount); err != nil { + t.Fatalf("historie zaehlen: %v", err) + } + if historyCount != 0 { + t.Fatalf("laufende instanz hat historieneintraege, darf nicht sein: %d", historyCount) + } +} + +// TestOpenCases_AkzeptanzkriteriumListeOffenerInstanzen prueft +// Akzeptanzkriterium 3/Pruefung 2: die Liste offener Instanzen vor dem +// Cutover ist vollstaendig und enthaelt ausschliesslich nicht abgeschlossene +// Vorgaenge. +func TestOpenCases_AkzeptanzkriteriumListeOffenerInstanzen(t *testing.T) { + migrator, _ := setupMigratorTest(t) + ctx := context.Background() + + open, err := migrator.OpenCases(ctx) + if err != nil { + t.Fatalf("OpenCases: %v", err) + } + if len(open) != 1 || open[0].SourceCaseID != "archivdms-case-3" { + t.Fatalf("erwartet genau archivdms-case-3 als offene instanz, habe %+v", open) + } +} + +// TestMigrateCompleted_AkzeptanzkriteriumWiederholbar prueft, dass ein +// erneuter Lauf gegen dieselbe Quelle keine Duplikate erzeugt. +func TestMigrateCompleted_AkzeptanzkriteriumWiederholbar(t *testing.T) { + migrator, pool := setupMigratorTest(t) + ctx := context.Background() + + if _, err := migrator.MigrateCompleted(ctx); err != nil { + t.Fatalf("erster lauf: %v", err) + } + res2, err := migrator.MigrateCompleted(ctx) + if err != nil { + t.Fatalf("zweiter lauf: %v", err) + } + if res2.InstancesMigrated != 0 || res2.InstancesSkippedExisting != 2 { + t.Fatalf("zweiter lauf haette nichts neu migrieren duerfen, habe %+v", res2) + } + + var instanceCount, historyCount int + if err := pool.QueryRow(ctx, `SELECT count(*) FROM workflow_instances`).Scan(&instanceCount); err != nil { + t.Fatalf("instanzen zaehlen: %v", err) + } + if err := pool.QueryRow(ctx, `SELECT count(*) FROM workflow_instance_history`).Scan(&historyCount); err != nil { + t.Fatalf("historie zaehlen: %v", err) + } + if instanceCount != 2 || historyCount != 5 { + t.Fatalf("erneuter lauf hat duplikate erzeugt: %d instanzen/%d historieneintraege", instanceCount, historyCount) + } +} + +// TestMigrateCompleted_FailClosedBeiUnbekanntemAkteur prueft, dass ein +// nicht aufloesbarer Akteur die Migration dieses Vorgangs mit einem klaren +// Fehler abbricht statt ihn mit einem Platzhalter zu uebernehmen. +func TestMigrateCompleted_FailClosedBeiUnbekanntemAkteur(t *testing.T) { + migrator, _ := setupMigratorTest(t) + ctx := context.Background() + + src := migrator.source.(*memorySource) + src.history["archivdms-case-1"][0].SourceActor = "archivdms-user-unbekannt" + + _, err := migrator.MigrateCompleted(ctx) + if !errors.Is(err, ErrUnresolvedActor) { + t.Fatalf("erwartet ErrUnresolvedActor, habe %v", err) + } +} -- 2.47.3