From 61ff4a4fa2f774a98bc48854f1a82df195d912ac Mon Sep 17 00:00:00 2001 From: mkelvers Date: Sun, 21 Jun 2026 17:17:57 +0200 Subject: [PATCH 1/9] feat: add jikan cache cleanup worker --- internal/database/database.go | 1 + internal/database/database_test.go | 80 ++++++++++++++++++++++++ internal/database/jikan_cache_cleanup.go | 72 +++++++++++++++++++++ 3 files changed, 153 insertions(+) create mode 100644 internal/database/jikan_cache_cleanup.go diff --git a/internal/database/database.go b/internal/database/database.go index 347a9474..0b4e4849 100644 --- a/internal/database/database.go +++ b/internal/database/database.go @@ -22,6 +22,7 @@ var Module = fx.Options( ProvideSQLDB, ProvideQueries, ), + fx.Invoke(RegisterJikanCacheCleanupWorker), ) func ProvideSQLDB(cfg config.Config) (*sql.DB, error) { diff --git a/internal/database/database_test.go b/internal/database/database_test.go index 74af7080..c32e6a4a 100644 --- a/internal/database/database_test.go +++ b/internal/database/database_test.go @@ -3,6 +3,8 @@ package database import ( "context" "database/sql" + "mal/internal/db" + "mal/internal/observability" "testing" _ "github.com/mattn/go-sqlite3" @@ -43,3 +45,81 @@ func TestRunMigrationsCreatesHotPathIndexes(t *testing.T) { }) } } + +func TestCleanupExpiredJikanCache(t *testing.T) { + sqlDB := newMigratedTestDB(t) + defer closeTestDB(t, sqlDB) + + ctx := context.Background() + for _, row := range []struct { + key string + expiresAt string + }{ + {key: "expired", expiresAt: "2000-01-01T00:00:00Z"}, + {key: "fresh", expiresAt: "2999-01-01T00:00:00Z"}, + } { + _, err := sqlDB.ExecContext(ctx, `INSERT INTO jikan_cache (key, data, expires_at) VALUES (?, ?, ?)`, row.key, "{}", row.expiresAt) + if err != nil { + t.Fatalf("insert %s cache row: %v", row.key, err) + } + } + + cleanupExpiredJikanCache(ctx, db.New(sqlDB), observability.NewMetrics()) + + keys := jikanCacheKeys(ctx, t, sqlDB) + if len(keys) != 1 || keys[0] != "fresh" { + t.Fatalf("remaining cache keys = %v, want [fresh]", keys) + } +} + +func newMigratedTestDB(t *testing.T) *sql.DB { + t.Helper() + + sqlDB, err := sql.Open("sqlite3", ":memory:") + if err != nil { + t.Fatalf("open sqlite: %v", err) + } + sqlDB.SetMaxOpenConns(1) + + if err := RunMigrations(sqlDB); err != nil { + closeTestDB(t, sqlDB) + t.Fatalf("RunMigrations: %v", err) + } + + return sqlDB +} + +func closeTestDB(t *testing.T, sqlDB *sql.DB) { + t.Helper() + + if err := sqlDB.Close(); err != nil { + t.Errorf("close sqlite: %v", err) + } +} + +func jikanCacheKeys(ctx context.Context, t *testing.T, sqlDB *sql.DB) []string { + t.Helper() + + var keys []string + rows, err := sqlDB.QueryContext(ctx, `SELECT key FROM jikan_cache ORDER BY key`) + if err != nil { + t.Fatalf("query cache keys: %v", err) + } + defer func() { + if err := rows.Close(); err != nil { + t.Errorf("close rows: %v", err) + } + }() + for rows.Next() { + var key string + if err := rows.Scan(&key); err != nil { + t.Fatalf("scan key: %v", err) + } + keys = append(keys, key) + } + if err := rows.Err(); err != nil { + t.Fatalf("iterate keys: %v", err) + } + + return keys +} diff --git a/internal/database/jikan_cache_cleanup.go b/internal/database/jikan_cache_cleanup.go new file mode 100644 index 00000000..2d1b0f7a --- /dev/null +++ b/internal/database/jikan_cache_cleanup.go @@ -0,0 +1,72 @@ +package database + +import ( + "context" + "mal/internal/db" + "mal/internal/observability" + "time" + + "go.uber.org/fx" +) + +const ( + jikanCacheCleanupInterval = time.Hour + jikanCacheCleanupTimeout = 30 * time.Second + jikanCacheCleanupWorker = "jikan_cache_cleanup" +) + +func RegisterJikanCacheCleanupWorker(lc fx.Lifecycle, queries *db.Queries, metrics *observability.Metrics) { + ctx, cancel := context.WithCancel(context.Background()) + + lc.Append(fx.Hook{ + OnStart: func(startCtx context.Context) error { + go func() { + <-startCtx.Done() + cancel() + }() + go runJikanCacheCleanupWorker(ctx, queries, metrics) + return nil + }, + OnStop: func(context.Context) error { + cancel() + return nil + }, + }) +} + +func runJikanCacheCleanupWorker(ctx context.Context, queries *db.Queries, metrics *observability.Metrics) { + observability.Info("jikan_cache_cleanup_worker_start", "database", "", nil) + + ticker := time.NewTicker(jikanCacheCleanupInterval) + defer ticker.Stop() + + for { + cleanupExpiredJikanCache(ctx, queries, metrics) + + select { + case <-ticker.C: + case <-ctx.Done(): + observability.Info("jikan_cache_cleanup_worker_stop", "database", "", nil) + return + } + } +} + +func cleanupExpiredJikanCache(ctx context.Context, queries *db.Queries, metrics *observability.Metrics) { + cleanupCtx, cancel := context.WithTimeout(ctx, jikanCacheCleanupTimeout) + defer cancel() + + err := queries.DeleteExpiredJikanCache(cleanupCtx) + metrics.ObserveWorkerTick(jikanCacheCleanupWorker, err) + if err != nil { + observability.Warn( + "jikan_cache_cleanup_failed", + "database", + "", + map[string]any{ + "worker": jikanCacheCleanupWorker, + }, + err, + ) + } +} From b87f743615442aaf98bb347365c4fe6b6e6e2408 Mon Sep 17 00:00:00 2001 From: mkelvers Date: Sun, 21 Jun 2026 17:18:05 +0200 Subject: [PATCH 2/9] feat: add expired failed mapping cleanup query --- internal/db/querier.go | 1 + internal/db/queries.sql | 5 +++++ internal/db/queries.sql.go | 11 +++++++++++ 3 files changed, 17 insertions(+) diff --git a/internal/db/querier.go b/internal/db/querier.go index f6521f6a..5753c8e1 100644 --- a/internal/db/querier.go +++ b/internal/db/querier.go @@ -15,6 +15,7 @@ type Querier interface { CreateSession(ctx context.Context, arg CreateSessionParams) (Session, error) DeleteAnimeFetchRetry(ctx context.Context, animeID int64) error DeleteContinueWatchingEntry(ctx context.Context, arg DeleteContinueWatchingEntryParams) error + DeleteExpiredFailedEpisodeProviderMappings(ctx context.Context) error DeleteExpiredJikanCache(ctx context.Context) error DeleteSession(ctx context.Context, id string) error DeleteWatchListEntry(ctx context.Context, arg DeleteWatchListEntryParams) error diff --git a/internal/db/queries.sql b/internal/db/queries.sql index 000f4490..03ccc290 100644 --- a/internal/db/queries.sql +++ b/internal/db/queries.sql @@ -333,6 +333,11 @@ SELECT anime_id, provider, provider_show_id, failed_until, last_error, updated_a FROM episode_provider_mapping WHERE anime_id = ? AND provider = ? LIMIT 1; +-- name: DeleteExpiredFailedEpisodeProviderMappings :exec +DELETE FROM episode_provider_mapping +WHERE provider_show_id = '' + AND failed_until <= CURRENT_TIMESTAMP; + -- name: GetTrackedAiringAnimeIDsDueForEpisodeRefresh :many WITH tracked AS ( SELECT DISTINCT w.anime_id diff --git a/internal/db/queries.sql.go b/internal/db/queries.sql.go index d84cf87a..025f1d35 100644 --- a/internal/db/queries.sql.go +++ b/internal/db/queries.sql.go @@ -149,6 +149,17 @@ func (q *Queries) DeleteContinueWatchingEntry(ctx context.Context, arg DeleteCon return err } +const deleteExpiredFailedEpisodeProviderMappings = `-- name: DeleteExpiredFailedEpisodeProviderMappings :exec +DELETE FROM episode_provider_mapping +WHERE provider_show_id = '' + AND failed_until <= CURRENT_TIMESTAMP +` + +func (q *Queries) DeleteExpiredFailedEpisodeProviderMappings(ctx context.Context) error { + _, err := q.db.ExecContext(ctx, deleteExpiredFailedEpisodeProviderMappings) + return err +} + const deleteExpiredJikanCache = `-- name: DeleteExpiredJikanCache :exec DELETE FROM jikan_cache WHERE datetime(expires_at) <= CURRENT_TIMESTAMP ` From dbe029e15e1d2716d55b9821cb7b3aae90f463f3 Mon Sep 17 00:00:00 2001 From: mkelvers Date: Sun, 21 Jun 2026 17:18:11 +0200 Subject: [PATCH 3/9] fix: ignore expired anime in random pool cache --- integrations/jikan/client_test.go | 47 +++++++++++++++++++++++++++++++ internal/db/queries.sql | 2 +- internal/db/queries.sql.go | 2 +- 3 files changed, 49 insertions(+), 2 deletions(-) diff --git a/integrations/jikan/client_test.go b/integrations/jikan/client_test.go index 5f7e4dfa..7f8de793 100644 --- a/integrations/jikan/client_test.go +++ b/integrations/jikan/client_test.go @@ -86,6 +86,32 @@ func TestGetWithCacheAllowsEmptySearchResults(t *testing.T) { } } +func TestLoadCachedRandomPoolIgnoresExpiredAnimeCache(t *testing.T) { + sqlDB := newTestCacheDB(t) + defer func() { + if err := sqlDB.Close(); err != nil { + t.Errorf("close sqlite: %v", err) + } + }() + + queries := db.New(sqlDB) + client := NewClient(config.Config{}, queries, observability.NewMetrics()) + insertCachedAnime(t, sqlDB, "anime:1", Anime{MalID: 1, Title: "fresh"}, time.Now().Add(time.Hour)) + insertCachedAnime(t, sqlDB, "anime:2", Anime{MalID: 2, Title: "expired"}, time.Now().Add(-time.Hour)) + + client.loadCachedRandomPool(context.Background()) + + client.poolMu.RLock() + defer client.poolMu.RUnlock() + + if len(client.randomPool) != 1 { + t.Fatalf("randomPool length = %d, want 1", len(client.randomPool)) + } + if client.randomPool[0].MalID != 1 || client.randomPool[0].Title != "fresh" { + t.Fatalf("randomPool[0] = %+v, want fresh anime", client.randomPool[0]) + } +} + func newTestCacheDB(t *testing.T) *sql.DB { t.Helper() ctx := context.Background() @@ -135,6 +161,27 @@ func insertCachedResponse(t *testing.T, sqlDB *sql.DB, key string, value TopAnim } } +func insertCachedAnime(t *testing.T, sqlDB *sql.DB, key string, value Anime, expiresAt time.Time) { + t.Helper() + ctx := context.Background() + + encoded, err := json.Marshal(value) + if err != nil { + t.Fatalf("marshal cached anime: %v", err) + } + + _, err = sqlDB.ExecContext( + ctx, + `INSERT INTO jikan_cache (key, data, expires_at) VALUES (?, ?, ?)`, + key, + string(encoded), + expiresAt, + ) + if err != nil { + t.Fatalf("insert cached anime: %v", err) + } +} + func waitForFreshCache(t *testing.T, sqlDB *sql.DB, client *Client, key string) { t.Helper() diff --git a/internal/db/queries.sql b/internal/db/queries.sql index 000f4490..3e08fd15 100644 --- a/internal/db/queries.sql +++ b/internal/db/queries.sql @@ -357,4 +357,4 @@ LIMIT ?; -- name: GetAllCachedAnime :many SELECT data FROM jikan_cache -WHERE key LIKE 'anime:%' LIMIT 1000; +WHERE key LIKE 'anime:%' AND datetime(expires_at) > CURRENT_TIMESTAMP LIMIT 1000; diff --git a/internal/db/queries.sql.go b/internal/db/queries.sql.go index d84cf87a..a3c0b58c 100644 --- a/internal/db/queries.sql.go +++ b/internal/db/queries.sql.go @@ -227,7 +227,7 @@ func (q *Queries) GetAPITokenByHash(ctx context.Context, tokenHash string) (ApiT const getAllCachedAnime = `-- name: GetAllCachedAnime :many SELECT data FROM jikan_cache -WHERE key LIKE 'anime:%' LIMIT 1000 +WHERE key LIKE 'anime:%' AND datetime(expires_at) > CURRENT_TIMESTAMP LIMIT 1000 ` func (q *Queries) GetAllCachedAnime(ctx context.Context) ([]string, error) { From a0fff22e90407b2aff5e5f651298e2802b0be44b Mon Sep 17 00:00:00 2001 From: mkelvers Date: Sun, 21 Jun 2026 17:18:23 +0200 Subject: [PATCH 4/9] fix: limit sqlite connection pool --- internal/db/sqlite.go | 3 +++ 1 file changed, 3 insertions(+) diff --git a/internal/db/sqlite.go b/internal/db/sqlite.go index c73f35ee..3f90b22f 100644 --- a/internal/db/sqlite.go +++ b/internal/db/sqlite.go @@ -19,6 +19,9 @@ func Open(dbFile string) (*sql.DB, error) { if err != nil { return nil, fmt.Errorf("failed to open db: %w", err) } + db.SetMaxOpenConns(1) + db.SetMaxIdleConns(1) + // WAL improves concurrency between readers and writers. if _, err := db.ExecContext(context.Background(), "PRAGMA journal_mode=WAL;"); err != nil { return nil, fmt.Errorf("failed to enable WAL mode: %w", err) From 3d855ab1ebcfb7a1c6268a6a52f4fa898361ff87 Mon Sep 17 00:00:00 2001 From: mkelvers Date: Sun, 21 Jun 2026 17:18:29 +0200 Subject: [PATCH 5/9] fix: bound stale jikan cache reads --- internal/db/queries.sql | 2 +- internal/db/queries.sql.go | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/internal/db/queries.sql b/internal/db/queries.sql index 000f4490..f1fac293 100644 --- a/internal/db/queries.sql +++ b/internal/db/queries.sql @@ -233,7 +233,7 @@ WHERE key = ? AND datetime(expires_at) > CURRENT_TIMESTAMP LIMIT 1; -- name: GetJikanCacheStale :one SELECT data FROM jikan_cache -WHERE key = ? LIMIT 1; +WHERE key = ? AND datetime(expires_at) > datetime(CURRENT_TIMESTAMP, '-14 days') LIMIT 1; -- name: SetJikanCache :exec INSERT INTO jikan_cache (key, data, expires_at) diff --git a/internal/db/queries.sql.go b/internal/db/queries.sql.go index d84cf87a..2e5201f7 100644 --- a/internal/db/queries.sql.go +++ b/internal/db/queries.sql.go @@ -567,7 +567,7 @@ func (q *Queries) GetJikanCache(ctx context.Context, key string) (string, error) const getJikanCacheStale = `-- name: GetJikanCacheStale :one SELECT data FROM jikan_cache -WHERE key = ? LIMIT 1 +WHERE key = ? AND datetime(expires_at) > datetime(CURRENT_TIMESTAMP, '-14 days') LIMIT 1 ` func (q *Queries) GetJikanCacheStale(ctx context.Context, key string) (string, error) { From 577f289f8168048f1d09ea12750c5a8101d81762 Mon Sep 17 00:00:00 2001 From: mkelvers Date: Sun, 21 Jun 2026 17:18:36 +0200 Subject: [PATCH 6/9] refactor: share episode cache decoding --- internal/episodes/service/cache_store.go | 110 ++++++++++------------- 1 file changed, 45 insertions(+), 65 deletions(-) diff --git a/internal/episodes/service/cache_store.go b/internal/episodes/service/cache_store.go index 13748037..3d7041f4 100644 --- a/internal/episodes/service/cache_store.go +++ b/internal/episodes/service/cache_store.go @@ -132,41 +132,7 @@ func (s *EpisodeService) markFailure(ctx context.Context, anime domain.Anime, ca } func (s *EpisodeService) getCached(ctx context.Context, anime domain.Anime) (domain.CanonicalEpisodeList, bool) { - row, err := s.queries.GetEpisodeAvailabilityCache(ctx, int64(anime.MalID)) - if err != nil { - s.metrics.ObserveCache("episode_availability", "miss") - return domain.CanonicalEpisodeList{}, false - } - var payload domain.CanonicalEpisodeList - if err := json.Unmarshal([]byte(row.Data), &payload); err != nil { - s.metrics.ObserveCache("episode_availability", "miss") - observability.Warn( - "episodes_cached_payload_invalid", - "episodes", - "", - map[string]any{ - "anime_id": anime.MalID, - }, - err, - ) - return domain.CanonicalEpisodeList{}, false - } - if !isCanonicalEpisodePayloadValid(payload, anime.Episodes) { - s.metrics.ObserveCache("episode_availability", "miss") - observability.Info( - "episodes_cached_payload_rejected", - "episodes", - "", - map[string]any{ - "anime_id": anime.MalID, - "expected_count": anime.Episodes, - "cached_episodes": len(payload.Episodes), - }, - ) - return domain.CanonicalEpisodeList{}, false - } - s.metrics.ObserveCache("episode_availability", "hit") - return payload, true + return s.getDecodedCached(ctx, anime, "episode_availability") } func (s *EpisodeService) getFreshCached(ctx context.Context, anime domain.Anime) (domain.CanonicalEpisodeList, bool) { @@ -181,24 +147,10 @@ func (s *EpisodeService) getFreshCached(ctx context.Context, anime domain.Anime) return domain.CanonicalEpisodeList{}, false } - payload, ok := s.decodeFreshCachedPayload(anime, row.Data) + payload, ok := s.decodeCachedPayload(anime, row.Data, "episode_availability_fresh") if !ok { return domain.CanonicalEpisodeList{}, false } - if !isCanonicalEpisodePayloadValid(payload, anime.Episodes) { - s.metrics.ObserveCache("episode_availability_fresh", "miss") - observability.Info( - "episodes_cached_payload_rejected", - "episodes", - "", - map[string]any{ - "anime_id": anime.MalID, - "expected_count": anime.Episodes, - "cached_episodes": len(payload.Episodes), - }, - ) - return domain.CanonicalEpisodeList{}, false - } s.metrics.ObserveCache("episode_availability_fresh", "hit") observability.Info( "episodes_cache_served", @@ -213,6 +165,20 @@ func (s *EpisodeService) getFreshCached(ctx context.Context, anime domain.Anime) return payload, true } +func (s *EpisodeService) getDecodedCached(ctx context.Context, anime domain.Anime, metric string) (domain.CanonicalEpisodeList, bool) { + row, err := s.queries.GetEpisodeAvailabilityCache(ctx, int64(anime.MalID)) + if err != nil { + s.metrics.ObserveCache(metric, "miss") + return domain.CanonicalEpisodeList{}, false + } + payload, ok := s.decodeCachedPayload(anime, row.Data, metric) + if !ok { + return domain.CanonicalEpisodeList{}, false + } + s.metrics.ObserveCache(metric, "hit") + return payload, true +} + func (s *EpisodeService) isFreshEpisodeCache(anime domain.Anime, row db.EpisodeAvailabilityCache, now time.Time) bool { if row.NextRefreshAt.Valid && !row.NextRefreshAt.Time.After(now) { s.metrics.ObserveCache("episode_availability_fresh", "miss") @@ -243,22 +209,36 @@ func (s *EpisodeService) isFreshEpisodeCache(anime domain.Anime, row db.EpisodeA return true } -func (s *EpisodeService) decodeFreshCachedPayload(anime domain.Anime, raw string) (domain.CanonicalEpisodeList, bool) { +func (s *EpisodeService) decodeCachedPayload(anime domain.Anime, raw string, metric string) (domain.CanonicalEpisodeList, bool) { var payload domain.CanonicalEpisodeList - err := json.Unmarshal([]byte(raw), &payload) - if err == nil { - return payload, true + if err := json.Unmarshal([]byte(raw), &payload); err != nil { + s.metrics.ObserveCache(metric, "miss") + observability.Warn( + "episodes_cached_payload_invalid", + "episodes", + "", + map[string]any{ + "anime_id": anime.MalID, + }, + err, + ) + return domain.CanonicalEpisodeList{}, false } - s.metrics.ObserveCache("episode_availability_fresh", "miss") - observability.Warn( - "episodes_cached_payload_invalid", - "episodes", - "", - map[string]any{ - "anime_id": anime.MalID, - }, - err, - ) - return domain.CanonicalEpisodeList{}, false + if !isCanonicalEpisodePayloadValid(payload, anime.Episodes) { + s.metrics.ObserveCache(metric, "miss") + observability.Info( + "episodes_cached_payload_rejected", + "episodes", + "", + map[string]any{ + "anime_id": anime.MalID, + "expected_count": anime.Episodes, + "cached_episodes": len(payload.Episodes), + }, + ) + return domain.CanonicalEpisodeList{}, false + } + + return payload, true } From 0fe3a7162738385e01e9b3a06ceabfb90ee035bc Mon Sep 17 00:00:00 2001 From: mkelvers Date: Sun, 21 Jun 2026 17:18:43 +0200 Subject: [PATCH 7/9] 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 8/9] 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 9/9] 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) +}