Compare commits

...

10 Commits

Author SHA1 Message Date
rosemariejebbjtxbfp a75318b811 feat(api): Redis read cache for product details and search
CI / Go (api) (pull_request) Successful in 54s
CI / Python (ingestion) (pull_request) Successful in 19s
CI / Migrations (postgres) (pull_request) Successful in 27s
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>
2026-06-24 06:52:56 +00:00
lixu 20b9eb0fc9 Merge pull request 'ci: 修复 main 上 Python CI 历史失败(在 python3.12+node 容器内运行)' (#28) from devin/1782280936-fix-python-ci into main
CI / Go (api) (push) Successful in 54s
CI / Python (ingestion) (push) Successful in 23s
CI / Migrations (postgres) (push) Successful in 27s
2026-06-24 14:32:51 +08:00
rosemariejebbjtxbfp 17bc0ed680 ci: remove GitHub-hosted actions, manual checkout + Go from CN mirror
CI / Go (api) (pull_request) Successful in 51s
CI / Python (ingestion) (pull_request) Successful in 21s
CI / Migrations (postgres) (pull_request) Successful in 26s
Self-hosted Gitea runner has flaky/blocked access to github.com, causing
actions/checkout and actions/setup-go to time out intermittently across all
jobs. Replace them with:
- manual git checkout against $GITHUB_SERVER_URL (the Gitea host, reachable
  from job containers)
- Go installed from mirrors.aliyun.com

Python job stays on the python3.12-nodejs20 container (fixes the original
PEP 660 editable-install failure). No more github.com network dependency.

Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>
2026-06-24 06:30:03 +00:00
rosemariejebbjtxbfp 5020fdcc19 ci: re-trigger with public gitea actions url
CI / Go (api) (pull_request) Failing after 11s
CI / Python (ingestion) (pull_request) Failing after 31s
CI / Migrations (postgres) (pull_request) Failing after 11s
Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>
2026-06-24 06:24:52 +00:00
rosemariejebbjtxbfp 885f6c2b01 ci: re-trigger with self-hosted actions
CI / Go (api) (pull_request) Failing after 12s
CI / Python (ingestion) (pull_request) Failing after 1m31s
CI / Migrations (postgres) (pull_request) Failing after 32s
Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>
2026-06-24 06:21:01 +00:00
rosemariejebbjtxbfp 2aff6287d2 ci: re-trigger after configuring action proxy
CI / Go (api) (pull_request) Failing after 13s
CI / Python (ingestion) (pull_request) Failing after 31s
CI / Migrations (postgres) (pull_request) Failing after 1m32s
Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>
2026-06-24 06:10:22 +00:00
rosemariejebbjtxbfp 3364a50c68 ci: run Python job in python3.12+node container
CI / Go (api) (pull_request) Failing after 1m32s
CI / Python (ingestion) (pull_request) Failing after 31s
CI / Migrations (postgres) (pull_request) Failing after 19m16s
setup-python@v5 can't fetch CPython 3.12 on the self-hosted Gitea runner
(it queries the Gitea API for actions/python-versions, which 404s), so the
job silently fell back to the image's system Python 3.10 + pip 22.0.2. That
old pip lacks PEP 660 editable support, so 'pip install -e' failed with
'build backend is missing the build_editable hook', and 3.10 is below the
project's requires-python>=3.11 (code uses datetime.UTC).

Run the job inside nikolaik/python-nodejs:python3.12-nodejs20 which bundles
Python 3.12, Node 20 (for actions/checkout) and modern pip, removing the
GitHub download dependency entirely.

Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>
2026-06-24 06:02:28 +00:00
lixu 936828abae Merge pull request 'perf(search): 浏览路径部分复合索引 (0012)' (#27) from devin/1782279211-search-indexes into main
CI / Go (api) (push) Successful in 49s
CI / Python (ingestion) (push) Failing after 15s
CI / Migrations (postgres) (push) Successful in 31s
2026-06-24 13:35:50 +08:00
rosemariejebbjtxbfp ab7a964934 perf(search): 为浏览路径补部分复合索引 (0012)
CI / Go (api) (pull_request) Successful in 58s
CI / Python (ingestion) (pull_request) Failing after 21s
CI / Migrations (postgres) (pull_request) Successful in 32s
Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>
2026-06-24 05:33:31 +00:00
lixu 649fcc711c Merge pull request 'docs: 可扩展性设计与路线图 (scalability roadmap)' (#26) from devin/1782278890-scalability-doc into main
CI / Go (api) (push) Successful in 52s
CI / Python (ingestion) (push) Failing after 16s
CI / Migrations (postgres) (push) Successful in 26s
2026-06-24 13:31:21 +08:00
14 changed files with 465 additions and 17 deletions
+45 -13
View File
@@ -5,6 +5,10 @@ on:
branches: [main]
pull_request:
env:
GO_VERSION: "1.23.12"
GOPROXY: "https://goproxy.cn,direct"
jobs:
go:
name: Go (api)
@@ -25,11 +29,22 @@ jobs:
env:
OPENGOODS_DATABASE_URL: postgres://opengoods:opengoods@postgres:5432/opengoods?sslmode=disable
steps:
- uses: actions/checkout@v4
- uses: actions/setup-go@v5
with:
go-version: "1.23"
cache-dependency-path: api/go.sum
- name: Checkout
working-directory: ${{ github.workspace }}
run: |
git config --global --add safe.directory '*'
git init -q .
git remote add origin "${GITHUB_SERVER_URL}/${GITHUB_REPOSITORY}.git"
git -c protocol.version=2 fetch -q --no-tags --depth 1 origin "${GITHUB_REF}"
git checkout -q --force FETCH_HEAD
- name: Setup Go
working-directory: ${{ github.workspace }}
run: |
curl -fsSL -o /tmp/go.tgz "https://mirrors.aliyun.com/golang/go${GO_VERSION}.linux-amd64.tar.gz"
rm -rf /usr/local/go
tar -C /usr/local -xzf /tmp/go.tgz
echo "/usr/local/go/bin" >> "$GITHUB_PATH"
echo "$HOME/go/bin" >> "$GITHUB_PATH"
- name: Apply migrations
working-directory: .
run: |
@@ -44,14 +59,20 @@ jobs:
python:
name: Python (ingestion)
runs-on: ubuntu-latest
container:
image: nikolaik/python-nodejs:python3.12-nodejs20
defaults:
run:
working-directory: ingestion
steps:
- uses: actions/checkout@v4
- uses: actions/setup-python@v5
with:
python-version: "3.12"
- name: Checkout
working-directory: ${{ github.workspace }}
run: |
git config --global --add safe.directory '*'
git init -q .
git remote add origin "${GITHUB_SERVER_URL}/${GITHUB_REPOSITORY}.git"
git -c protocol.version=2 fetch -q --no-tags --depth 1 origin "${GITHUB_REF}"
git checkout -q --force FETCH_HEAD
- name: Install
run: pip install -e ".[dev]"
- name: Ruff lint
@@ -77,10 +98,21 @@ jobs:
env:
DBURL: postgres://opengoods:opengoods@postgres:5432/opengoods?sslmode=disable
steps:
- uses: actions/checkout@v4
- uses: actions/setup-go@v5
with:
go-version: "1.23"
- name: Checkout
working-directory: ${{ github.workspace }}
run: |
git config --global --add safe.directory '*'
git init -q .
git remote add origin "${GITHUB_SERVER_URL}/${GITHUB_REPOSITORY}.git"
git -c protocol.version=2 fetch -q --no-tags --depth 1 origin "${GITHUB_REF}"
git checkout -q --force FETCH_HEAD
- name: Setup Go
run: |
curl -fsSL -o /tmp/go.tgz "https://mirrors.aliyun.com/golang/go${GO_VERSION}.linux-amd64.tar.gz"
rm -rf /usr/local/go
tar -C /usr/local -xzf /tmp/go.tgz
echo "/usr/local/go/bin" >> "$GITHUB_PATH"
echo "$HOME/go/bin" >> "$GITHUB_PATH"
- name: Install golang-migrate
run: go install -tags 'postgres' github.com/golang-migrate/migrate/v4/cmd/migrate@v4.18.1
- name: Migrate up
+7 -1
View File
@@ -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)
+126
View File
@@ -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()
}
+84
View File
@@ -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")
}
}
+77 -3
View File
@@ -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.
+52
View File
@@ -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
+3
View File
@@ -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
+3
View File
@@ -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
+3
View File
@@ -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
+3
View File
@@ -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}"
+1
View File
@@ -6,6 +6,7 @@ requires-python = ">=3.11"
dependencies = [
"httpx>=0.27",
"psycopg[binary]>=3.2",
"redis>=5.0,<6",
]
[project.optional-dependencies]
+44
View File
@@ -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
+2
View File
@@ -0,0 +1,2 @@
DROP INDEX IF EXISTS idx_product_active_category;
DROP INDEX IF EXISTS idx_product_active_quality;
+15
View File
@@ -0,0 +1,15 @@
-- 检索优化(可扩展性路线图 阶段0):为最常见的"浏览"路径补部分/复合索引。
-- 搜索查询恒带 WHERE status = 'active';无关键词时按 quality_score DESC, name 排序。
-- 现有索引无法同时满足"过滤 active + 按 quality_score/name 排序",深翻页时需要对全部
-- active 行排序。下面的部分复合索引让规划器直接走索引顺序扫描,省掉排序、加速深翻页与
-- count(*)。索引只覆盖 active 行,体积更小。
-- 默认浏览(无关键词、无品类):ORDER BY quality_score DESC, name
CREATE INDEX IF NOT EXISTS idx_product_active_quality
ON product (quality_score DESC, name)
WHERE status = 'active';
-- 品类内浏览:先按 category_id 收敛,再按 quality_score 排序
CREATE INDEX IF NOT EXISTS idx_product_active_category
ON product (category_id, quality_score DESC)
WHERE status = 'active';