nexarch/workflow/internal/gates/engine/qa02_test.go

// Package engine ist das Pruefgate fuer QA-02: Zustandsautomat (ENG-01),
// Prozessdefinitions-Format (ENG-03), Modul-Adapter-Schnittstelle (ENG-02)
// und Fehlerbehandlung/Retry (ENG-05) werden hier nicht isoliert, sondern
// im Zusammenspiel gegen eine echte, frisch migrierte Testdatenbank
// gefahren.
package engine

import (
	"context"
	"errors"
	"os"
	"path/filepath"
	"sync"
	"testing"
	"time"

	"github.com/jackc/pgx/v5/pgxpool"

	"gitea.perlbach24.de/scripte/nexarch/workflow/internal/migrate"
	"gitea.perlbach24.de/scripte/nexarch/workflow/internal/moduleadapter"
	"gitea.perlbach24.de/scripte/nexarch/workflow/internal/processdef"
	"gitea.perlbach24.de/scripte/nexarch/workflow/internal/statemachine"
	"gitea.perlbach24.de/scripte/nexarch/workflow/internal/stepretry"
	"gitea.perlbach24.de/scripte/nexarch/workflow/internal/testdb"
)

func moduleRoot(t *testing.T) string {
	t.Helper()
	dir, err := os.Getwd()
	if err != nil {
		t.Fatalf("getwd: %v", err)
	}
	for {
		if _, err := os.Stat(filepath.Join(dir, "go.mod")); err == nil {
			return dir
		}
		parent := filepath.Dir(dir)
		if parent == dir {
			t.Fatal("go.mod nicht gefunden")
		}
		dir = parent
	}
}

func newPool(t *testing.T) *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)
	migrations, err := migrate.Load(filepath.Join(moduleRoot(t), "migrations", "tenant"))
	if err != nil {
		t.Fatalf("migrationen laden: %v", err)
	}
	if _, err := migrate.Up(ctx, pool, migrations); err != nil {
		t.Fatalf("migrationen anwenden: %v", err)
	}
	return pool
}

// rechnungsfreigabeJSON ist eine rein fiktive Prozessdefinition (ENG-03
// Format) mit 6 Zustaenden und 7 Uebergaengen, darunter ein Zyklus
// (nacharbeit<->pruefung) und zwei verschiedene Abbruchpfade — genug
// Struktur, um jeden Uebergang ueber mehrere Instanzen hinweg mindestens
// einmal real zu durchlaufen (Akzeptanzkriterium 1).
const rechnungsfreigabeJSON = `{
	"name": "rechnungsfreigabe-qa02",
	"states": ["antrag", "pruefung", "nacharbeit", "freigegeben", "abgelehnt", "storniert"],
	"initial": "antrag",
	"transitions": [
		{"from": "antrag", "to": "pruefung"},
		{"from": "pruefung", "to": "freigegeben"},
		{"from": "pruefung", "to": "nacharbeit"},
		{"from": "nacharbeit", "to": "pruefung"},
		{"from": "pruefung", "to": "abgelehnt"},
		{"from": "antrag", "to": "storniert"},
		{"from": "nacharbeit", "to": "storniert"}
	]
}`

type transitionKey struct{ from, to string }

// AK1 + Pruefung 1: alle sieben deklarierten Uebergaenge werden ueber vier
// Instanzen hinweg real durchlaufen, und ein unzulaessiger Uebergang wird
// abgewiesen.
func TestZustandsautomat_AlleUebergaengeDurchlaufenUndUnzulaessigerUebergangAbgewiesen(t *testing.T) {
	pool := newPool(t)
	ctx := context.Background()

	registry := processdef.NewRegistry(pool)
	def, _, err := registry.Load(ctx, []byte(rechnungsfreigabeJSON))
	if err != nil {
		t.Fatalf("prozessdefinition laden: %v", err)
	}
	engine, err := registry.BuildEngine(ctx)
	if err != nil {
		t.Fatalf("engine bauen: %v", err)
	}

	visited := map[transitionKey]bool{}
	walk := func(actor string, steps ...string) string {
		inst, err := registry.Start(ctx, engine, def.Name, actor)
		if err != nil {
			t.Fatalf("instanz starten: %v", err)
		}
		from := inst.State
		for _, to := range steps {
			if err := engine.Transition(ctx, inst.ID, to, actor, "qa02"); err != nil {
				t.Fatalf("uebergang %s->%s: %v", from, to, err)
			}
			visited[transitionKey{from, to}] = true
			from = to
		}
		return inst.ID
	}

	walk("alice", "pruefung", "freigegeben")
	walk("bob", "pruefung", "nacharbeit", "pruefung", "abgelehnt")
	walk("carol", "storniert")
	walk("dave", "pruefung", "nacharbeit", "storniert")

	for _, tr := range def.Transitions {
		if !visited[transitionKey{tr.From, tr.To}] {
			t.Errorf("uebergang %s->%s wurde in keiner Instanz durchlaufen", tr.From, tr.To)
		}
	}

	// Pruefung 1: unzulaessiger Uebergang (antrag -> freigegeben ist nicht
	// deklariert) wird abgewiesen, der Zustand bleibt unveraendert.
	rejectedID := walk("erin")
	if err := engine.Transition(ctx, rejectedID, "freigegeben", "erin", "verbotener versuch"); !errors.Is(err, statemachine.ErrIllegalTransition) {
		t.Fatalf("erwartet ErrIllegalTransition, habe %v", err)
	}
	got, err := engine.Get(ctx, rejectedID)
	if err != nil {
		t.Fatalf("get: %v", err)
	}
	if got.State != "antrag" {
		t.Fatalf("zustand nach abgewiesenem uebergang = %q, will unveraendert antrag", got.State)
	}
}

