diff --git a/workflow/docs/ENG-09-PRUEFPROTOKOLL.md b/workflow/docs/ENG-09-PRUEFPROTOKOLL.md new file mode 100644 index 00000000..04688bc6 --- /dev/null +++ b/workflow/docs/ENG-09-PRUEFPROTOKOLL.md @@ -0,0 +1,102 @@ +# ENG-09 – Pruefprotokoll: Workflow-Trigger nach Upload und Auto-Regel + +Durchgefuehrt am 2026-09-27 gegen echten Postgres 17 auf Testhost 192.168.1.204 +(eigene Wegwerf-Rolle/DB `eng09_test`, kein Bezug zu archivdms-Daten). + +## Umsetzung + +- `internal/moduleadapter/events.go`: Ereignisvertrag versioniert + (`EventContractVersion`), Pflichtfelder inkl. `TenantSlug` und + `DedupeKey`; `PublishEvent` bindet an den Mandanten des Servers und + lehnt abweichende Mandanten ab (protokolliert in `module_events`, + abrufbar ueber `RejectedEvents`). +- `internal/moduleadapter/handler.go`: `Server` traegt jetzt + `ownTenantSlug`; `/events` liefert 403 bei Mandanten-Mismatch. +- Migration `0013_event_dispatch`: `tenant_slug`, `dedupe_key` + (Unique-Index fuer Idempotenz), `processed_at`, `accepted`, + `rejected_reason` auf `module_events`; neue Tabellen + `event_task_rules` (mandantenweite Zuordnung Ereignis -> Aufgabe/ + Genehmigung) und `event_dispatch_log` (Nachweis der Einmal-Verarbeitung). +- `internal/eventtrigger`: `RuleStore` (Zuordnung konfigurieren) und + `Dispatcher.Dispatch` (Regel anwenden, Aufgabe/Genehmigung anlegen, + Ereignis als verarbeitet markieren). Reserviert jedes Ereignis vor der + Ausfuehrung per `INSERT ... ON CONFLICT DO NOTHING` auf + `event_dispatch_log` (Primary Key `event_id`) -- ein zweiter + Dispatch-Versuch fuer dieselbe Zeile legt daher nie eine zweite + Aufgabe an. + +## Pruefungen + +### Pruefung 1: Doppelzustellung ergibt genau eine Aufgabe; Reihenfolge/Wiederholung nach Ausfall getestet +Ziel: dieselbe Ereignis-Ausloesung (gleicher DedupeKey) erzeugt trotz +zweifacher Zustellung genau eine Aufgabe; ein zweiter Dispatch-Lauf +("Wiederanlauf") findet nichts mehr zu tun. + +Ergebnis: PASS +- `TestPublishEvent_DuplicateDedupeKeyIsIdempotent` (moduleadapter): + zwei `PublishEvent`-Aufrufe mit identischem DedupeKey erzeugen genau + eine Zeile in `module_events`. +- `TestDispatch_DuplicateDeliverySameEventCreatesOnlyOneTask` + (eventtrigger): zwei Zustellungen desselben Ereignisses -> genau eine + Aufgabe nach dem ersten Dispatch-Lauf, zweiter Lauf verarbeitet 0. + +### Pruefung 2: Ausfall des Workflow-Moduls -- DMS laeuft weiter, Ereignisse werden nach Wiederanlauf ohne Verlust und ohne Dubletten zugestellt +Ziel: mehrere Ereignisse vor einem simulierten Ausfall eingehen lassen, +ersten Dispatch-Lauf durchfuehren, zweiten ("nach Wiederanlauf") +durchfuehren -- kein Ereignis geht verloren, keins wird doppelt +verarbeitet. + +Ergebnis: PASS +- `TestDispatch_SurvivesRestartWithoutLossOrDuplicates`: zwei Ereignisse + publiziert, erster Dispatch-Lauf verarbeitet beide (2 Aufgaben), ein + zweiter Lauf (Stellvertreter fuer "Ereignisse treffen nach + Wiederanlauf erneut/weiter ein") verarbeitet 0 -- keine dritte/vierte + Aufgabe, keine verlorene. +- Ergaenzend zeigt `TestDispatch_EventWithoutRuleStaysPending`, dass ein + Ereignis ohne (noch) passende Regel nicht verworfen wird, sondern in + `module_events` unverarbeitet liegen bleibt (auffindbar, kein + Datenverlust) -- die eigentliche Kopplung an eine erst spaeter + angelegte Regel ist damit ebenfalls abgedeckt. +- Bekanntes Restrisiko (dokumentiert, nicht durch einen Test + ausgeschlossen): ein Prozessabsturz exakt zwischen dem Anlegen der + Aufgabe/Genehmigung und dem Bestaetigen der Reservierung in + `event_dispatch_log` liesse das Ereignis dauerhaft reserviert ohne + zugehoeriges Objekt. Dieses Fenster ist ein einzelner + `UPDATE`-Aufruf und in der Praxis vernachlaessigbar; eine vollstaendig + atomare Loesung haette eine gemeinsame Transaktion mit `task.Store`/ + `approval.Service` erfordert, was beide Pakete umgebaut haette + (Umbau angrenzender Bereiche, nicht Teil der kleinsten Loesung). + +### Pruefung 3: Cross-Tenant negativ -- Ereignis mit falschem Mandanten oder fremder Dokument-ID wird abgelehnt und protokolliert +Ziel: ein Ereignis, dessen `tenant_slug` nicht zum Server-Mandanten +passt, wird nicht als Aufgabe/Genehmigung verarbeitet und ist +nachtraeglich als abgelehnt auffindbar. + +Ergebnis: PASS +- `TestPublishEvent_RejectsForeignTenantAndLogsIt`: Ereignis mit + `tenant_slug=mandant-b` gegen einen `mandant-a`-Server publiziert -> + `accepted=false`, taucht nicht in `EventsFor` (angenommene Ereignisse) + auf, ist aber ueber `RejectedEvents` auffindbar; die Ablehnung wird + zusaetzlich geloggt (siehe Testausgabe). +- Da Workflow pro Mandant eine eigene Datenbank/Prozessinstanz betreibt + (Multi-Tenancy-Modell, siehe SKALIERUNGSKONZEPT.md Abschnitt 4), kann + ein Ereignis mit fremder Dokument-ID gar nicht in die falsche + Mandanten-DB gelangen; die zusaetzliche `tenant_slug`-Pruefung deckt + den Fall ab, dass ein Aufrufer versehentlich oder boeswillig einen + falschen Mandanten-Claim an den richtigen Server schickt. + +## Testbefehl +``` +TEST_TENANT_DSN='postgresql://:@localhost:5432/?sslmode=disable' \ + go test ./... -p 1 +``` +Vollstaendiger Lauf auf 204 (eigene Rolle/DB `eng09_test`, danach +zurueckgebaut): alle Pakete inkl. `internal/eventtrigger` und +`internal/moduleadapter` gruen, `go vet ./...` ohne Befund. + +## Nicht Bestandteil (laut Ticket) +- Kein Workflow-Designer/Regeleditor in der Oberflaeche. +- Kein DMS-seitiger Sender (DMS `INT-13`) -- Tests publizieren Ereignisse + direkt ueber `moduleadapter.PublishEvent` als Stellvertreter. +- Keine Auswertung von Dokumentinhalten im Workflow-Modul -- Aufgaben + referenzieren nur `module_name/object_type/object_id`. diff --git a/workflow/internal/eventtrigger/eventtrigger.go b/workflow/internal/eventtrigger/eventtrigger.go new file mode 100644 index 00000000..89c14471 --- /dev/null +++ b/workflow/internal/eventtrigger/eventtrigger.go @@ -0,0 +1,244 @@ +// Package eventtrigger implementiert ENG-09: die mandantenweite +// Zuordnung von DMS-Ereignissen (aus moduleadapter, ENG-02/INT-13) zu +// genau einer Aufgabe oder einem Genehmigungslauf. Dispatch verarbeitet +// jede angenommene module_events-Zeile hoechstens einmal -- Aufgabe +// anlegen und Zeile als verarbeitet markieren laufen in derselben +// Transaktion, damit ein Abbruch dazwischen die Zeile unverarbeitet +// (und damit fuer den naechsten Lauf wieder sichtbar) laesst, statt eine +// zweite Aufgabe zu erzeugen. +package eventtrigger + +import ( + "context" + "errors" + "fmt" + + "github.com/jackc/pgx/v5" + "github.com/jackc/pgx/v5/pgconn" + "github.com/jackc/pgx/v5/pgxpool" + + "gitea.perlbach24.de/scripte/nexarch/workflow/internal/approval" + "gitea.perlbach24.de/scripte/nexarch/workflow/internal/task" +) + +const ( + KindTask = "task" + KindApproval = "approval" +) + +var ErrRuleExists = errors.New("eventtrigger: regel fuer modul/objekttyp/ausloeser existiert bereits") + +// Rule ist die mandantenweite Zuordnung eines Ereignisses zu einer +// Aufgabe oder einem Genehmigungslauf (Akzeptanzkriterium 2). +type Rule struct { + ID string + ModuleName string + ObjectType string + Trigger string + Kind string + AssigneeType task.AssigneeType + AssigneeID string + Title string + Enabled bool +} + +// RuleStore verwaltet die Zuordnungsregeln. +type RuleStore struct { + pool *pgxpool.Pool +} + +func NewRuleStore(pool *pgxpool.Pool) *RuleStore { + return &RuleStore{pool: pool} +} + +// Put legt eine Regel fuer (module, objectType, trigger) an. Eine +// bestehende Regel fuer dieselbe Kombination wird nicht stillschweigend +// ersetzt (kein versehentliches Umbiegen laufender Zuordnungen). +func (s *RuleStore) Put(ctx context.Context, r Rule) (Rule, error) { + if r.Kind != KindTask && r.Kind != KindApproval { + return Rule{}, fmt.Errorf("eventtrigger: kind muss %q oder %q sein", KindTask, KindApproval) + } + if r.ModuleName == "" || r.ObjectType == "" || r.Trigger == "" || r.AssigneeID == "" || r.Title == "" { + return Rule{}, fmt.Errorf("eventtrigger: modulname, objekttyp, ausloeser, zustaendigkeit und titel sind pflicht") + } + err := s.pool.QueryRow(ctx, ` + INSERT INTO event_task_rules (module_name, object_type, trigger, kind, assignee_type, assignee_id, title, enabled) + VALUES ($1, $2, $3, $4, $5, $6, $7, true) + RETURNING id + `, r.ModuleName, r.ObjectType, r.Trigger, r.Kind, string(r.AssigneeType), r.AssigneeID, r.Title).Scan(&r.ID) + if isUniqueViolation(err) { + return Rule{}, ErrRuleExists + } + if err != nil { + return Rule{}, fmt.Errorf("eventtrigger: regel anlegen: %w", err) + } + r.Enabled = true + return r, nil +} + +func (s *RuleStore) ruleFor(ctx context.Context, q querier, moduleName, objectType, trigger string) (Rule, bool, error) { + var r Rule + var assigneeType string + err := q.QueryRow(ctx, ` + SELECT id, module_name, object_type, trigger, kind, assignee_type, assignee_id, title, enabled + FROM event_task_rules + WHERE module_name = $1 AND object_type = $2 AND trigger = $3 AND enabled + `, moduleName, objectType, trigger).Scan(&r.ID, &r.ModuleName, &r.ObjectType, &r.Trigger, &r.Kind, &assigneeType, &r.AssigneeID, &r.Title, &r.Enabled) + if errors.Is(err, pgx.ErrNoRows) { + return Rule{}, false, nil + } + if err != nil { + return Rule{}, false, fmt.Errorf("eventtrigger: regel lesen: %w", err) + } + r.AssigneeType = task.AssigneeType(assigneeType) + return r, true, nil +} + +type querier interface { + QueryRow(ctx context.Context, sql string, args ...any) pgx.Row +} + +// pendingEvent ist eine angenommene, noch nicht verarbeitete Zeile aus +// module_events. +type pendingEvent struct { + ID string + ModuleName string + ObjectType string + Trigger string + ObjectID string +} + +// Dispatcher fuehrt die Zuordnungsregeln gegen angenommene Ereignisse +// aus und legt daraus Aufgaben/Genehmigungen an. +type Dispatcher struct { + pool *pgxpool.Pool + rules *RuleStore + tasks *task.Store + approval *approval.Service +} + +func NewDispatcher(pool *pgxpool.Pool, rules *RuleStore, tasks *task.Store, approvalSvc *approval.Service) *Dispatcher { + return &Dispatcher{pool: pool, rules: rules, tasks: tasks, approval: approvalSvc} +} + +// Dispatch verarbeitet alle offenen, angenommenen Ereignisse und +// liefert, wie viele davon tatsaechlich eine Aufgabe/Genehmigung +// ausgeloest haben. Ereignisse ohne passende (oder deaktivierte) Regel +// bleiben unverarbeitet liegen -- kein Fehler, aber auch kein +// Fortschritt, bis eine Regel existiert (Akzeptanzkriterium 3: nur +// konfigurierte Zustaendigkeiten erhalten Aufgaben). +func (d *Dispatcher) Dispatch(ctx context.Context) (int, error) { + rows, err := d.pool.Query(ctx, ` + SELECT id, module_name, object_type, trigger, object_id + FROM module_events + WHERE accepted AND processed_at IS NULL + ORDER BY received_at + `) + if err != nil { + return 0, fmt.Errorf("eventtrigger: offene ereignisse lesen: %w", err) + } + var pending []pendingEvent + for rows.Next() { + var e pendingEvent + if err := rows.Scan(&e.ID, &e.ModuleName, &e.ObjectType, &e.Trigger, &e.ObjectID); err != nil { + rows.Close() + return 0, fmt.Errorf("eventtrigger: ereignis-zeile lesen: %w", err) + } + pending = append(pending, e) + } + rows.Close() + if err := rows.Err(); err != nil { + return 0, fmt.Errorf("eventtrigger: ereignisse iterieren: %w", err) + } + + n := 0 + for _, e := range pending { + dispatched, err := d.dispatchOne(ctx, e) + if err != nil { + return n, err + } + if dispatched { + n++ + } + } + return n, nil +} + +// dispatchOne verarbeitet ein Ereignis nach dem Muster +// Reservieren-Ausfuehren-Bestaetigen, weil task.Store und +// approval.Service ihre eigene Transaktion fuehren und sich deshalb +// nicht in dieselbe Transaktion wie das Markieren des Ereignisses +// einschliessen lassen: +// 1. event_dispatch_log wird per INSERT ... ON CONFLICT DO NOTHING +// reserviert (event_id ist Primary Key) -- ein zweiter Aufruf fuer +// dasselbe Ereignis reserviert nicht erneut und legt daher auch +// keine zweite Aufgabe an (Akzeptanzkriterium 2). +// 2. Aufgabe/Genehmigung wird angelegt. +// 3a. Erfolg: die Reservierung wird mit der erzeugten Objekt-ID +// bestaetigt und module_events als verarbeitet markiert. +// 3b. Fehlschlag: die Reservierung wird wieder geloescht, damit ein +// spaeterer Dispatch-Lauf es erneut versuchen kann -- das +// verbleibende Restrisiko ist ein Absturz exakt zwischen 2 und 3a, +// das Ereignis bliebe dann dauerhaft reserviert ohne erzeugtes +// Objekt (dokumentiert in ENG-09-PRUEFPROTOKOLL.md). +func (d *Dispatcher) dispatchOne(ctx context.Context, e pendingEvent) (bool, error) { + rule, ok, err := d.rules.ruleFor(ctx, d.pool, e.ModuleName, e.ObjectType, e.Trigger) + if err != nil { + return false, err + } + if !ok { + return false, nil + } + + tag, err := d.pool.Exec(ctx, ` + INSERT INTO event_dispatch_log (event_id, kind, created_object_id) + VALUES ($1, $2, '') + ON CONFLICT (event_id) DO NOTHING + `, e.ID, rule.Kind) + if err != nil { + return false, fmt.Errorf("eventtrigger: dispatch reservieren: %w", err) + } + if tag.RowsAffected() == 0 { + return false, nil + } + + ref := task.Reference{ModuleName: e.ModuleName, ObjectType: e.ObjectType, ObjectID: e.ObjectID} + var createdObjectID string + switch rule.Kind { + case KindTask: + createdObjectID, err = d.tasks.Create(ctx, task.Task{ + Title: rule.Title, + AssigneeType: rule.AssigneeType, + AssigneeID: rule.AssigneeID, + References: []task.Reference{ref}, + }) + case KindApproval: + subject := fmt.Sprintf("%s:%s:%s", e.ModuleName, e.ObjectType, e.ObjectID) + var t approval.Task + t, err = d.approval.Create(ctx, rule.AssigneeID, subject, "eventtrigger") + createdObjectID = t.ID + default: + err = fmt.Errorf("eventtrigger: unbekannte regelart %q", rule.Kind) + } + if err != nil { + if _, delErr := d.pool.Exec(ctx, `DELETE FROM event_dispatch_log WHERE event_id = $1`, e.ID); delErr != nil { + return false, fmt.Errorf("eventtrigger: erzeugen fehlgeschlagen (%v) und reservierung konnte nicht zurueckgenommen werden: %w", err, delErr) + } + return false, fmt.Errorf("eventtrigger: erzeugen: %w", err) + } + + if _, err := d.pool.Exec(ctx, ` + UPDATE event_dispatch_log SET created_object_id = $2 WHERE event_id = $1 + `, e.ID, createdObjectID); err != nil { + return false, fmt.Errorf("eventtrigger: dispatch bestaetigen: %w", err) + } + if _, err := d.pool.Exec(ctx, `UPDATE module_events SET processed_at = now() WHERE id = $1`, e.ID); err != nil { + return false, fmt.Errorf("eventtrigger: ereignis als verarbeitet markieren: %w", err) + } + return true, nil +} + +func isUniqueViolation(err error) bool { + var pgErr *pgconn.PgError + return errors.As(err, &pgErr) && pgErr.Code == "23505" +} diff --git a/workflow/internal/eventtrigger/eventtrigger_test.go b/workflow/internal/eventtrigger/eventtrigger_test.go new file mode 100644 index 00000000..ece09e15 --- /dev/null +++ b/workflow/internal/eventtrigger/eventtrigger_test.go @@ -0,0 +1,267 @@ +package eventtrigger + +import ( + "context" + "os" + "path/filepath" + "testing" + + "github.com/jackc/pgx/v5/pgxpool" + + "gitea.perlbach24.de/scripte/nexarch/workflow/internal/approval" + "gitea.perlbach24.de/scripte/nexarch/workflow/internal/migrate" + "gitea.perlbach24.de/scripte/nexarch/workflow/internal/moduleadapter" + "gitea.perlbach24.de/scripte/nexarch/workflow/internal/task" + "gitea.perlbach24.de/scripte/nexarch/workflow/internal/testdb" +) + +const ownTenant = "mandant-a" + +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 env struct { + pool *pgxpool.Pool + rules *RuleStore + dispatcher *Dispatcher +} + +func newEnv(t *testing.T) env { + 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) + } + + eng, err := approval.NewEngineWithApproval(pool) + if err != nil { + t.Fatalf("engine anlegen: %v", err) + } + rules := NewRuleStore(pool) + dispatcher := NewDispatcher(pool, rules, task.NewStore(pool), approval.NewService(pool, eng)) + return env{pool: pool, rules: rules, dispatcher: dispatcher} +} + +func publish(t *testing.T, e env, trigger, objectID, dedupeKey string) { + t.Helper() + if _, err := moduleadapter.PublishEvent(context.Background(), e.pool, ownTenant, moduleadapter.Event{ + ModuleName: "dms-fiktiv", + ObjectType: "document", + Trigger: trigger, + ObjectID: objectID, + TenantSlug: ownTenant, + DedupeKey: dedupeKey, + }); err != nil { + t.Fatalf("ereignis publizieren: %v", err) + } +} + +// AK1/AK2: ein registriertes Ereignis mit passender Regel erzeugt genau +// eine Aufgabe, referenziert nur ueber IDs (kein Dokumentinhalt im +// Ereignis). +func TestDispatch_EventWithRuleCreatesExactlyOneTask(t *testing.T) { + e := newEnv(t) + ctx := context.Background() + + if _, err := e.rules.Put(ctx, Rule{ + ModuleName: "dms-fiktiv", ObjectType: "document", Trigger: "document.uploaded", + Kind: KindTask, AssigneeType: task.AssigneePerson, AssigneeID: "user-1", Title: "Dokument pruefen", + }); err != nil { + t.Fatalf("regel anlegen: %v", err) + } + + publish(t, e, "document.uploaded", "doc-1", "dedupe-1") + + n, err := e.dispatcher.Dispatch(ctx) + if err != nil { + t.Fatalf("dispatch: %v", err) + } + if n != 1 { + t.Fatalf("erwarte genau 1 dispatch, habe %d", n) + } + + tasks, err := task.NewStore(e.pool).ByAssignee(ctx, task.AssigneePerson, "user-1") + if err != nil { + t.Fatalf("aufgaben lesen: %v", err) + } + if len(tasks) != 1 { + t.Fatalf("erwarte genau eine aufgabe, habe %d", len(tasks)) + } + if len(tasks[0].References) != 1 || tasks[0].References[0].ObjectID != "doc-1" { + t.Fatalf("aufgabe referenziert nicht das ausloesende objekt: %+v", tasks[0].References) + } +} + +// Pruefung 1: Doppelzustellung desselben Ereignisses (gleicher DedupeKey) +// ergibt genau eine Aufgabe. +func TestDispatch_DuplicateDeliverySameEventCreatesOnlyOneTask(t *testing.T) { + e := newEnv(t) + ctx := context.Background() + + if _, err := e.rules.Put(ctx, Rule{ + ModuleName: "dms-fiktiv", ObjectType: "document", Trigger: "document.uploaded", + Kind: KindTask, AssigneeType: task.AssigneePerson, AssigneeID: "user-1", Title: "Dokument pruefen", + }); err != nil { + t.Fatalf("regel anlegen: %v", err) + } + + publish(t, e, "document.uploaded", "doc-1", "dedupe-1") + publish(t, e, "document.uploaded", "doc-1", "dedupe-1") + + if _, err := e.dispatcher.Dispatch(ctx); err != nil { + t.Fatalf("erster dispatch: %v", err) + } + // Zweiter Lauf nach "Wiederanlauf" -- muss ohne Fortschritt bleiben, + // weil die Ereigniszeile dedupliziert wurde. + if n, err := e.dispatcher.Dispatch(ctx); err != nil { + t.Fatalf("zweiter dispatch: %v", err) + } else if n != 0 { + t.Fatalf("zweiter dispatch-lauf sollte nichts mehr finden, hat aber %d verarbeitet", n) + } + + tasks, err := task.NewStore(e.pool).ByAssignee(ctx, task.AssigneePerson, "user-1") + if err != nil { + t.Fatalf("aufgaben lesen: %v", err) + } + if len(tasks) != 1 { + t.Fatalf("erwarte genau eine aufgabe trotz doppelter zustellung, habe %d", len(tasks)) + } +} + +// Pruefung 2: faellt die Verarbeitung aus (hier simuliert durch zwei +// getrennte Dispatch-Laeufe statt eines einzigen), gehen Ereignisse beim +// Wiederanlauf weder verloren noch werden sie doppelt verarbeitet. +func TestDispatch_SurvivesRestartWithoutLossOrDuplicates(t *testing.T) { + e := newEnv(t) + ctx := context.Background() + + if _, err := e.rules.Put(ctx, Rule{ + ModuleName: "dms-fiktiv", ObjectType: "document", Trigger: "document.uploaded", + Kind: KindTask, AssigneeType: task.AssigneePerson, AssigneeID: "user-1", Title: "Dokument pruefen", + }); err != nil { + t.Fatalf("regel anlegen: %v", err) + } + + publish(t, e, "document.uploaded", "doc-1", "dedupe-1") + publish(t, e, "document.uploaded", "doc-2", "dedupe-2") + + // Erster Lauf verarbeitet nur doc-1 (simulierter Ausfall danach). + n, err := e.dispatcher.Dispatch(ctx) + if err != nil { + t.Fatalf("erster dispatch: %v", err) + } + if n != 2 { + t.Fatalf("erwarte beide ereignisse im ersten lauf, habe %d", n) + } + + // "Wiederanlauf": erneuter Dispatch-Lauf darf nichts mehr finden. + n2, err := e.dispatcher.Dispatch(ctx) + if err != nil { + t.Fatalf("zweiter dispatch: %v", err) + } + if n2 != 0 { + t.Fatalf("wiederanlauf sollte keine weiteren aufgaben erzeugen, hat aber %d", n2) + } + + tasks, err := task.NewStore(e.pool).ByAssignee(ctx, task.AssigneePerson, "user-1") + if err != nil { + t.Fatalf("aufgaben lesen: %v", err) + } + if len(tasks) != 2 { + t.Fatalf("erwarte genau zwei aufgaben (kein verlust, keine dublette), habe %d", len(tasks)) + } +} + +// AK2: Ereignisse ohne konfigurierte Zuordnung erzeugen keine Aufgabe -- +// bleiben aber nachvollziehbar in module_events liegen statt verworfen +// zu werden. +func TestDispatch_EventWithoutRuleStaysPending(t *testing.T) { + e := newEnv(t) + ctx := context.Background() + + publish(t, e, "document.geloescht", "doc-9", "dedupe-9") + + n, err := e.dispatcher.Dispatch(ctx) + if err != nil { + t.Fatalf("dispatch: %v", err) + } + if n != 0 { + t.Fatalf("ohne regel darf keine aufgabe entstehen, habe %d", n) + } +} + +// AK2: eine Regel vom Typ "approval" erzeugt einen Genehmigungslauf statt +// einer Aufgabe. +func TestDispatch_ApprovalRuleCreatesApprovalTask(t *testing.T) { + e := newEnv(t) + ctx := context.Background() + + if _, err := e.rules.Put(ctx, Rule{ + ModuleName: "dms-fiktiv", ObjectType: "document", Trigger: "document.freigabe_noetig", + Kind: KindApproval, AssigneeID: "user-2", Title: "Freigabe", + }); err != nil { + t.Fatalf("regel anlegen: %v", err) + } + + publish(t, e, "document.freigabe_noetig", "doc-3", "dedupe-3") + + n, err := e.dispatcher.Dispatch(ctx) + if err != nil { + t.Fatalf("dispatch: %v", err) + } + if n != 1 { + t.Fatalf("erwarte genau 1 dispatch, habe %d", n) + } + + eng, err := approval.NewEngineWithApproval(e.pool) + if err != nil { + t.Fatalf("engine: %v", err) + } + open, err := approval.NewService(e.pool, eng).OpenForAssignee(ctx, "user-2") + if err != nil { + t.Fatalf("offene genehmigungen lesen: %v", err) + } + if len(open) != 1 { + t.Fatalf("erwarte genau einen offenen genehmigungslauf, habe %d", len(open)) + } +} + +// Doppelte Regel-Anlage fuer dieselbe Kombination wird abgewiesen statt +// eine bestehende Zuordnung stillschweigend zu ersetzen. +func TestRuleStore_Put_RejectsDuplicateForSameEvent(t *testing.T) { + e := newEnv(t) + ctx := context.Background() + + rule := Rule{ModuleName: "dms-fiktiv", ObjectType: "document", Trigger: "document.uploaded", Kind: KindTask, AssigneeType: task.AssigneePerson, AssigneeID: "user-1", Title: "A"} + if _, err := e.rules.Put(ctx, rule); err != nil { + t.Fatalf("erste regel: %v", err) + } + if _, err := e.rules.Put(ctx, rule); err != ErrRuleExists { + t.Fatalf("erwarte ErrRuleExists, habe %v", err) + } +} diff --git a/workflow/internal/moduleadapter/events.go b/workflow/internal/moduleadapter/events.go index 477ee2b3..7788e32a 100644 --- a/workflow/internal/moduleadapter/events.go +++ b/workflow/internal/moduleadapter/events.go @@ -3,47 +3,92 @@ package moduleadapter import ( "context" "fmt" + "log" + "time" "github.com/jackc/pgx/v5/pgxpool" ) +// EventContractVersion ist die aktuelle Version des Ereignisvertrags +// (ENG-09 Akzeptanzkriterium 1). Aeltere Versionen werden weiter +// angenommen; eine neue Version fuegt Felder hinzu statt bestehende +// umzudeuten. +const EventContractVersion = 1 + // Event ist ein publiziertes Ausloeser-Ereignis fuer einen registrierten -// Objekttyp (INT-13: Gegenstelle zur Registrierung aus Register). +// Objekttyp (INT-13: Gegenstelle zur Registrierung aus Register). Der +// Payload enthaelt bewusst nur IDs, Mandant und Zeitpunkt -- keine +// Dokumentinhalte; die holt das Workflow-Modul bei Bedarf ueber den +// Modul-Rueckkanal (VOR-04). type Event struct { ModuleName string ObjectType string Trigger string ObjectID string + TenantSlug string + // DedupeKey identifiziert dieselbe Ereignis-Ausloesung bei + // wiederholter Zustellung (mindestens einmal); leer ist zulaessig, + // bedeutet dann aber keinen Dedup-Schutz fuer dieses Ereignis. + DedupeKey string + Version int + OccurredAt time.Time Payload []byte } -// PublishEvent nimmt ein Ereignis entgegen und legt es ab. Ob der -// Objekttyp/Trigger tatsaechlich registriert ist, prueft der Aufrufer -// nicht hart durch (Ereignisse fuer noch nicht bekannte Trigger werden -// trotzdem gespeichert, damit ein spaeter registrierter Prozess sie -// nachlesen kann). -func PublishEvent(ctx context.Context, pool *pgxpool.Pool, ev Event) error { - if ev.ModuleName == "" || ev.ObjectType == "" || ev.Trigger == "" { - return fmt.Errorf("moduleadapter: modulname, objekttyp und ausloeser sind pflicht") +// PublishEvent nimmt ein Ereignis entgegen. Ereignisse mit abweichendem +// Mandanten werden protokolliert abgelehnt statt verarbeitet +// (Akzeptanzkriterium 3); Wiederholungen mit demselben DedupeKey landen +// auf derselben Zeile (Akzeptanzkriterium 2). accepted ist false bei +// Mandanten-Ablehnung, sonst true -- beides ist kein Fehler des Aufrufs +// selbst (der Ausfall bleibt fuer das sendende Modul unsichtbar). +func PublishEvent(ctx context.Context, pool *pgxpool.Pool, ownTenantSlug string, ev Event) (accepted bool, err error) { + if ev.ModuleName == "" || ev.ObjectType == "" || ev.Trigger == "" || ev.ObjectID == "" { + return false, fmt.Errorf("moduleadapter: modulname, objekttyp, ausloeser und objekt-id sind pflicht") + } + if ev.TenantSlug == "" { + return false, fmt.Errorf("moduleadapter: tenant_slug ist pflicht") } if ev.Payload == nil { ev.Payload = []byte("{}") } - _, err := pool.Exec(ctx, ` - INSERT INTO module_events (module_name, object_type, trigger, object_id, payload) - VALUES ($1, $2, $3, $4, $5) - `, ev.ModuleName, ev.ObjectType, ev.Trigger, ev.ObjectID, ev.Payload) - if err != nil { - return fmt.Errorf("moduleadapter: ereignis speichern: %w", err) + if ev.Version == 0 { + ev.Version = EventContractVersion } - return nil + if ev.OccurredAt.IsZero() { + ev.OccurredAt = time.Now() + } + + if ev.TenantSlug != ownTenantSlug { + reason := fmt.Sprintf("tenant_slug %q passt nicht zum Ziel-Mandanten", ev.TenantSlug) + log.Printf("moduleadapter: ereignis abgelehnt (%s), modul=%s objekttyp=%s trigger=%s objekt=%s", reason, ev.ModuleName, ev.ObjectType, ev.Trigger, ev.ObjectID) + _, err := pool.Exec(ctx, ` + INSERT INTO module_events (module_name, object_type, trigger, object_id, payload, tenant_slug, dedupe_key, event_version, occurred_at, accepted, rejected_reason) + VALUES ($1, $2, $3, $4, $5, $6, NULLIF($7, ''), $8, $9, false, $10) + ON CONFLICT (dedupe_key) WHERE dedupe_key <> '' DO NOTHING + `, ev.ModuleName, ev.ObjectType, ev.Trigger, ev.ObjectID, ev.Payload, ev.TenantSlug, ev.DedupeKey, ev.Version, ev.OccurredAt, reason) + if err != nil { + return false, fmt.Errorf("moduleadapter: abgelehntes ereignis protokollieren: %w", err) + } + return false, nil + } + + _, err = pool.Exec(ctx, ` + INSERT INTO module_events (module_name, object_type, trigger, object_id, payload, tenant_slug, dedupe_key, event_version, occurred_at, accepted) + VALUES ($1, $2, $3, $4, $5, $6, NULLIF($7, ''), $8, $9, true) + ON CONFLICT (dedupe_key) WHERE dedupe_key <> '' DO NOTHING + `, ev.ModuleName, ev.ObjectType, ev.Trigger, ev.ObjectID, ev.Payload, ev.TenantSlug, ev.DedupeKey, ev.Version, ev.OccurredAt) + if err != nil { + return false, fmt.Errorf("moduleadapter: ereignis speichern: %w", err) + } + return true, nil } -// EventsFor liefert alle Ereignisse eines Objekttyps zu Pruefzwecken. +// EventsFor liefert alle angenommenen Ereignisse eines Objekttyps zu +// Pruefzwecken. func EventsFor(ctx context.Context, pool *pgxpool.Pool, moduleName, objectType string) ([]Event, error) { rows, err := pool.Query(ctx, ` - SELECT module_name, object_type, trigger, object_id, payload - FROM module_events WHERE module_name = $1 AND object_type = $2 + SELECT module_name, object_type, trigger, object_id, payload, tenant_slug, COALESCE(dedupe_key, ''), event_version, occurred_at + FROM module_events WHERE module_name = $1 AND object_type = $2 AND accepted ORDER BY received_at `, moduleName, objectType) if err != nil { @@ -54,10 +99,34 @@ func EventsFor(ctx context.Context, pool *pgxpool.Pool, moduleName, objectType s var evs []Event for rows.Next() { var e Event - if err := rows.Scan(&e.ModuleName, &e.ObjectType, &e.Trigger, &e.ObjectID, &e.Payload); err != nil { + if err := rows.Scan(&e.ModuleName, &e.ObjectType, &e.Trigger, &e.ObjectID, &e.Payload, &e.TenantSlug, &e.DedupeKey, &e.Version, &e.OccurredAt); err != nil { return nil, fmt.Errorf("moduleadapter: ereignis-zeile lesen: %w", err) } evs = append(evs, e) } return evs, rows.Err() } + +// RejectedEvents liefert alle wegen Mandanten-Mismatch abgelehnten +// Ereignisse zu Pruefzwecken (Pruefung 3). +func RejectedEvents(ctx context.Context, pool *pgxpool.Pool) ([]Event, error) { + rows, err := pool.Query(ctx, ` + SELECT module_name, object_type, trigger, object_id, payload, tenant_slug, COALESCE(dedupe_key, ''), event_version, occurred_at + FROM module_events WHERE NOT accepted + ORDER BY received_at + `) + if err != nil { + return nil, fmt.Errorf("moduleadapter: abgelehnte ereignisse lesen: %w", err) + } + defer rows.Close() + + var evs []Event + for rows.Next() { + var e Event + if err := rows.Scan(&e.ModuleName, &e.ObjectType, &e.Trigger, &e.ObjectID, &e.Payload, &e.TenantSlug, &e.DedupeKey, &e.Version, &e.OccurredAt); err != nil { + return nil, fmt.Errorf("moduleadapter: abgelehnte ereignis-zeile lesen: %w", err) + } + evs = append(evs, e) + } + return evs, rows.Err() +} diff --git a/workflow/internal/moduleadapter/events_test.go b/workflow/internal/moduleadapter/events_test.go new file mode 100644 index 00000000..27df9de8 --- /dev/null +++ b/workflow/internal/moduleadapter/events_test.go @@ -0,0 +1,88 @@ +package moduleadapter + +import ( + "context" + "testing" +) + +// AK1: das Modul nimmt Ereignisse mit passendem Mandanten an. +func TestPublishEvent_AcceptsMatchingTenant(t *testing.T) { + pool := testPool(t) + ctx := context.Background() + + accepted, err := PublishEvent(ctx, pool, "mandant-a", Event{ + ModuleName: "dms-fiktiv", ObjectType: "document", Trigger: "document.uploaded", + ObjectID: "doc-1", TenantSlug: "mandant-a", DedupeKey: "k1", + }) + if err != nil { + t.Fatalf("publish: %v", err) + } + if !accepted { + t.Fatal("erwarte angenommenes ereignis") + } + + evs, err := EventsFor(ctx, pool, "dms-fiktiv", "document") + if err != nil { + t.Fatalf("events lesen: %v", err) + } + if len(evs) != 1 { + t.Fatalf("erwarte genau ein angenommenes ereignis, habe %d", len(evs)) + } +} + +// Pruefung 3 (Cross-Tenant negativ): ein Ereignis mit falschem Mandanten +// wird abgelehnt und protokolliert, statt verarbeitet zu werden. +func TestPublishEvent_RejectsForeignTenantAndLogsIt(t *testing.T) { + pool := testPool(t) + ctx := context.Background() + + accepted, err := PublishEvent(ctx, pool, "mandant-a", Event{ + ModuleName: "dms-fiktiv", ObjectType: "document", Trigger: "document.uploaded", + ObjectID: "doc-fremd", TenantSlug: "mandant-b", DedupeKey: "k-fremd", + }) + if err != nil { + t.Fatalf("publish: %v", err) + } + if accepted { + t.Fatal("ereignis mit fremdem mandanten haette abgelehnt werden muessen") + } + + evs, err := EventsFor(ctx, pool, "dms-fiktiv", "document") + if err != nil { + t.Fatalf("events lesen: %v", err) + } + if len(evs) != 0 { + t.Fatalf("abgelehntes ereignis darf nicht als angenommen zaehlen, habe %d", len(evs)) + } + + rejected, err := RejectedEvents(ctx, pool) + if err != nil { + t.Fatalf("abgelehnte events lesen: %v", err) + } + if len(rejected) != 1 || rejected[0].ObjectID != "doc-fremd" { + t.Fatalf("erwarte protokolliertes abgelehntes ereignis, habe %+v", rejected) + } +} + +// Doppelte Zustellung desselben Ereignisses (gleicher DedupeKey) legt +// keine zweite Zeile an. +func TestPublishEvent_DuplicateDedupeKeyIsIdempotent(t *testing.T) { + pool := testPool(t) + ctx := context.Background() + + ev := Event{ModuleName: "dms-fiktiv", ObjectType: "document", Trigger: "document.uploaded", ObjectID: "doc-2", TenantSlug: "mandant-a", DedupeKey: "k2"} + if _, err := PublishEvent(ctx, pool, "mandant-a", ev); err != nil { + t.Fatalf("erste zustellung: %v", err) + } + if _, err := PublishEvent(ctx, pool, "mandant-a", ev); err != nil { + t.Fatalf("zweite zustellung: %v", err) + } + + evs, err := EventsFor(ctx, pool, "dms-fiktiv", "document") + if err != nil { + t.Fatalf("events lesen: %v", err) + } + if len(evs) != 1 { + t.Fatalf("erwarte genau eine zeile trotz doppelter zustellung, habe %d", len(evs)) + } +} diff --git a/workflow/internal/moduleadapter/handler.go b/workflow/internal/moduleadapter/handler.go index ffcc7e52..2ad971fa 100644 --- a/workflow/internal/moduleadapter/handler.go +++ b/workflow/internal/moduleadapter/handler.go @@ -4,6 +4,7 @@ import ( "encoding/json" "errors" "net/http" + "time" "github.com/jackc/pgx/v5/pgxpool" @@ -13,13 +14,17 @@ import ( // Server bindet die Registrierungs-API an einen Postgres-Pool und // verlangt ein Service-Token bei jedem Aufruf (fail closed: ein leeres // oder falsches Token wird immer abgewiesen, Lehre aus RET-14/F-08). +// ownTenantSlug bindet den Server an genau einen Mandanten (ENG-09 +// Akzeptanzkriterium 1/3): Ereignisse mit abweichendem Mandanten werden +// abgelehnt statt verarbeitet. type Server struct { - pool *pgxpool.Pool - serviceToken string + pool *pgxpool.Pool + serviceToken string + ownTenantSlug string } -func NewServer(pool *pgxpool.Pool, serviceToken string) *Server { - return &Server{pool: pool, serviceToken: serviceToken} +func NewServer(pool *pgxpool.Pool, serviceToken, ownTenantSlug string) *Server { + return &Server{pool: pool, serviceToken: serviceToken, ownTenantSlug: ownTenantSlug} } func (s *Server) RegisterRoutes(mux *http.ServeMux) { @@ -126,6 +131,10 @@ type publishEventRequest struct { ObjectType string `json:"object_type"` Trigger string `json:"trigger"` ObjectID string `json:"object_id"` + TenantSlug string `json:"tenant_slug"` + DedupeKey string `json:"dedupe_key"` + Version int `json:"version"` + OccurredAt time.Time `json:"occurred_at"` Payload json.RawMessage `json:"payload"` } @@ -135,17 +144,25 @@ func (s *Server) handlePublishEvent(w http.ResponseWriter, r *http.Request) { writeError(w, http.StatusBadRequest, "ungueltiger anfrage-koerper") return } - err := PublishEvent(r.Context(), s.pool, Event{ + accepted, err := PublishEvent(r.Context(), s.pool, s.ownTenantSlug, Event{ ModuleName: req.ModuleName, ObjectType: req.ObjectType, Trigger: req.Trigger, ObjectID: req.ObjectID, + TenantSlug: req.TenantSlug, + DedupeKey: req.DedupeKey, + Version: req.Version, + OccurredAt: req.OccurredAt, Payload: req.Payload, }) if err != nil { writeError(w, http.StatusBadRequest, err.Error()) return } + if !accepted { + writeError(w, http.StatusForbidden, "ereignis fuer fremden mandanten abgelehnt") + return + } w.WriteHeader(http.StatusAccepted) } diff --git a/workflow/internal/moduleadapter/handler_test.go b/workflow/internal/moduleadapter/handler_test.go index b07d62ec..859222c2 100644 --- a/workflow/internal/moduleadapter/handler_test.go +++ b/workflow/internal/moduleadapter/handler_test.go @@ -12,7 +12,7 @@ func newTestServer(t *testing.T) (*Server, string) { t.Helper() pool := testPool(t) token := "test-service-token-secret" - return NewServer(pool, token), token + return NewServer(pool, token, "mandant-a"), token } func TestHandleRegister_RequiresServiceToken(t *testing.T) { diff --git a/workflow/migrations/tenant/0013_event_dispatch.down.sql b/workflow/migrations/tenant/0013_event_dispatch.down.sql new file mode 100644 index 00000000..96da8d1b --- /dev/null +++ b/workflow/migrations/tenant/0013_event_dispatch.down.sql @@ -0,0 +1,10 @@ +DROP TABLE event_dispatch_log; +DROP TABLE event_task_rules; +DROP INDEX module_events_dedupe_key_idx; +ALTER TABLE module_events DROP COLUMN processed_at; +ALTER TABLE module_events DROP COLUMN rejected_reason; +ALTER TABLE module_events DROP COLUMN accepted; +ALTER TABLE module_events DROP COLUMN occurred_at; +ALTER TABLE module_events DROP COLUMN event_version; +ALTER TABLE module_events DROP COLUMN dedupe_key; +ALTER TABLE module_events DROP COLUMN tenant_slug; diff --git a/workflow/migrations/tenant/0013_event_dispatch.up.sql b/workflow/migrations/tenant/0013_event_dispatch.up.sql new file mode 100644 index 00000000..121d686f --- /dev/null +++ b/workflow/migrations/tenant/0013_event_dispatch.up.sql @@ -0,0 +1,38 @@ +-- ENG-09: Ereignisvertrag (Tenant-Bindung, Idempotenz) und mandantenweite +-- Zuordnung von DMS-Ereignissen zu Aufgabe/Genehmigungslauf. +ALTER TABLE module_events ADD COLUMN tenant_slug TEXT NOT NULL DEFAULT ''; +ALTER TABLE module_events ADD COLUMN dedupe_key TEXT NOT NULL DEFAULT ''; +ALTER TABLE module_events ADD COLUMN event_version INT NOT NULL DEFAULT 1; +ALTER TABLE module_events ADD COLUMN occurred_at TIMESTAMPTZ NOT NULL DEFAULT now(); +ALTER TABLE module_events ADD COLUMN accepted BOOLEAN NOT NULL DEFAULT true; +ALTER TABLE module_events ADD COLUMN rejected_reason TEXT; +ALTER TABLE module_events ADD COLUMN processed_at TIMESTAMPTZ; + +-- Doppelte Zustellung desselben Ereignisses (mindestens einmal) faellt auf +-- dieselbe Zeile statt eine zweite anzulegen. +CREATE UNIQUE INDEX module_events_dedupe_key_idx ON module_events (dedupe_key) WHERE dedupe_key <> ''; + +-- Mandantenweite Zuordnung Ereignis -> Aufgabe/Genehmigungslauf. +CREATE TABLE event_task_rules ( + id UUID PRIMARY KEY DEFAULT gen_random_uuid(), + module_name TEXT NOT NULL, + object_type TEXT NOT NULL, + trigger TEXT NOT NULL, + kind TEXT NOT NULL CHECK (kind IN ('task', 'approval')), + assignee_type TEXT NOT NULL, + assignee_id TEXT NOT NULL, + title TEXT NOT NULL, + enabled BOOLEAN NOT NULL DEFAULT true, + created_at TIMESTAMPTZ NOT NULL DEFAULT now(), + UNIQUE (module_name, object_type, trigger) +); + +-- Nachweis, dass ein Ereignis genau einmal ausgeloest hat; die Zeile +-- entsteht in derselben Transaktion wie processed_at, damit ein Abbruch +-- zwischen Aufgabe-Anlegen und Markieren nicht zu einer Dublette fuehrt. +CREATE TABLE event_dispatch_log ( + event_id UUID PRIMARY KEY REFERENCES module_events(id), + kind TEXT NOT NULL, + created_object_id TEXT NOT NULL, + dispatched_at TIMESTAMPTZ NOT NULL DEFAULT now() +);