Serve catalogue scale from one exact, cached snapshot

AGENTS.md

@@ -148,6 +148,7 @@ Each is self-contained and can be read independently.
 - **API keys:** Hashed at rest (SHA-256), scoped `full` or `cv`, and mintable only by an account with a verified address. Key management (create/list/revoke) and password change are cookie-only. A key does not carry the session generation, so a `token_version` bump does not revoke it — that is intentional for sign-out-everywhere and wrong for a takeover, so the seizure and the mailed-code password reset delete the rows in the same statement
 - **Enrichment:** Queue-driven (`enrichment_outbox`), provider-agnostic LLM, `Sanitize` + `Validate` gate
 - **Embeddings:** Queue-driven (`semantic_outbox`), incremental (`cmd/embed`) — pgvector-backed `job_semantic_chunks` plus a legacy single-vector column, no search index. Reconciled by bumping the embedder-model version string, which re-enqueues the whole catalogue through the existing staleness check
+- **Catalogue scale:** Every public figure describing how big the catalogue is — `GET /api/v1/stats/catalog`, the jobs list's `meta.total`, the `/about` and `/open` strips — reads ONE snapshot published by `cmd/rollup-stats` (`internal/catalogstats`). Never count on a request path: `catalogstats.Load` takes no exact counter, so it cannot. A read never fails; a cold cache, an unreachable Redis or no cache at all degrades to the planner estimate with `exact: false`, which zeroes the figures that exist only in the database — pass `exact` through to whatever renders them, because a zero must not reach a page as if it were a measurement
 - **Dictionaries:** All facet dictionaries are dict-only in production — never guess, emit nothing for unknowns
 - **Job deletion:** The lifecycle only soft-closes; `cmd/prune` is the sole hard-delete path
 - **In-app assistant:** a bounded tool-calling loop in-process (`internal/assistant`), streamed over SSE, open to every signed-in user. Tools act as the authenticated caller — no credential is minted for an agent

cmd/rollup-stats/main.go

