melosh/mal
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge branch 'codex/edb6' into integrate-codex-work

Browse Source 
# Conflicts:
#	integrations/jikan/client_test.go
This commit is contained in:
mkelvers
2026-06-21 17:20:46 +02:00
parent 52b04b17d1 8792cdc457
commit a5875de32e
7 changed files with 173 additions and 1 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
internal/db/querier.go
View File

@@ -31,6 +31,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)

7
internal/db/queries.sql
View File

@@ -235,6 +235,13 @@ WHERE key = ? AND datetime(expires_at) > CURRENT_TIMESTAMP LIMIT 1;
SELECT data FROM jikan_cache
WHERE key = ? AND datetime(expires_at) > datetime(CURRENT_TIMESTAMP, '-14 days') 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 (?, ?, ?)

21
internal/db/queries.sql.go
View File

@@ -576,6 +576,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 = ? AND datetime(expires_at) > datetime(CURRENT_TIMESTAMP, '-14 days') LIMIT 1

92
internal/observability/metrics.go
View File

@@ -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

4
internal/observability/metrics_test.go
View File

@@ -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) {