Add correlation IDs to vote processing pipeline

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)

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
 }

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)
 		}

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)
 	}
 }
 

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"
@@ -17,9 +19,16 @@ 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"
 )
 
+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"
@@ -582,6 +591,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.

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)

internal/app/overlay.go

@@ -41,15 +41,15 @@ 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
 	}
 
 	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)
 	}
 

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)

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)}
+}

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")
+}

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)
 }