nexarch/workflow/internal/loadtest/loadtest_test.go

// Package loadtest ist QA-07: Last- und Leistungstest fuer den
// Ereigniszustellungspfad (ENG-02), die Genehmigungs-Engine (APR) und
// das regelbasierte Routing (RTE) unter gleichzeitiger Last aus zwei
// simulierten Modulen (DMS, Mail). Kein Produktionscode -- reiner
// Lastgenerator mit Messung, siehe workflow/docs/QA-07-PRUEFPROTOKOLL.md
// fuer die gemessenen Werte und die Begruendung der Ziel-Latenz.
package loadtest

import (
	"context"
	"errors"
	"fmt"
	"math"
	"os"
	"path/filepath"
	"sort"
	"sync"
	"sync/atomic"
	"testing"
	"time"

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

	"gitea.perlbach24.de/scripte/nexarch/common/testdb"
	"gitea.perlbach24.de/scripte/nexarch/workflow/internal/approval"
	"gitea.perlbach24.de/scripte/nexarch/workflow/internal/eventtrigger"
	"gitea.perlbach24.de/scripte/nexarch/workflow/internal/migrate"
	"gitea.perlbach24.de/scripte/nexarch/workflow/internal/moduleadapter"
	"gitea.perlbach24.de/scripte/nexarch/workflow/internal/routingengine"
	"gitea.perlbach24.de/scripte/nexarch/workflow/internal/routingrule"
	"gitea.perlbach24.de/scripte/nexarch/workflow/internal/task"
)

const ownTenant = "mandant-last"

// targetP95DeliveryLatency: Dispatcher.Dispatch verarbeitet offene
// Ereignisse seriell in einer Goroutine (siehe eventtrigger.go), daher
// steigt die Latenz bei einem Burst von 600 gleichzeitigen Ereignissen
// mit der Position in der Warteschlange. Wert anhand gemessener
// Werte festgelegt, siehe QA-07-PRUEFPROTOKOLL.md.
const (
	targetP95DeliveryLatency = 25 * time.Second
	dispatchPollInterval     = 50 * time.Millisecond
)

func findModuleRoot(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
	}
}

type loadEnv struct {
	pool        *pgxpool.Pool
	rules       *eventtrigger.RuleStore
	dispatcher  *eventtrigger.Dispatcher
	approvalSvc *approval.Service
	tasks       *task.Store
	routeStore  *routingrule.Store
	routeEngine *routingengine.Engine
}

