kafka/internal/cache/cache.go
Franz Kafka 94322ceff4 feat: Valkey cache for search results
- Add internal/cache package using go-redis/v9 (Valkey-compatible)
- Cache keys are deterministic SHA-256 hashes of search parameters
- Cache wraps the Search() method: check cache → miss → execute → store
- Gracefully disabled if Valkey is unreachable or unconfigured
- Configurable TTL (default 5m), address, password, and DB index
- Environment variable overrides: VALKEY_ADDRESS, VALKEY_PASSWORD,
  VALKEY_DB, VALKEY_CACHE_TTL
- Structured JSON logging via slog throughout cache layer
- Refactored service.go: extract executeSearch() from Search() for clarity
- Update config.example.toml with [cache] section
- Add cache package tests (key generation, nop behavior)
2026-03-21 15:43:47 +00:00

162 lines
4.1 KiB
Go

package cache
import (
"context"
"crypto/sha256"
"encoding/hex"
"encoding/json"
"fmt"
"log/slog"
"time"
"github.com/ashie/gosearch/internal/contracts"
"github.com/redis/go-redis/v9"
)
// Config holds Valkey/Redis connection settings.
type Config struct {
// Address is the Valkey server address (e.g. "localhost:6379").
Address string
// Password for authentication (empty = no auth).
Password string
// Database index (default 0).
DB int
// Default TTL for cached search results.
DefaultTTL time.Duration
}
// Cache provides a Valkey-backed cache for search responses.
// It is safe for concurrent use.
// If the Valkey connection is nil or fails, cache operations are no-ops.
type Cache struct {
client *redis.Client
ttl time.Duration
logger *slog.Logger
}
// New creates a new Cache. If cfg.Address is empty, returns a no-op cache.
func New(cfg Config, logger *slog.Logger) *Cache {
if logger == nil {
logger = slog.Default()
}
if cfg.Address == "" {
logger.Debug("cache disabled: no valkey address configured")
return &Cache{logger: logger}
}
ttl := cfg.DefaultTTL
if ttl <= 0 {
ttl = 5 * time.Minute
}
client := redis.NewClient(&redis.Options{
Addr: cfg.Address,
Password: cfg.Password,
DB: cfg.DB,
})
// Verify connectivity with a short timeout.
ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second)
defer cancel()
if err := client.Ping(ctx).Err(); err != nil {
logger.Warn("cache disabled: valkey ping failed", "addr", cfg.Address, "error", err)
return &Cache{logger: logger}
}
logger.Info("cache connected", "addr", cfg.Address, "db", cfg.DB, "ttl", ttl)
return &Cache{client: client, ttl: ttl, logger: logger}
}
// Enabled returns true if the cache has a live Valkey connection.
func (c *Cache) Enabled() bool {
return c.client != nil
}
// Get retrieves a cached search response. Returns (response, true) on hit,
// (zero, false) on miss or error.
func (c *Cache) Get(ctx context.Context, key string) (contracts.SearchResponse, bool) {
if !c.Enabled() {
return contracts.SearchResponse{}, false
}
fullKey := "gosearch:" + key
data, err := c.client.Get(ctx, fullKey).Bytes()
if err != nil {
if err != redis.Nil {
c.logger.Debug("cache miss (error)", "key", fullKey, "error", err)
}
return contracts.SearchResponse{}, false
}
var resp contracts.SearchResponse
if err := json.Unmarshal(data, &resp); err != nil {
c.logger.Warn("cache hit but unmarshal failed", "key", fullKey, "error", err)
return contracts.SearchResponse{}, false
}
c.logger.Debug("cache hit", "key", fullKey)
return resp, true
}
// Set stores a search response in the cache with the default TTL.
func (c *Cache) Set(ctx context.Context, key string, resp contracts.SearchResponse) {
if !c.Enabled() {
return
}
data, err := json.Marshal(resp)
if err != nil {
c.logger.Warn("cache set: marshal failed", "key", key, "error", err)
return
}
fullKey := "gosearch:" + key
if err := c.client.Set(ctx, fullKey, data, c.ttl).Err(); err != nil {
c.logger.Warn("cache set failed", "key", fullKey, "error", err)
}
}
// Invalidate removes a specific key from the cache.
func (c *Cache) Invalidate(ctx context.Context, key string) {
if !c.Enabled() {
return
}
fullKey := "gosearch:" + key
c.client.Del(ctx, fullKey)
}
// Close closes the Valkey connection.
func (c *Cache) Close() error {
if c.client == nil {
return nil
}
return c.client.Close()
}
// Key generates a deterministic cache key from search parameters.
// The key is a SHA-256 hash of the normalized parameters, prefixed for readability.
func Key(req contracts.SearchRequest) string {
h := sha256.New()
fmt.Fprintf(h, "q=%s|", req.Query)
fmt.Fprintf(h, "format=%s|", req.Format)
fmt.Fprintf(h, "pageno=%d|", req.Pageno)
fmt.Fprintf(h, "safesearch=%d|", req.Safesearch)
fmt.Fprintf(h, "lang=%s|", req.Language)
if req.TimeRange != nil {
fmt.Fprintf(h, "tr=%s|", *req.TimeRange)
}
for _, e := range req.Engines {
fmt.Fprintf(h, "e=%s|", e)
}
for _, cat := range req.Categories {
fmt.Fprintf(h, "c=%s|", cat)
}
return hex.EncodeToString(h.Sum(nil))[:32]
}