From 2a94f3c86bb830cb4910aaf8c69d05fd68f2cd17 Mon Sep 17 00:00:00 2001 From: Patrick Scheid Date: Sun, 22 Feb 2026 23:20:50 +0100 Subject: [PATCH 1/4] Add correlation IDs to vote processing pipeline Introduce a correlation package with a custom slog.Handler that automatically injects correlation_id from context into log entries. Each webhook notification and ticker refresh gets a unique 8-char hex ID, enabling end-to-end tracing across webhook receipt, vote processing, and WebSocket publish. --- internal/adapter/eventpublisher/publisher.go | 4 +- internal/adapter/twitch/webhook.go | 17 ++-- internal/adapter/websocket/publisher.go | 2 +- internal/app/overlay.go | 4 +- internal/app/ticker.go | 17 ++-- internal/platform/correlation/correlation.go | 62 ++++++++++++++ .../platform/correlation/correlation_test.go | 84 +++++++++++++++++++ internal/platform/logging/logger.go | 4 + 8 files changed, 175 insertions(+), 19 deletions(-) create mode 100644 internal/platform/correlation/correlation.go create mode 100644 internal/platform/correlation/correlation_test.go diff --git a/internal/adapter/eventpublisher/publisher.go b/internal/adapter/eventpublisher/publisher.go index 468cfce..b8791b3 100644 --- a/internal/adapter/eventpublisher/publisher.go +++ b/internal/adapter/eventpublisher/publisher.go @@ -36,10 +36,10 @@ func (ep *EventPublisher) PublishSentimentUpdated(ctx context.Context, broadcast func (ep *EventPublisher) PublishConfigChanged(ctx context.Context, broadcasterID string) error { if err := ep.configCache.InvalidateCache(ctx, broadcasterID); err != nil { - slog.Warn("Failed to invalidate Redis config cache", "broadcaster_id", broadcasterID, "error", err) + slog.WarnContext(ctx, "Failed to invalidate Redis config cache", "broadcaster_id", broadcasterID, "error", err) } if err := redis.PublishConfigInvalidation(ctx, ep.redisClient, broadcasterID); err != nil { - slog.Warn("Failed to publish config invalidation", "broadcaster_id", broadcasterID, "error", err) + slog.WarnContext(ctx, "Failed to publish config invalidation", "broadcaster_id", broadcasterID, "error", err) } return nil } diff --git a/internal/adapter/twitch/webhook.go b/internal/adapter/twitch/webhook.go index bbcc86e..de096a6 100644 --- a/internal/adapter/twitch/webhook.go +++ b/internal/adapter/twitch/webhook.go @@ -10,6 +10,7 @@ import ( "github.com/Its-donkey/kappopher/helix" "github.com/pscheid92/chatpulse/internal/adapter/metrics" "github.com/pscheid92/chatpulse/internal/domain" + "github.com/pscheid92/chatpulse/internal/platform/correlation" ) const webhookProcessingTimeout = 5 * time.Second @@ -54,25 +55,27 @@ func NewWebhookHandler(secret string, overlay domain.Overlay, presence ViewerPre } func (wh *WebhookHandler) handleNotification(msg *helix.EventSubWebhookMessage) { + ctx := correlation.WithID(context.Background(), correlation.NewID()) + if msg.SubscriptionType != helix.EventSubTypeChannelChatMessage { return } event, err := helix.ParseEventSubEvent[helix.ChannelChatMessageEvent](msg) if err != nil { - slog.Error("Failed to parse chat message event", "error", err) + slog.ErrorContext(ctx, "Failed to parse chat message event", "error", err) return } if !wh.presence.HasViewers(event.BroadcasterUserID) { - slog.Debug("Skipping vote: no viewers", "broadcaster", event.BroadcasterUserID) + slog.DebugContext(ctx, "Skipping vote: no viewers", "broadcaster", event.BroadcasterUserID) if wh.voteMetrics != nil { wh.voteMetrics.VotesProcessed.WithLabelValues("no_viewers").Inc() } return } - ctx, cancel := context.WithTimeout(context.Background(), webhookProcessingTimeout) + ctx, cancel := context.WithTimeout(ctx, webhookProcessingTimeout) defer cancel() start := time.Now() @@ -82,14 +85,14 @@ func (wh *WebhookHandler) handleNotification(msg *helix.EventSubWebhookMessage) } if errors.Is(err, context.DeadlineExceeded) { - slog.Warn("ProcessMessage timed out", "broadcaster", event.BroadcasterUserID, "timeout", webhookProcessingTimeout) + slog.WarnContext(ctx, "ProcessMessage timed out", "broadcaster", event.BroadcasterUserID, "timeout", webhookProcessingTimeout) if wh.voteMetrics != nil { wh.voteMetrics.VotesProcessed.WithLabelValues("timeout").Inc() } return } if err != nil { - slog.Error("ProcessMessage failed", "broadcaster", event.BroadcasterUserID, "error", err) + slog.ErrorContext(ctx, "ProcessMessage failed", "broadcaster", event.BroadcasterUserID, "error", err) if wh.voteMetrics != nil { wh.voteMetrics.VotesProcessed.WithLabelValues("error").Inc() } @@ -117,13 +120,13 @@ func (wh *WebhookHandler) handleNotification(msg *helix.EventSubWebhookMessage) wh.onVoteApplied(event.BroadcasterUserID) } - slog.Info("Vote processed via webhook", "user", event.ChatterUserID, "forRatio", snapshot.ForRatio, "againstRatio", snapshot.AgainstRatio, "totalVotes", snapshot.TotalVotes) + slog.InfoContext(ctx, "Vote processed via webhook", "user", event.ChatterUserID, "forRatio", snapshot.ForRatio, "againstRatio", snapshot.AgainstRatio, "totalVotes", snapshot.TotalVotes) if wh.publisher == nil { return } if pubErr := wh.publisher.PublishSentimentUpdated(ctx, event.BroadcasterUserID, snapshot); pubErr != nil { - slog.Error("Failed to publish sentiment update", "broadcaster", event.BroadcasterUserID, "error", pubErr) + slog.ErrorContext(ctx, "Failed to publish sentiment update", "broadcaster", event.BroadcasterUserID, "error", pubErr) } } diff --git a/internal/adapter/websocket/publisher.go b/internal/adapter/websocket/publisher.go index fbe2fa9..402748e 100644 --- a/internal/adapter/websocket/publisher.go +++ b/internal/adapter/websocket/publisher.go @@ -40,7 +40,7 @@ func (p *Publisher) PublishSentiment(ctx context.Context, broadcasterID string, config, err := p.configSource.GetConfigByBroadcaster(ctx, broadcasterID) if err != nil { - slog.Warn("Failed to get config for publish, using defaults", "broadcaster_id", broadcasterID, "error", err) + slog.WarnContext(ctx, "Failed to get config for publish, using defaults", "broadcaster_id", broadcasterID, "error", err) displayMode = string(domain.DisplayModeCombined) } else { displayMode = string(config.DisplayMode) diff --git a/internal/app/overlay.go b/internal/app/overlay.go index 57b8047..7f647f9 100644 --- a/internal/app/overlay.go +++ b/internal/app/overlay.go @@ -41,7 +41,7 @@ func (o *Overlay) ProcessMessage(ctx context.Context, broadcasterUserID, chatter debounced, err := o.debounce.IsDebounced(ctx, broadcasterUserID, chatterUserID) if err != nil { // Fail-open: let votes through on debounce errors rather than silently dropping them - slog.Warn("Debounce check failed, allowing vote", "broadcaster", broadcasterUserID, "chatter", chatterUserID, "error", err) + slog.WarnContext(ctx, "Debounce check failed, allowing vote", "broadcaster", broadcasterUserID, "chatter", chatterUserID, "error", err) } if debounced { return nil, domain.VoteDebounced, target, nil @@ -49,7 +49,7 @@ func (o *Overlay) ProcessMessage(ctx context.Context, broadcasterUserID, chatter snapshot, err := o.sentiment.RecordVote(ctx, broadcasterUserID, target, config.MemorySeconds) if err != nil { - slog.Error("RecordVote error", "error", err) + slog.ErrorContext(ctx, "RecordVote error", "error", err) return nil, domain.VoteNoMatch, target, fmt.Errorf("record vote failed: %w", err) } diff --git a/internal/app/ticker.go b/internal/app/ticker.go index 8762065..971917a 100644 --- a/internal/app/ticker.go +++ b/internal/app/ticker.go @@ -7,6 +7,7 @@ import ( "time" "github.com/pscheid92/chatpulse/internal/domain" + "github.com/pscheid92/chatpulse/internal/platform/correlation" ) const defaultTickInterval = 2 * time.Second @@ -63,25 +64,27 @@ func (t *SnapshotTicker) refresh(ctx context.Context) { t.mu.Unlock() for _, id := range ids { - cfg, err := t.configs.GetConfigByBroadcaster(ctx, id) + tickCtx := correlation.WithID(ctx, correlation.NewID()) + + cfg, err := t.configs.GetConfigByBroadcaster(tickCtx, id) if err != nil { - slog.Debug("Ticker: config lookup failed, removing broadcaster", "broadcaster", id, "error", err) + slog.DebugContext(tickCtx, "Ticker: config lookup failed, removing broadcaster", "broadcaster", id, "error", err) t.untrack(id) continue } - snap, err := t.store.GetSnapshot(ctx, id, cfg.MemorySeconds) + snap, err := t.store.GetSnapshot(tickCtx, id, cfg.MemorySeconds) if err != nil { - slog.Warn("Ticker: snapshot failed", "broadcaster", id, "error", err) + slog.WarnContext(tickCtx, "Ticker: snapshot failed", "broadcaster", id, "error", err) continue } - if err := t.publisher.PublishSentimentUpdated(ctx, id, snap); err != nil { - slog.Warn("Ticker: publish failed", "broadcaster", id, "error", err) + if err := t.publisher.PublishSentimentUpdated(tickCtx, id, snap); err != nil { + slog.WarnContext(tickCtx, "Ticker: publish failed", "broadcaster", id, "error", err) continue } - slog.Debug("Ticker: refreshed snapshot", "broadcaster", id, "forRatio", snap.ForRatio, "againstRatio", snap.AgainstRatio, "totalVotes", snap.TotalVotes) + slog.DebugContext(tickCtx, "Ticker: refreshed snapshot", "broadcaster", id, "forRatio", snap.ForRatio, "againstRatio", snap.AgainstRatio, "totalVotes", snap.TotalVotes) if snap.TotalVotes == 0 { t.untrack(id) diff --git a/internal/platform/correlation/correlation.go b/internal/platform/correlation/correlation.go new file mode 100644 index 0000000..58e8a5f --- /dev/null +++ b/internal/platform/correlation/correlation.go @@ -0,0 +1,62 @@ +package correlation + +import ( + "context" + "crypto/rand" + "encoding/hex" + "fmt" + "log/slog" +) + +type contextKey struct{} + +// NewID generates an 8-character hex correlation ID (4 random bytes). +func NewID() string { + b := make([]byte, 4) + _, _ = rand.Read(b) + return hex.EncodeToString(b) +} + +// WithID returns a new context carrying the given correlation ID. +func WithID(ctx context.Context, id string) context.Context { + return context.WithValue(ctx, contextKey{}, id) +} + +// ID extracts the correlation ID from ctx, returning ("", false) if not present. +func ID(ctx context.Context) (string, bool) { + id, ok := ctx.Value(contextKey{}).(string) + return id, ok && id != "" +} + +// Handler wraps an existing slog.Handler to automatically inject a +// "correlation_id" attribute when the context carries one. +type Handler struct { + inner slog.Handler +} + +// NewHandler creates a correlation-aware handler wrapping the given handler. +func NewHandler(inner slog.Handler) *Handler { + return &Handler{inner: inner} +} + +func (h *Handler) Enabled(ctx context.Context, level slog.Level) bool { + return h.inner.Enabled(ctx, level) +} + +func (h *Handler) Handle(ctx context.Context, r slog.Record) error { + if id, ok := ID(ctx); ok { + r.AddAttrs(slog.String("correlation_id", id)) + } + if err := h.inner.Handle(ctx, r); err != nil { + return fmt.Errorf("correlation handler: %w", err) + } + return nil +} + +func (h *Handler) WithAttrs(attrs []slog.Attr) slog.Handler { + return &Handler{inner: h.inner.WithAttrs(attrs)} +} + +func (h *Handler) WithGroup(name string) slog.Handler { + return &Handler{inner: h.inner.WithGroup(name)} +} diff --git a/internal/platform/correlation/correlation_test.go b/internal/platform/correlation/correlation_test.go new file mode 100644 index 0000000..9952ce4 --- /dev/null +++ b/internal/platform/correlation/correlation_test.go @@ -0,0 +1,84 @@ +package correlation + +import ( + "bytes" + "context" + "log/slog" + "testing" + + "github.com/stretchr/testify/assert" +) + +func TestNewID_Length(t *testing.T) { + id := NewID() + assert.Len(t, id, 8) +} + +func TestNewID_Unique(t *testing.T) { + ids := make(map[string]struct{}, 100) + for range 100 { + ids[NewID()] = struct{}{} + } + assert.Len(t, ids, 100) +} + +func TestWithID_and_ID_Roundtrip(t *testing.T) { + ctx := WithID(context.Background(), "abc12345") + id, ok := ID(ctx) + assert.True(t, ok) + assert.Equal(t, "abc12345", id) +} + +func TestID_Missing(t *testing.T) { + id, ok := ID(context.Background()) + assert.False(t, ok) + assert.Empty(t, id) +} + +func TestID_EmptyString(t *testing.T) { + ctx := WithID(context.Background(), "") + id, ok := ID(ctx) + assert.False(t, ok) + assert.Empty(t, id) +} + +func TestHandler_AddsCorrelationID(t *testing.T) { + var buf bytes.Buffer + inner := slog.NewTextHandler(&buf, &slog.HandlerOptions{Level: slog.LevelDebug}) + handler := NewHandler(inner) + logger := slog.New(handler) + + ctx := WithID(context.Background(), "test1234") + logger.InfoContext(ctx, "test message", "key", "value") + + output := buf.String() + assert.Contains(t, output, "correlation_id=test1234") + assert.Contains(t, output, "key=value") + assert.Contains(t, output, "test message") +} + +func TestHandler_NoCorrelationID_WhenMissing(t *testing.T) { + var buf bytes.Buffer + inner := slog.NewTextHandler(&buf, &slog.HandlerOptions{Level: slog.LevelDebug}) + handler := NewHandler(inner) + logger := slog.New(handler) + + logger.InfoContext(context.Background(), "no correlation") + + output := buf.String() + assert.NotContains(t, output, "correlation_id") +} + +func TestHandler_WithAttrs_PreservesCorrelation(t *testing.T) { + var buf bytes.Buffer + inner := slog.NewTextHandler(&buf, &slog.HandlerOptions{Level: slog.LevelDebug}) + handler := NewHandler(inner) + logger := slog.New(handler).With("component", "test") + + ctx := WithID(context.Background(), "attr1234") + logger.InfoContext(ctx, "with attrs") + + output := buf.String() + assert.Contains(t, output, "correlation_id=attr1234") + assert.Contains(t, output, "component=test") +} diff --git a/internal/platform/logging/logger.go b/internal/platform/logging/logger.go index 72eb949..53760bd 100644 --- a/internal/platform/logging/logger.go +++ b/internal/platform/logging/logger.go @@ -3,6 +3,8 @@ package logging import ( "log/slog" "os" + + "github.com/pscheid92/chatpulse/internal/platform/correlation" ) // Logger is the application-wide structured logger instance. @@ -39,6 +41,8 @@ func InitLogger(level, format string) { handler = slog.NewTextHandler(os.Stdout, opts) } + handler = correlation.NewHandler(handler) + Logger = slog.New(handler) slog.SetDefault(Logger) } From 11464b50709d6ad2e570d2e823ddf982a427c2e0 Mon Sep 17 00:00:00 2001 From: Patrick Scheid Date: Sun, 22 Feb 2026 23:28:54 +0100 Subject: [PATCH 2/4] Add correlation ID test assertion and fix overlay CSP Verify that webhook handler injects a correlation ID into the context passed to ProcessMessage. Also allow unpkg.com in connect-src CSP directive to unblock Centrifuge JS source map fetches. --- internal/adapter/httpserver/handlers_overlay.go | 2 +- internal/adapter/twitch/webhook_test.go | 5 +++++ 2 files changed, 6 insertions(+), 1 deletion(-) diff --git a/internal/adapter/httpserver/handlers_overlay.go b/internal/adapter/httpserver/handlers_overlay.go index 2604d0c..ed4c28e 100644 --- a/internal/adapter/httpserver/handlers_overlay.go +++ b/internal/adapter/httpserver/handlers_overlay.go @@ -27,7 +27,7 @@ func overlaySecurityHeaders() echo.MiddlewareFunc { "script-src 'self' 'unsafe-inline' https://unpkg.com; "+ "style-src 'self' 'unsafe-inline' https://fonts.googleapis.com; "+ "font-src 'self' https://fonts.gstatic.com; "+ - "connect-src 'self' wss: ws:; "+ + "connect-src 'self' wss: ws: https://unpkg.com; "+ "frame-ancestors *") return next(c) } diff --git a/internal/adapter/twitch/webhook_test.go b/internal/adapter/twitch/webhook_test.go index 3c681cf..960911e 100644 --- a/internal/adapter/twitch/webhook_test.go +++ b/internal/adapter/twitch/webhook_test.go @@ -17,6 +17,7 @@ import ( "github.com/Its-donkey/kappopher/helix" "github.com/pscheid92/chatpulse/internal/app" "github.com/pscheid92/chatpulse/internal/domain" + "github.com/pscheid92/chatpulse/internal/platform/correlation" "github.com/stretchr/testify/assert" ) @@ -582,6 +583,10 @@ func TestWebhook_ProcessMessageContextHasTimeout(t *testing.T) { assert.True(t, hasDeadline, "Context passed to ProcessMessage should have a deadline") assert.WithinDuration(t, time.Now().Add(webhookProcessingTimeout), deadline, 2*time.Second, "Context deadline should be approximately webhookProcessingTimeout from now") + + corrID, hasCorrID := correlation.ID(ctx) + assert.True(t, hasCorrID, "Context should carry a correlation ID") + assert.Len(t, corrID, 8, "Correlation ID should be 8 hex chars") } // TestWebhook_NoViewersSkipsProcessMessage verifies that votes are skipped when no viewers are watching. From 02f93e24ca8c9cc6aeb52866cbeaa33b656ad11c Mon Sep 17 00:00:00 2001 From: Patrick Scheid Date: Sun, 22 Feb 2026 23:34:53 +0100 Subject: [PATCH 3/4] Install correlation handler in webhook tests via TestMain Without this, the default slog handler ignores context values and correlation_id never appears in test log output. --- internal/adapter/twitch/webhook_test.go | 8 ++++++++ 1 file changed, 8 insertions(+) diff --git a/internal/adapter/twitch/webhook_test.go b/internal/adapter/twitch/webhook_test.go index 960911e..9ef32dd 100644 --- a/internal/adapter/twitch/webhook_test.go +++ b/internal/adapter/twitch/webhook_test.go @@ -6,8 +6,10 @@ import ( "crypto/sha256" "encoding/hex" "encoding/json" + "log/slog" "net/http" "net/http/httptest" + "os" "strconv" "strings" "sync" @@ -21,6 +23,12 @@ import ( "github.com/stretchr/testify/assert" ) +func TestMain(m *testing.M) { + handler := correlation.NewHandler(slog.NewTextHandler(os.Stderr, &slog.HandlerOptions{Level: slog.LevelDebug})) + slog.SetDefault(slog.New(handler)) + os.Exit(m.Run()) +} + const ( testWebhookSecret = "test-webhook-secret-1234567890" testChatterID = "chatter-1" From 0a24d1258e52dc2180276e5457475d027c7dc754 Mon Sep 17 00:00:00 2001 From: Patrick Scheid Date: Sun, 22 Feb 2026 23:48:36 +0100 Subject: [PATCH 4/4] Stop deleting EventSub conduit on shutdown The conduit and all its subscriptions were deleted on every graceful shutdown, requiring users to re-login via OAuth to restore them. Keeping the conduit alive across restarts lets findOrCreateConduit reuse it and configureShard update the webhook URL, so existing subscriptions survive container restarts. --- cmd/server/main.go | 7 ++----- 1 file changed, 2 insertions(+), 5 deletions(-) diff --git a/cmd/server/main.go b/cmd/server/main.go index a278feb..8b4048a 100644 --- a/cmd/server/main.go +++ b/cmd/server/main.go @@ -168,7 +168,7 @@ func initWebhookHandler(cfg *config.Config, ovl *app.Overlay, presence twitch.Vi return twitch.NewWebhookHandler(cfg.WebhookSecret, ovl, presence, publisher, onVoteApplied, voteMetrics) } -func runGracefulShutdown(srv *httpserver.Server, node *centrifuge.Node, conduitMgr *twitch.EventSubManager) <-chan struct{} { +func runGracefulShutdown(srv *httpserver.Server, node *centrifuge.Node) <-chan struct{} { done := make(chan struct{}) sigChan := make(chan os.Signal, 1) signal.Notify(sigChan, os.Interrupt, syscall.SIGTERM) @@ -186,9 +186,6 @@ func runGracefulShutdown(srv *httpserver.Server, node *centrifuge.Node, conduitM if err := node.Shutdown(ctx); err != nil { slog.Error("Centrifuge node shutdown error", "error", err) } - if err := conduitMgr.Cleanup(ctx); err != nil { - slog.Error("Failed to clean up conduit", "error", err) - } close(done) }() @@ -276,7 +273,7 @@ func main() { os.Exit(1) } - done := runGracefulShutdown(srv, ws.node, eventsubMgr) + done := runGracefulShutdown(srv, ws.node) if err := srv.Start(); err != nil && !errors.Is(err, http.ErrServerClosed) { slog.Error("Server error", "error", err)