ENG-09: workflow-trigger nach upload und auto-regel

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
2026-09-27 09:34:14 +02:00
1 parent b8b8197ec2
commit 6e586b9087
@@ -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://<rolle>:<pw>@localhost:5432/<db>?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`.
@@ -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"
+}
@@ -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)
+	}
+}
@@ -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()
+}
@@ -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))
+	}
+}
@@ -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)
 }

@@ -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) {
@@ -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;
@@ -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()
+);