From 0fe3a7162738385e01e9b3a06ceabfb90ee035bc Mon Sep 17 00:00:00 2001 From: mkelvers Date: Sun, 21 Jun 2026 17:18:43 +0200 Subject: [PATCH 1/3] feat: add jikan cache stats query --- integrations/jikan/client_test.go | 27 +++++++++++++++++++++++++++ internal/db/querier.go | 1 + internal/db/queries.sql | 7 +++++++ internal/db/queries.sql.go | 21 +++++++++++++++++++++ 4 files changed, 56 insertions(+) diff --git a/integrations/jikan/client_test.go b/integrations/jikan/client_test.go index 5f7e4dfa..edeee534 100644 --- a/integrations/jikan/client_test.go +++ b/integrations/jikan/client_test.go @@ -86,6 +86,33 @@ func TestGetWithCacheAllowsEmptySearchResults(t *testing.T) { } } +func TestGetJikanCacheStatsCountsRowsAndExpiry(t *testing.T) { + sqlDB := newTestCacheDB(t) + defer func() { + if err := sqlDB.Close(); err != nil { + t.Errorf("close sqlite: %v", err) + } + }() + + oldest := time.Date(2026, 6, 20, 12, 0, 0, 0, time.UTC) + insertCachedResponse(t, sqlDB, "expired", TopAnimeResponse{Data: []Anime{{MalID: 1}}}, oldest) + insertCachedResponse(t, sqlDB, "fresh", TopAnimeResponse{Data: []Anime{{MalID: 2}}}, time.Now().Add(time.Hour)) + + stats, err := db.New(sqlDB).GetJikanCacheStats(context.Background()) + if err != nil { + t.Fatalf("GetJikanCacheStats: %v", err) + } + if stats.TotalRows != 2 { + t.Fatalf("TotalRows = %d, want 2", stats.TotalRows) + } + if stats.ExpiredRows != 1 { + t.Fatalf("ExpiredRows = %d, want 1", stats.ExpiredRows) + } + if stats.OldestExpiresAtSeconds != oldest.Unix() { + t.Fatalf("OldestExpiresAtSeconds = %d, want %d", stats.OldestExpiresAtSeconds, oldest.Unix()) + } +} + func newTestCacheDB(t *testing.T) *sql.DB { t.Helper() ctx := context.Background() diff --git a/internal/db/querier.go b/internal/db/querier.go index f6521f6a..211177f2 100644 --- a/internal/db/querier.go +++ b/internal/db/querier.go @@ -30,6 +30,7 @@ type Querier interface { GetEpisodeAvailabilityCache(ctx context.Context, animeID int64) (EpisodeAvailabilityCache, error) GetEpisodeProviderMapping(ctx context.Context, arg GetEpisodeProviderMappingParams) (EpisodeProviderMapping, error) GetJikanCache(ctx context.Context, key string) (string, error) + GetJikanCacheStats(ctx context.Context) (GetJikanCacheStatsRow, error) GetJikanCacheStale(ctx context.Context, key string) (string, error) GetSession(ctx context.Context, id string) (Session, error) GetTrackedAiringAnimeIDsDueForEpisodeRefresh(ctx context.Context, limit int64) ([]int64, error) diff --git a/internal/db/queries.sql b/internal/db/queries.sql index 000f4490..dba4f4ef 100644 --- a/internal/db/queries.sql +++ b/internal/db/queries.sql @@ -235,6 +235,13 @@ WHERE key = ? AND datetime(expires_at) > CURRENT_TIMESTAMP LIMIT 1; SELECT data FROM jikan_cache WHERE key = ? LIMIT 1; +-- name: GetJikanCacheStats :one +SELECT + COUNT(*) AS total_rows, + COUNT(*) FILTER (WHERE datetime(expires_at) <= CURRENT_TIMESTAMP) AS expired_rows, + COALESCE(unixepoch(MIN(expires_at)), 0) AS oldest_expires_at_seconds +FROM jikan_cache; + -- name: SetJikanCache :exec INSERT INTO jikan_cache (key, data, expires_at) VALUES (?, ?, ?) diff --git a/internal/db/queries.sql.go b/internal/db/queries.sql.go index d84cf87a..881d4451 100644 --- a/internal/db/queries.sql.go +++ b/internal/db/queries.sql.go @@ -565,6 +565,27 @@ func (q *Queries) GetJikanCache(ctx context.Context, key string) (string, error) return data, err } +const getJikanCacheStats = `-- name: GetJikanCacheStats :one +SELECT + COUNT(*) AS total_rows, + COUNT(*) FILTER (WHERE datetime(expires_at) <= CURRENT_TIMESTAMP) AS expired_rows, + COALESCE(unixepoch(MIN(expires_at)), 0) AS oldest_expires_at_seconds +FROM jikan_cache +` + +type GetJikanCacheStatsRow struct { + TotalRows int64 `json:"total_rows"` + ExpiredRows int64 `json:"expired_rows"` + OldestExpiresAtSeconds int64 `json:"oldest_expires_at_seconds"` +} + +func (q *Queries) GetJikanCacheStats(ctx context.Context) (GetJikanCacheStatsRow, error) { + row := q.db.QueryRowContext(ctx, getJikanCacheStats) + var i GetJikanCacheStatsRow + err := row.Scan(&i.TotalRows, &i.ExpiredRows, &i.OldestExpiresAtSeconds) + return i, err +} + const getJikanCacheStale = `-- name: GetJikanCacheStale :one SELECT data FROM jikan_cache WHERE key = ? LIMIT 1 From d3ec9d9f6e1bb2b7ec444b154bfb4d331aaf7e6b Mon Sep 17 00:00:00 2001 From: mkelvers Date: Sun, 21 Jun 2026 17:18:52 +0200 Subject: [PATCH 2/3] feat: expose jikan cache metrics --- internal/observability/metrics.go | 92 +++++++++++++++++++++++++- internal/observability/metrics_test.go | 4 ++ 2 files changed, 95 insertions(+), 1 deletion(-) diff --git a/internal/observability/metrics.go b/internal/observability/metrics.go index d9ee2f11..0419b0ce 100644 --- a/internal/observability/metrics.go +++ b/internal/observability/metrics.go @@ -37,6 +37,11 @@ type histogramSample struct { sum float64 } +type gaugeSample struct { + labels map[string]string + value float64 +} + type counterVec struct { mu sync.Mutex labelNames []string @@ -50,6 +55,12 @@ type histogramVec struct { samples map[string]*histogramSample } +type gaugeVec struct { + mu sync.Mutex + labelNames []string + samples map[string]*gaugeSample +} + type Metrics struct { httpRequests *counterVec httpRequestLatency *histogramVec @@ -59,6 +70,8 @@ type Metrics struct { dbQueryLatency *histogramVec workerTicks *counterVec cacheOperations *counterVec + jikanCacheRows *gaugeVec + jikanCacheOldest *gaugeVec } func NewMetrics() *Metrics { @@ -71,6 +84,8 @@ func NewMetrics() *Metrics { dbQueryLatency: newHistogramVec(defaultDurationBuckets, "operation", "result"), workerTicks: newCounterVec("worker", "result"), cacheOperations: newCounterVec("cache", "result"), + jikanCacheRows: newGaugeVec("state"), + jikanCacheOldest: newGaugeVec(), } } @@ -119,6 +134,12 @@ func (m *Metrics) ObserveCache(cache string, result string) { m.cacheOperations.Inc(cache, result) } +func (m *Metrics) ObserveJikanCacheStats(totalRows int64, expiredRows int64, oldestExpiresAtSeconds int64) { + m.jikanCacheRows.Set(float64(totalRows), "total") + m.jikanCacheRows.Set(float64(expiredRows), "expired") + m.jikanCacheOldest.Set(float64(oldestExpiresAtSeconds)) +} + func (m *Metrics) writePrometheus(w http.ResponseWriter) error { if err := writeCounterMetric(w, "mal_http_requests_total", "Total HTTP requests by method, route, and status.", m.httpRequests.snapshot()); err != nil { return err @@ -141,7 +162,13 @@ func (m *Metrics) writePrometheus(w http.ResponseWriter) error { if err := writeCounterMetric(w, "mal_worker_ticks_total", "Total background worker ticks by worker and result.", m.workerTicks.snapshot()); err != nil { return err } - return writeCounterMetric(w, "mal_cache_operations_total", "Total cache hits and misses by cache name.", m.cacheOperations.snapshot()) + if err := writeCounterMetric(w, "mal_cache_operations_total", "Total cache hits and misses by cache name.", m.cacheOperations.snapshot()); err != nil { + return err + } + if err := writeGaugeMetric(w, "mal_jikan_cache_rows", "Current jikan_cache row count by state.", m.jikanCacheRows.snapshot()); err != nil { + return err + } + return writeGaugeMetric(w, "mal_jikan_cache_oldest_expires_at_seconds", "Unix timestamp for the oldest jikan_cache expires_at value, or 0 when empty.", m.jikanCacheOldest.snapshot()) } func newCounterVec(labelNames ...string) *counterVec { @@ -237,6 +264,45 @@ func (h *histogramVec) snapshot() []histogramSample { return out } +func newGaugeVec(labelNames ...string) *gaugeVec { + return &gaugeVec{ + labelNames: append([]string(nil), labelNames...), + samples: make(map[string]*gaugeSample), + } +} + +func (g *gaugeVec) Set(value float64, labelValues ...string) { + g.mu.Lock() + defer g.mu.Unlock() + + key, labels := buildLabelKey(g.labelNames, labelValues) + if labels == nil { + return + } + sample, ok := g.samples[key] + if !ok { + sample = &gaugeSample{labels: labels} + g.samples[key] = sample + } + sample.value = value +} + +func (g *gaugeVec) snapshot() []gaugeSample { + g.mu.Lock() + defer g.mu.Unlock() + + keys := sortedGaugeSampleKeys(g.samples) + out := make([]gaugeSample, 0, len(keys)) + for _, key := range keys { + sample := g.samples[key] + out = append(out, gaugeSample{ + labels: copyLabels(sample.labels), + value: sample.value, + }) + } + return out +} + func buildLabelKey(labelNames []string, labelValues []string) (string, map[string]string) { if len(labelNames) != len(labelValues) { return "", nil @@ -276,6 +342,15 @@ func sortedHistogramSampleKeys(samples map[string]*histogramSample) []string { return keys } +func sortedGaugeSampleKeys(samples map[string]*gaugeSample) []string { + keys := make([]string, 0, len(samples)) + for key := range samples { + keys = append(keys, key) + } + sort.Strings(keys) + return keys +} + func writeCounterMetric(w http.ResponseWriter, name string, help string, samples []counterSample) error { if _, err := fmt.Fprintf(w, "# HELP %s %s\n", name, help); err != nil { return err @@ -291,6 +366,21 @@ func writeCounterMetric(w http.ResponseWriter, name string, help string, samples return nil } +func writeGaugeMetric(w http.ResponseWriter, name string, help string, samples []gaugeSample) error { + if _, err := fmt.Fprintf(w, "# HELP %s %s\n", name, help); err != nil { + return err + } + if _, err := fmt.Fprintf(w, "# TYPE %s gauge\n", name); err != nil { + return err + } + for _, sample := range samples { + if _, err := fmt.Fprintf(w, "%s%s %s\n", name, formatLabels(sample.labels), formatFloat(sample.value)); err != nil { + return err + } + } + return nil +} + func writeHistogramMetric(w http.ResponseWriter, name string, help string, samples []histogramSample, bounds []float64) error { if _, err := fmt.Fprintf(w, "# HELP %s %s\n", name, help); err != nil { return err diff --git a/internal/observability/metrics_test.go b/internal/observability/metrics_test.go index 4998c160..6157c647 100644 --- a/internal/observability/metrics_test.go +++ b/internal/observability/metrics_test.go @@ -21,6 +21,7 @@ func TestMetricsHandlerRendersPrometheusFamilies(t *testing.T) { metrics.ObserveWorkerTick("episodes_availability", nil) metrics.ObserveCache("jikan", "hit") metrics.ObserveCache("episode_availability", "miss") + metrics.ObserveJikanCacheStats(12, 3, 1770000000) req := httptest.NewRequestWithContext(context.Background(), http.MethodGet, "/metrics", nil) rec := httptest.NewRecorder() @@ -39,6 +40,9 @@ func TestMetricsHandlerRendersPrometheusFamilies(t *testing.T) { assertContains(t, text, `mal_db_query_duration_seconds_count{operation="query",result="success"} 1`) assertContains(t, text, `mal_worker_ticks_total{result="success",worker="episodes_availability"} 1`) assertContains(t, text, `mal_cache_operations_total{cache="episode_availability",result="miss"} 1`) + assertContains(t, text, `mal_jikan_cache_rows{state="total"} 12`) + assertContains(t, text, `mal_jikan_cache_rows{state="expired"} 3`) + assertContains(t, text, `mal_jikan_cache_oldest_expires_at_seconds 1770000000`) } func TestInstrumentDBRecordsQueryLatency(t *testing.T) { From 8792cdc457ac2b89bd23901f899380a4db8e2d22 Mon Sep 17 00:00:00 2001 From: mkelvers Date: Sun, 21 Jun 2026 17:19:01 +0200 Subject: [PATCH 3/3] feat: observe jikan cache stats --- integrations/jikan/cache/store.go | 22 ++++++++++++++++++++++ 1 file changed, 22 insertions(+) diff --git a/integrations/jikan/cache/store.go b/integrations/jikan/cache/store.go index a90058fa..3dd8fe78 100644 --- a/integrations/jikan/cache/store.go +++ b/integrations/jikan/cache/store.go @@ -22,6 +22,7 @@ func NewStore(queries db.Querier, metrics *observability.Metrics) *Store { func (s *Store) Get(parentCtx context.Context, key string, out any) bool { ctx, cancel := context.WithTimeout(parentCtx, 2*time.Second) defer cancel() + defer s.observeStats(parentCtx) data, err := s.db.GetJikanCache(ctx, key) if err != nil { @@ -42,6 +43,7 @@ func (s *Store) Get(parentCtx context.Context, key string, out any) bool { func (s *Store) GetStale(parentCtx context.Context, key string, out any) bool { ctx, cancel := context.WithTimeout(parentCtx, 2*time.Second) defer cancel() + defer s.observeStats(parentCtx) data, err := s.db.GetJikanCacheStale(ctx, key) if err != nil { @@ -62,6 +64,7 @@ func (s *Store) GetStale(parentCtx context.Context, key string, out any) bool { func (s *Store) Set(parentCtx context.Context, key string, data any, ttl time.Duration) { ctx, cancel := context.WithTimeout(parentCtx, 2*time.Second) defer cancel() + defer s.observeStats(parentCtx) bytes, err := json.Marshal(data) if err != nil { @@ -84,3 +87,22 @@ func (s *Store) Set(parentCtx context.Context, key string, data any, ttl time.Du ) } } + +func (s *Store) observeStats(parentCtx context.Context) { + ctx, cancel := context.WithTimeout(parentCtx, 2*time.Second) + defer cancel() + + stats, err := s.db.GetJikanCacheStats(ctx) + if err != nil { + observability.LogJSON( + observability.LogLevelWarn, + "jikan_cache_stats", + "jikan", + "", + nil, + err, + ) + return + } + s.metrics.ObserveJikanCacheStats(stats.TotalRows, stats.ExpiredRows, stats.OldestExpiresAtSeconds) +}