124 lines
3.0 KiB
Go
124 lines
3.0 KiB
Go
package database
|
|
|
|
import (
|
|
"context"
|
|
"encoding/json"
|
|
"log/slog"
|
|
"time"
|
|
|
|
"github.com/jmoiron/sqlx"
|
|
)
|
|
|
|
// CacheRepo manages the source_cache table for remote subscription caching.
|
|
type CacheRepo struct {
|
|
db *sqlx.DB
|
|
}
|
|
|
|
func NewCacheRepo(db *sqlx.DB) *CacheRepo {
|
|
return &CacheRepo{db: db}
|
|
}
|
|
|
|
type cacheRow struct {
|
|
CacheKey string `db:"cache_key"`
|
|
Content string `db:"content"`
|
|
Metadata string `db:"metadata"`
|
|
CachedAt int64 `db:"cached_at"`
|
|
TTL int `db:"ttl"`
|
|
}
|
|
|
|
// CacheEntry holds cached content and its metadata.
|
|
type CacheEntry struct {
|
|
Content string
|
|
Metadata map[string]any
|
|
CachedAt int64
|
|
TTL int
|
|
}
|
|
|
|
// Get retrieves a cache entry by key. Returns nil if not found or expired.
|
|
func (r *CacheRepo) Get(key string) (*CacheEntry, error) {
|
|
var row cacheRow
|
|
err := r.db.Get(&row, "SELECT * FROM source_cache WHERE cache_key = ?", key)
|
|
if err != nil {
|
|
return nil, nil // cache miss is not an error
|
|
}
|
|
// Check expiry
|
|
if time.Now().Unix() > row.CachedAt+int64(row.TTL) {
|
|
return nil, nil
|
|
}
|
|
var metadata map[string]any
|
|
json.Unmarshal([]byte(row.Metadata), &metadata)
|
|
if metadata == nil {
|
|
metadata = map[string]any{}
|
|
}
|
|
return &CacheEntry{
|
|
Content: row.Content,
|
|
Metadata: metadata,
|
|
CachedAt: row.CachedAt,
|
|
TTL: row.TTL,
|
|
}, nil
|
|
}
|
|
|
|
// Put stores a cache entry.
|
|
func (r *CacheRepo) Put(key, content string, metadata map[string]any, ttl int) error {
|
|
defer func() {
|
|
if rec := recover(); rec != nil {
|
|
slog.Warn("cache put panicked", "error", rec)
|
|
}
|
|
}()
|
|
now := time.Now().Unix()
|
|
metaJSON := marshalJSON(metadata)
|
|
_, err := r.db.Exec(
|
|
`INSERT INTO source_cache (cache_key, content, metadata, cached_at, ttl)
|
|
VALUES (?, ?, ?, ?, ?)
|
|
ON CONFLICT(cache_key) DO UPDATE SET
|
|
content = excluded.content, metadata = excluded.metadata,
|
|
cached_at = excluded.cached_at, ttl = excluded.ttl`,
|
|
key, content, metaJSON, now, ttl,
|
|
)
|
|
return err
|
|
}
|
|
|
|
// SafeGet wraps Get with panic recovery — per review-resolution #35.
|
|
func (r *CacheRepo) SafeGet(key string) (*CacheEntry, bool) {
|
|
defer func() {
|
|
if rec := recover(); rec != nil {
|
|
slog.Warn("cache get panicked", "error", rec)
|
|
}
|
|
}()
|
|
entry, err := r.Get(key)
|
|
if err != nil || entry == nil {
|
|
return nil, false
|
|
}
|
|
return entry, true
|
|
}
|
|
|
|
// SafePut wraps Put with panic recovery — per review-resolution #35.
|
|
func (r *CacheRepo) SafePut(key, content string, metadata map[string]any, ttl int) {
|
|
defer func() {
|
|
if rec := recover(); rec != nil {
|
|
slog.Warn("cache put panicked", "error", rec)
|
|
}
|
|
}()
|
|
_ = r.Put(key, content, metadata, ttl)
|
|
}
|
|
|
|
// StartCacheCleaner runs a background goroutine that periodically removes expired cache entries.
|
|
// Per review-resolution #32.
|
|
func StartCacheCleaner(ctx context.Context, db *sqlx.DB, interval time.Duration) {
|
|
ticker := time.NewTicker(interval)
|
|
go func() {
|
|
defer ticker.Stop()
|
|
for {
|
|
select {
|
|
case <-ticker.C:
|
|
_, _ = db.ExecContext(ctx,
|
|
"DELETE FROM source_cache WHERE cached_at + ttl < ?",
|
|
time.Now().Unix(),
|
|
)
|
|
case <-ctx.Done():
|
|
return
|
|
}
|
|
}
|
|
}()
|
|
}
|