// AK2 + Pruefung 2: zwei fiktive Modul-Adapter werden gleichzeitig
// registriert und unabhaengig voneinander geprueft.
func TestAdapterRegistrierung_ZweiFiktiveModuleGleichzeitig(t *testing.T) {
	pool := newPool(t)
	ctx := context.Background()

	type result struct {
		reg moduleadapter.Registration
		err error
	}
	results := make([]result, 2)
	var wg sync.WaitGroup
	wg.Add(2)
	go func() {
		defer wg.Done()
		reg, err := moduleadapter.Register(ctx, pool, "fiktivmodul-a", "rechnung", []string{"rechnung.eingereicht"})
		results[0] = result{reg, err}
	}()
	go func() {
		defer wg.Done()
		reg, err := moduleadapter.Register(ctx, pool, "fiktivmodul-b", "urlaubsantrag", []string{"antrag.gestellt"})
		results[1] = result{reg, err}
	}()
	wg.Wait()

	for i, r := range results {
		if r.err != nil {
			t.Fatalf("registrierung %d: %v", i, r.err)
		}
		if r.reg.ActiveVersion != 1 {
			t.Fatalf("registrierung %d: version = %d, will 1", i, r.reg.ActiveVersion)
		}
	}

	regs, err := moduleadapter.ListRegistrations(ctx, pool)
	if err != nil {
		t.Fatalf("auflisten: %v", err)
	}
	if len(regs) != 2 {
		t.Fatalf("erwarte 2 registrierungen, habe %d: %+v", len(regs), regs)
	}

	triggersA, err := moduleadapter.TriggersFor(ctx, pool, "fiktivmodul-a", "rechnung")
	if err != nil {
		t.Fatalf("trigger fiktivmodul-a: %v", err)
	}
	if len(triggersA) != 1 || triggersA[0] != "rechnung.eingereicht" {
		t.Fatalf("trigger fiktivmodul-a = %v, will [rechnung.eingereicht]", triggersA)
	}

	triggersB, err := moduleadapter.TriggersFor(ctx, pool, "fiktivmodul-b", "urlaubsantrag")
	if err != nil {
		t.Fatalf("trigger fiktivmodul-b: %v", err)
	}
	if len(triggersB) != 1 || triggersB[0] != "antrag.gestellt" {
		t.Fatalf("trigger fiktivmodul-b = %v, will [antrag.gestellt]", triggersB)
	}
}

