2026-09-27 19:05:21 +02:00
5 changed files with 466 additions and 0 deletions
@@ -0,0 +1,67 @@
+# TSK-05 – Prüfprotokoll
+
+Eskalations-Benachrichtigung, ruft Core-Dispatcher (CFG-02).
+
+Datum: 2026-09-27. Host: Entwicklungshost 192.168.1.204, Wegwerf-Cluster
+(`TEST_TENANT_DSN` aus `/etc/nexarch-dev/dms-dev.env`, `dms_dev`-Rolle, je
+Testlauf frisches Schema über `internal/testdb`). Branch
+`feature/tsk-05-eskalations-benachrichtigung-ruft-core-dispatcher`, Basis
+`origin/main` (Head `d936dca7`).
+
+## Umsetzung
+
+- `workflow/internal/notifyclient` — schlanker HTTP-Client für Core CFG-02
+  (`internal/notifyapi`, `POST /notify/enqueue`, `X-Service-Token`-Header).
+  Workflow ist ein physisch getrenntes Go-Modul (eigenes `go.mod`) und kann
+  CFG-02 daher nicht direkt importieren, gleiches Muster wie
+  `archive/internal/notifyclient` und `mail/internal/notifyclient`.
+- `workflow/internal/escalationnotify` — verbindet
+  `escalation.Store.ProcessDue` (TSK-03) mit dem neuen Client:
+  `DispatchDue` ruft für jede fällige Eskalation genau einmal
+  `notifyclient.Enqueue` auf (Ereignistyp `workflow.task.escalated`, Kanal
+  `email`, Empfänger = Aufgaben-Assignee aus `task.Store.Get`).
+  `RunAt` bündelt `ProcessDue` + `DispatchDue` in einem Aufruf, damit kein
+  Aufrufer ProcessDue-Ergebnisse zwischenspeichert und dadurch versehentlich
+  erneut verarbeitet.
+- Ein Fehlschlag von `Enqueue` (Netzwerk, Timeout, Fehlerstatus) wird
+  protokolliert (`log.Printf`) und übersprungen — die Verarbeitung der
+  restlichen Eskalationen und der Aufrufer selbst laufen unverändert weiter.
+- Keine Doppelbenachrichtigung: `escalation.Store.ProcessDue` markiert jede
+  Zeile atomar (`UPDATE ... WHERE fired_at IS NULL ... RETURNING`) — ein
+  zweiter Aufruf von `RunAt` mit demselben oder späterem `now` liefert die
+  bereits ausgelöste Eskalation nicht erneut, `DispatchDue` bekommt für sie
+  keinen Eintrag mehr und ruft den Dispatcher entsprechend nicht erneut auf.
+
+## Akzeptanzkriterien / Prüfungen
+
+| # | Akzeptanzkriterium | Prüfung | Ergebnis |
+|---|---|---|---|
+| 1 | Eskalation ruft den Core-Dispatcher mit korrektem Ereignistyp auf | Testaufruf gegen simulierten Core-Dispatcher erfolgreich | **Bestanden.** `TestRunAt_DispatchesEscalationWithCorrectEventType`: `httptest`-Server simuliert CFG-02, empfangenes `event_type` = `workflow.task.escalated`, `channel` = `email`, `dispatched` = 1. |
+| 2 | Fehlschlag des Dispatcher-Aufrufs bricht die Eskalation nicht ab | Nichterreichbarkeit des Dispatchers führt zu protokolliertem Fehler statt Absturz | **Bestanden.** `TestRunAt_DispatcherFailureDoesNotAbort`: Client zeigt auf unerreichbaren Port (127.0.0.1:1), `RunAt` liefert `err == nil` (kein Absturz/Abbruch der Eskalationsverarbeitung), Fehler erscheint protokolliert (`log.Printf`, im Testlauf sichtbar), `dispatched` = 0. |
+| 3 | Keine Doppelbenachrichtigung bei mehrfachem Joblauf | Wiederholter Joblauf erzeugt keinen doppelten Aufruf | **Bestanden.** `TestRunAt_RepeatedRunDoesNotDoubleDispatch`: zwei aufeinanderfolgende `RunAt`-Aufrufe für dieselbe fällige Eskalation — erster Lauf `dispatched` = 1, zweiter Lauf `dispatched` = 0, Gesamtzahl der Dispatcher-Aufrufe am simulierten Server = 1. |
+
+## Testläufe
+
+```
+go test ./internal/escalationnotify/... ./internal/notifyclient/... -v
+--- PASS: TestRunAt_DispatchesEscalationWithCorrectEventType
+--- PASS: TestRunAt_DispatcherFailureDoesNotAbort
+--- PASS: TestRunAt_RepeatedRunDoesNotDoubleDispatch
+--- PASS: TestEnqueue_SendsRequestAndParsesResponse
+--- PASS: TestEnqueue_ReturnsErrorOnServerFailure
+--- PASS: TestEnqueue_UnreachableEndpointReturnsErrorWithoutHanging
+ok
+```
+
+Zusätzlich vollständiger Modul-Lauf `go test ./...` im `workflow`-Modul:
+alle Pakete `ok` (keine Regression durch die neuen Pakete), `go build ./...`
+und `go vet ./...` ohne Befund.
+
+## Offene Punkte
+
+- Die konkrete Kanalwahl (E-Mail vs. In-App) und die Nutzerpräferenz-Prüfung
+  liegen laut Ticket-DNA bei CFG-02/CFG-03 (Core) und sind hier bewusst
+  nicht dupliziert.
+- Empfänger-Auflösung nutzt aktuell die Aufgaben-Assignee-ID 1:1 als
+  `user_id`/`recipient` an CFG-02 — eine feinere Auflösung (z.B. Gruppen auf
+  Mitgliederlisten) ist nicht Bestandteil dieser Kachel.
@@ -0,0 +1,67 @@
+// Package escalationnotify verbindet escalation.Store.ProcessDue (TSK-03)
+// mit Core CFG-02 (TSK-05): jede fällige Eskalation löst GENAU EINE
+// Benachrichtigung über den Core-Dispatcher aus, kein eigener Versandweg.
+package escalationnotify
+
+import (
+	"context"
+	"log"
+	"time"
+
+	"gitea.perlbach24.de/scripte/nexarch/workflow/internal/escalation"
+	"gitea.perlbach24.de/scripte/nexarch/workflow/internal/notifyclient"
+	"gitea.perlbach24.de/scripte/nexarch/workflow/internal/task"
+)
+
+// EventType ist der an CFG-02 gemeldete Ereignistyp (AC 1).
+const EventType = "workflow.task.escalated"
+
+// Channel ist der bei CFG-02 angefragte Kanal; die Kanal-Auswahl selbst
+// (CFG-03) liegt bei Core, nicht bei Workflow.
+const Channel = "email"
+
+// DispatchDue ruft für jede übergebene, bereits als fällig markierte
+// Eskalation (escalation.Store.ProcessDue) den Core-Dispatcher auf. Ein
+// wiederholter Joblauf reicht ProcessDue erneut auf — da ProcessDue jede
+// Zeile nur einmal als fällig zurückgibt (fired_at wird atomar gesetzt),
+// erzeugt ein wiederholter Aufruf von DispatchDue mit leerer escalations-
+// Liste keinen weiteren Dispatcher-Aufruf (AC 3).
+//
+// Ein Fehlschlag des Dispatcher-Aufrufs für eine Eskalation wird
+// protokolliert, bricht aber weder die restlichen Eskalationen noch den
+// Aufrufer ab (AC 2).
+func DispatchDue(ctx context.Context, escalations []escalation.Fired, tasks *task.Store, client *notifyclient.Client, tenantSlug string, logger *log.Logger) (dispatched int) {
+	if logger == nil {
+		logger = log.Default()
+	}
+	for _, e := range escalations {
+		t, err := tasks.Get(ctx, e.TaskID)
+		if err != nil {
+			logger.Printf("escalationnotify: aufgabe %q lesen fehlgeschlagen: %v", e.TaskID, err)
+			continue
+		}
+		payload := map[string]any{
+			"task_id":           e.TaskID,
+			"process_type":      e.ProcessType,
+			"escalation_action": e.EscalationAction,
+		}
+		if _, err := client.Enqueue(ctx, tenantSlug, t.AssigneeID, EventType, Channel, t.AssigneeID, payload); err != nil {
+			logger.Printf("escalationnotify: dispatcher-aufruf fuer aufgabe %q fehlgeschlagen: %v", e.TaskID, err)
+			continue
+		}
+		dispatched++
+	}
+	return dispatched
+}
+
+// RunAt ist der Einstiegspunkt für den zeitgesteuerten Job: ProcessDue
+// (TSK-03) und Dispatch (TSK-05) in einem Aufruf, damit kein Aufrufer
+// ProcessDue-Ergebnisse zwischenspeichern und dadurch versehentlich erneut
+// verarbeiten kann.
+func RunAt(ctx context.Context, store *escalation.Store, tasks *task.Store, client *notifyclient.Client, tenantSlug string, logger *log.Logger, now time.Time) (dispatched int, err error) {
+	_, escalations, err := store.ProcessDue(ctx, now)
+	if err != nil {
+		return 0, err
+	}
+	return DispatchDue(ctx, escalations, tasks, client, tenantSlug, logger), nil
+}
@@ -0,0 +1,190 @@
+package escalationnotify
+
+import (
+	"context"
+	"encoding/json"
+	"log"
+	"net/http"
+	"net/http/httptest"
+	"os"
+	"path/filepath"
+	"sync/atomic"
+	"testing"
+	"time"
+
+	"gitea.perlbach24.de/scripte/nexarch/workflow/internal/escalation"
+	"gitea.perlbach24.de/scripte/nexarch/workflow/internal/migrate"
+	"gitea.perlbach24.de/scripte/nexarch/workflow/internal/notifyclient"
+	"gitea.perlbach24.de/scripte/nexarch/workflow/internal/task"
+	"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 newTestStores(t *testing.T) (*escalation.Store, *task.Store) {
+	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)
+	}
+	return escalation.NewStore(pool), task.NewStore(pool)
+}
+
+func newDueTask(t *testing.T, ts *task.Store, es *escalation.Store, processType string, dueDate time.Time) string {
+	t.Helper()
+	ctx := context.Background()
+	id, err := ts.Create(ctx, task.Task{Title: "Testaufgabe", AssigneeType: task.AssigneePerson, AssigneeID: "user-escnotify"})
+	if err != nil {
+		t.Fatalf("Create: %v", err)
+	}
+	if err := ts.SetDueDate(ctx, id, dueDate); err != nil {
+		t.Fatalf("SetDueDate: %v", err)
+	}
+	if err := es.SetRule(ctx, escalation.Rule{ProcessType: processType, ReminderDaysBefore: []int{1}, EscalationAction: "notify-manager"}); err != nil {
+		t.Fatalf("SetRule: %v", err)
+	}
+	if err := es.Schedule(ctx, id, processType); err != nil {
+		t.Fatalf("Schedule: %v", err)
+	}
+	return id
+}
+
+// fakeDispatcher simuliert Core CFG-02s POST /notify/enqueue.
+func fakeDispatcher(t *testing.T, fail bool) (*httptest.Server, *int32) {
+	t.Helper()
+	var calls int32
+	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		if r.URL.Path != "/notify/enqueue" || r.Method != http.MethodPost {
+			http.NotFound(w, r)
+			return
+		}
+		if r.Header.Get("X-Service-Token") != "test-token" {
+			http.Error(w, "unauthorized", http.StatusUnauthorized)
+			return
+		}
+		atomic.AddInt32(&calls, 1)
+		if fail {
+			http.Error(w, "boom", http.StatusInternalServerError)
+			return
+		}
+		w.Header().Set("Content-Type", "application/json")
+		_ = json.NewEncoder(w).Encode(map[string]any{"job_id": "job-1", "skipped": false})
+	}))
+	return srv, &calls
+}
+
+// Prüfung 1 / AC1: Testaufruf gegen simulierten Core-Dispatcher erfolgreich,
+// mit korrektem Ereignistyp.
+func TestRunAt_DispatchesEscalationWithCorrectEventType(t *testing.T) {
+	es, ts := newTestStores(t)
+	ctx := context.Background()
+
+	due := time.Now().Add(-time.Hour)
+	taskID := newDueTask(t, ts, es, "runat-ac1", due)
+
+	var gotEventType, gotChannel string
+	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		var body map[string]any
+		_ = json.NewDecoder(r.Body).Decode(&body)
+		gotEventType, _ = body["event_type"].(string)
+		gotChannel, _ = body["channel"].(string)
+		w.Header().Set("Content-Type", "application/json")
+		_ = json.NewEncoder(w).Encode(map[string]any{"job_id": "job-1", "skipped": false})
+	}))
+	defer srv.Close()
+
+	client := notifyclient.New(srv.URL, "test-token")
+	dispatched, err := RunAt(ctx, es, ts, client, "mandant-a", log.Default(), time.Now())
+	if err != nil {
+		t.Fatalf("RunAt: %v", err)
+	}
+	if dispatched != 1 {
+		t.Fatalf("erwartete 1 versendete eskalation, habe %d", dispatched)
+	}
+	if gotEventType != EventType {
+		t.Fatalf("erwarteter ereignistyp %q, habe %q", EventType, gotEventType)
+	}
+	if gotChannel != Channel {
+		t.Fatalf("erwarteter kanal %q, habe %q", Channel, gotChannel)
+	}
+	_ = taskID
+}
+
+// Prüfung 2 / AC2: Nichterreichbarkeit des Dispatchers führt zu
+// protokolliertem Fehler statt Absturz -- die Eskalation selbst bleibt
+// verarbeitet (ProcessDue markiert fired_at unabhängig vom Versand).
+func TestRunAt_DispatcherFailureDoesNotAbort(t *testing.T) {
+	es, ts := newTestStores(t)
+	ctx := context.Background()
+
+	due := time.Now().Add(-time.Hour)
+	newDueTask(t, ts, es, "runat-ac2", due)
+
+	client := notifyclient.New("http://127.0.0.1:1", "test-token") // garantiert unerreichbar
+	dispatched, err := RunAt(ctx, es, ts, client, "mandant-a", log.Default(), time.Now())
+	if err != nil {
+		t.Fatalf("RunAt soll trotz Dispatcher-Fehler nicht abbrechen, habe: %v", err)
+	}
+	if dispatched != 0 {
+		t.Fatalf("erwartete 0 erfolgreich versendete eskalationen, habe %d", dispatched)
+	}
+}
+
+// Prüfung 3 / AC3: Wiederholter Joblauf erzeugt keinen doppelten Aufruf.
+func TestRunAt_RepeatedRunDoesNotDoubleDispatch(t *testing.T) {
+	es, ts := newTestStores(t)
+	ctx := context.Background()
+
+	due := time.Now().Add(-time.Hour)
+	newDueTask(t, ts, es, "runat-ac3", due)
+
+	srv, calls := fakeDispatcher(t, false)
+	defer srv.Close()
+	client := notifyclient.New(srv.URL, "test-token")
+
+	now := time.Now()
+	firstDispatched, err := RunAt(ctx, es, ts, client, "mandant-a", log.Default(), now)
+	if err != nil {
+		t.Fatalf("RunAt (1. lauf): %v", err)
+	}
+	if firstDispatched != 1 {
+		t.Fatalf("erwartete 1 versendete eskalation im ersten lauf, habe %d", firstDispatched)
+	}
+
+	secondDispatched, err := RunAt(ctx, es, ts, client, "mandant-a", log.Default(), now)
+	if err != nil {
+		t.Fatalf("RunAt (2. lauf): %v", err)
+	}
+	if secondDispatched != 0 {
+		t.Fatalf("erwartete 0 versendete eskalationen im zweiten lauf (keine doppelbenachrichtigung), habe %d", secondDispatched)
+	}
+	if got := atomic.LoadInt32(calls); got != 1 {
+		t.Fatalf("erwartete genau 1 dispatcher-aufruf insgesamt, habe %d", got)
+	}
+}
@@ -0,0 +1,83 @@
+// Package notifyclient ist Workflows Kopplung an Core CFG-02
+// (internal/notifyapi, POST /notify/enqueue). Workflow ist ein physisch
+// getrenntes Go-Modul und kann Core CFG-02 daher nicht direkt importieren
+// — nur über die HTTP-Schnittstelle aufrufen (gleiches Muster wie
+// archive/internal/notifyclient und mail/internal/notifyclient).
+package notifyclient
+
+import (
+	"bytes"
+	"context"
+	"encoding/json"
+	"fmt"
+	"io"
+	"net/http"
+	"time"
+)
+
+type Client struct {
+	baseURL      string
+	serviceToken string
+	httpClient   *http.Client
+}
+
+func New(baseURL, serviceToken string) *Client {
+	return &Client{
+		baseURL:      baseURL,
+		serviceToken: serviceToken,
+		httpClient:   &http.Client{Timeout: 5 * time.Second},
+	}
+}
+
+type enqueueRequest struct {
+	TenantSlug string         `json:"tenant_slug"`
+	UserID     string         `json:"user_id"`
+	EventType  string         `json:"event_type"`
+	Channel    string         `json:"channel"`
+	Recipient  string         `json:"recipient"`
+	Payload    map[string]any `json:"payload"`
+}
+
+// EnqueueResult spiegelt CFG-02s enqueueResponse.
+type EnqueueResult struct {
+	JobID   string
+	Skipped bool
+}
+
+// Enqueue löst ein Benachrichtigungs-Ereignis über CFG-02 aus. Jeder Fehler
+// (Transport, Timeout, unerwarteter Status) wird zurückgegeben — der
+// Aufrufer entscheidet, ob er ihn protokolliert und die Eskalation trotzdem
+// fortsetzt (AC 2).
+func (c *Client) Enqueue(ctx context.Context, tenantSlug, userID, eventType, channel, recipient string, payload map[string]any) (EnqueueResult, error) {
+	body, err := json.Marshal(enqueueRequest{
+		TenantSlug: tenantSlug, UserID: userID, EventType: eventType, Channel: channel, Recipient: recipient, Payload: payload,
+	})
+	if err != nil {
+		return EnqueueResult{}, fmt.Errorf("notifyclient: request kodieren: %w", err)
+	}
+	req, err := http.NewRequestWithContext(ctx, http.MethodPost, c.baseURL+"/notify/enqueue", bytes.NewReader(body))
+	if err != nil {
+		return EnqueueResult{}, fmt.Errorf("notifyclient: request bauen: %w", err)
+	}
+	req.Header.Set("Content-Type", "application/json")
+	req.Header.Set("X-Service-Token", c.serviceToken)
+
+	resp, err := c.httpClient.Do(req)
+	if err != nil {
+		return EnqueueResult{}, fmt.Errorf("notifyclient: aufruf fehlgeschlagen: %w", err)
+	}
+	defer func() { _ = resp.Body.Close() }()
+
+	if resp.StatusCode != http.StatusOK {
+		respBody, _ := io.ReadAll(resp.Body)
+		return EnqueueResult{}, fmt.Errorf("notifyclient: unerwarteter status %d: %s", resp.StatusCode, string(respBody))
+	}
+	var out struct {
+		JobID   string `json:"job_id"`
+		Skipped bool   `json:"skipped"`
+	}
+	if err := json.NewDecoder(resp.Body).Decode(&out); err != nil {
+		return EnqueueResult{}, fmt.Errorf("notifyclient: antwort dekodieren: %w", err)
+	}
+	return EnqueueResult{JobID: out.JobID, Skipped: out.Skipped}, nil
+}
@@ -0,0 +1,59 @@
+package notifyclient
+
+import (
+	"context"
+	"encoding/json"
+	"net/http"
+	"net/http/httptest"
+	"testing"
+)
+
+func TestEnqueue_SendsRequestAndParsesResponse(t *testing.T) {
+	var gotToken string
+	var gotBody map[string]any
+	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		if r.Method != http.MethodPost || r.URL.Path != "/notify/enqueue" {
+			http.NotFound(w, r)
+			return
+		}
+		gotToken = r.Header.Get("X-Service-Token")
+		_ = json.NewDecoder(r.Body).Decode(&gotBody)
+		w.Header().Set("Content-Type", "application/json")
+		_ = json.NewEncoder(w).Encode(map[string]any{"job_id": "job-42", "skipped": false})
+	}))
+	defer srv.Close()
+
+	client := New(srv.URL, "test-token")
+	res, err := client.Enqueue(context.Background(), "mandant-a", "user-1", "workflow.task.escalated", "email", "user-1", map[string]any{"task_id": "t-1"})
+	if err != nil {
+		t.Fatalf("Enqueue: %v", err)
+	}
+	if res.JobID != "job-42" || res.Skipped {
+		t.Fatalf("unerwartetes ergebnis: %+v", res)
+	}
+	if gotToken != "test-token" {
+		t.Fatalf("erwartete service-token-header, habe: %q", gotToken)
+	}
+	if gotBody["event_type"] != "workflow.task.escalated" {
+		t.Fatalf("unerwarteter event_type: %v", gotBody["event_type"])
+	}
+}
+
+func TestEnqueue_ReturnsErrorOnServerFailure(t *testing.T) {
+	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		w.WriteHeader(http.StatusInternalServerError)
+	}))
+	defer srv.Close()
+
+	client := New(srv.URL, "test-token")
+	if _, err := client.Enqueue(context.Background(), "mandant-a", "user-1", "workflow.task.escalated", "email", "user-1", nil); err == nil {
+		t.Fatal("erwartete fehler bei 500")
+	}
+}
+
+func TestEnqueue_UnreachableEndpointReturnsErrorWithoutHanging(t *testing.T) {
+	client := New("http://127.0.0.1:1", "test-token")
+	if _, err := client.Enqueue(context.Background(), "mandant-a", "user-1", "workflow.task.escalated", "email", "user-1", nil); err == nil {
+		t.Fatal("erwartete fehler bei nicht erreichbarem endpunkt")
+	}
+}