diff --git a/workflow/docs/TSK-05-PRUEFPROTOKOLL.md b/workflow/docs/TSK-05-PRUEFPROTOKOLL.md new file mode 100644 index 00000000..e2c3c337 --- /dev/null +++ b/workflow/docs/TSK-05-PRUEFPROTOKOLL.md @@ -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. diff --git a/workflow/internal/escalationnotify/escalationnotify.go b/workflow/internal/escalationnotify/escalationnotify.go new file mode 100644 index 00000000..4443484b --- /dev/null +++ b/workflow/internal/escalationnotify/escalationnotify.go @@ -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 +} diff --git a/workflow/internal/escalationnotify/escalationnotify_test.go b/workflow/internal/escalationnotify/escalationnotify_test.go new file mode 100644 index 00000000..36c36a08 --- /dev/null +++ b/workflow/internal/escalationnotify/escalationnotify_test.go @@ -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) + } +} diff --git a/workflow/internal/notifyclient/notifyclient.go b/workflow/internal/notifyclient/notifyclient.go new file mode 100644 index 00000000..c6cc0a0b --- /dev/null +++ b/workflow/internal/notifyclient/notifyclient.go @@ -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 +} diff --git a/workflow/internal/notifyclient/notifyclient_test.go b/workflow/internal/notifyclient/notifyclient_test.go new file mode 100644 index 00000000..ac40407b --- /dev/null +++ b/workflow/internal/notifyclient/notifyclient_test.go @@ -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") + } +}