// AK3 + Pruefung 3: ein simulierter Fehlerfall loest Retry mit Backoff aus,
// gebunden an eine echte Workflow-Instanz; nach erschoepften Versuchen
// landet der Schritt im Dead-Letter und ist manuell reaktivierbar.
func TestFehlerbehandlungUndRetry_SimulierterFehlerLoestRetryAus(t *testing.T) {
	pool := newPool(t)
	ctx := context.Background()

	def := statemachine.ProcessDefinition{
		Name:    "benachrichtigung-qa02",
		States:  []string{"offen", "erledigt"},
		Initial: "offen",
		Transitions: []statemachine.Transition{
			{From: "offen", To: "erledigt"},
		},
	}
	engine, err := statemachine.NewEngine(pool, def)
	if err != nil {
		t.Fatalf("engine anlegen: %v", err)
	}
	inst, err := engine.Start(ctx, "benachrichtigung-qa02", "qa02")
	if err != nil {
		t.Fatalf("instanz starten: %v", err)
	}

	store := stepretry.NewStore(pool)
	calls := 0
	fn := func(context.Context) error {
		calls++
		if calls < 3 {
			return errors.New("simulierter dienstausfall")
		}
		return nil
	}

	step, err := store.Schedule(ctx, inst.ID, "benachrichtigung-versand", 5)
	if err != nil {
		t.Fatalf("schedule: %v", err)
	}
	if err := stepretry.Execute(ctx, store, step, fn); err == nil {
		t.Fatal("erster versuch haette fehlschlagen sollen")
	}
	got, err := store.Get(ctx, step.ID)
	if err != nil {
		t.Fatalf("get nach versuch 1: %v", err)
	}
	if got.Status != stepretry.StatusPending || got.Attempts != 1 || !got.NextAttemptAt.After(time.Now()) {
		t.Fatalf("nach fehlgeschlagenem versuch: %+v", got)
	}

	// Zeitraffer statt echtem Warten: ClaimDue mit einem weit in der Zukunft
	// liegenden "jetzt" macht den Schritt sofort faellig.
	due, err := store.ClaimDue(ctx, time.Now().Add(10*time.Minute), 10)
	if err != nil {
		t.Fatalf("claim due: %v", err)
	}
	var claimed *stepretry.Step
	for i := range due {
		if due[i].ID == step.ID {
			claimed = &due[i]
		}
	}
	if claimed == nil {
		t.Fatal("schritt haette nach zeitraffer faellig sein muessen")
	}
	if err := stepretry.Execute(ctx, store, *claimed, fn); err == nil {
		t.Fatal("zweiter versuch haette ebenfalls fehlschlagen sollen")
	}
	if err := stepretry.Execute(ctx, store, stepretry.Step{ID: step.ID}, fn); err != nil {
		t.Fatalf("dritter versuch (sollte gelingen): %v", err)
	}
	got, err = store.Get(ctx, step.ID)
	if err != nil {
		t.Fatalf("get nach erfolg: %v", err)
	}
	if got.Status != stepretry.StatusSucceeded {
		t.Fatalf("status nach erfolg = %q, will succeeded", got.Status)
	}
	if calls != 3 {
		t.Fatalf("fn wurde %d mal aufgerufen, will 3", calls)
	}

	if err := engine.Transition(ctx, inst.ID, "erledigt", "qa02", "benachrichtigt"); err != nil {
		t.Fatalf("abschliessender uebergang nach erfolgreichem retry: %v", err)
	}

	// Dauerhaft fehlschlagender zweiter Schritt: Dead-Letter + Reaktivierung.
	deadStep, err := store.Schedule(ctx, inst.ID, "erp-meldung", 2)
	if err != nil {
		t.Fatalf("schedule dead-letter-schritt: %v", err)
	}
	alwaysFail := func(context.Context) error { return errors.New("erp dauerhaft nicht erreichbar") }
	for i := 0; i < 2; i++ {
		cur, err := store.Get(ctx, deadStep.ID)
		if err != nil {
			t.Fatalf("get vor versuch %d: %v", i+1, err)
		}
		if err := stepretry.Execute(ctx, store, cur, alwaysFail); err == nil {
			t.Fatalf("versuch %d haette fehlschlagen muessen", i+1)
		}
	}
	cur, err := store.Get(ctx, deadStep.ID)
	if err != nil {
		t.Fatalf("get nach erschoepften versuchen: %v", err)
	}
	if cur.Status != stepretry.StatusDeadLetter {
		t.Fatalf("status = %q, will dead_letter", cur.Status)
	}
	dead, err := store.ListDeadLetter(ctx)
	if err != nil {
		t.Fatalf("list dead letter: %v", err)
	}
	found := false
	for _, d := range dead {
		if d.ID == deadStep.ID {
			found = true
		}
	}
	if !found {
		t.Fatal("dead-letter-schritt nicht gefunden")
	}
	if err := store.Reactivate(ctx, deadStep.ID); err != nil {
		t.Fatalf("reactivate: %v", err)
	}
	cur, err = store.Get(ctx, deadStep.ID)
	if err != nil {
		t.Fatalf("get nach reactivate: %v", err)
	}
	if cur.Status != stepretry.StatusPending || cur.Attempts != 0 {
		t.Fatalf("nach reactivate: %+v, will pending/attempts=0", cur)
	}
}