func newLoadEnv(t *testing.T) loadEnv {
	t.Helper()
	dsn := os.Getenv("TEST_TENANT_DSN")
	if dsn == "" {
		t.Skip("TEST_TENANT_DSN nicht gesetzt, Lasttest uebersprungen")
	}
	ctx := context.Background()

	pool, _ := testdb.PoolWithSchema(t, dsn, func(cfg *pgxpool.Config) {
		cfg.MaxConns = 40
		cfg.MinConns = 5
	})

	migrations, err := migrate.Load(filepath.Join(findModuleRoot(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)
	}

	eng, err := approval.NewEngineWithApproval(pool)
	if err != nil {
		t.Fatalf("engine anlegen: %v", err)
	}
	rules := eventtrigger.NewRuleStore(pool)
	tasks := task.NewStore(pool)
	approvalSvc := approval.NewService(pool, eng)
	dispatcher := eventtrigger.NewDispatcher(pool, rules, tasks, approvalSvc)
	routeStore := routingrule.NewStore(pool)
	routeEngine := routingengine.NewEngine(routeStore, pool)

	return loadEnv{
		pool: pool, rules: rules, dispatcher: dispatcher,
		approvalSvc: approvalSvc, tasks: tasks,
		routeStore: routeStore, routeEngine: routeEngine,
	}
}

// runBounded fuehrt n Aufrufe von fn parallel aus, begrenzt auf maxConcurrency
// gleichzeitige Goroutinen (Pool-Erschoepfung vermeiden, aber trotzdem
// echte Nebenlaeufigkeit simulieren).
func runBounded(n, maxConcurrency int, fn func(i int)) {
	sem := make(chan struct{}, maxConcurrency)
	var wg sync.WaitGroup
	for i := 0; i < n; i++ {
		wg.Add(1)
		sem <- struct{}{}
		go func(i int) {
			defer wg.Done()
			defer func() { <-sem }()
			fn(i)
		}(i)
	}
	wg.Wait()
}

func percentile(sorted []time.Duration, p float64) time.Duration {
	if len(sorted) == 0 {
		return 0
	}
	idx := int(math.Ceil(p/100*float64(len(sorted)))) - 1
	if idx < 0 {
		idx = 0
	}
	if idx >= len(sorted) {
		idx = len(sorted) - 1
	}
	return sorted[idx]
}

func summarizeLatencies(t *testing.T, label string, latencies []time.Duration, wall time.Duration) (p50, p95, p99, max time.Duration) {
	t.Helper()
	sorted := append([]time.Duration(nil), latencies...)
	sort.Slice(sorted, func(i, j int) bool { return sorted[i] < sorted[j] })
	p50 = percentile(sorted, 50)
	p95 = percentile(sorted, 95)
	p99 = percentile(sorted, 99)
	if len(sorted) > 0 {
		max = sorted[len(sorted)-1]
	}
	throughput := float64(len(sorted)) / wall.Seconds()
	t.Logf("%s: n=%d wall=%s p50=%s p95=%s p99=%s max=%s durchsatz=%.1f/s",
		label, len(sorted), wall, p50, p95, p99, max, throughput)
	return
}

// TestQA07_WorkflowLastUndLeistungstest ist der QA-07-Lasttest: realistische
// gleichzeitige Last aus zwei simulierten Modulen (DMS, Mail) gegen
// Modul-Adapter-Ereignispfad (ENG-02), Genehmigungs-Engine (APR) und
// Routing (RTE).
func TestQA07_WorkflowLastUndLeistungstest(t *testing.T) {
	e := newLoadEnv(t)
	ctx := context.Background()

	t.Run("ENG02_Ereigniszustellung_DMS_und_Mail_parallel", func(t *testing.T) {
		testEventDeliveryUnderLoad(t, ctx, e)
	})
	t.Run("APR_KeineDoppeltenOderVerlorenenFreigaben", func(t *testing.T) {
		testApprovalUnderLoad(t, ctx, e)
	})
	t.Run("RTE_RoutingDeterminismusUnterLast", func(t *testing.T) {
		testRoutingDeterminismUnderLoad(t, ctx, e)
	})
}

// testEventDeliveryUnderLoad ist Pruefung 1: Last mit simulierten
// parallelen Ereignissen aus DMS und Mail, Ziel-Latenz fuer ENG-02 geprueft.
func testEventDeliveryUnderLoad(t *testing.T, ctx context.Context, e loadEnv) {
	const (
		producersPerModule = 20
		eventsPerProducer  = 15
	)
	total := producersPerModule * eventsPerProducer * 2

	if _, err := e.rules.Put(ctx, eventtrigger.Rule{
		ModuleName: "dms-last", ObjectType: "document", Trigger: "document.hochgeladen",
		Kind: eventtrigger.KindTask, AssigneeType: task.AssigneePerson, AssigneeID: "sachbearbeiter-dms",
		Title: "Dokument pruefen",
	}); err != nil {
		t.Fatalf("regel dms anlegen: %v", err)
	}
	if _, err := e.rules.Put(ctx, eventtrigger.Rule{
		ModuleName: "mail-last", ObjectType: "message", Trigger: "mail.eingegangen",
		Kind: eventtrigger.KindTask, AssigneeType: task.AssigneePerson, AssigneeID: "sachbearbeiter-mail",
		Title: "E-Mail bearbeiten",
	}); err != nil {
		t.Fatalf("regel mail anlegen: %v", err)
	}

	stopDispatch := make(chan struct{})
	dispatchDone := make(chan struct{})
	var dispatchedTotal atomic.Int64
	go func() {
		defer close(dispatchDone)
		ticker := time.NewTicker(dispatchPollInterval)
		defer ticker.Stop()
		for {
			select {
			case <-ticker.C:
				n, err := e.dispatcher.Dispatch(ctx)
				if err != nil {
					t.Errorf("dispatch: %v", err)
					return
				}
				dispatchedTotal.Add(int64(n))
			case <-stopDispatch:
				n, err := e.dispatcher.Dispatch(ctx)
				if err != nil {
					t.Errorf("dispatch (abschluss): %v", err)
				}
				dispatchedTotal.Add(int64(n))
				return
			}
		}
	}()

	var publishErrs atomic.Int64
	wallStart := time.Now()

	producer := func(module, objectType, trigger string, idx int) {
		for i := 0; i < eventsPerProducer; i++ {
			objectID := fmt.Sprintf("%s-%d-%d", objectType, idx, i)
			_, err := moduleadapter.PublishEvent(ctx, e.pool, ownTenant, moduleadapter.Event{
				ModuleName: module, ObjectType: objectType, Trigger: trigger, ObjectID: objectID,
				TenantSlug: ownTenant, DedupeKey: fmt.Sprintf("dedupe-%s-%s", module, objectID),
			})
			if err != nil {
				publishErrs.Add(1)
			}
		}
	}

	var wg sync.WaitGroup
	for i := 0; i < producersPerModule; i++ {
		wg.Add(2)
		go func(i int) { defer wg.Done(); producer("dms-last", "document", "document.hochgeladen", i) }(i)
		go func(i int) { defer wg.Done(); producer("mail-last", "message", "mail.eingegangen", i) }(i)
	}
	wg.Wait()

	if n := publishErrs.Load(); n != 0 {
		t.Fatalf("%d von %d publish-aufrufen fehlgeschlagen", n, total)
	}

	deadline := time.Now().Add(30 * time.Second)
	for {
		var pending int
		if err := e.pool.QueryRow(ctx, `
			SELECT count(*) FROM module_events
			WHERE module_name IN ('dms-last', 'mail-last') AND processed_at IS NULL
		`).Scan(&pending); err != nil {
			t.Fatalf("offene ereignisse zaehlen: %v", err)
		}
		if pending == 0 {
			break
		}
		if time.Now().After(deadline) {
			t.Fatalf("timeout: %d ereignisse nach 30s noch unverarbeitet", pending)
		}
		time.Sleep(20 * time.Millisecond)
	}
	close(stopDispatch)
	<-dispatchDone
	wall := time.Since(wallStart)

	rows, err := e.pool.Query(ctx, `
		SELECT received_at, processed_at FROM module_events
		WHERE module_name IN ('dms-last', 'mail-last') AND processed_at IS NOT NULL
	`)
	if err != nil {
		t.Fatalf("latenzen lesen: %v", err)
	}
	defer rows.Close()
	var latencies []time.Duration
	for rows.Next() {
		var receivedAt, processedAt time.Time
		if err := rows.Scan(&receivedAt, &processedAt); err != nil {
			t.Fatalf("latenz-zeile lesen: %v", err)
		}
		latencies = append(latencies, processedAt.Sub(receivedAt))
	}
	if err := rows.Err(); err != nil {
		t.Fatalf("latenzen iterieren: %v", err)
	}

	if len(latencies) != total {
		t.Fatalf("verarbeitete ereignisse = %d, erwarte %d (keines darf verloren gehen)", len(latencies), total)
	}

	_, p95, _, _ := summarizeLatencies(t, "ENG-02 Ereigniszustellung DMS+Mail", latencies, wall)
	if p95 > targetP95DeliveryLatency {
		t.Fatalf("p95-latenz %s ueberschreitet Ziel-Latenz %s", p95, targetP95DeliveryLatency)
	}

	var taskCount int
	if err := e.pool.QueryRow(ctx, `SELECT count(*) FROM tasks WHERE assignee_id IN ('sachbearbeiter-dms', 'sachbearbeiter-mail')`).Scan(&taskCount); err != nil {
		t.Fatalf("aufgaben zaehlen: %v", err)
	}
	if taskCount != total {
		t.Fatalf("aufgaben angelegt = %d, erwarte %d", taskCount, total)
	}
}

// testApprovalUnderLoad ist Pruefung 2: doppelte Zustellung desselben
// Ereignisses (Wiederholung durch das sendende Modul) darf keine zweite
// Genehmigungsanfrage erzeugen, und gleichzeitige Entscheidungen auf
// derselben Aufgabe duerfen nur genau einmal durchgehen.
func testApprovalUnderLoad(t *testing.T, ctx context.Context, e loadEnv) {
	if _, err := e.rules.Put(ctx, eventtrigger.Rule{
		ModuleName: "dms-last", ObjectType: "document", Trigger: "document.freigabe_angefordert",
		Kind: eventtrigger.KindApproval, AssigneeType: task.AssigneePerson, AssigneeID: "alice",
		Title: "Freigabe Dokument",
	}); err != nil {
		t.Fatalf("regel anlegen: %v", err)
	}

	const (
		numDocuments     = 20
		redeliveryFanout = 5
	)

	var publishErrs atomic.Int64
	for d := 0; d < numDocuments; d++ {
		objectID := fmt.Sprintf("freigabe-doc-%d", d)
		dedupeKey := fmt.Sprintf("dedupe-freigabe-%d", d)
		runBounded(redeliveryFanout, redeliveryFanout, func(i int) {
			_, err := moduleadapter.PublishEvent(ctx, e.pool, ownTenant, moduleadapter.Event{
				ModuleName: "dms-last", ObjectType: "document", Trigger: "document.freigabe_angefordert",
				ObjectID: objectID, TenantSlug: ownTenant, DedupeKey: dedupeKey,
			})
			if err != nil {
				publishErrs.Add(1)
			}
		})
	}
	if n := publishErrs.Load(); n != 0 {
		t.Fatalf("%d publish-aufrufe (mehrfachzustellung) fehlgeschlagen", n)
	}

	var eventRows int
	if err := e.pool.QueryRow(ctx, `
		SELECT count(*) FROM module_events WHERE module_name = 'dms-last' AND trigger = 'document.freigabe_angefordert'
	`).Scan(&eventRows); err != nil {
		t.Fatalf("ereignis-zeilen zaehlen: %v", err)
	}
	if eventRows != numDocuments {
		t.Fatalf("ereignis-zeilen = %d, erwarte %d (wiederholte zustellung muss auf dieselbe zeile fallen)", eventRows, numDocuments)
	}

	n, err := e.dispatcher.Dispatch(ctx)
	if err != nil {
		t.Fatalf("dispatch: %v", err)
	}
	if n != numDocuments {
		t.Fatalf("dispatch hat %d aufgaben ausgeloest, erwarte %d", n, numDocuments)
	}

	var approvalCount int
	if err := e.pool.QueryRow(ctx, `SELECT count(*) FROM approval_tasks WHERE assignee = 'alice'`).Scan(&approvalCount); err != nil {
		t.Fatalf("genehmigungsaufgaben zaehlen: %v", err)
	}
	if approvalCount != numDocuments {
		t.Fatalf("genehmigungsaufgaben = %d, erwarte genau %d (keine doppelten, keine verlorenen)", approvalCount, numDocuments)
	}

	rows, err := e.pool.Query(ctx, `SELECT id FROM approval_tasks WHERE assignee = 'alice'`)
	if err != nil {
		t.Fatalf("aufgaben-ids lesen: %v", err)
	}
	var taskIDs []string
	for rows.Next() {
		var id string
		if err := rows.Scan(&id); err != nil {
			t.Fatalf("aufgaben-id lesen: %v", err)
		}
		taskIDs = append(taskIDs, id)
	}
	rows.Close()
	if err := rows.Err(); err != nil {
		t.Fatalf("aufgaben-ids iterieren: %v", err)
	}

	const concurrentDecidersPerTask = 8
	var (
		mu                    sync.Mutex
		latencies             []time.Duration
		totalSuccesses        int
		totalAlreadyDecided   int
		totalUnexpectedErrors int
	)
	wallStart := time.Now()
	for _, id := range taskIDs {
		runBounded(concurrentDecidersPerTask, concurrentDecidersPerTask, func(i int) {
			decision := approval.Approved
			if i%2 == 1 {
				decision = approval.Rejected
			}
			start := time.Now()
			_, err := e.approvalSvc.Decide(ctx, id, "alice", decision, "last-leistungstest")
			d := time.Since(start)

			mu.Lock()
			defer mu.Unlock()
			latencies = append(latencies, d)
			switch {
			case err == nil:
				totalSuccesses++
			case errors.Is(err, approval.ErrAlreadyDecided):
				totalAlreadyDecided++
			default:
				totalUnexpectedErrors++
				t.Errorf("unerwarteter fehler bei decide: %v", err)
			}
		})
	}
	wall := time.Since(wallStart)
	summarizeLatencies(t, "APR Decide unter Mehrfachzustellung", latencies, wall)

	wantTotal := numDocuments * concurrentDecidersPerTask
	if totalSuccesses != numDocuments {
		t.Fatalf("erfolgreiche entscheidungen = %d, erwarte genau %d (eine je aufgabe)", totalSuccesses, numDocuments)
	}
	if totalUnexpectedErrors != 0 {
		t.Fatalf("%d unerwartete fehler bei entscheidungen", totalUnexpectedErrors)
	}
	if totalSuccesses+totalAlreadyDecided != wantTotal {
		t.Fatalf("erfolge+bereits-entschieden = %d, erwarte %d (keine anfrage darf verloren gehen)", totalSuccesses+totalAlreadyDecided, wantTotal)
	}

	var decidedCount int
	if err := e.pool.QueryRow(ctx, `
		SELECT count(*) FROM approval_tasks
		WHERE assignee = 'alice' AND status IN ('approved', 'rejected') AND decided_at IS NOT NULL
	`).Scan(&decidedCount); err != nil {
		t.Fatalf("entschiedene aufgaben zaehlen: %v", err)
	}
	if decidedCount != numDocuments {
		t.Fatalf("entschiedene aufgaben in db = %d, erwarte %d (kein inkonsistenter zustand)", decidedCount, numDocuments)
	}
}

// testRoutingDeterminismUnterLast ist Pruefung 3: dasselbe Eingabeobjekt
// liefert unter Nebenlaeufigkeit immer dieselbe Routing-Entscheidung.
func testRoutingDeterminismUnderLoad(t *testing.T, ctx context.Context, e loadEnv) {
	rulesSpec := []struct {
		name     string
		priority int
		cond     []routingrule.Condition
		action   routingrule.Action
	}{
		{"Eilrechnung", 100,
			[]routingrule.Condition{
				{Field: "object_type", Operator: "equals", Value: "invoice"},
				{Field: "tags", Operator: "contains", Value: "eilig"},
			},
			routingrule.Action{Type: "route_to_person", Target: "approver-express"}},
		{"Rechnung Standard", 50,
			[]routingrule.Condition{{Field: "object_type", Operator: "equals", Value: "invoice"}},
			routingrule.Action{Type: "route_to_process", ProcessID: "p-invoice-standard"}},
		{"Mail Beschwerde", 80,
			[]routingrule.Condition{
				{Field: "object_type", Operator: "equals", Value: "mail-message"},
				{Field: "category", Operator: "equals", Value: "beschwerde"},
			},
			routingrule.Action{Type: "route_to_person", Target: "support-lead"}},
		{"Mail Standard", 10,
			[]routingrule.Condition{{Field: "object_type", Operator: "equals", Value: "mail-message"}},
			routingrule.Action{Type: "route_to_process", ProcessID: "p-mail-standard"}},
		{"Vertrag", 60,
			[]routingrule.Condition{{Field: "object_type", Operator: "equals", Value: "contract"}},
			routingrule.Action{Type: "route_to_person", Target: "legal-team"}},
	}
	for _, r := range rulesSpec {
		if _, err := e.routeStore.Create(ctx, r.name, r.priority, r.cond, []routingrule.Action{r.action}); err != nil {
			t.Fatalf("regel %q anlegen: %v", r.name, err)
		}
	}

	type variant struct {
		objectType string
		object     map[string]any
		wantRule   string // "" bedeutet: keine regel, klaerungskorb
	}
	variants := []variant{
		{"invoice", map[string]any{"object_type": "invoice", "tags": []string{"eilig"}, "amount": 500.0}, "Eilrechnung"},
		{"invoice", map[string]any{"object_type": "invoice", "tags": []string{"normal"}, "amount": 200.0}, "Rechnung Standard"},
		{"mail-message", map[string]any{"object_type": "mail-message", "category": "beschwerde"}, "Mail Beschwerde"},
		{"mail-message", map[string]any{"object_type": "mail-message", "category": "anfrage"}, "Mail Standard"},
		{"unbekannt", map[string]any{"object_type": "unbekannt"}, ""},
	}

	const callsPerVariant = 200
	var (
		mu        sync.Mutex
		latencies []time.Duration
		outcomes  = make(map[int][]string)
	)
	wallStart := time.Now()
	totalCalls := callsPerVariant * len(variants)
	runBounded(totalCalls, 50, func(i int) {
		vi := i % len(variants)
		v := variants[vi]
		objectID := fmt.Sprintf("route-%d-%d", vi, i)

		start := time.Now()
		decision, err := e.routeEngine.Execute(ctx, v.objectType, objectID, v.object)
		d := time.Since(start)
		if err != nil {
			t.Errorf("execute: %v", err)
			return
		}

		outcome := ""
		if decision.Matched {
			outcome = decision.RuleName
		}

		mu.Lock()
		defer mu.Unlock()
		latencies = append(latencies, d)
		outcomes[vi] = append(outcomes[vi], outcome)
	})
	wall := time.Since(wallStart)
	summarizeLatencies(t, "RTE Routing unter Last", latencies, wall)

	for vi, v := range variants {
		results := outcomes[vi]
		if len(results) != callsPerVariant {
			t.Fatalf("variante %d: %d ergebnisse, erwarte %d", vi, len(results), callsPerVariant)
		}
		for _, got := range results {
			if got != v.wantRule {
				t.Fatalf("variante %d (object_type=%s): ergebnis %q, erwarte durchgehend %q -- routing nicht deterministisch unter last",
					vi, v.objectType, got, v.wantRule)
			}
		}
	}

	var basketCount int
	if err := e.pool.QueryRow(ctx, `SELECT count(*) FROM routing_clarification_basket WHERE object_type = 'unbekannt'`).Scan(&basketCount); err != nil {
		t.Fatalf("klaerungskorb zaehlen: %v", err)
	}
	if basketCount != callsPerVariant {
		t.Fatalf("klaerungskorb-eintraege = %d, erwarte %d (kein objekt ohne regel darf verloren gehen)", basketCount, callsPerVariant)
	}
}