From 065e379f7dca9052636cb69e232ca09a50649264 Mon Sep 17 00:00:00 2001 From: Patrick Scheid Date: Sun, 22 Feb 2026 23:02:59 +0100 Subject: [PATCH] Add Prometheus metrics for HTTP, votes, cache, and WebSocket Instrument the application with Prometheus metrics covering HTTP request duration/count, vote pipeline throughput and latency, config cache hit rates, and WebSocket connection tracking. Exposes a /metrics endpoint with standard go_* and process_* collectors alongside chatpulse_* metrics. Expands ProcessMessage to return VoteTarget for per-direction vote counting. --- cmd/server/main.go | 25 ++++++--- go.mod | 2 +- go.sum | 2 + internal/adapter/httpserver/routes.go | 7 +++ internal/adapter/httpserver/server.go | 15 ++++-- internal/adapter/metrics/cache.go | 37 +++++++++++++ internal/adapter/metrics/http.go | 70 +++++++++++++++++++++++++ internal/adapter/metrics/metrics.go | 24 +++++++++ internal/adapter/metrics/vote.go | 35 +++++++++++++ internal/adapter/metrics/websocket.go | 30 +++++++++++ internal/adapter/redis/config_cache.go | 29 ++++++++-- internal/adapter/twitch/webhook.go | 34 +++++++++++- internal/adapter/twitch/webhook_test.go | 10 ++-- internal/adapter/websocket/node.go | 14 +++-- internal/adapter/websocket/publisher.go | 10 +++- internal/app/overlay.go | 14 ++--- internal/app/overlay_test.go | 23 +++++--- internal/app/service_test.go | 4 +- internal/domain/overlay.go | 2 +- 19 files changed, 340 insertions(+), 47 deletions(-) create mode 100644 internal/adapter/metrics/cache.go create mode 100644 internal/adapter/metrics/http.go create mode 100644 internal/adapter/metrics/metrics.go create mode 100644 internal/adapter/metrics/vote.go create mode 100644 internal/adapter/metrics/websocket.go diff --git a/cmd/server/main.go b/cmd/server/main.go index ad0dd2a..a278feb 100644 --- a/cmd/server/main.go +++ b/cmd/server/main.go @@ -17,6 +17,7 @@ import ( "github.com/jackc/pgx/v5/pgxpool" "github.com/pscheid92/chatpulse/internal/adapter/eventpublisher" "github.com/pscheid92/chatpulse/internal/adapter/httpserver" + "github.com/pscheid92/chatpulse/internal/adapter/metrics" "github.com/pscheid92/chatpulse/internal/adapter/postgres" "github.com/pscheid92/chatpulse/internal/adapter/redis" "github.com/pscheid92/chatpulse/internal/adapter/twitch" @@ -125,8 +126,8 @@ type realtimeInfra struct { presence *websocket.PresenceChecker } -func setupWebSocket(streamerRepo domain.StreamerRepository, redisAddr string, configCache domain.ConfigSource) (realtimeInfra, error) { - node, err := websocket.NewNode(streamerRepo) +func setupWebSocket(streamerRepo domain.StreamerRepository, redisAddr string, configCache domain.ConfigSource, wsMetrics *metrics.WebSocketMetrics) (realtimeInfra, error) { + node, err := websocket.NewNode(streamerRepo, wsMetrics) if err != nil { return realtimeInfra{}, fmt.Errorf("create Centrifuge node: %w", err) } @@ -142,7 +143,7 @@ func setupWebSocket(streamerRepo domain.StreamerRepository, redisAddr string, co return realtimeInfra{ node: node, - publisher: websocket.NewPublisher(node, configCache), + publisher: websocket.NewPublisher(node, configCache, wsMetrics), presence: websocket.NewPresenceChecker(node), }, nil } @@ -163,8 +164,8 @@ func initEventSubManager(cfg *config.Config, eventSubRepo *postgres.EventSubRepo return mgr, nil } -func initWebhookHandler(cfg *config.Config, ovl *app.Overlay, presence twitch.ViewerPresence, publisher domain.EventPublisher, onVoteApplied func(string)) *twitch.WebhookHandler { - return twitch.NewWebhookHandler(cfg.WebhookSecret, ovl, presence, publisher, onVoteApplied) +func initWebhookHandler(cfg *config.Config, ovl *app.Overlay, presence twitch.ViewerPresence, publisher domain.EventPublisher, onVoteApplied func(string), voteMetrics *metrics.VoteMetrics) *twitch.WebhookHandler { + return twitch.NewWebhookHandler(cfg.WebhookSecret, ovl, presence, publisher, onVoteApplied, voteMetrics) } func runGracefulShutdown(srv *httpserver.Server, node *centrifuge.Node, conduitMgr *twitch.EventSubManager) <-chan struct{} { @@ -229,9 +230,17 @@ func main() { defer rc.stopEviction() defer rc.cancelPubsub() + // Metrics setup + reg := metrics.NewRegistry() + httpMetrics := metrics.NewHTTPMetrics(reg) + voteMetrics := metrics.NewVoteMetrics(reg) + cacheMetrics := metrics.NewCacheMetrics(reg) + wsMetrics := metrics.NewWebSocketMetrics(reg) + rc.configCache.SetCacheMetrics(cacheMetrics) + ovl := app.NewOverlay(rc.configCache, rc.sentiment, rc.debouncer) - ws, err := setupWebSocket(db.streamerRepo, rc.client.Options().Addr, rc.configCache) + ws, err := setupWebSocket(db.streamerRepo, rc.client.Options().Addr, rc.configCache, wsMetrics) if err != nil { slog.Error("WebSocket setup failed", "error", err) os.Exit(1) @@ -252,7 +261,7 @@ func main() { } slog.Info("EventSub configured", "callback_url", cfg.WebhookCallbackURL) - webhookHdlr := initWebhookHandler(cfg, ovl, ws.presence, publisher, ticker.Track) + webhookHdlr := initWebhookHandler(cfg, ovl, ws.presence, publisher, ticker.Track, voteMetrics) appSvc := app.NewService(db.streamerRepo, db.configRepo, ovl, eventsubMgr, publisher) healthChecks := slices.Concat(db.healthChecks, rc.healthChecks) @@ -261,7 +270,7 @@ func main() { CheckOrigin: websocket.NewCheckOrigin(cfg.WebhookCallbackURL, cfg.AppEnv == "development"), }) - srv, err := httpserver.NewServer(cfg, appSvc, ws.presence, wsHandler, http.HandlerFunc(webhookHdlr.HandleEventSub), eventsubMgr, healthChecks) + srv, err := httpserver.NewServer(cfg, appSvc, ws.presence, wsHandler, http.HandlerFunc(webhookHdlr.HandleEventSub), eventsubMgr, healthChecks, httpMetrics, metrics.Handler(reg)) if err != nil { slog.Error("Failed to create server", "error", err) os.Exit(1) diff --git a/go.mod b/go.mod index e20acd5..36d47d2 100644 --- a/go.mod +++ b/go.mod @@ -10,6 +10,7 @@ require ( github.com/jackc/tern/v2 v2.3.5 github.com/joho/godotenv v1.5.1 github.com/labstack/echo/v4 v4.15.0 + github.com/prometheus/client_golang v1.23.2 github.com/pscheid92/uuid v0.1.1 github.com/redis/go-redis/v9 v9.18.0 github.com/stretchr/testify v1.11.1 @@ -86,7 +87,6 @@ require ( github.com/planetscale/vtprotobuf v0.6.1-0.20240319094008-0393e58bdf10 // indirect github.com/pmezard/go-difflib v1.0.0 // indirect github.com/power-devops/perfstat v0.0.0-20240221224432-82ca36839d55 // indirect - github.com/prometheus/client_golang v1.23.2 // indirect github.com/prometheus/client_model v0.6.2 // indirect github.com/prometheus/common v0.67.5 // indirect github.com/prometheus/procfs v0.19.2 // indirect diff --git a/go.sum b/go.sum index 78c62fe..b5f844d 100644 --- a/go.sum +++ b/go.sum @@ -117,6 +117,8 @@ github.com/kr/pretty v0.3.1 h1:flRD4NNwYAUpkphVc1HcthR4KEIFJ65n8Mw5qdRn3LE= github.com/kr/pretty v0.3.1/go.mod h1:hoEshYVHaxMs3cyo3Yncou5ZscifuDolrwPKZanG3xk= github.com/kr/text v0.2.0 h1:5Nx0Ya0ZqY2ygV366QzturHI13Jq95ApcVaJBhpS+AY= github.com/kr/text v0.2.0/go.mod h1:eLer722TekiGuMkidMxC/pM04lWEeraHUUmBw8l2grE= +github.com/kylelemons/godebug v1.1.0 h1:RPNrshWIDI6G2gRW9EHilWtl7Z6Sb1BR0xunSBf0SNc= +github.com/kylelemons/godebug v1.1.0/go.mod h1:9/0rRGxNHcop5bhtWyNeEfOS8JIWk580+fNqagV/RAw= github.com/labstack/echo/v4 v4.15.0 h1:hoRTKWcnR5STXZFe9BmYun9AMTNeSbjHi2vtDuADJ24= github.com/labstack/echo/v4 v4.15.0/go.mod h1:xmw1clThob0BSVRX1CRQkGQ/vjwcpOMjQZSZa9fKA/c= github.com/labstack/gommon v0.4.2 h1:F8qTUNXgG1+6WQmqoUWnz8WiEU60mXVVw0P4ht1WRA0= diff --git a/internal/adapter/httpserver/routes.go b/internal/adapter/httpserver/routes.go index 1182fcc..eb26df1 100644 --- a/internal/adapter/httpserver/routes.go +++ b/internal/adapter/httpserver/routes.go @@ -9,6 +9,9 @@ import ( ) func (s *Server) registerRoutes() { + if s.httpMetrics != nil { + s.echo.Use(s.httpMetrics.Middleware()) + } s.echo.Use(s.setupRequestLoggerMiddleware()) s.echo.Use(middleware.Recover()) s.echo.Use(ErrorHandlingMiddleware()) @@ -35,6 +38,10 @@ func (s *Server) registerRoutes() { s.echo.GET("/", s.handleLanding) + if s.metricsHandler != nil { + s.echo.GET("/metrics", echo.WrapHandler(s.metricsHandler)) + } + s.registerHealthRoutes() s.registerAuthRoutes(csrfMiddleware, authRL) s.registerDashboardRoutes(csrfMiddleware, dashboardRL) diff --git a/internal/adapter/httpserver/server.go b/internal/adapter/httpserver/server.go index e6f2d11..437cdd7 100644 --- a/internal/adapter/httpserver/server.go +++ b/internal/adapter/httpserver/server.go @@ -11,6 +11,7 @@ import ( "github.com/gorilla/sessions" "github.com/labstack/echo/v4" + "github.com/pscheid92/chatpulse/internal/adapter/metrics" "github.com/pscheid92/chatpulse/internal/adapter/twitch" "github.com/pscheid92/chatpulse/internal/app" "github.com/pscheid92/chatpulse/internal/domain" @@ -42,13 +43,15 @@ type Server struct { templates *template.Template - oauthClient twitchOAuthClient - sessionStore *sessions.CookieStore - healthChecks []HealthCheck - startTime time.Time + oauthClient twitchOAuthClient + sessionStore *sessions.CookieStore + healthChecks []HealthCheck + startTime time.Time + httpMetrics *metrics.HTTPMetrics + metricsHandler http.Handler } -func NewServer(cfg *config.Config, app appService, presence twitch.ViewerPresence, websocketHandler http.Handler, webhookHandler http.Handler, twitchService domain.EventSubService, healthChecks []HealthCheck) (*Server, error) { +func NewServer(cfg *config.Config, app appService, presence twitch.ViewerPresence, websocketHandler http.Handler, webhookHandler http.Handler, twitchService domain.EventSubService, healthChecks []HealthCheck, httpMetrics *metrics.HTTPMetrics, metricsHandler http.Handler) (*Server, error) { templates, err := template.ParseFS(web.TemplateFiles, "templates/*.html") if err != nil { return nil, fmt.Errorf("failed to parse templates: %w", err) @@ -73,6 +76,8 @@ func NewServer(cfg *config.Config, app appService, presence twitch.ViewerPresenc templates: templates, healthChecks: healthChecks, startTime: time.Now(), + httpMetrics: httpMetrics, + metricsHandler: metricsHandler, } srv.registerRoutes() diff --git a/internal/adapter/metrics/cache.go b/internal/adapter/metrics/cache.go new file mode 100644 index 0000000..449b819 --- /dev/null +++ b/internal/adapter/metrics/cache.go @@ -0,0 +1,37 @@ +package metrics + +import "github.com/prometheus/client_golang/prometheus" + +// CacheMetrics holds Prometheus metrics for config cache performance. +type CacheMetrics struct { + Hits *prometheus.CounterVec + Misses *prometheus.CounterVec + Invalidations prometheus.Counter +} + +// NewCacheMetrics creates and registers cache metrics on the given registry. +func NewCacheMetrics(reg prometheus.Registerer) *CacheMetrics { + m := &CacheMetrics{ + Hits: prometheus.NewCounterVec(prometheus.CounterOpts{ + Namespace: namespace, + Subsystem: "config_cache", + Name: "hits_total", + Help: "Total number of config cache hits, by layer.", + }, []string{"layer"}), + Misses: prometheus.NewCounterVec(prometheus.CounterOpts{ + Namespace: namespace, + Subsystem: "config_cache", + Name: "misses_total", + Help: "Total number of config cache misses, by layer.", + }, []string{"layer"}), + Invalidations: prometheus.NewCounter(prometheus.CounterOpts{ + Namespace: namespace, + Subsystem: "config_cache", + Name: "invalidations_total", + Help: "Total number of config cache invalidations.", + }), + } + + reg.MustRegister(m.Hits, m.Misses, m.Invalidations) + return m +} diff --git a/internal/adapter/metrics/http.go b/internal/adapter/metrics/http.go new file mode 100644 index 0000000..66e0f9a --- /dev/null +++ b/internal/adapter/metrics/http.go @@ -0,0 +1,70 @@ +package metrics + +import ( + "strconv" + "strings" + + "github.com/labstack/echo/v4" + "github.com/prometheus/client_golang/prometheus" +) + +// HTTPMetrics holds Prometheus metrics for HTTP request tracking. +type HTTPMetrics struct { + RequestDuration *prometheus.HistogramVec + RequestsTotal *prometheus.CounterVec + InFlightGauge prometheus.Gauge +} + +// NewHTTPMetrics creates and registers HTTP metrics on the given registry. +func NewHTTPMetrics(reg prometheus.Registerer) *HTTPMetrics { + m := &HTTPMetrics{ + RequestDuration: prometheus.NewHistogramVec(prometheus.HistogramOpts{ + Namespace: namespace, + Subsystem: "http", + Name: "request_duration_seconds", + Help: "Duration of HTTP requests in seconds.", + Buckets: prometheus.DefBuckets, + }, []string{"method", "route", "status_code"}), + RequestsTotal: prometheus.NewCounterVec(prometheus.CounterOpts{ + Namespace: namespace, + Subsystem: "http", + Name: "requests_total", + Help: "Total number of HTTP requests.", + }, []string{"method", "route", "status_code"}), + InFlightGauge: prometheus.NewGauge(prometheus.GaugeOpts{ + Namespace: namespace, + Subsystem: "http", + Name: "in_flight_requests", + Help: "Number of HTTP requests currently being processed.", + }), + } + + reg.MustRegister(m.RequestDuration, m.RequestsTotal, m.InFlightGauge) + return m +} + +// Middleware returns an Echo middleware that records HTTP metrics. +// It skips /metrics and /health/* endpoints. +func (m *HTTPMetrics) Middleware() echo.MiddlewareFunc { + return func(next echo.HandlerFunc) echo.HandlerFunc { + return func(c echo.Context) error { + path := c.Path() + if path == "/metrics" || strings.HasPrefix(path, "/health/") { + return next(c) + } + + m.InFlightGauge.Inc() + defer m.InFlightGauge.Dec() + + timer := prometheus.NewTimer(prometheus.ObserverFunc(func(v float64) { + status := strconv.Itoa(c.Response().Status) + m.RequestDuration.WithLabelValues(c.Request().Method, path, status).Observe(v) + m.RequestsTotal.WithLabelValues(c.Request().Method, path, status).Inc() + })) + + err := next(c) + timer.ObserveDuration() + return err + } + } +} diff --git a/internal/adapter/metrics/metrics.go b/internal/adapter/metrics/metrics.go new file mode 100644 index 0000000..a6f1efe --- /dev/null +++ b/internal/adapter/metrics/metrics.go @@ -0,0 +1,24 @@ +package metrics + +import ( + "net/http" + + "github.com/prometheus/client_golang/prometheus" + "github.com/prometheus/client_golang/prometheus/collectors" + "github.com/prometheus/client_golang/prometheus/promhttp" +) + +const namespace = "chatpulse" + +// NewRegistry creates a Prometheus registry with Go runtime and process collectors. +func NewRegistry() *prometheus.Registry { + reg := prometheus.NewRegistry() + reg.MustRegister(collectors.NewGoCollector()) + reg.MustRegister(collectors.NewProcessCollector(collectors.ProcessCollectorOpts{})) + return reg +} + +// Handler returns an http.Handler that serves Prometheus metrics. +func Handler(reg *prometheus.Registry) http.Handler { + return promhttp.HandlerFor(reg, promhttp.HandlerOpts{}) +} diff --git a/internal/adapter/metrics/vote.go b/internal/adapter/metrics/vote.go new file mode 100644 index 0000000..849ef74 --- /dev/null +++ b/internal/adapter/metrics/vote.go @@ -0,0 +1,35 @@ +package metrics + +import "github.com/prometheus/client_golang/prometheus" + +// VoteMetrics holds Prometheus metrics for the vote processing pipeline. +type VoteMetrics struct { + VotesProcessed *prometheus.CounterVec + ProcessingDuration prometheus.Histogram + VotesByTarget *prometheus.CounterVec +} + +// NewVoteMetrics creates and registers vote pipeline metrics on the given registry. +func NewVoteMetrics(reg prometheus.Registerer) *VoteMetrics { + m := &VoteMetrics{ + VotesProcessed: prometheus.NewCounterVec(prometheus.CounterOpts{ + Namespace: namespace, + Name: "votes_processed_total", + Help: "Total number of votes processed, by result.", + }, []string{"result"}), + ProcessingDuration: prometheus.NewHistogram(prometheus.HistogramOpts{ + Namespace: namespace, + Name: "votes_processing_duration_seconds", + Help: "Duration of vote processing in seconds.", + Buckets: []float64{0.0005, 0.001, 0.005, 0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1.0}, + }), + VotesByTarget: prometheus.NewCounterVec(prometheus.CounterOpts{ + Namespace: namespace, + Name: "votes_by_target_total", + Help: "Total number of applied votes, by target direction.", + }, []string{"target"}), + } + + reg.MustRegister(m.VotesProcessed, m.ProcessingDuration, m.VotesByTarget) + return m +} diff --git a/internal/adapter/metrics/websocket.go b/internal/adapter/metrics/websocket.go new file mode 100644 index 0000000..4d2442c --- /dev/null +++ b/internal/adapter/metrics/websocket.go @@ -0,0 +1,30 @@ +package metrics + +import "github.com/prometheus/client_golang/prometheus" + +// WebSocketMetrics holds Prometheus metrics for WebSocket connections. +type WebSocketMetrics struct { + ActiveConnections prometheus.Gauge + MessagesPublished prometheus.Counter +} + +// NewWebSocketMetrics creates and registers WebSocket metrics on the given registry. +func NewWebSocketMetrics(reg prometheus.Registerer) *WebSocketMetrics { + m := &WebSocketMetrics{ + ActiveConnections: prometheus.NewGauge(prometheus.GaugeOpts{ + Namespace: namespace, + Subsystem: "websocket", + Name: "active_connections", + Help: "Number of active WebSocket connections.", + }), + MessagesPublished: prometheus.NewCounter(prometheus.CounterOpts{ + Namespace: namespace, + Subsystem: "websocket", + Name: "messages_published_total", + Help: "Total number of WebSocket messages published.", + }), + } + + reg.MustRegister(m.ActiveConnections, m.MessagesPublished) + return m +} diff --git a/internal/adapter/redis/config_cache.go b/internal/adapter/redis/config_cache.go index 4ea4a11..6674c20 100644 --- a/internal/adapter/redis/config_cache.go +++ b/internal/adapter/redis/config_cache.go @@ -11,15 +11,17 @@ import ( goredis "github.com/redis/go-redis/v9" + "github.com/pscheid92/chatpulse/internal/adapter/metrics" "github.com/pscheid92/chatpulse/internal/domain" ) const configCacheTTL = 1 * time.Hour type ConfigCacheRepo struct { - rdb goredis.Cmdable - configs domain.ConfigRepository - mem *memoryCache + rdb goredis.Cmdable + configs domain.ConfigRepository + mem *memoryCache + cacheMetrics *metrics.CacheMetrics } func NewConfigCacheRepo(rdb goredis.Cmdable, configs domain.ConfigRepository, memCacheTTL time.Duration) *ConfigCacheRepo { @@ -57,17 +59,34 @@ func (r *ConfigCacheRepo) StartEvictionTimer(interval time.Duration) func() { } } +// SetCacheMetrics configures cache metrics for observability. +func (r *ConfigCacheRepo) SetCacheMetrics(m *metrics.CacheMetrics) { + r.cacheMetrics = m +} + func (r *ConfigCacheRepo) GetConfigByBroadcaster(ctx context.Context, broadcasterID string) (domain.OverlayConfig, error) { // Layer 1: in-memory cache if config, ok := r.mem.get(broadcasterID); ok { + if r.cacheMetrics != nil { + r.cacheMetrics.Hits.WithLabelValues("memory").Inc() + } return config, nil } + if r.cacheMetrics != nil { + r.cacheMetrics.Misses.WithLabelValues("memory").Inc() + } // Layer 2: Redis cache if config, ok := r.getCached(ctx, broadcasterID); ok { + if r.cacheMetrics != nil { + r.cacheMetrics.Hits.WithLabelValues("redis").Inc() + } r.mem.set(broadcasterID, config) return config, nil } + if r.cacheMetrics != nil { + r.cacheMetrics.Misses.WithLabelValues("redis").Inc() + } // Layer 3: PostgreSQL result, err := r.configs.GetByBroadcasterID(ctx, broadcasterID) @@ -84,6 +103,10 @@ func (r *ConfigCacheRepo) GetConfigByBroadcaster(ctx context.Context, broadcaste func (r *ConfigCacheRepo) InvalidateCache(ctx context.Context, broadcasterID string) error { r.mem.invalidate(broadcasterID) + if r.cacheMetrics != nil { + r.cacheMetrics.Invalidations.Inc() + } + ck := configCacheKey(broadcasterID) if err := r.rdb.Del(ctx, ck).Err(); err != nil { return fmt.Errorf("failed to invalidate config cache: %w", err) diff --git a/internal/adapter/twitch/webhook.go b/internal/adapter/twitch/webhook.go index 7a79f7c..bbcc86e 100644 --- a/internal/adapter/twitch/webhook.go +++ b/internal/adapter/twitch/webhook.go @@ -8,6 +8,7 @@ import ( "time" "github.com/Its-donkey/kappopher/helix" + "github.com/pscheid92/chatpulse/internal/adapter/metrics" "github.com/pscheid92/chatpulse/internal/domain" ) @@ -25,14 +26,16 @@ type WebhookHandler struct { presence ViewerPresence publisher domain.EventPublisher onVoteApplied func(broadcasterID string) + voteMetrics *metrics.VoteMetrics } -func NewWebhookHandler(secret string, overlay domain.Overlay, presence ViewerPresence, publisher domain.EventPublisher, onVoteApplied func(string)) *WebhookHandler { +func NewWebhookHandler(secret string, overlay domain.Overlay, presence ViewerPresence, publisher domain.EventPublisher, onVoteApplied func(string), voteMetrics *metrics.VoteMetrics) *WebhookHandler { wh := &WebhookHandler{ overlay: overlay, presence: presence, publisher: publisher, onVoteApplied: onVoteApplied, + voteMetrics: voteMetrics, } wh.handler = helix.NewEventSubWebhookHandler( @@ -63,26 +66,53 @@ func (wh *WebhookHandler) handleNotification(msg *helix.EventSubWebhookMessage) if !wh.presence.HasViewers(event.BroadcasterUserID) { slog.Debug("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) defer cancel() - snapshot, result, err := wh.overlay.ProcessMessage(ctx, event.BroadcasterUserID, event.ChatterUserID, event.Message.Text) + start := time.Now() + snapshot, result, target, err := wh.overlay.ProcessMessage(ctx, event.BroadcasterUserID, event.ChatterUserID, event.Message.Text) + if wh.voteMetrics != nil { + wh.voteMetrics.ProcessingDuration.Observe(time.Since(start).Seconds()) + } + if errors.Is(err, context.DeadlineExceeded) { slog.Warn("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) + if wh.voteMetrics != nil { + wh.voteMetrics.VotesProcessed.WithLabelValues("error").Inc() + } return } + if wh.voteMetrics != nil { + wh.voteMetrics.VotesProcessed.WithLabelValues(result.String()).Inc() + } + if result != domain.VoteApplied { return } + if wh.voteMetrics != nil { + switch target { + case domain.VoteTargetPositive: + wh.voteMetrics.VotesByTarget.WithLabelValues("for").Inc() + case domain.VoteTargetNegative: + wh.voteMetrics.VotesByTarget.WithLabelValues("against").Inc() + } + } + if wh.onVoteApplied != nil { wh.onVoteApplied(event.BroadcasterUserID) } diff --git a/internal/adapter/twitch/webhook_test.go b/internal/adapter/twitch/webhook_test.go index e97cb2c..3c681cf 100644 --- a/internal/adapter/twitch/webhook_test.go +++ b/internal/adapter/twitch/webhook_test.go @@ -241,12 +241,12 @@ func newContextCapturingOverlay() *contextCapturingOverlay { return &contextCapturingOverlay{called: make(chan struct{}, 1)} } -func (o *contextCapturingOverlay) ProcessMessage(ctx context.Context, _, _, _ string) (*domain.WindowSnapshot, domain.VoteResult, error) { +func (o *contextCapturingOverlay) ProcessMessage(ctx context.Context, _, _, _ string) (*domain.WindowSnapshot, domain.VoteResult, domain.VoteTarget, error) { o.mu.Lock() o.capturedCtx = ctx o.mu.Unlock() o.called <- struct{}{} - return nil, domain.VoteNoMatch, nil + return nil, domain.VoteNoMatch, domain.VoteTargetNone, nil } func (o *contextCapturingOverlay) Reset(_ context.Context, _ string) error { @@ -278,7 +278,7 @@ func setupWebhookTest(t *testing.T) (*WebhookHandler, *testStore, string) { debouncer := newTestDebouncer() ovl := app.NewOverlay(store, store, debouncer) - handler := NewWebhookHandler(testWebhookSecret, ovl, alwaysHasViewers, nil, nil) + handler := NewWebhookHandler(testWebhookSecret, ovl, alwaysHasViewers, nil, nil, nil) return handler, store, broadcasterID } @@ -558,7 +558,7 @@ func TestWebhook_ReplayAttack(t *testing.T) { // TestWebhook_ProcessMessageContextHasTimeout verifies that ProcessMessage receives a context with a deadline set. func TestWebhook_ProcessMessageContextHasTimeout(t *testing.T) { ovl := newContextCapturingOverlay() - handler := NewWebhookHandler(testWebhookSecret, ovl, alwaysHasViewers, nil, nil) + handler := NewWebhookHandler(testWebhookSecret, ovl, alwaysHasViewers, nil, nil, nil) body := makeChatMessageBody("broadcaster-123", "yes") req := makeSignedNotification(testWebhookSecret, body) @@ -592,7 +592,7 @@ func TestWebhook_NoViewersSkipsProcessMessage(t *testing.T) { ovl := app.NewOverlay(store, store, newTestDebouncer()) // presence always returns false — no viewers connected - handler := NewWebhookHandler(testWebhookSecret, ovl, viewerPresenceFunc(func(_ string) bool { return false }), nil, nil) + handler := NewWebhookHandler(testWebhookSecret, ovl, viewerPresenceFunc(func(_ string) bool { return false }), nil, nil, nil) body := makeChatMessageBody(broadcasterID, "yes") req := makeSignedNotification(testWebhookSecret, body) diff --git a/internal/adapter/websocket/node.go b/internal/adapter/websocket/node.go index 6b3e266..c75531c 100644 --- a/internal/adapter/websocket/node.go +++ b/internal/adapter/websocket/node.go @@ -6,11 +6,12 @@ import ( "log/slog" "github.com/centrifugal/centrifuge" + "github.com/pscheid92/chatpulse/internal/adapter/metrics" "github.com/pscheid92/chatpulse/internal/domain" "github.com/pscheid92/uuid" ) -func NewNode(userRepo domain.StreamerRepository) (*centrifuge.Node, error) { +func NewNode(userRepo domain.StreamerRepository, wsMetrics *metrics.WebSocketMetrics) (*centrifuge.Node, error) { conf := centrifuge.Config{LogLevel: centrifuge.LogLevelInfo, LogHandler: slogHandler} node, err := centrifuge.New(conf) if err != nil { @@ -18,7 +19,7 @@ func NewNode(userRepo domain.StreamerRepository) (*centrifuge.Node, error) { } node.OnConnecting(onConnecting(userRepo)) - node.OnConnect(onConnect()) + node.OnConnect(onConnect(wsMetrics)) return node, nil } @@ -55,10 +56,14 @@ func onConnecting(userRepo domain.StreamerRepository) func(ctx context.Context, } } -func onConnect() func(client *centrifuge.Client) { +func onConnect(wsMetrics *metrics.WebSocketMetrics) func(client *centrifuge.Client) { return func(client *centrifuge.Client) { slog.Debug("Client connected", "client_id", client.ID(), "user_id", client.UserID()) + if wsMetrics != nil { + wsMetrics.ActiveConnections.Inc() + } + client.OnSubscribe(func(e centrifuge.SubscribeEvent, cb centrifuge.SubscribeCallback) { options := centrifuge.SubscribeOptions{EmitPresence: true} cb(centrifuge.SubscribeReply{Options: options}, nil) @@ -66,6 +71,9 @@ func onConnect() func(client *centrifuge.Client) { client.OnDisconnect(func(e centrifuge.DisconnectEvent) { slog.Debug("Client disconnected", "client_id", client.ID(), "reason", e.Reason) + if wsMetrics != nil { + wsMetrics.ActiveConnections.Dec() + } }) } } diff --git a/internal/adapter/websocket/publisher.go b/internal/adapter/websocket/publisher.go index fd3e8cc..fbe2fa9 100644 --- a/internal/adapter/websocket/publisher.go +++ b/internal/adapter/websocket/publisher.go @@ -8,6 +8,7 @@ import ( "time" "github.com/centrifugal/centrifuge" + "github.com/pscheid92/chatpulse/internal/adapter/metrics" "github.com/pscheid92/chatpulse/internal/domain" ) @@ -24,10 +25,11 @@ type sessionUpdate struct { type Publisher struct { node *centrifuge.Node configSource domain.ConfigSource + wsMetrics *metrics.WebSocketMetrics } -func NewPublisher(node *centrifuge.Node, configSource domain.ConfigSource) *Publisher { - return &Publisher{node: node, configSource: configSource} +func NewPublisher(node *centrifuge.Node, configSource domain.ConfigSource, wsMetrics *metrics.WebSocketMetrics) *Publisher { + return &Publisher{node: node, configSource: configSource, wsMetrics: wsMetrics} } func (p *Publisher) PublishSentiment(ctx context.Context, broadcasterID string, snapshot *domain.WindowSnapshot) error { @@ -62,5 +64,9 @@ func (p *Publisher) PublishSentiment(ctx context.Context, broadcasterID string, return fmt.Errorf("publish to channel %s: %w", channel, err) } + if p.wsMetrics != nil { + p.wsMetrics.MessagesPublished.Inc() + } + return nil } diff --git a/internal/app/overlay.go b/internal/app/overlay.go index 9530e58..57b8047 100644 --- a/internal/app/overlay.go +++ b/internal/app/overlay.go @@ -24,18 +24,18 @@ func NewOverlay(configSource domain.ConfigSource, sentiment domain.SentimentStor } } -func (o *Overlay) ProcessMessage(ctx context.Context, broadcasterUserID, chatterUserID, messageText string) (*domain.WindowSnapshot, domain.VoteResult, error) { +func (o *Overlay) ProcessMessage(ctx context.Context, broadcasterUserID, chatterUserID, messageText string) (*domain.WindowSnapshot, domain.VoteResult, domain.VoteTarget, error) { config, err := o.getConfig(ctx, broadcasterUserID) if errors.Is(err, domain.ErrConfigNotFound) { - return nil, domain.VoteNoMatch, nil + return nil, domain.VoteNoMatch, domain.VoteTargetNone, nil } if err != nil { - return nil, domain.VoteNoMatch, fmt.Errorf("config lookup failed: %w", err) + return nil, domain.VoteNoMatch, domain.VoteTargetNone, fmt.Errorf("config lookup failed: %w", err) } target := matchTrigger(messageText, config) if target == domain.VoteTargetNone { - return nil, domain.VoteNoMatch, nil + return nil, domain.VoteNoMatch, domain.VoteTargetNone, nil } debounced, err := o.debounce.IsDebounced(ctx, broadcasterUserID, chatterUserID) @@ -44,16 +44,16 @@ func (o *Overlay) ProcessMessage(ctx context.Context, broadcasterUserID, chatter slog.Warn("Debounce check failed, allowing vote", "broadcaster", broadcasterUserID, "chatter", chatterUserID, "error", err) } if debounced { - return nil, domain.VoteDebounced, nil + 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) - return nil, domain.VoteNoMatch, fmt.Errorf("record vote failed: %w", err) + return nil, domain.VoteNoMatch, target, fmt.Errorf("record vote failed: %w", err) } - return snapshot, domain.VoteApplied, nil + return snapshot, domain.VoteApplied, target, nil } func (o *Overlay) Reset(ctx context.Context, broadcasterID string) error { diff --git a/internal/app/overlay_test.go b/internal/app/overlay_test.go index 3917b73..f4d49ab 100644 --- a/internal/app/overlay_test.go +++ b/internal/app/overlay_test.go @@ -103,9 +103,10 @@ func TestProcessMessage_Success(t *testing.T) { configSource, sentiment, debounce := newHappyPathMocks() ovl := NewOverlay(configSource, sentiment, debounce) - snap, result, err := ovl.ProcessMessage(context.Background(), "broadcaster-1", "chatter-1", "yes") + snap, result, target, err := ovl.ProcessMessage(context.Background(), "broadcaster-1", "chatter-1", "yes") require.NoError(t, err) assert.Equal(t, domain.VoteApplied, result) + assert.Equal(t, domain.VoteTargetPositive, target) assert.NotNil(t, snap) assert.Equal(t, 1.0, snap.ForRatio) assert.Equal(t, 1, snap.TotalVotes) @@ -115,9 +116,10 @@ func TestProcessMessage_AgainstVote(t *testing.T) { configSource, sentiment, debounce := newHappyPathMocks() ovl := NewOverlay(configSource, sentiment, debounce) - snap, result, err := ovl.ProcessMessage(context.Background(), "broadcaster-1", "chatter-1", "no") + snap, result, target, err := ovl.ProcessMessage(context.Background(), "broadcaster-1", "chatter-1", "no") require.NoError(t, err) assert.Equal(t, domain.VoteApplied, result) + assert.Equal(t, domain.VoteTargetNegative, target) assert.NotNil(t, snap) assert.Equal(t, 1.0, snap.AgainstRatio) } @@ -129,9 +131,10 @@ func TestProcessMessage_NoConfig(t *testing.T) { } ovl := NewOverlay(configSource, sentiment, debounce) - snap, result, err := ovl.ProcessMessage(context.Background(), "broadcaster-1", "chatter-1", "yes") + snap, result, target, err := ovl.ProcessMessage(context.Background(), "broadcaster-1", "chatter-1", "yes") require.NoError(t, err) assert.Equal(t, domain.VoteNoMatch, result) + assert.Equal(t, domain.VoteTargetNone, target) assert.Nil(t, snap) } @@ -142,9 +145,10 @@ func TestProcessMessage_ConfigLookupError(t *testing.T) { } ovl := NewOverlay(configSource, sentiment, debounce) - snap, result, err := ovl.ProcessMessage(context.Background(), "broadcaster-1", "chatter-1", "yes") + snap, result, target, err := ovl.ProcessMessage(context.Background(), "broadcaster-1", "chatter-1", "yes") assert.Error(t, err) assert.Equal(t, domain.VoteNoMatch, result) + assert.Equal(t, domain.VoteTargetNone, target) assert.Nil(t, snap) } @@ -152,9 +156,10 @@ func TestProcessMessage_NoTriggerMatch(t *testing.T) { configSource, sentiment, debounce := newHappyPathMocks() ovl := NewOverlay(configSource, sentiment, debounce) - snap, result, err := ovl.ProcessMessage(context.Background(), "broadcaster-1", "chatter-1", "hello world") + snap, result, target, err := ovl.ProcessMessage(context.Background(), "broadcaster-1", "chatter-1", "hello world") require.NoError(t, err) assert.Equal(t, domain.VoteNoMatch, result) + assert.Equal(t, domain.VoteTargetNone, target) assert.Nil(t, snap) } @@ -165,9 +170,10 @@ func TestProcessMessage_Debounced(t *testing.T) { } ovl := NewOverlay(configSource, sentiment, debounce) - snap, result, err := ovl.ProcessMessage(context.Background(), "broadcaster-1", "chatter-1", "yes") + snap, result, target, err := ovl.ProcessMessage(context.Background(), "broadcaster-1", "chatter-1", "yes") require.NoError(t, err) assert.Equal(t, domain.VoteDebounced, result) + assert.Equal(t, domain.VoteTargetPositive, target) assert.Nil(t, snap) } @@ -180,9 +186,10 @@ func TestProcessMessage_RecordVoteError(t *testing.T) { } ovl := NewOverlay(configSource, sentiment, debounce) - snap, result, err := ovl.ProcessMessage(context.Background(), "broadcaster-1", "chatter-1", "yes") + snap, result, target, err := ovl.ProcessMessage(context.Background(), "broadcaster-1", "chatter-1", "yes") assert.Error(t, err) assert.Equal(t, domain.VoteNoMatch, result) + assert.Equal(t, domain.VoteTargetPositive, target) assert.Nil(t, snap) } @@ -212,7 +219,7 @@ func TestProcessMessage_PassesWindowSeconds(t *testing.T) { } ovl := NewOverlay(configSource, sentiment, debounce) - _, _, err := ovl.ProcessMessage(context.Background(), "broadcaster-1", "chatter-1", "yes") + _, _, _, err := ovl.ProcessMessage(context.Background(), "broadcaster-1", "chatter-1", "yes") require.NoError(t, err) assert.Equal(t, 60, capturedWindowSeconds) } diff --git a/internal/app/service_test.go b/internal/app/service_test.go index e4402d1..6c9c802 100644 --- a/internal/app/service_test.go +++ b/internal/app/service_test.go @@ -80,8 +80,8 @@ type mockOverlay struct { resetFn func(ctx context.Context, broadcasterID string) error } -func (m *mockOverlay) ProcessMessage(context.Context, string, string, string) (*domain.WindowSnapshot, domain.VoteResult, error) { - return nil, domain.VoteNoMatch, nil +func (m *mockOverlay) ProcessMessage(context.Context, string, string, string) (*domain.WindowSnapshot, domain.VoteResult, domain.VoteTarget, error) { + return nil, domain.VoteNoMatch, domain.VoteTargetNone, nil } func (m *mockOverlay) Reset(ctx context.Context, broadcasterID string) error { diff --git a/internal/domain/overlay.go b/internal/domain/overlay.go index 4f03681..23bafee 100644 --- a/internal/domain/overlay.go +++ b/internal/domain/overlay.go @@ -52,6 +52,6 @@ func (r VoteResult) String() string { } type Overlay interface { - ProcessMessage(ctx context.Context, broadcasterUserID, chatterUserID, messageText string) (*WindowSnapshot, VoteResult, error) + ProcessMessage(ctx context.Context, broadcasterUserID, chatterUserID, messageText string) (*WindowSnapshot, VoteResult, VoteTarget, error) Reset(ctx context.Context, broadcasterID string) error }