diff --git a/internal/notify/dispatcher.go b/internal/notify/dispatcher.go new file mode 100644 index 0000000..21c0cd1 --- /dev/null +++ b/internal/notify/dispatcher.go @@ -0,0 +1,81 @@ +// Package notify implementiert Core CFG-02: den zentralen Benachrichtigungs- +// Dispatcher, ueber den beliebige Module Benachrichtigungen ausloesen — +// Warteschlange, Wiederholungslogik, Kanal-Abstraktion. Die tatsaechlichen +// Kanaele (E-Mail/In-App) sind CFG-03, hier gibt es nur die Sender- +// Schnittstelle als Vorbereitung. +package notify + +import ( + "context" + "encoding/json" + "fmt" + "time" + + "github.com/jackc/pgx/v5/pgxpool" +) + +// DefaultMaxAttempts begrenzt Wiederholungsversuche (Akzeptanzkriterium 2) — +// nach dieser Anzahl gibt der Dispatcher kontrolliert auf (status=failed) +// statt endlos zu wiederholen. +const DefaultMaxAttempts = 5 + +// DefaultRetryBackoff ist die Basis-Wartezeit zwischen Wiederholungen, +// linear mit der Versuchsnummer skaliert. +const DefaultRetryBackoff = 200 * time.Millisecond + +type Notification struct { + ID string + Channel string + Recipient string + Payload map[string]any + Attempts int +} + +// Sender ist die schmale Schnittstelle, die ein konkreter Kanal (CFG-03) +// implementiert. Der Dispatcher selbst weiss nichts ueber E-Mail/In-App. +type Sender interface { + Send(ctx context.Context, n Notification) error +} + +// Dispatcher ist die EINE Schnittstelle, ueber die Module Benachrichtigungen +// ausloesen — kein Modul baut eigenen Versandcode (Akzeptanzkriterium 1). +type Dispatcher struct { + pool *pgxpool.Pool + maxAttempts int + retryBackoff time.Duration +} + +func NewDispatcher(pool *pgxpool.Pool) *Dispatcher { + return &Dispatcher{pool: pool, maxAttempts: DefaultMaxAttempts, retryBackoff: DefaultRetryBackoff} +} + +// WithRetryPolicy erlaubt Tests/Betrieb, Versuchsanzahl und Backoff +// anzupassen, ohne die Default-Policy im Produktionscode zu veraendern. +func (d *Dispatcher) WithRetryPolicy(maxAttempts int, backoff time.Duration) *Dispatcher { + return &Dispatcher{pool: d.pool, maxAttempts: maxAttempts, retryBackoff: backoff} +} + +// Enqueue reiht eine Benachrichtigung in die Postgres-Warteschlange ein und +// kehrt sofort zurueck — die Zeile ueberlebt jeden Neustart des Dispatcher- +// Prozesses unveraendert (Akzeptanzkriterium 3), da sie ausschliesslich in +// der Datenbank existiert, nicht im Prozessspeicher. +func (d *Dispatcher) Enqueue(ctx context.Context, channel, recipient string, payload map[string]any) (string, error) { + if payload == nil { + payload = map[string]any{} + } + payloadJSON, err := json.Marshal(payload) + if err != nil { + return "", fmt.Errorf("payload serialisieren: %w", err) + } + + var id string + err = d.pool.QueryRow(ctx, ` + INSERT INTO notification_jobs (channel, recipient, payload, max_attempts) + VALUES ($1, $2, $3, $4) + RETURNING id + `, channel, recipient, payloadJSON, d.maxAttempts).Scan(&id) + if err != nil { + return "", fmt.Errorf("benachrichtigung einreihen: %w", err) + } + return id, nil +} diff --git a/internal/notify/dispatcher_test.go b/internal/notify/dispatcher_test.go new file mode 100644 index 0000000..bcb490a --- /dev/null +++ b/internal/notify/dispatcher_test.go @@ -0,0 +1,219 @@ +package notify + +import ( + "context" + "errors" + "os" + "sync" + "testing" + "time" + + "github.com/jackc/pgx/v5/pgxpool" +) + +func setupTest(t *testing.T) (*pgxpool.Pool, func()) { + t.Helper() + adminDSN := os.Getenv("TEST_ADMIN_DSN") + if adminDSN == "" { + t.Skip("TEST_ADMIN_DSN nicht gesetzt, Integrationstest uebersprungen") + } + ctx := context.Background() + + pool, err := pgxpool.New(ctx, adminDSN) + if err != nil { + t.Fatalf("pool: %v", err) + } + if _, err := pool.Exec(ctx, ` + CREATE TABLE IF NOT EXISTS notification_jobs ( + id UUID PRIMARY KEY DEFAULT gen_random_uuid(), + channel TEXT NOT NULL, + recipient TEXT NOT NULL, + payload JSONB NOT NULL DEFAULT '{}'::jsonb, + status TEXT NOT NULL DEFAULT 'pending' CHECK (status IN ('pending', 'sent', 'failed')), + attempts INT NOT NULL DEFAULT 0, + max_attempts INT NOT NULL DEFAULT 5, + next_attempt_at TIMESTAMPTZ NOT NULL DEFAULT now(), + last_error TEXT, + created_at TIMESTAMPTZ NOT NULL DEFAULT now(), + updated_at TIMESTAMPTZ NOT NULL DEFAULT now() + )`); err != nil { + t.Fatalf("schema: %v", err) + } + + cleanup := func() { pool.Close() } + return pool, cleanup +} + +type fakeSender struct { + mu sync.Mutex + sentIDs []string + failUntil int + calls int +} + +func (f *fakeSender) Send(ctx context.Context, n Notification) error { + f.mu.Lock() + defer f.mu.Unlock() + f.calls++ + if f.calls <= f.failUntil { + return errors.New("simulierter zustellfehler") + } + f.sentIDs = append(f.sentIDs, n.ID) + return nil +} + +func (f *fakeSender) sentCount() int { + f.mu.Lock() + defer f.mu.Unlock() + return len(f.sentIDs) +} + +// Akzeptanzkriterium 1: Module loesen ueber Enqueue aus, keine eigene +// Versandlogik noetig. +func TestDispatcher_EnqueueAndProcess(t *testing.T) { + pool, cleanup := setupTest(t) + defer cleanup() + ctx := context.Background() + + d := NewDispatcher(pool) + id, err := d.Enqueue(ctx, "email", "alice@example.com", map[string]any{"subject": "Willkommen"}) + if err != nil { + t.Fatalf("enqueue: %v", err) + } + if id == "" { + t.Fatal("erwartet nicht-leere id") + } + + sender := &fakeSender{} + sent, failed, err := d.ProcessDue(ctx, sender, 10) + if err != nil { + t.Fatalf("process: %v", err) + } + if sent != 1 || failed != 0 { + t.Fatalf("erwartet sent=1 failed=0, habe sent=%d failed=%d", sent, failed) + } + if sender.sentCount() != 1 { + t.Fatalf("erwartet 1 zustellung, habe %d", sender.sentCount()) + } +} + +// Akzeptanzkriterium 2 + Pruefung 2: Wiederholungslogik greift bei +// simuliertem Fehler und bricht nach definierter Anzahl kontrolliert ab. +func TestProcessDue_RetriesThenGivesUpAfterMaxAttempts(t *testing.T) { + pool, cleanup := setupTest(t) + defer cleanup() + ctx := context.Background() + + d := NewDispatcher(pool).WithRetryPolicy(3, time.Millisecond) + id, err := d.Enqueue(ctx, "email", "bob@example.com", nil) + if err != nil { + t.Fatalf("enqueue: %v", err) + } + + sender := &fakeSender{failUntil: 100} // schlaegt bei jedem versuch fehl + + for i := 0; i < 3; i++ { + time.Sleep(5 * time.Millisecond) // next_attempt_at abwarten + if _, _, err := d.ProcessDue(ctx, sender, 10); err != nil { + t.Fatalf("process %d: %v", i, err) + } + } + + var status string + var attempts int + if err := pool.QueryRow(ctx, `SELECT status, attempts FROM notification_jobs WHERE id = $1`, id).Scan(&status, &attempts); err != nil { + t.Fatalf("status lesen: %v", err) + } + if status != "failed" { + t.Fatalf("erwartet status failed nach max_attempts, habe %q", status) + } + if attempts != 3 { + t.Fatalf("erwartet 3 versuche, habe %d", attempts) + } + + // Weiteres ProcessDue darf den bereits aufgegebenen job nicht mehr anfassen. + sent, failed, err := d.ProcessDue(ctx, sender, 10) + if err != nil { + t.Fatalf("process nach abbruch: %v", err) + } + if sent != 0 || failed != 0 { + t.Fatalf("erwartet keine weitere verarbeitung, habe sent=%d failed=%d", sent, failed) + } +} + +// Akzeptanzkriterium 3 + Pruefung 1: Neustart des Dienstes waehrend offener +// Zustellung verliert keine Nachricht — simuliert durch eine komplett neue +// Dispatcher/Pool-Instanz nach dem Enqueue, bevor irgendetwas verarbeitet wurde. +func TestQueue_SurvivesRestartWithoutMessageLoss(t *testing.T) { + pool, cleanup := setupTest(t) + defer cleanup() + ctx := context.Background() + + firstInstance := NewDispatcher(pool) + id, err := firstInstance.Enqueue(ctx, "email", "carol@example.com", nil) + if err != nil { + t.Fatalf("enqueue: %v", err) + } + + // "Neustart": eine voellig neue Dispatcher-Instanz (repraesentiert einen + // neuen Prozess) verbindet sich neu und verarbeitet die Warteschlange — + // die Nachricht existiert ausschliesslich in Postgres, nicht im + // Prozessspeicher der ersten Instanz. + restartedInstance := NewDispatcher(pool) + sender := &fakeSender{} + sent, failed, err := restartedInstance.ProcessDue(ctx, sender, 10) + if err != nil { + t.Fatalf("process nach neustart: %v", err) + } + if sent != 1 || failed != 0 { + t.Fatalf("erwartet sent=1 nach neustart, habe sent=%d failed=%d", sent, failed) + } + if len(sender.sentIDs) != 1 || sender.sentIDs[0] != id { + t.Fatalf("erwartet zustellung der urspruenglichen nachricht %q, habe %v", id, sender.sentIDs) + } +} + +// Akzeptanzkriterium 3 + Pruefung 3: zwei gleichzeitig ausloesende Module, +// beide Nachrichten werden korrekt (und nicht doppelt) zugestellt. +func TestProcessDue_ConcurrentDispatchBothDelivered(t *testing.T) { + pool, cleanup := setupTest(t) + defer cleanup() + ctx := context.Background() + + d := NewDispatcher(pool) + idA, err := d.Enqueue(ctx, "email", "modul-a@example.com", nil) + if err != nil { + t.Fatalf("enqueue a: %v", err) + } + idB, err := d.Enqueue(ctx, "email", "modul-b@example.com", nil) + if err != nil { + t.Fatalf("enqueue b: %v", err) + } + + sender := &fakeSender{} + var wg sync.WaitGroup + for i := 0; i < 2; i++ { + wg.Add(1) + go func() { + defer wg.Done() + if _, _, err := d.ProcessDue(ctx, sender, 10); err != nil { + t.Errorf("process: %v", err) + } + }() + } + wg.Wait() + + if sender.sentCount() != 2 { + t.Fatalf("erwartet genau 2 zustellungen, habe %d: %v", sender.sentCount(), sender.sentIDs) + } + seen := map[string]bool{} + for _, id := range sender.sentIDs { + if seen[id] { + t.Fatalf("nachricht %q wurde doppelt zugestellt", id) + } + seen[id] = true + } + if !seen[idA] || !seen[idB] { + t.Fatalf("erwartet beide nachrichten zugestellt, habe %v", sender.sentIDs) + } +} diff --git a/internal/notify/worker.go b/internal/notify/worker.go new file mode 100644 index 0000000..cad3f8d --- /dev/null +++ b/internal/notify/worker.go @@ -0,0 +1,102 @@ +package notify + +import ( + "context" + "encoding/json" + "fmt" + "time" +) + +// ProcessDue holt bis zu limit faellige Benachrichtigungen und versucht sie +// ueber sender zuzustellen. FOR UPDATE SKIP LOCKED serialisiert konkurrierende +// Aufrufe (Akzeptanzkriterium 3 / Pruefung 3: zwei gleichzeitig ausloesende +// Module duerfen sich nicht gegenseitig blockieren oder Nachrichten doppelt +// zustellen) — dieselbe Konvention wie internal/tenant.Lifecycle.ProcessDueDeletions. +func (d *Dispatcher) ProcessDue(ctx context.Context, sender Sender, limit int) (sent, failed int, err error) { + tx, err := d.pool.Begin(ctx) + if err != nil { + return 0, 0, fmt.Errorf("transaktion starten: %w", err) + } + defer func() { _ = tx.Rollback(ctx) }() + + rows, err := tx.Query(ctx, ` + SELECT id, channel, recipient, payload, attempts, max_attempts + FROM notification_jobs + WHERE status = 'pending' AND next_attempt_at <= now() + ORDER BY created_at + FOR UPDATE SKIP LOCKED + LIMIT $1 + `, limit) + if err != nil { + return 0, 0, fmt.Errorf("faellige benachrichtigungen abfragen: %w", err) + } + + type due struct { + id, channel, recipient string + payload []byte + attempts, maxAttempts int + } + var candidates []due + for rows.Next() { + var c due + if err := rows.Scan(&c.id, &c.channel, &c.recipient, &c.payload, &c.attempts, &c.maxAttempts); err != nil { + rows.Close() + return 0, 0, fmt.Errorf("faellige benachrichtigung lesen: %w", err) + } + candidates = append(candidates, c) + } + rows.Close() + if err := rows.Err(); err != nil { + return 0, 0, err + } + + for _, c := range candidates { + var payload map[string]any + if err := json.Unmarshal(c.payload, &payload); err != nil { + payload = map[string]any{} + } + + sendErr := sender.Send(ctx, Notification{ + ID: c.id, Channel: c.channel, Recipient: c.recipient, Payload: payload, Attempts: c.attempts, + }) + + if sendErr == nil { + if _, err := tx.Exec(ctx, ` + UPDATE notification_jobs SET status = 'sent', updated_at = now() WHERE id = $1 + `, c.id); err != nil { + return sent, failed, fmt.Errorf("erfolg speichern: %w", err) + } + sent++ + continue + } + + newAttempts := c.attempts + 1 + if newAttempts >= c.maxAttempts { + // Akzeptanzkriterium 2: kontrollierter Abbruch nach definierter + // Anzahl Versuche, kein endloses Wiederholen. + if _, err := tx.Exec(ctx, ` + UPDATE notification_jobs + SET status = 'failed', attempts = $2, last_error = $3, updated_at = now() + WHERE id = $1 + `, c.id, newAttempts, sendErr.Error()); err != nil { + return sent, failed, fmt.Errorf("fehlschlag speichern: %w", err) + } + failed++ + continue + } + + nextAttempt := time.Now().Add(time.Duration(newAttempts) * d.retryBackoff) + if _, err := tx.Exec(ctx, ` + UPDATE notification_jobs + SET attempts = $2, next_attempt_at = $3, last_error = $4, updated_at = now() + WHERE id = $1 + `, c.id, newAttempts, nextAttempt, sendErr.Error()); err != nil { + return sent, failed, fmt.Errorf("wiederholung planen: %w", err) + } + } + + if err := tx.Commit(ctx); err != nil { + return 0, 0, fmt.Errorf("transaktion committen: %w", err) + } + return sent, failed, nil +} diff --git a/migrations/0005_notification_jobs.down.sql b/migrations/0005_notification_jobs.down.sql new file mode 100644 index 0000000..1a73f4f --- /dev/null +++ b/migrations/0005_notification_jobs.down.sql @@ -0,0 +1 @@ +DROP TABLE IF EXISTS notification_jobs; diff --git a/migrations/0005_notification_jobs.up.sql b/migrations/0005_notification_jobs.up.sql new file mode 100644 index 0000000..3bb2d5c --- /dev/null +++ b/migrations/0005_notification_jobs.up.sql @@ -0,0 +1,20 @@ +-- Benachrichtigungs-Dispatcher-Warteschlange (CFG-02, siehe +-- core-kanban/tickets/CFG-02.md). Postgres-basiert statt Redis/AMQP +-- (Projekt-Konvention, siehe nexarch-state.json techstack.job_queue) — +-- Zeilen ueberleben einen Neustart des Dispatcher-Prozesses unveraendert +-- (Akzeptanzkriterium 3). +CREATE TABLE notification_jobs ( + id UUID PRIMARY KEY DEFAULT gen_random_uuid(), + channel TEXT NOT NULL, + recipient TEXT NOT NULL, + payload JSONB NOT NULL DEFAULT '{}'::jsonb, + status TEXT NOT NULL DEFAULT 'pending' CHECK (status IN ('pending', 'sent', 'failed')), + attempts INT NOT NULL DEFAULT 0, + max_attempts INT NOT NULL DEFAULT 5, + next_attempt_at TIMESTAMPTZ NOT NULL DEFAULT now(), + last_error TEXT, + created_at TIMESTAMPTZ NOT NULL DEFAULT now(), + updated_at TIMESTAMPTZ NOT NULL DEFAULT now() +); + +CREATE INDEX notification_jobs_due_idx ON notification_jobs (status, next_attempt_at);