sysopsandClaude Sonnet 5
6f532d8350
internal/notify: Dispatcher.Enqueue ist die EINE schmale Schnittstelle, ueber die Module Benachrichtigungen ausloesen (Akzeptanzkriterium 1) — kein Modul baut eigenen Versandcode. Warteschlange ist die Postgres-Tabelle notification_jobs (Projekt-Konvention statt Redis/AMQP), existiert ausschliesslich in der Datenbank, nicht im Prozessspeicher. Dispatcher.ProcessDue holt faellige Jobs per FOR UPDATE SKIP LOCKED (dieselbe Konvention wie internal/tenant.Lifecycle.ProcessDueDeletions) — serialisiert konkurrierende Worker/Module, verhindert doppelte Zustellung. Fehlschlag erhoeht attempts und plant next_attempt_at mit linearem Backoff; nach max_attempts wird der Job kontrolliert auf status=failed gesetzt statt endlos wiederholt zu werden (Akzeptanzkriterium 2). Sender ist eine schmale Schnittstelle fuer die eigentlichen Kanaele (E-Mail/In-App = CFG-03, nicht Teil dieser Kachel) — der Dispatcher kennt nur "zustellen oder nicht", keine Kanal-Details. Pruefungen (ausgefuehrt auf root@192.168.1.131, go build/vet/test PASS): 1. Neustart waehrend offener Zustellung verliert keine Nachricht — TestQueue_SurvivesRestartWithoutMessageLoss: Enqueue durch eine Dispatcher-Instanz, Verarbeitung durch eine komplett neue (simulierter Neustart), Nachricht wird trotzdem zugestellt. PASS. 2. Wiederholungslogik greift bei simuliertem Fehler und bricht kontrolliert ab — TestProcessDue_RetriesThenGivesUpAfterMaxAttempts: 3 Versuche bei max_attempts=3, danach status=failed, keine weitere Verarbeitung. PASS. 3. Zwei Module loesen gleichzeitig aus, beide korrekt zugestellt — TestProcessDue_ConcurrentDispatchBothDelivered: zwei parallele ProcessDue-Aufrufe, beide Nachrichten je genau einmal zugestellt, keine Doppelzustellung. PASS. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
82 lines
2.7 KiB
Go
82 lines
2.7 KiB
Go
// 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
|
|
}
|