@@ -47,7 +47,7 @@ func main() {
 }
 
 func run() int {
-	ctx, _, pool, cleanup, err := worker.Bootstrap(context.Background())
+	ctx, cfg, pool, cleanup, err := worker.Bootstrap(context.Background())
 	if err != nil {
 		log.Printf("database: %v", err)
 		return 1
@@ -97,9 +97,42 @@ func run() int {
 	}
 
 	log.Printf("rollup-stats: rebuilt job_daily_stats (%d active day rows) and insights_* rollups", days)
+
+	// The catalogue-scale snapshot rides along after the rollups have committed. It is
+	// a separate concern with its own failure mode, so it neither joins their
+	// transaction nor changes this run's exit code: the rollups are the worker's job,
+	// and they are already done.
+	publish(ctx, cfg.RedisURL, db.New(pool))
+
 	return 0
 }
 
+// publish measures the catalogue and stores the snapshot every public surface reads.
+// Every failure here is logged and swallowed — see the call site.
+func publish(ctx context.Context, redisURL string, q *db.Queries) {
+	c, closeCache, err := snapshotCache(redisURL)
+	if err != nil {
+		log.Printf("rollup-stats: catalogue snapshot not published: %v", err)
+		return
+	}
+	defer closeCache()
+
+	// A missing or unreadable channel file costs one stat, not the snapshot. The
+	// counts are why this exists; publishing them with a zero channel count beats
+	// publishing nothing and leaving every surface on the estimate.
+	channels, err := configuredTelegramChannels()
+	if err != nil {
+		log.Printf("rollup-stats: telegram channel count unavailable, publishing without it: %v", err)
+	}
+
+	if err := publishSnapshot(ctx, q, c, channels); err != nil {
+		log.Printf("rollup-stats: catalogue snapshot not published: %v", err)
+		return
+	}
+
+	log.Printf("rollup-stats: published the catalogue-scale snapshot")
+}
+
 // rebuildInsights clears and recomputes the four insights_* rollups inside the
 // caller's transaction. prevTs (the growth-window start) and minSalarySample are
 // passed to the SQL so the window and sample floor live here, not in the queries.

cmd/rollup-stats/publish.go

@@ -0,0 +1,55 @@
+package main
+
+import (
+	"context"
+	"fmt"
+
+	"github.com/redis/go-redis/v9"
+
+	"github.com/strelov1/freehire/internal/cache"
+	"github.com/strelov1/freehire/internal/catalogstats"
+	"github.com/strelov1/freehire/internal/telegram"
+)
+
+// publishSnapshot measures the catalogue and publishes the figures every public surface
+// quotes (internal/catalogstats).
+//
+// It lives in this worker because the exact counts are a full catalogue scan and this
+// worker is already scanning jobs for the rollups — one more aggregate on a run that is
+// already doing heavier work, against a new cron unit that would cost real ops surface.
+//
+// The error is returned rather than acted on: the rollups are this worker's primary job
+// and have already committed by the time this runs, so whether a failed snapshot should
+// fail the run is the caller's decision, and the caller's answer is no.
+func publishSnapshot(ctx context.Context, counts catalogstats.ExactCounter, c cache.Cache, telegramChannels int) error {
+	snapshot, err := catalogstats.Compute(ctx, counts, telegramChannels)
+	if err != nil {
+		return err
+	}
+	if err := catalogstats.Store(ctx, c, snapshot); err != nil {
+		return fmt.Errorf("publishing the catalogue snapshot: %w", err)
+	}
+	return nil
+}
+
+// snapshotCache builds the shared cache the snapshot is published to. A malformed
+// REDIS_URL is reported, not fatal: this worker's primary job needs no cache at all.
+func snapshotCache(redisURL string) (cache.Cache, func(), error) {
+	opts, err := redis.ParseURL(redisURL)
+	if err != nil {
+		return nil, nil, fmt.Errorf("redis: %w", err)
+	}
+	client := redis.NewClient(opts)
+	return cache.NewRedisCache(client), func() { _ = client.Close() }, nil
+}
+
+// configuredTelegramChannels counts the channels the crawler is configured to read.
+// Resolving it means reading sources/telegram.yml relative to the worker's working
+// directory, which is why catalogstats takes the count rather than finding it itself.
+func configuredTelegramChannels() (int, error) {
+	cfg, err := telegram.LoadChannels()
+	if err != nil {
+		return 0, err
+	}
+	return len(cfg.Channels), nil
+}

cmd/rollup-stats/publish_test.go

@@ -0,0 +1,78 @@
+package main
+
+import (
+	"context"
+	"errors"
+	"testing"
+	"time"
+
+	"github.com/strelov1/freehire/internal/cache"
+	"github.com/strelov1/freehire/internal/catalogstats"
+	"github.com/strelov1/freehire/internal/db"
+)
+
+type stubCounter struct {
+	row db.CountCatalogueScaleRow
+	err error
+}
+
+func (s stubCounter) CountCatalogueScale(context.Context) (db.CountCatalogueScaleRow, error) {
+	return s.row, s.err
+}
+
+func TestPublishSnapshotStoresAReadableSnapshot(t *testing.T) {
+	c := cache.NewMemory()
+	ctx := context.Background()
+	counts := stubCounter{row: db.CountCatalogueScaleRow{OpenJobs: 3_300_658, Companies: 294_282}}
+
+	if err := publishSnapshot(ctx, counts, c, 95); err != nil {
+		t.Fatalf("publishSnapshot: %v", err)
+	}
+
+	got := catalogstats.Load(ctx, c, failingEstimator{})
+	if !got.Exact {
+		t.Fatal("Load reports a degraded snapshot right after publishing one")
+	}
+	if got.OpenJobs != 3_300_658 || got.Companies != 294_282 {
+		t.Errorf("OpenJobs/Companies = %d/%d, want 3300658/294282", got.OpenJobs, got.Companies)
+	}
+	if got.TelegramChannels != 95 {
+		t.Errorf("TelegramChannels = %d, want the 95 passed in", got.TelegramChannels)
+	}
+}
+
+// The rollups are this worker's primary job and commit in their own transaction. A
+// snapshot that cannot be computed or stored is worth logging, not worth failing a run
+// that already did its work — so the failure must arrive as a value the caller chooses
+// what to do with, not as a panic or a process exit.
+func TestPublishSnapshotReportsFailureWithoutPanicking(t *testing.T) {
+	ctx := context.Background()
+
+	t.Run("counting fails", func(t *testing.T) {
+		err := publishSnapshot(ctx, stubCounter{err: errors.New("scan failed")}, cache.NewMemory(), 95)
+		if err == nil {
+			t.Error("publishSnapshot returned nil when the count failed — the caller has nothing to log")
+		}
+	})
+
+	t.Run("storing fails", func(t *testing.T) {
+		counts := stubCounter{row: db.CountCatalogueScaleRow{OpenJobs: 1, Companies: 1}}
+		err := publishSnapshot(ctx, counts, unwritableCache{}, 95)
+		if err == nil {
+			t.Error("publishSnapshot returned nil when the store failed")
+		}
+	})
+}
+
+type failingEstimator struct{}
+
+func (failingEstimator) EstimateOpenJobs(context.Context) (int64, error) {
+	return 0, errors.New("the estimate must not be reached when a snapshot exists")
+}
+
+type unwritableCache struct{ cache.Cache }
+
+func (unwritableCache) Get(context.Context, string) ([]byte, bool, error) { return nil, false, nil }
+func (unwritableCache) Set(context.Context, string, []byte, time.Duration) error {
+	return errors.New("backend unreachable")
+}

cmd/server/main.go

@@ -16,6 +16,7 @@ import (
 	appleauth "github.com/strelov1/freehire/internal/auth/apple"
 	"github.com/strelov1/freehire/internal/auth/oauth"
 	"github.com/strelov1/freehire/internal/blobstore"
+	"github.com/strelov1/freehire/internal/cache"
 	"github.com/strelov1/freehire/internal/config"
 	"github.com/strelov1/freehire/internal/credits"
 	"github.com/strelov1/freehire/internal/cv"
@@ -257,6 +258,7 @@ func main() {
 	handler.Register(app, handler.Config{
 		Pool:                        pool,
 		Throttler:                   throttler,
+		Cache:                       cache.NewRedisCache(redisClient),
 		FrontendOrigin:              cfg.FrontendOrigin,
 		JWTSecret:                   cfg.JWTSecret,
 		JWTTTL:                      cfg.JWTTTL,

internal/cache/aliasing_test.go

@@ -0,0 +1,70 @@
+package cache
+
+import (
+	"context"
+	"testing"
+	"time"
+)
+
+// The two implementations must not differ in whether a caller can reach into the
+// cache's own state. RedisCache serializes over a socket, so its callers physically
+// cannot; Memory hands out whatever it was given. If Memory aliases, code that is
+// correct against Redis silently corrupts the cache when run against Memory — the worst
+// shape of leaky abstraction, because tests use Memory and production uses Redis.
+//
+// Run against both implementations so neither can drift from the other.
+func forEachCache(t *testing.T, fn func(t *testing.T, c Cache)) {
+	t.Helper()
+	t.Run("Memory", func(t *testing.T) {
+		m, _ := newTestMemory(t)
+		fn(t, m)
+	})
+	t.Run("Redis", func(t *testing.T) {
+		c, _ := newTestRedisCache(t)
+		fn(t, c)
+	})
+}
+
+func TestCacheDoesNotAliasStoredValue(t *testing.T) {
+	forEachCache(t, func(t *testing.T, c Cache) {
+		ctx := context.Background()
+		val := []byte("original")
+
+		if err := c.Set(ctx, "k", val, time.Minute); err != nil {
+			t.Fatalf("Set: %v", err)
+		}
+		val[0] = 'X' // the caller reuses its buffer, as callers do
+
+		got, _, err := c.Get(ctx, "k")
+		if err != nil {
+			t.Fatalf("Get: %v", err)
+		}
+		if string(got) != "original" {
+			t.Errorf("Get = %q, want %q — mutating the slice passed to Set changed the cached entry", got, "original")
+		}
+	})
+}
+
+func TestCacheDoesNotAliasReturnedValue(t *testing.T) {
+	forEachCache(t, func(t *testing.T, c Cache) {
+		ctx := context.Background()
+
+		if err := c.Set(ctx, "k", []byte("original"), time.Minute); err != nil {
+			t.Fatalf("Set: %v", err)
+		}
+
+		first, _, err := c.Get(ctx, "k")
+		if err != nil {
+			t.Fatalf("first Get: %v", err)
+		}
+		first[0] = 'X' // a caller decoding in place, or just reusing the slice
+
+		second, _, err := c.Get(ctx, "k")
+		if err != nil {
+			t.Fatalf("second Get: %v", err)
+		}
+		if string(second) != "original" {
+			t.Errorf("Get = %q, want %q — mutating a returned slice changed the cached entry", second, "original")
+		}
+	})
+}

internal/cache/cache.go

@@ -0,0 +1,39 @@
+// Package cache provides the shared best-effort key-value cache used for values that
+// are expensive to compute, identical for every caller, and tolerable slightly stale.
+//
+// Nothing here is a source of truth. Every value in a Cache can also be obtained (or
+// approximated) some other way, and a caller that cannot read the cache is expected to
+// do exactly that — which is why the interface reports errors rather than swallowing
+// them, and why no method promises the value it stored is still there.
+package cache
+
+import (
+	"context"
+	"time"
+)
+
+// Cache stores bytes under a key for a bounded time.
+//
+// A caller MUST treat both a miss and an error as "no cached value" and fall back to
+// its own source of truth. The interface surfaces the error instead of hiding it so
+// that decision stays with the caller — the same split ratelimit.Throttler uses, where
+// the implementation reports and one caller decides to fail open. An implementation
+// that swallowed errors would make "the backend is down" indistinguishable from "this
+// key was never written", and those want different logging even though they want the
+// same fallback.
+//
+// Implementations are safe for concurrent use, and never share storage with the caller:
+// mutating a slice passed to Set, or one returned by Get, does not change what is
+// cached. RedisCache gets that for free by round-tripping through a socket, so Memory
+// copies to match — otherwise code exercised against Memory in tests would behave
+// differently against Redis in production.
+type Cache interface {
+	// Get returns the stored bytes and whether a live entry was found. A missing or
+	// expired key returns (nil, false, nil): a miss is an outcome, not a failure.
+	Get(ctx context.Context, key string) ([]byte, bool, error)
+
+	// Set stores val under key, to be forgotten after ttl elapses. A ttl of zero or
+	// less stores nothing — a caller asking for an already-expired entry gets its
+	// wish, not an error.
+	Set(ctx context.Context, key string, val []byte, ttl time.Duration) error
+}

internal/cache/json.go

@@ -0,0 +1,41 @@
+package cache
+
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+	"time"
+)
+
+// GetJSON reads key and decodes it into T. It is a free function rather than a method
+// because Go does not permit type parameters on methods.
+//
+// A miss, an unreadable backend, and a payload that no longer decodes into T all report
+// found == false, because all three mean the same thing to a caller: there is no usable
+// cached value, recompute. The error is still returned for the last two, so a stale
+// incompatible payload — typically a deploy that changed T while entries written by the
+// previous build are still live — is visible in logs rather than looking like ordinary
+// cache churn.
+func GetJSON[T any](ctx context.Context, c Cache, key string) (T, bool, error) {
+	var zero T
+
+	raw, found, err := c.Get(ctx, key)
+	if err != nil || !found {
+		return zero, false, err
+	}
+
+	var val T
+	if err := json.Unmarshal(raw, &val); err != nil {
+		return zero, false, fmt.Errorf("cache: decoding %q: %w", key, err)
+	}
+	return val, true, nil
+}
+
+// SetJSON encodes val as JSON and stores it under key for ttl.
+func SetJSON[T any](ctx context.Context, c Cache, key string, val T, ttl time.Duration) error {
+	raw, err := json.Marshal(val)
+	if err != nil {
+		return fmt.Errorf("cache: encoding %q: %w", key, err)
+	}
+	return c.Set(ctx, key, raw, ttl)
+}