RTE-02: routing engine ausfuehrung
Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
This commit is contained in:
rootandClaude Sonnet 5 committed
1 parent
a684b46c5c
commit
772fa9caa5
4 files changed
+374
No files matched your search
@@ -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
|
||||
}
|
||||
@@ -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)
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,2 @@
|
||||
DROP TABLE IF EXISTS routing_clarification_basket;
|
||||
DROP TABLE IF EXISTS routing_decisions;
|
||||
@@ -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;
|
||||
Reference in new issue
Block a user