From 772fa9caa577cde3ecbf6d224b0a9804f0e4afca Mon Sep 17 00:00:00 2001 From: root Date: Sat, 26 Sep 2026 09:15:49 +0000 Subject: [PATCH] RTE-02: routing engine ausfuehrung Co-Authored-By: Claude Sonnet 5 --- .../internal/routingengine/routingengine.go | 154 ++++++++++++++ .../routingengine/routingengine_test.go | 194 ++++++++++++++++++ .../tenant/0013_routing_execution.down.sql | 2 + .../tenant/0013_routing_execution.up.sql | 24 +++ 4 files changed, 374 insertions(+) create mode 100644 workflow/internal/routingengine/routingengine.go create mode 100644 workflow/internal/routingengine/routingengine_test.go create mode 100644 workflow/migrations/tenant/0013_routing_execution.down.sql create mode 100644 workflow/migrations/tenant/0013_routing_execution.up.sql diff --git a/workflow/internal/routingengine/routingengine.go b/workflow/internal/routingengine/routingengine.go new file mode 100644 index 00000000..bdc8e772 --- /dev/null +++ b/workflow/internal/routingengine/routingengine.go @@ -0,0 +1,154 @@ +// Package routingengine implementiert RTE-02: fuehrt das Regelwerk aus +// RTE-01 gegen eingehende Objekte aus. Erste zutreffende Regel nach +// Prioritaet gewinnt, sonst landet das Objekt im Klaerungskorb statt +// verloren zu gehen. +package routingengine + +import ( + "context" + "encoding/json" + "fmt" + "time" + + "github.com/jackc/pgx/v5" + "github.com/jackc/pgx/v5/pgxpool" + + "gitea.perlbach24.de/scripte/nexarch/workflow/internal/routingrule" +) + +// Decision ist das Ergebnis einer Ausfuehrung: entweder eine passende Regel +// mit ihrer Aktion, oder keine Regel (Klaerungskorb). +type Decision struct { + Matched bool + RuleID string + RuleName string + Action routingrule.Action + InBasket bool + BasketID string +} + +type Engine struct { + rules *routingrule.Store + pool *pgxpool.Pool +} + +func NewEngine(rules *routingrule.Store, pool *pgxpool.Pool) *Engine { + return &Engine{rules: rules, pool: pool} +} + +// Execute wertet das Regelwerk live gegen object aus (Pruefung 3: eine +// Regeländerung wirkt beim naechsten Execute-Aufruf, da ListEnabled bei +// jedem Aufruf frisch aus der Datenbank liest — kein In-Memory-Cache, also +// kein Neustart noetig). objectType/objectID identifizieren das Objekt fuer +// Protokoll und Klaerungskorb, unabhaengig vom aufrufenden Modul. +func (e *Engine) Execute(ctx context.Context, objectType, objectID string, object map[string]any) (Decision, error) { + rules, err := e.rules.ListEnabled(ctx) + if err != nil { + return Decision{}, fmt.Errorf("routingengine: regeln laden: %w", err) + } + + for _, r := range rules { + if !r.Matches(object) { + continue + } + if len(r.Actions) == 0 { + continue + } + action := r.Actions[0] + if err := e.recordDecision(ctx, objectType, objectID, &r.ID, action); err != nil { + return Decision{}, err + } + return Decision{Matched: true, RuleID: r.ID, RuleName: r.Name, Action: action}, nil + } + + // Akzeptanzkriterium 1: keine zutreffende Regel -> Klaerungskorb statt + // Verlust des Objekts. + basketID, err := e.addToBasket(ctx, objectType, objectID, "keine zutreffende regel") + if err != nil { + return Decision{}, err + } + if err := e.recordDecision(ctx, objectType, objectID, nil, routingrule.Action{}); err != nil { + return Decision{}, err + } + return Decision{Matched: false, InBasket: true, BasketID: basketID}, nil +} + +// recordDecision protokolliert JEDE Weiterleitung nachvollziehbar +// (Akzeptanzkriterium 2), auch die ohne Treffer. +func (e *Engine) recordDecision(ctx context.Context, objectType, objectID string, ruleID *string, action routingrule.Action) error { + var actionJSON []byte + if ruleID != nil { + b, err := json.Marshal(action) + if err != nil { + return fmt.Errorf("routingengine: aktion kodieren: %w", err) + } + actionJSON = b + } + _, err := e.pool.Exec(ctx, ` + INSERT INTO routing_decisions (object_type, object_id, matched_rule_id, matched_action) + VALUES ($1, $2, $3, $4) + `, objectType, objectID, ruleID, actionJSON) + if err != nil { + return fmt.Errorf("routingengine: entscheidung protokollieren: %w", err) + } + return nil +} + +func (e *Engine) addToBasket(ctx context.Context, objectType, objectID, reason string) (string, error) { + var id string + err := e.pool.QueryRow(ctx, ` + INSERT INTO routing_clarification_basket (object_type, object_id, reason) + VALUES ($1, $2, $3) RETURNING id + `, objectType, objectID, reason).Scan(&id) + if err != nil { + return "", fmt.Errorf("routingengine: klaerungskorb eintragen: %w", err) + } + return id, nil +} + +// UnresolvedBasketEntries liefert alle offenen Klaerungskorb-Eintraege +// (Pruefung 2: kein Testobjekt geht verloren -- nachweisbar ueber diese +// Liste). +func (e *Engine) UnresolvedBasketEntries(ctx context.Context) ([]BasketEntry, error) { + rows, err := e.pool.Query(ctx, ` + SELECT id, object_type, object_id, reason, created_at + FROM routing_clarification_basket + WHERE resolved_at IS NULL + ORDER BY created_at ASC + `) + if err != nil { + return nil, fmt.Errorf("routingengine: klaerungskorb lesen: %w", err) + } + defer rows.Close() + + var out []BasketEntry + for rows.Next() { + var b BasketEntry + if err := rows.Scan(&b.ID, &b.ObjectType, &b.ObjectID, &b.Reason, &b.CreatedAt); err != nil { + return nil, fmt.Errorf("routingengine: zeile lesen: %w", err) + } + out = append(out, b) + } + return out, rows.Err() +} + +// ResolveBasketEntry markiert einen Klaerungskorb-Eintrag als aufgeloest +// (z. B. nachdem manuell weitergeleitet wurde). +func (e *Engine) ResolveBasketEntry(ctx context.Context, id string) error { + res, err := e.pool.Exec(ctx, `UPDATE routing_clarification_basket SET resolved_at = now() WHERE id = $1 AND resolved_at IS NULL`, id) + if err != nil { + return fmt.Errorf("routingengine: klaerungskorb aufloesen: %w", err) + } + if res.RowsAffected() == 0 { + return pgx.ErrNoRows + } + return nil +} + +type BasketEntry struct { + ID string + ObjectType string + ObjectID string + Reason string + CreatedAt time.Time +} diff --git a/workflow/internal/routingengine/routingengine_test.go b/workflow/internal/routingengine/routingengine_test.go new file mode 100644 index 00000000..5bcf7fdb --- /dev/null +++ b/workflow/internal/routingengine/routingengine_test.go @@ -0,0 +1,194 @@ +package routingengine + +import ( + "context" + "os" + "path/filepath" + "testing" + + "github.com/jackc/pgx/v5/pgxpool" + + "gitea.perlbach24.de/scripte/nexarch/workflow/internal/migrate" + "gitea.perlbach24.de/scripte/nexarch/workflow/internal/routingrule" + "gitea.perlbach24.de/scripte/nexarch/workflow/internal/testdb" +) + +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 + } +} + +func setup(t *testing.T) (*Engine, *routingrule.Store, *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(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) + } + + store := routingrule.NewStore(pool) + return NewEngine(store, pool), store, pool +} + +// Akzeptanzkriterium 1: nicht zuordenbares Objekt landet im Klaerungskorb, +// geht also nicht verloren (Pruefung 2). +func TestExecute_NoMatch_LandsInClarificationBasket(t *testing.T) { + ctx := context.Background() + engine, store, _ := setup(t) + + _, err := store.Create(ctx, "nur rechnungen", 10, []routingrule.Condition{ + {Field: "object_type", Operator: "equals", Value: "invoice"}, + }, []routingrule.Action{{Type: "route_to_process", Target: "approval"}}) + if err != nil { + t.Fatalf("regel anlegen: %v", err) + } + + decision, err := engine.Execute(ctx, "document", "doc-no-match-1", map[string]any{"object_type": "contract"}) + if err != nil { + t.Fatalf("execute: %v", err) + } + if decision.Matched { + t.Fatal("erwartet: kein Treffer, aber Matched=true") + } + if !decision.InBasket || decision.BasketID == "" { + t.Fatal("erwartet: Objekt landet im Klaerungskorb") + } + + entries, err := engine.UnresolvedBasketEntries(ctx) + if err != nil { + t.Fatalf("basket lesen: %v", err) + } + found := false + for _, e := range entries { + if e.ID == decision.BasketID && e.ObjectID == "doc-no-match-1" { + found = true + } + } + if !found { + t.Fatal("Objekt nicht im Klaerungskorb wiederauffindbar -- waere verloren gegangen") + } +} + +// Akzeptanzkriterium: erste zutreffende Regel nach Prioritaet gewinnt. +func TestExecute_FirstMatchingRuleByPriorityWins(t *testing.T) { + ctx := context.Background() + engine, store, _ := setup(t) + + _, err := store.Create(ctx, "niedrige prioritaet", 1, []routingrule.Condition{ + {Field: "object_type", Operator: "equals", Value: "invoice"}, + }, []routingrule.Action{{Type: "route_to_process", Target: "low"}}) + if err != nil { + t.Fatalf("niedrige regel anlegen: %v", err) + } + _, err = store.Create(ctx, "hohe prioritaet", 100, []routingrule.Condition{ + {Field: "object_type", Operator: "equals", Value: "invoice"}, + }, []routingrule.Action{{Type: "route_to_process", Target: "high"}}) + if err != nil { + t.Fatalf("hohe regel anlegen: %v", err) + } + + decision, err := engine.Execute(ctx, "document", "doc-priority-1", map[string]any{"object_type": "invoice"}) + if err != nil { + t.Fatalf("execute: %v", err) + } + if !decision.Matched { + t.Fatal("erwartet: Treffer") + } + if decision.Action.Target != "high" { + t.Fatalf("erwartet Ziel 'high' (hoehere Prioritaet), bekam %q", decision.Action.Target) + } +} + +// Akzeptanzkriterium 2: jede Weiterleitung ist nachvollziehbar protokolliert +// -- sowohl Treffer als auch Klaerungskorb-Faelle. +func TestExecute_EveryDecisionIsLogged(t *testing.T) { + ctx := context.Background() + engine, store, pool := setup(t) + + rule, err := store.Create(ctx, "log-test-regel", 5, []routingrule.Condition{ + {Field: "object_type", Operator: "equals", Value: "einvoice"}, + }, []routingrule.Action{{Type: "route_to_process", Target: "approval"}}) + if err != nil { + t.Fatalf("regel anlegen: %v", err) + } + + if _, err := engine.Execute(ctx, "document", "doc-log-1", map[string]any{"object_type": "einvoice"}); err != nil { + t.Fatalf("execute (treffer): %v", err) + } + if _, err := engine.Execute(ctx, "document", "doc-log-2", map[string]any{"object_type": "unbekannt"}); err != nil { + t.Fatalf("execute (kein treffer): %v", err) + } + + var matchedCount, unmatchedCount int + if err := pool.QueryRow(ctx, `SELECT count(*) FROM routing_decisions WHERE object_id = 'doc-log-1' AND matched_rule_id = $1`, rule.ID).Scan(&matchedCount); err != nil { + t.Fatalf("protokoll (treffer) lesen: %v", err) + } + if matchedCount != 1 { + t.Fatalf("erwartet 1 protokollierte Treffer-Entscheidung, bekam %d", matchedCount) + } + if err := pool.QueryRow(ctx, `SELECT count(*) FROM routing_decisions WHERE object_id = 'doc-log-2' AND matched_rule_id IS NULL`).Scan(&unmatchedCount); err != nil { + t.Fatalf("protokoll (kein treffer) lesen: %v", err) + } + if unmatchedCount != 1 { + t.Fatalf("erwartet 1 protokollierte Nicht-Treffer-Entscheidung, bekam %d", unmatchedCount) + } +} + +// Akzeptanzkriterium 3 / Pruefung 3: eine Regeländerung wirkt ohne Neustart, +// da Execute bei jedem Aufruf frisch aus der Datenbank liest. +func TestExecute_RuleChangeTakesEffectWithoutRestart(t *testing.T) { + ctx := context.Background() + engine, store, _ := setup(t) + + rule, err := store.Create(ctx, "aenderbare regel", 5, []routingrule.Condition{ + {Field: "object_type", Operator: "equals", Value: "mail"}, + }, []routingrule.Action{{Type: "route_to_process", Target: "process-a"}}) + if err != nil { + t.Fatalf("regel anlegen: %v", err) + } + + decision1, err := engine.Execute(ctx, "message", "msg-1", map[string]any{"object_type": "mail"}) + if err != nil { + t.Fatalf("execute vor aenderung: %v", err) + } + if decision1.Action.Target != "process-a" { + t.Fatalf("erwartet 'process-a' vor Aenderung, bekam %q", decision1.Action.Target) + } + + // Aenderung der Regel zur Laufzeit -- dieselbe Engine-Instanz, kein Neustart. + if _, err := store.Update(ctx, rule.ID, []routingrule.Condition{ + {Field: "object_type", Operator: "equals", Value: "mail"}, + }, []routingrule.Action{{Type: "route_to_process", Target: "process-b"}}); err != nil { + t.Fatalf("regel aendern: %v", err) + } + + decision2, err := engine.Execute(ctx, "message", "msg-2", map[string]any{"object_type": "mail"}) + if err != nil { + t.Fatalf("execute nach aenderung: %v", err) + } + if decision2.Action.Target != "process-b" { + t.Fatalf("erwartet 'process-b' nach Aenderung ohne Neustart, bekam %q", decision2.Action.Target) + } +} diff --git a/workflow/migrations/tenant/0013_routing_execution.down.sql b/workflow/migrations/tenant/0013_routing_execution.down.sql new file mode 100644 index 00000000..16bfd414 --- /dev/null +++ b/workflow/migrations/tenant/0013_routing_execution.down.sql @@ -0,0 +1,2 @@ +DROP TABLE IF EXISTS routing_clarification_basket; +DROP TABLE IF EXISTS routing_decisions; diff --git a/workflow/migrations/tenant/0013_routing_execution.up.sql b/workflow/migrations/tenant/0013_routing_execution.up.sql new file mode 100644 index 00000000..2ff41647 --- /dev/null +++ b/workflow/migrations/tenant/0013_routing_execution.up.sql @@ -0,0 +1,24 @@ +-- RTE-02: Ausfuehrung des Regelwerks aus RTE-01 gegen eingehende Objekte. +-- routing_decisions protokolliert jede Weiterleitung nachvollziehbar +-- (Akzeptanzkriterium 2), unabhaengig davon ob eine Regel griff oder nicht. +CREATE TABLE routing_decisions ( + id UUID PRIMARY KEY DEFAULT gen_random_uuid(), + object_type TEXT NOT NULL, + object_id TEXT NOT NULL, + matched_rule_id UUID REFERENCES routing_rules(id), + matched_action JSONB, + decided_at TIMESTAMPTZ NOT NULL DEFAULT now() +); +CREATE INDEX routing_decisions_object_idx ON routing_decisions (object_type, object_id); + +-- Klaerungskorb (Akzeptanzkriterium 1): Objekte ohne zutreffende Regel +-- landen hier statt verloren zu gehen, bis sie manuell aufgeloest werden. +CREATE TABLE routing_clarification_basket ( + id UUID PRIMARY KEY DEFAULT gen_random_uuid(), + object_type TEXT NOT NULL, + object_id TEXT NOT NULL, + reason TEXT NOT NULL, + created_at TIMESTAMPTZ NOT NULL DEFAULT now(), + resolved_at TIMESTAMPTZ +); +CREATE INDEX routing_clarification_basket_unresolved_idx ON routing_clarification_basket (created_at) WHERE resolved_at IS NULL;