feat(api): Redis read cache for product details and search

Stage-0 caching from docs/scalability.md. The Go API now caches hot
product-detail and search-result reads in Redis with a fail-open,
epoch-versioned scheme; Python ingestion bumps the epoch after a write
run to invalidate the cache globally in O(1).

- api/internal/cache: fail-open Cache (GetJSON/SetJSON) namespaced by an
  epoch counter (og:cache:epoch); disabled when Redis is unconfigured.
- store: ProductByID/ProductByGTIN (24h TTL) and SearchProducts (1h TTL)
  read-through the cache via WithCache.
- ingestion: bump_cache_epoch() called after update_off/seed_off/
  import_bypos/dedup commits when rows changed; best-effort, never fails
  a run. Adds redis dependency.

Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>

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