Merge pull request 'feat(api): Redis read cache for product details and search (stage 0)' (#29) from devin/1782283345-redis-read-cache into main

api/cmd/server/main.go

@@ -9,6 +9,7 @@ import (
 
 	"github.com/jackc/pgx/v5/pgxpool"
 
+	"github.com/baicai2026-baicai/goods/api/internal/cache"
 	"github.com/baicai2026-baicai/goods/api/internal/config"
 	"github.com/baicai2026-baicai/goods/api/internal/handler"
 	"github.com/baicai2026-baicai/goods/api/internal/publicweb"
@@ -36,7 +37,12 @@ func main() {
 	if !limiter.Enabled() {
 		log.Print("warning: Redis not configured; public API rate limiting disabled")
 	}
-	h := handler.New(store.New(pool), publicweb.Dist()).
+
+	readCache := cache.New(cfg.RedisURL)
+	if !readCache.Enabled() {
+		log.Print("warning: Redis not configured; public API read cache disabled")
+	}
+	h := handler.New(store.New(pool).WithCache(readCache), publicweb.Dist()).
 		WithRateLimit(limiter, cfg.AnonRateLimitPerMin).
 		WithQuotas(cfg.AnonTotalQuota, cfg.RegisteredRateLimitPerMin, cfg.RegisteredQuotaTotal)
 

api/internal/cache/cache.go

@@ -0,0 +1,126 @@
+// Package cache is a Redis-backed, fail-open read cache for the public API.
+//
+// It caches hot product details and search results so repeated reads avoid
+// PostgreSQL. Like the ratelimit package, every operation fails open: if Redis
+// is unavailable or misconfigured the caller simply falls back to the database,
+// so the cache can never take the API down or serve stale data after Redis loss.
+//
+// Invalidation is global and O(1): keys are namespaced by an epoch counter
+// stored in Redis (og:cache:epoch). The Python ingestion bumps that counter
+// after a write run, which logically invalidates every cached entry at once
+// while old keys age out via their TTL. The epoch is read at most once per
+// refresh interval per process, so it adds no per-request round trip.
+package cache
+
+import (
+	"context"
+	"encoding/json"
+	"errors"
+	"log"
+	"strconv"
+	"sync"
+	"time"
+
+	"github.com/redis/go-redis/v9"
+)
+
+// epochKey is the Redis key holding the global cache generation counter.
+const epochKey = "og:cache:epoch"
+
+// epochRefresh bounds how often a process re-reads the epoch from Redis.
+const epochRefresh = 10 * time.Second
+
+// opTimeout caps any single Redis operation so a slow backend never blocks a
+// request beyond this; on timeout the cache fails open.
+const opTimeout = 150 * time.Millisecond
+
+// Cache wraps a Redis client. A nil-backed Cache (Redis unconfigured) disables
+// caching: every Get misses and every Set is a no-op.
+type Cache struct {
+	rdb *redis.Client
+
+	mu         sync.RWMutex
+	epoch      int64
+	epochSetAt time.Time
+	epochOK    bool
+}
+
+// New builds a Cache from a redis:// URL. On a parse error it logs and returns a
+// disabled (fail-open) cache so the server still boots.
+func New(redisURL string) *Cache {
+	opt, err := redis.ParseURL(redisURL)
+	if err != nil {
+		log.Printf("cache: invalid redis url %q: %v (caching disabled)", redisURL, err)
+		return &Cache{}
+	}
+	return &Cache{rdb: redis.NewClient(opt)}
+}
+
+// Enabled reports whether a Redis backend is configured.
+func (c *Cache) Enabled() bool { return c != nil && c.rdb != nil }
+
+// epochNow returns the current cache generation, reading it from Redis at most
+// once per epochRefresh. On any Redis error it keeps the last known value and
+// throttles re-reads so a down backend cannot slow the hot path.
+func (c *Cache) epochNow(ctx context.Context) int64 {
+	c.mu.RLock()
+	if c.epochOK && time.Since(c.epochSetAt) < epochRefresh {
+		e := c.epoch
+		c.mu.RUnlock()
+		return e
+	}
+	c.mu.RUnlock()
+
+	cctx, cancel := context.WithTimeout(ctx, opTimeout)
+	defer cancel()
+	n, err := c.rdb.Get(cctx, epochKey).Int64()
+
+	c.mu.Lock()
+	defer c.mu.Unlock()
+	switch {
+	case err == nil:
+		c.epoch = n
+	case errors.Is(err, redis.Nil):
+		c.epoch = 0
+	}
+	c.epochSetAt = time.Now()
+	c.epochOK = true
+	return c.epoch
+}
+
+// key namespaces a logical suffix under the current epoch.
+func (c *Cache) key(ctx context.Context, suffix string) string {
+	return "og:v" + strconv.FormatInt(c.epochNow(ctx), 10) + ":" + suffix
+}
+
+// GetJSON unmarshals the cached value for suffix into dest and reports a hit.
+// Any miss, decode error, or Redis error returns false (fail-open).
+func (c *Cache) GetJSON(ctx context.Context, suffix string, dest any) bool {
+	if !c.Enabled() {
+		return false
+	}
+	k := c.key(ctx, suffix)
+	cctx, cancel := context.WithTimeout(ctx, opTimeout)
+	defer cancel()
+	b, err := c.rdb.Get(cctx, k).Bytes()
+	if err != nil {
+		return false
+	}
+	return json.Unmarshal(b, dest) == nil
+}
+
+// SetJSON stores val (JSON-encoded) for suffix with the given TTL. Best effort:
+// marshal or Redis errors are ignored.
+func (c *Cache) SetJSON(ctx context.Context, suffix string, val any, ttl time.Duration) {
+	if !c.Enabled() {
+		return
+	}
+	b, err := json.Marshal(val)
+	if err != nil {
+		return
+	}
+	k := c.key(ctx, suffix)
+	cctx, cancel := context.WithTimeout(ctx, opTimeout)
+	defer cancel()
+	_ = c.rdb.Set(cctx, k, b, ttl).Err()
+}

api/internal/cache/cache_test.go

@@ -0,0 +1,84 @@
+package cache
+
+import (
+	"context"
+	"fmt"
+	"os"
+	"testing"
+	"time"
+)
+
+// TestDisabledFailsOpen verifies a Cache without a Redis backend never panics,
+// always misses, and silently drops writes.
+func TestDisabledFailsOpen(t *testing.T) {
+	c := New("not-a-valid-url") // parse error => disabled
+	if c.Enabled() {
+		t.Fatal("expected cache to be disabled for invalid url")
+	}
+	c.SetJSON(context.Background(), "k", map[string]int{"a": 1}, time.Minute)
+	var dst map[string]int
+	if c.GetJSON(context.Background(), "k", &dst) {
+		t.Fatalf("disabled cache must always miss, got %+v", dst)
+	}
+}
+
+func testCache(t *testing.T) *Cache {
+	t.Helper()
+	url := os.Getenv("OPENGOODS_REDIS_URL")
+	if url == "" {
+		url = "redis://localhost:6379/0"
+	}
+	c := New(url)
+	if !c.Enabled() {
+		t.Skip("redis not configured")
+	}
+	ctx, cancel := context.WithTimeout(context.Background(), time.Second)
+	defer cancel()
+	if err := c.rdb.Ping(ctx).Err(); err != nil {
+		t.Skipf("redis not reachable: %v", err)
+	}
+	return c
+}
+
+// TestRoundTrip stores then reads a value back.
+func TestRoundTrip(t *testing.T) {
+	c := testCache(t)
+	ctx := context.Background()
+	suffix := fmt.Sprintf("test:rt:%d", time.Now().UnixNano())
+
+	c.SetJSON(ctx, suffix, map[string]any{"name": "foo", "n": float64(3)}, time.Minute)
+	got := map[string]any{}
+	if !c.GetJSON(ctx, suffix, &got) {
+		t.Fatal("expected cache hit after set")
+	}
+	if got["name"] != "foo" || got["n"] != float64(3) {
+		t.Fatalf("unexpected payload: %+v", got)
+	}
+}
+
+// TestEpochInvalidation verifies that bumping the epoch counter logically drops
+// every previously cached entry.
+func TestEpochInvalidation(t *testing.T) {
+	c := testCache(t)
+	ctx := context.Background()
+	suffix := fmt.Sprintf("test:epoch:%d", time.Now().UnixNano())
+
+	c.SetJSON(ctx, suffix, map[string]int{"v": 1}, time.Minute)
+	var dst map[string]int
+	if !c.GetJSON(ctx, suffix, &dst) {
+		t.Fatal("expected hit before epoch bump")
+	}
+
+	// Simulate an ingestion write bumping the global epoch.
+	if err := c.rdb.Incr(ctx, epochKey).Err(); err != nil {
+		t.Fatalf("incr epoch: %v", err)
+	}
+	// Force the process to re-read the epoch rather than use its cached value.
+	c.mu.Lock()
+	c.epochOK = false
+	c.mu.Unlock()
+
+	if c.GetJSON(ctx, suffix, &dst) {
+		t.Fatal("entry should be invisible after epoch bump")
+	}
+}

api/internal/store/store.go

@@ -4,21 +4,35 @@ package store
 
 import (
 	"context"
+	"crypto/sha1"
+	"encoding/hex"
 	"encoding/json"
 	"errors"
 	"strconv"
 	"strings"
+	"time"
 
 	"github.com/jackc/pgx/v5"
 	"github.com/jackc/pgx/v5/pgxpool"
+
+	"github.com/baicai2026-baicai/goods/api/internal/cache"
+)
+
+// Cache TTLs for the public read cache. Product details are far less volatile
+// than search result sets, so they live longer; both are also invalidated
+// wholesale whenever ingestion bumps the cache epoch.
+const (
+	productCacheTTL = 24 * time.Hour
+	searchCacheTTL  = time.Hour
 )
 
 // ErrNotFound is returned when a requested row does not exist.
 var ErrNotFound = errors.New("not found")
 
-// Store wraps a PostgreSQL connection pool.
+// Store wraps a PostgreSQL connection pool and an optional read cache.
 type Store struct {
-	pool *pgxpool.Pool
+	pool  *pgxpool.Pool
+	cache *cache.Cache
 }
 
 // New constructs a Store from an existing pgx pool.
@@ -26,6 +40,30 @@ func New(pool *pgxpool.Pool) *Store {
 	return &Store{pool: pool}
 }
 
+// WithCache attaches a Redis-backed read cache. A nil or disabled cache leaves
+// the Store reading straight from PostgreSQL.
+func (s *Store) WithCache(c *cache.Cache) *Store {
+	s.cache = c
+	return s
+}
+
+// cacheGet reads a cached JSON value into dest, reporting a hit. It is a no-op
+// miss when no cache is attached.
+func (s *Store) cacheGet(ctx context.Context, suffix string, dest any) bool {
+	if s.cache == nil {
+		return false
+	}
+	return s.cache.GetJSON(ctx, suffix, dest)
+}
+
+// cacheSet stores a JSON value when a cache is attached.
+func (s *Store) cacheSet(ctx context.Context, suffix string, val any, ttl time.Duration) {
+	if s.cache == nil {
+		return
+	}
+	s.cache.SetJSON(ctx, suffix, val, ttl)
+}
+
 // Ping verifies database connectivity.
 func (s *Store) Ping(ctx context.Context) error {
 	return s.pool.Ping(ctx)
@@ -222,6 +260,10 @@ func stringifyAttr(v any) string {
 
 // ProductByGTIN looks up an active product by any of its barcodes.
 func (s *Store) ProductByGTIN(ctx context.Context, gtin string) (*Product, error) {
+	const suffix = "prod:gtin:"
+	if cached := new(Product); s.cacheGet(ctx, suffix+gtin, cached) {
+		return cached, nil
+	}
 	row := s.pool.QueryRow(ctx, productSelect+`
  WHERE p.status = 'active'
    AND (p.gtin = $1 OR EXISTS (
@@ -238,11 +280,16 @@ func (s *Store) ProductByGTIN(ctx context.Context, gtin string) (*Product, error
 	if p.Specs, err = s.buildSpecs(ctx, p.ArchiveKind, attrs); err != nil {
 		return nil, err
 	}
+	s.cacheSet(ctx, suffix+gtin, p, productCacheTTL)
 	return p, nil
 }
 
 // ProductByID looks up a product by its UUID.
 func (s *Store) ProductByID(ctx context.Context, id string) (*Product, error) {
+	const suffix = "prod:id:"
+	if cached := new(Product); s.cacheGet(ctx, suffix+id, cached) {
+		return cached, nil
+	}
 	row := s.pool.QueryRow(ctx, productSelect+" WHERE p.id = $1", id)
 	p, attrs, err := scanProduct(row)
 	if err != nil {
@@ -254,6 +301,7 @@ func (s *Store) ProductByID(ctx context.Context, id string) (*Product, error) {
 	if p.Specs, err = s.buildSpecs(ctx, p.ArchiveKind, attrs); err != nil {
 		return nil, err
 	}
+	s.cacheSet(ctx, suffix+id, p, productCacheTTL)
 	return p, nil
 }
 
@@ -268,6 +316,11 @@ const fuzzyThreshold = "0.42"
 // similarity blended with quality_score so the best, most-complete records
 // surface first. Without a query, results are ordered by quality_score.
 func (s *Store) SearchProducts(ctx context.Context, f SearchFilters, limit, offset int) ([]ProductSummary, int, error) {
+	suffix := searchCacheSuffix(f, limit, offset)
+	if entry := new(searchCacheEntry); s.cacheGet(ctx, suffix, entry) {
+		return entry.Items, entry.Total, nil
+	}
+
 	args := []any{}
 	where := "WHERE p.status = 'active'"
 
@@ -334,7 +387,28 @@ LEFT JOIN category c ON c.id = p.category_id `
 		}
 		out = append(out, ps)
 	}
-	return out, total, rows.Err()
+	if err := rows.Err(); err != nil {
+		return nil, 0, err
+	}
+	s.cacheSet(ctx, suffix, searchCacheEntry{Items: out, Total: total}, searchCacheTTL)
+	return out, total, nil
+}
+
+// searchCacheEntry is the cached payload for a SearchProducts call.
+type searchCacheEntry struct {
+	Items []ProductSummary `json:"items"`
+	Total int              `json:"total"`
+}
+
+// searchCacheSuffix derives a stable cache key from the full filter set and
+// paging window so distinct queries never collide.
+func searchCacheSuffix(f SearchFilters, limit, offset int) string {
+	raw := strings.Join([]string{
+		f.Query, f.Category, f.Brand, f.Country,
+		strconv.Itoa(limit), strconv.Itoa(offset),
+	}, "\x1f")
+	sum := sha1.Sum([]byte(raw))
+	return "search:" + hex.EncodeToString(sum[:])
 }
 
 // Nutriments returns just the nutrition payload for a product.

ingestion/opengoods/cache.py

@@ -0,0 +1,52 @@
+"""Read-cache invalidation signal for the public API.
+
+The Go API caches hot product details and search results in Redis, namespacing
+every key by a global generation counter (``og:cache:epoch``). Bumping that
+counter logically invalidates the entire cache in one O(1) operation while the
+old keys age out via their TTL.
+
+Ingestion is the only writer to the database, so after a run that changed data
+it calls :func:`bump_cache_epoch` to make those changes visible immediately
+instead of waiting for per-key TTLs to expire.
+
+Like the API's cache, this is strictly best effort and fails open: if Redis is
+unconfigured or unreachable the ingestion run still succeeds, and stale entries
+simply expire on their own.
+"""
+
+from __future__ import annotations
+
+import logging
+import os
+
+logger = logging.getLogger(__name__)
+
+EPOCH_KEY = "og:cache:epoch"
+
+
+def default_redis_url() -> str | None:
+    """Return the configured Redis URL, or ``None`` when caching is disabled."""
+    return os.environ.get("OPENGOODS_REDIS_URL") or None
+
+
+def bump_cache_epoch(url: str | None = None) -> bool:
+    """Increment the API cache generation counter.
+
+    Returns ``True`` if the counter was bumped, ``False`` if caching is disabled
+    or Redis was unreachable. Never raises: invalidation failures must not fail
+    an ingestion run.
+    """
+    url = url or default_redis_url()
+    if not url:
+        return False
+    try:
+        import redis  # imported lazily so the dependency is optional at runtime
+
+        client = redis.Redis.from_url(url, socket_timeout=2, socket_connect_timeout=2)
+        new_epoch = client.incr(EPOCH_KEY)
+        client.close()
+        logger.info("bumped API cache epoch to %s", new_epoch)
+        return True
+    except Exception as exc:  # noqa: BLE001 - invalidation is best effort
+        logger.warning("cache epoch bump skipped: %s", exc)
+        return False

ingestion/opengoods/jobs/dedup.py

@@ -12,6 +12,7 @@ import sys
 
 import psycopg
 
+from opengoods.cache import bump_cache_epoch
 from opengoods.etl.dedup import dedup_all
 from opengoods.etl.load import default_dsn
 
@@ -23,6 +24,8 @@ def run(args: argparse.Namespace) -> int:
             conn.rollback()
         else:
             conn.commit()
+    if not args.dry_run and summary["merged"]:
+        bump_cache_epoch()
     mode = "dry-run" if args.dry_run else "applied"
     print(f"{mode} groups={summary['groups']} merged={summary['merged']}")
     return 0

ingestion/opengoods/jobs/import_bypos.py

@@ -25,6 +25,7 @@ from collections.abc import Iterator
 import psycopg
 
 from opengoods.adapters.bypos import transform_bypos
+from opengoods.cache import bump_cache_epoch
 from opengoods.etl.load import default_dsn, ensure_bypos_source, load_bypos_record_safe
 
 
@@ -59,6 +60,8 @@ def run(args: argparse.Namespace) -> int:
             else:
                 errored += 1
         conn.commit()
+    if loaded:
+        bump_cache_epoch()
     print(f"loaded={loaded} skipped={skipped} errored={errored}")
     return 0
 

ingestion/opengoods/jobs/seed_off.py

@@ -28,6 +28,7 @@ from opengoods.adapters.openfoodfacts import (
     is_cn_gs1,
     read_dump,
 )
+from opengoods.cache import bump_cache_epoch
 from opengoods.etl.load import default_dsn, ensure_source, load_record_safe
 from opengoods.etl.transform import transform
 
@@ -67,6 +68,8 @@ def run(args: argparse.Namespace) -> int:
             else:
                 errored += 1
         conn.commit()
+    if loaded:
+        bump_cache_epoch()
     print(f"loaded={loaded} skipped={skipped} errored={errored}")
     return 0
 

ingestion/opengoods/jobs/update_off.py

@@ -17,6 +17,7 @@ import sys
 import psycopg
 
 from opengoods.adapters.openfoodfacts import SOURCE_NAME, OpenFoodFactsAdapter
+from opengoods.cache import bump_cache_epoch
 from opengoods.etl.load import default_dsn, ensure_source, load_record_safe
 from opengoods.etl.state import get_watermark, set_watermark
 from opengoods.etl.transform import transform
@@ -49,6 +50,8 @@ def run(args: argparse.Namespace) -> int:
             stats={"loaded": loaded, "skipped": skipped, "errored": errored, "since": since},
         )
         conn.commit()
+    if loaded:
+        bump_cache_epoch()
     print(
         f"since={since} loaded={loaded} skipped={skipped} "
         f"errored={errored} watermark={high_watermark}"

ingestion/pyproject.toml

@@ -6,6 +6,7 @@ requires-python = ">=3.11"
 dependencies = [
     "httpx>=0.27",
     "psycopg[binary]>=3.2",
+    "redis>=5.0,<6",
 ]
 
 [project.optional-dependencies]

ingestion/tests/test_cache.py

@@ -0,0 +1,44 @@
+"""Tests for the best-effort API cache invalidation signal."""
+
+from __future__ import annotations
+
+import sys
+
+import opengoods.cache as cache
+
+
+def test_disabled_when_no_url(monkeypatch):
+    monkeypatch.delenv("OPENGOODS_REDIS_URL", raising=False)
+    assert cache.default_redis_url() is None
+    assert cache.bump_cache_epoch() is False
+
+
+def test_bump_fails_open_when_unreachable(monkeypatch):
+    # An unroutable URL must not raise; the run still succeeds.
+    monkeypatch.setenv("OPENGOODS_REDIS_URL", "redis://127.0.0.1:1/0")
+    assert cache.bump_cache_epoch() is False
+
+
+def test_bump_increments_epoch(monkeypatch):
+    calls = {}
+
+    class FakeClient:
+        def incr(self, key):
+            calls["key"] = key
+            return 7
+
+        def close(self):
+            calls["closed"] = True
+
+    class FakeRedis:
+        @staticmethod
+        def from_url(url, **kwargs):
+            calls["url"] = url
+            return FakeClient()
+
+    monkeypatch.setitem(sys.modules, "redis", type("M", (), {"Redis": FakeRedis}))
+
+    assert cache.bump_cache_epoch("redis://example:6379/0") is True
+    assert calls["key"] == "og:cache:epoch"
+    assert calls["url"] == "redis://example:6379/0"
+    assert calls["closed"] is True