diff --git a/AGENTS.md b/AGENTS.md index 0f3d8171f..09787af4c 100644 --- a/AGENTS.md +++ b/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 diff --git a/cmd/rollup-stats/main.go b/cmd/rollup-stats/main.go index 9e93ebce8..8ae7e8112 100644 --- a/cmd/rollup-stats/main.go +++ b/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. diff --git a/cmd/rollup-stats/publish.go b/cmd/rollup-stats/publish.go new file mode 100644 index 000000000..fedf84d74 --- /dev/null +++ b/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 +} diff --git a/cmd/rollup-stats/publish_test.go b/cmd/rollup-stats/publish_test.go new file mode 100644 index 000000000..04a26d266 --- /dev/null +++ b/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") +} diff --git a/cmd/server/main.go b/cmd/server/main.go index 77edff63c..08c4b86c0 100644 --- a/cmd/server/main.go +++ b/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, diff --git a/internal/cache/aliasing_test.go b/internal/cache/aliasing_test.go new file mode 100644 index 000000000..281b93115 --- /dev/null +++ b/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") + } + }) +} diff --git a/internal/cache/cache.go b/internal/cache/cache.go new file mode 100644 index 000000000..fa68f1aa7 --- /dev/null +++ b/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 +} diff --git a/internal/cache/json.go b/internal/cache/json.go new file mode 100644 index 000000000..fee9bc824 --- /dev/null +++ b/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) +} diff --git a/internal/cache/json_test.go b/internal/cache/json_test.go new file mode 100644 index 000000000..193bb42e3 --- /dev/null +++ b/internal/cache/json_test.go @@ -0,0 +1,86 @@ +package cache + +import ( + "context" + "testing" + "time" +) + +type payload struct { + Name string `json:"name"` + Count int64 `json:"count"` +} + +func TestJSONRoundTrip(t *testing.T) { + m, _ := newTestMemory(t) + ctx := context.Background() + want := payload{Name: "catalogue", Count: 3_300_658} + + if err := SetJSON(ctx, m, "k", want, time.Minute); err != nil { + t.Fatalf("SetJSON: %v", err) + } + + got, found, err := GetJSON[payload](ctx, m, "k") + if err != nil { + t.Fatalf("GetJSON: %v", err) + } + if !found { + t.Fatal("GetJSON: found = false, want true for a key just set") + } + if got != want { + t.Errorf("GetJSON = %+v, want %+v", got, want) + } +} + +func TestJSONMissReturnsZeroValue(t *testing.T) { + m, _ := newTestMemory(t) + + got, found, err := GetJSON[payload](context.Background(), m, "absent") + if err != nil { + t.Fatalf("GetJSON on an absent key returned an error: %v", err) + } + if found { + t.Error("GetJSON: found = true, want false") + } + if got != (payload{}) { + t.Errorf("GetJSON = %+v, want the zero value on a miss", got) + } +} + +// A payload written by an older build may no longer decode into the current type. That +// must read as a miss so the caller recomputes, not as a hit carrying a half-filled +// value and not as a failure that wedges the caller until the key expires. The error +// still comes back, because "the cache holds something undecodable" is worth logging in +// a way an ordinary miss is not. +func TestJSONUndecodablePayloadIsAMiss(t *testing.T) { + m, _ := newTestMemory(t) + ctx := context.Background() + + if err := m.Set(ctx, "k", []byte(`{"count": "not a number"}`), time.Minute); err != nil { + t.Fatalf("Set: %v", err) + } + + got, found, err := GetJSON[payload](ctx, m, "k") + if found { + t.Error("GetJSON: found = true for an undecodable payload — the caller would use a half-filled value") + } + if got != (payload{}) { + t.Errorf("GetJSON = %+v, want the zero value when decoding fails", got) + } + if err == nil { + t.Error("GetJSON: err = nil for an undecodable payload — indistinguishable from an ordinary miss in logs") + } +} + +func TestJSONBackendErrorPropagates(t *testing.T) { + c, mr := newTestRedisCache(t) + mr.Close() + + _, found, err := GetJSON[payload](context.Background(), c, "k") + if found { + t.Error("GetJSON: found = true against a closed backend") + } + if err == nil { + t.Error("GetJSON: err = nil against a closed backend") + } +} diff --git a/internal/cache/memory.go b/internal/cache/memory.go new file mode 100644 index 000000000..d54fa739f --- /dev/null +++ b/internal/cache/memory.go @@ -0,0 +1,77 @@ +package cache + +import ( + "bytes" + "context" + "sync" + "time" +) + +// Memory is an in-process Cache. It exists for two callers: tests, which want a Cache +// without a Redis backend, and a deployment running without Redis at all. +// +// It is deliberately not a general-purpose cache — there is no eviction, so it suits a +// bounded set of long-lived keys (the catalogue-scale snapshot is one key) and not an +// unbounded keyspace. Expired entries are dropped when their key is read. +// +// A Memory is per-process, so two processes holding one do not agree with each other. +// Where that matters — a figure published to users, which must not differ between two +// web processes or reset on deploy — RedisCache is the implementation to use. +type Memory struct { + mu sync.RWMutex + entries map[string]memoryEntry + + // now is injectable so expiry is testable without sleeping. + now func() time.Time +} + +type memoryEntry struct { + val []byte + expiresAt time.Time +} + +// NewMemory returns an empty in-process Cache. +func NewMemory() *Memory { + return &Memory{entries: make(map[string]memoryEntry), now: time.Now} +} + +// Get implements Cache. +func (m *Memory) Get(_ context.Context, key string) ([]byte, bool, error) { + m.mu.RLock() + entry, ok := m.entries[key] + m.mu.RUnlock() + if !ok { + return nil, false, nil + } + + if !m.now().Before(entry.expiresAt) { + // Drop it rather than leaving a dead entry pinning its value. Re-checking + // under the write lock keeps a concurrent Set from being discarded. + m.mu.Lock() + if cur, still := m.entries[key]; still && !m.now().Before(cur.expiresAt) { + delete(m.entries, key) + } + m.mu.Unlock() + return nil, false, nil + } + + // Copy out for the same reason Set copies in: RedisCache round-trips through a + // socket, so its callers cannot reach its stored bytes. Memory must not be the + // implementation where they can, or code exercised against Memory in tests would + // behave differently against Redis in production. + return bytes.Clone(entry.val), true, nil +} + +// Set implements Cache. +func (m *Memory) Set(_ context.Context, key string, val []byte, ttl time.Duration) error { + if ttl <= 0 { + return nil + } + // Copy: the caller keeps its slice and may reuse the buffer. + stored := bytes.Clone(val) + + m.mu.Lock() + defer m.mu.Unlock() + m.entries[key] = memoryEntry{val: stored, expiresAt: m.now().Add(ttl)} + return nil +} diff --git a/internal/cache/memory_test.go b/internal/cache/memory_test.go new file mode 100644 index 000000000..6d1d9633c --- /dev/null +++ b/internal/cache/memory_test.go @@ -0,0 +1,122 @@ +package cache + +import ( + "context" + "testing" + "time" +) + +// fixedClock drives Memory's expiry deterministically. Sleeping past a TTL would make +// these tests slow and flaky for no gain — expiry is arithmetic on a timestamp, and the +// timestamp is worth injecting. +type fixedClock struct{ now time.Time } + +func (c *fixedClock) Now() time.Time { return c.now } +func (c *fixedClock) advance(d time.Duration) { c.now = c.now.Add(d) } + +func newTestMemory(t *testing.T) (*Memory, *fixedClock) { + t.Helper() + clock := &fixedClock{now: time.Unix(1_700_000_000, 0)} + m := NewMemory() + m.now = clock.Now + return m, clock +} + +func TestMemoryRoundTrip(t *testing.T) { + m, _ := newTestMemory(t) + ctx := context.Background() + + if err := m.Set(ctx, "k", []byte("v"), time.Minute); err != nil { + t.Fatalf("Set: %v", err) + } + + got, found, err := m.Get(ctx, "k") + if err != nil { + t.Fatalf("Get: %v", err) + } + if !found { + t.Fatal("Get: found = false, want true for a key just set") + } + if string(got) != "v" { + t.Errorf("Get = %q, want %q", got, "v") + } +} + +func TestMemoryMissingKeyIsNotAnError(t *testing.T) { + m, _ := newTestMemory(t) + + got, found, err := m.Get(context.Background(), "absent") + if err != nil { + t.Fatalf("Get on an absent key returned an error: %v — a miss is a normal outcome, not a failure", err) + } + if found { + t.Error("Get: found = true, want false for a key never set") + } + if got != nil { + t.Errorf("Get = %q, want nil on a miss", got) + } +} + +func TestMemoryEntryExpires(t *testing.T) { + m, clock := newTestMemory(t) + ctx := context.Background() + + if err := m.Set(ctx, "k", []byte("v"), time.Minute); err != nil { + t.Fatalf("Set: %v", err) + } + + clock.advance(59 * time.Second) + if _, found, _ := m.Get(ctx, "k"); !found { + t.Error("entry expired early: found = false one second before its TTL") + } + + clock.advance(2 * time.Second) + got, found, err := m.Get(ctx, "k") + if err != nil { + t.Fatalf("Get past TTL returned an error: %v — an expired entry is a miss", err) + } + if found { + t.Errorf("entry survived its TTL: found = true, value %q", got) + } +} + +func TestMemoryOverwrite(t *testing.T) { + m, _ := newTestMemory(t) + ctx := context.Background() + + if err := m.Set(ctx, "k", []byte("first"), time.Minute); err != nil { + t.Fatalf("Set first: %v", err) + } + if err := m.Set(ctx, "k", []byte("second"), time.Minute); err != nil { + t.Fatalf("Set second: %v", err) + } + + got, _, err := m.Get(ctx, "k") + if err != nil { + t.Fatalf("Get: %v", err) + } + if string(got) != "second" { + t.Errorf("Get = %q, want %q — a later Set must replace the entry", got, "second") + } +} + +// Memory backs a snapshot read on the hottest public path, so concurrent use is the +// normal case, not an edge one. Run with -race. +func TestMemoryConcurrentUse(t *testing.T) { + m, _ := newTestMemory(t) + ctx := context.Background() + + done := make(chan struct{}) + for i := range 8 { + go func() { + defer func() { done <- struct{}{} }() + for range 50 { + _ = m.Set(ctx, "k", []byte{byte(i)}, time.Minute) + _, _, _ = m.Get(ctx, "k") + } + }() + } + for range 8 { + <-done + } +} diff --git a/internal/cache/redis.go b/internal/cache/redis.go new file mode 100644 index 000000000..199e84bdc --- /dev/null +++ b/internal/cache/redis.go @@ -0,0 +1,46 @@ +package cache + +import ( + "context" + "errors" + "time" + + "github.com/redis/go-redis/v9" +) + +// RedisCache is the shared Cache: every process reading it sees the same value, which +// is the point for anything published to users. It carries no transport of its own — +// the caller supplies the client the process already builds. +type RedisCache struct { + client *redis.Client +} + +// NewRedisCache constructs a RedisCache over an existing client. +func NewRedisCache(client *redis.Client) *RedisCache { + return &RedisCache{client: client} +} + +// Get implements Cache. A key that is absent or has expired reports a miss; anything +// else — including an unreachable backend — is returned as an error, so the caller can +// log the difference even though it falls back the same way. +func (c *RedisCache) Get(ctx context.Context, key string) ([]byte, bool, error) { + val, err := c.client.Get(ctx, key).Bytes() + if errors.Is(err, redis.Nil) { + return nil, false, nil + } + if err != nil { + return nil, false, err + } + return val, true, nil +} + +// Set implements Cache. +func (c *RedisCache) Set(ctx context.Context, key string, val []byte, ttl time.Duration) error { + if ttl <= 0 { + // Redis rejects a zero expiry and treats a negative one as store-then-delete. + // Neither matches "this entry is already expired", and the round-trip buys + // nothing, so skip it. + return nil + } + return c.client.Set(ctx, key, val, ttl).Err() +} diff --git a/internal/cache/redis_test.go b/internal/cache/redis_test.go new file mode 100644 index 000000000..d017109bc --- /dev/null +++ b/internal/cache/redis_test.go @@ -0,0 +1,110 @@ +package cache + +import ( + "context" + "testing" + "time" + + "github.com/alicebob/miniredis/v2" + "github.com/redis/go-redis/v9" +) + +func newTestRedisCache(t *testing.T) (*RedisCache, *miniredis.Miniredis) { + t.Helper() + mr, err := miniredis.Run() + if err != nil { + t.Fatalf("miniredis.Run: %v", err) + } + t.Cleanup(mr.Close) + + client := redis.NewClient(&redis.Options{Addr: mr.Addr()}) + t.Cleanup(func() { _ = client.Close() }) + + return NewRedisCache(client), mr +} + +func TestRedisCacheRoundTrip(t *testing.T) { + c, _ := newTestRedisCache(t) + ctx := context.Background() + + if err := c.Set(ctx, "k", []byte("v"), time.Minute); err != nil { + t.Fatalf("Set: %v", err) + } + + got, found, err := c.Get(ctx, "k") + if err != nil { + t.Fatalf("Get: %v", err) + } + if !found { + t.Fatal("Get: found = false, want true for a key just set") + } + if string(got) != "v" { + t.Errorf("Get = %q, want %q", got, "v") + } +} + +func TestRedisCacheMissingKeyIsNotAnError(t *testing.T) { + c, _ := newTestRedisCache(t) + + got, found, err := c.Get(context.Background(), "absent") + if err != nil { + t.Fatalf("Get on an absent key returned an error: %v — redis.Nil is a miss, not a failure", err) + } + if found { + t.Error("Get: found = true, want false for a key never set") + } + if got != nil { + t.Errorf("Get = %q, want nil on a miss", got) + } +} + +func TestRedisCacheEntryExpires(t *testing.T) { + c, mr := newTestRedisCache(t) + ctx := context.Background() + + if err := c.Set(ctx, "k", []byte("v"), time.Minute); err != nil { + t.Fatalf("Set: %v", err) + } + + mr.FastForward(61 * time.Second) + + _, found, err := c.Get(ctx, "k") + if err != nil { + t.Fatalf("Get past TTL returned an error: %v — an expired key is a miss", err) + } + if found { + t.Error("entry survived its TTL: found = true") + } +} + +// A caller can only fall back deliberately if it can tell "the backend is unreachable" +// from "this key was never written". Both lead to the same fallback, but only one is +// worth logging. +func TestRedisCacheReportsBackendFailure(t *testing.T) { + c, mr := newTestRedisCache(t) + ctx := context.Background() + mr.Close() + + if _, _, err := c.Get(ctx, "k"); err == nil { + t.Error("Get against a closed backend returned no error — a caller cannot distinguish it from a miss") + } + if err := c.Set(ctx, "k", []byte("v"), time.Minute); err == nil { + t.Error("Set against a closed backend returned no error") + } +} + +func TestRedisCacheNonPositiveTTLStoresNothing(t *testing.T) { + c, _ := newTestRedisCache(t) + ctx := context.Background() + + if err := c.Set(ctx, "k", []byte("v"), 0); err != nil { + t.Fatalf("Set with a zero TTL: %v", err) + } + + // Redis treats SET with a zero expiry as an error and a negative one as "store + // then immediately delete"; neither is what a caller asking for an expired entry + // means. It must read back as a miss, not as a value that never expires. + if _, found, _ := c.Get(ctx, "k"); found { + t.Error("a non-positive TTL stored a live entry") + } +} diff --git a/internal/catalogstats/catalogstats.go b/internal/catalogstats/catalogstats.go new file mode 100644 index 000000000..81a29349a --- /dev/null +++ b/internal/catalogstats/catalogstats.go @@ -0,0 +1,102 @@ +// Package catalogstats owns the public catalogue-scale figures: how many open postings +// the catalogue holds, from how many companies, across how many platforms and channels. +// +// The figures are computed once by a scheduled worker and published as one Snapshot, so +// every surface that quotes catalogue scale quotes the same numbers. That is the point +// of the package: before it, /about and /open each took their own estimate at their own +// moment, and could disagree. +// +// The exact counts are catalogue-wide, so computing one is a sequential scan. It runs in +// the worker and never on a request path — see Load, which reads the published snapshot +// and degrades to an approximation rather than ever recomputing. +package catalogstats + +import ( + "context" + "fmt" + "slices" + "time" + + "github.com/strelov1/freehire/internal/db" + "github.com/strelov1/freehire/internal/sources" +) + +// Snapshot is the catalogue's scale at one instant. +type Snapshot struct { + // OpenJobs and Companies are exact counts over the set the public listings + // paginate: open, not duplicate-suppressed, not private. + OpenJobs int64 `json:"open_jobs"` + Companies int64 `json:"companies"` + + // Sources, ATSPlatforms and TelegramChannels describe reach rather than contents: + // what the crawler can read, whether or not each currently holds an open posting. + // + // Sources is every registered adapter; ATSPlatforms is the ATS subset. Both are + // published because they answer different questions — "how much of the market can + // you see" and "how many hiring systems do you speak" — and because quoting the + // wider number under the narrower label is the mistake this replaces. + Sources int `json:"sources"` + ATSPlatforms int `json:"ats_platforms"` + TelegramChannels int `json:"telegram_channels"` + + ComputedAt time.Time `json:"computed_at"` +} + +// ExactCounter reads the exact catalogue totals. It is the one thing Compute needs from +// the database, named narrowly so the dependency says so: *db.Queries satisfies it, and +// nothing else about the query layer is reachable from here. +type ExactCounter interface { + CountCatalogueScale(ctx context.Context) (db.CountCatalogueScaleRow, error) +} + +// Compute measures the catalogue now. +// +// The exact counts are a full scan, so this belongs in the scheduled worker and nowhere +// near a request. Load is the read path. +// +// telegramChannels is passed in rather than read here: resolving it means reading +// sources/telegram.yml, which is relative to the worker's working directory. Keeping +// that at the edge leaves Compute with no file system of its own. +func Compute(ctx context.Context, counts ExactCounter, telegramChannels int) (Snapshot, error) { + exact, err := counts.CountCatalogueScale(ctx) + if err != nil { + return Snapshot{}, fmt.Errorf("catalogstats: counting catalogue scale: %w", err) + } + + return Snapshot{ + OpenJobs: exact.OpenJobs, + Companies: exact.Companies, + Sources: Sources(), + ATSPlatforms: ATSPlatforms(), + TelegramChannels: telegramChannels, + ComputedAt: time.Now().UTC(), + }, nil +} + +// Sources counts every registered source adapter, of every kind — ATS platforms, +// aggregators, and single-company career feeds alike. It is the breadth figure: how many +// distinct places the crawler knows how to read. +func Sources() int { return len(sources.Taxonomy()) } + +// ATSPlatforms counts the registered multi-tenant applicant-tracking systems: adapters +// addressed by a board id that are not aggregators. +// +// Both exclusions are the label doing its job. An aggregator republishes many +// companies' postings and is not an ATS; a boardless adapter serves one company's own +// careers feed and is not a platform. The registry holds all three kinds, and counting +// all of them under "ATS platforms" is what the frontend constant this replaces did. +// +// It reads the taxonomy registry, which carries no transport, so this is pure in-process +// work — no network, no credentials, and the same answer on every host. +func ATSPlatforms() int { + registry := sources.Taxonomy() + aggregators := sources.AggregatorProviders(registry) + + n := 0 + for _, provider := range sources.BoardKeyedProviders(registry) { + if !slices.Contains(aggregators, provider) { + n++ + } + } + return n +} diff --git a/internal/catalogstats/compute_integration_test.go b/internal/catalogstats/compute_integration_test.go new file mode 100644 index 000000000..870677af2 --- /dev/null +++ b/internal/catalogstats/compute_integration_test.go @@ -0,0 +1,121 @@ +//go:build integration + +// Integration test for the exact half of a Snapshot. The whole reason this package +// exists is that the published open-job total described a wider set than the listing +// showed, so the one thing worth proving against a real Postgres is that the counts +// cover exactly the set GET /api/v1/jobs paginates. +// Run with: go test -tags=integration ./internal/catalogstats/ +package catalogstats + +import ( + "context" + "testing" + + "github.com/jackc/pgx/v5/pgxpool" + + "github.com/strelov1/freehire/internal/db" + "github.com/strelov1/freehire/internal/testdb" +) + +func seedJob(t *testing.T, q *db.Queries, source, externalID, companySlug string) db.Job { + t.Helper() + row, err := q.UpsertJob(context.Background(), db.UpsertJobParams{ + Source: source, + ExternalID: externalID, + URL: "https://example.test/" + externalID, + Title: "Go Engineer", + Company: companySlug, + CompanySlug: companySlug, + PublicSlug: "pslug-" + externalID, + Location: "Remote", + Remote: true, + Description: "Build things.", + }) + if err != nil { + t.Fatalf("seed %s: %v", externalID, err) + } + return row.Job +} + +func truncateJobs(t *testing.T, pool *pgxpool.Pool) { + t.Helper() + if _, err := pool.Exec(context.Background(), + "TRUNCATE enrichment_outbox, jobs, companies RESTART IDENTITY CASCADE"); err != nil { + t.Fatalf("truncate: %v", err) + } +} + +func TestComputeCountsOnlyThePaginatedSet(t *testing.T) { + pool := testdb.Pool(t) + q := db.New(pool) + ctx := context.Background() + truncateJobs(t, pool) + + // Three companies with a listed posting each, then one excluded posting of every + // kind. The excluded ones sit at companies that have no listed posting, so a + // company count that leaked would be wrong by more than the job count. + canonical := seedJob(t, q, "greenhouse", "acme:1", "acme") + seedJob(t, q, "greenhouse", "beta:1", "beta") + seedJob(t, q, "lever", "gamma:1", "gamma") + + seedJob(t, q, "greenhouse", "closedco:1", "closedco") + if _, err := pool.Exec(ctx, + `UPDATE jobs SET closed_at = now() WHERE external_id = 'closedco:1'`); err != nil { + t.Fatalf("close job: %v", err) + } + + seedJob(t, q, "greenhouse", "dupco:1", "dupco") + if _, err := pool.Exec(ctx, + `UPDATE jobs SET duplicate_of = $1 WHERE external_id = 'dupco:1'`, canonical.ID); err != nil { + t.Fatalf("suppress duplicate: %v", err) + } + + seedJob(t, q, "greenhouse", "privco:1", "privco") + if _, err := pool.Exec(ctx, + `UPDATE jobs SET is_private = true WHERE external_id = 'privco:1'`); err != nil { + t.Fatalf("mark private: %v", err) + } + + got, err := Compute(ctx, q, 7) + if err != nil { + t.Fatalf("Compute: %v", err) + } + + if got.OpenJobs != 3 { + t.Errorf("OpenJobs = %d, want 3 — the closed, duplicate-suppressed and private "+ + "postings must not be counted", got.OpenJobs) + } + if got.Companies != 3 { + t.Errorf("Companies = %d, want 3 — only companies with a listed posting count", got.Companies) + } + if got.TelegramChannels != 7 { + t.Errorf("TelegramChannels = %d, want the 7 passed in", got.TelegramChannels) + } + if got.Sources != Sources() || got.ATSPlatforms != ATSPlatforms() { + t.Errorf("Sources/ATSPlatforms = %d/%d, want the registry-derived %d/%d", + got.Sources, got.ATSPlatforms, Sources(), ATSPlatforms()) + } + if got.ComputedAt.IsZero() { + t.Error("ComputedAt is zero — consumers cannot tell how stale a snapshot is") + } +} + +// Both figures must describe the same instant. Reading them in separate statements +// would let an ingest land between them and publish a company count for a catalogue +// that no longer matches the job count beside it. +func TestComputeReadsBothCountsInOneStatement(t *testing.T) { + pool := testdb.Pool(t) + q := db.New(pool) + ctx := context.Background() + truncateJobs(t, pool) + + seedJob(t, q, "greenhouse", "acme:1", "acme") + + got, err := Compute(ctx, q, 0) + if err != nil { + t.Fatalf("Compute: %v", err) + } + if got.OpenJobs != 1 || got.Companies != 1 { + t.Fatalf("OpenJobs/Companies = %d/%d, want 1/1", got.OpenJobs, got.Companies) + } +} diff --git a/internal/catalogstats/derived_test.go b/internal/catalogstats/derived_test.go new file mode 100644 index 000000000..adb862778 --- /dev/null +++ b/internal/catalogstats/derived_test.go @@ -0,0 +1,81 @@ +package catalogstats + +import ( + "slices" + "testing" + + "github.com/strelov1/freehire/internal/sources" +) + +// The figure is labelled "ATS platforms" on /open, so it must count ATS platforms: the +// multi-tenant systems addressed by a board id, each serving many companies. The +// registry also holds aggregators (third-party feeds republishing many companies) and +// single-company career feeds, and counting those under this label is what the +// hardcoded frontend constant used to do. +func TestATSPlatformsCountsOnlyBoardKeyedNonAggregators(t *testing.T) { + got := ATSPlatforms() + + registry := sources.Taxonomy() + if got <= 0 { + t.Fatalf("ATSPlatforms = %d, want a positive count", got) + } + if got >= len(registry) { + t.Errorf("ATSPlatforms = %d against a %d-adapter registry — aggregators and "+ + "single-company feeds are being counted as ATS platforms", got, len(registry)) + } + + // Named adapters, so a predicate that drifts fails with a reason rather than a + // number. greenhouse is a board-keyed ATS; adzuna is board-keyed but aggregates + // other companies' postings; amazon is one company's own careers feed. + boardKeyed := sources.BoardKeyedProviders(registry) + aggregators := sources.AggregatorProviders(registry) + + for _, tc := range []struct { + provider string + wantATS bool + because string + }{ + {"greenhouse", true, "a multi-tenant ATS addressed by a board id"}, + {"adzuna", false, "an aggregator republishing many companies' postings"}, + {"amazon", false, "a single company's own careers feed"}, + } { + isATS := slices.Contains(boardKeyed, tc.provider) && !slices.Contains(aggregators, tc.provider) + if isATS != tc.wantATS { + t.Errorf("%s counted as an ATS platform = %v, want %v — it is %s", + tc.provider, isATS, tc.wantATS, tc.because) + } + } +} + +// The /open strip leads with total reach, so the snapshot carries it alongside the +// narrower ATS figure. Both are derived from the same registry, and the wider one must +// actually be wider — if they ever coincide, one of the two predicates is wrong. +func TestSourcesCountsEveryRegisteredAdapter(t *testing.T) { + got := Sources() + + if want := len(sources.Taxonomy()); got != want { + t.Errorf("Sources = %d, want %d — every registered adapter counts, whatever kind it is", got, want) + } + if ats := ATSPlatforms(); got <= ats { + t.Errorf("Sources = %d is not wider than ATSPlatforms = %d — aggregators and "+ + "single-company feeds are sources too, so the totals cannot coincide", got, ats) + } +} + +// Nothing about the count may live in a literal: adding an adapter must move it. +func TestATSPlatformsIsDerivedFromTheRegistry(t *testing.T) { + registry := sources.Taxonomy() + boardKeyed := sources.BoardKeyedProviders(registry) + aggregators := sources.AggregatorProviders(registry) + + want := 0 + for _, p := range boardKeyed { + if !slices.Contains(aggregators, p) { + want++ + } + } + + if got := ATSPlatforms(); got != want { + t.Errorf("ATSPlatforms = %d, want %d — the count must fall out of the registry, not a constant", got, want) + } +} diff --git a/internal/catalogstats/load.go b/internal/catalogstats/load.go new file mode 100644 index 000000000..64cbad4b3 --- /dev/null +++ b/internal/catalogstats/load.go @@ -0,0 +1,95 @@ +package catalogstats + +import ( + "context" + "log" + "time" + + "github.com/strelov1/freehire/internal/cache" +) + +// snapshotKey is where the published snapshot lives. One key, one value, read by every +// process — that shared read is what makes two surfaces agree. +const snapshotKey = "catalogstats:snapshot" + +// snapshotTTL deliberately outlives the worker's schedule by a wide margin. Setting it +// near the cron interval would mean one skipped or slow run drops every surface back to +// the estimate; a long window degrades a missed run to "stale but exact", which is +// strictly better than "fresh but wrong". ComputedAt travels in the snapshot, so the +// staleness is observable rather than hidden. +const snapshotTTL = 24 * time.Hour + +// readTimeout bounds a single cache read. Long enough for a healthy same-host round +// trip, short enough that an unreachable backend degrades the figure instead of holding +// up the response — mirroring the bound ratelimit puts on its own Redis call. +const readTimeout = 100 * time.Millisecond + +// Estimator supplies the approximate open-job count used when no snapshot is published. +// *db.Queries satisfies it via EstimateOpenJobs. +type Estimator interface { + EstimateOpenJobs(ctx context.Context) (int64, error) +} + +// Result is a snapshot plus whether it is the real thing. +// +// Exact distinguishes a published measurement from the degraded fallback. Consumers may +// render the two differently, and none of them should have to guess which they hold. +type Result struct { + Snapshot + Exact bool +} + +// Store publishes a snapshot for every reader. +func Store(ctx context.Context, c cache.Cache, s Snapshot) error { + return cache.SetJSON(ctx, c, snapshotKey, s, snapshotTTL) +} + +// Load reads the published snapshot. +// +// It never recomputes — that is enforced by this signature, which has no ExactCounter to +// reach for, not by a runtime check. A cold cache, an unreachable backend, and a payload +// left by an older build all degrade the same way: the approximate open-job count, the +// registry-derived figures (which cost nothing and need no cache), and Exact false. +// +// It returns no error. Every caller is rendering a page or a list, and for all of them a +// missing figure beats a failed response, so there is no decision left to delegate. +// Failures are logged here instead. +func Load(ctx context.Context, c cache.Cache, est Estimator) Result { + // No cache configured is the same situation as a cache with nothing in it, and it + // is a real deployment: the API can run without Redis, it just never sees a + // snapshot. Silently, because unlike a failure there is nothing to report. + if c == nil { + return Result{Snapshot: degraded(ctx, est), Exact: false} + } + + readCtx, cancel := context.WithTimeout(ctx, readTimeout) + defer cancel() + + snapshot, found, err := cache.GetJSON[Snapshot](readCtx, c, snapshotKey) + if err != nil { + // Worth a line: an unreachable cache or an undecodable payload is an + // operational fact, unlike an ordinary miss before the first worker run. + log.Printf("catalogstats: reading the published snapshot: %v (falling back to the estimate)", err) + } + if found { + return Result{Snapshot: snapshot, Exact: true} + } + + return Result{Snapshot: degraded(ctx, est), Exact: false} +} + +// degraded assembles the best snapshot obtainable without the published one: an +// approximate open-job count and the registry figures, which are pure in-process work. +// The counts that exist only in the database — companies, configured channels — stay +// zero, and Exact false is how a consumer knows not to show them. +func degraded(ctx context.Context, est Estimator) Snapshot { + s := Snapshot{Sources: Sources(), ATSPlatforms: ATSPlatforms()} + + openJobs, err := est.EstimateOpenJobs(ctx) + if err != nil { + log.Printf("catalogstats: estimating open jobs: %v (serving no figure)", err) + return s + } + s.OpenJobs = openJobs + return s +} diff --git a/internal/catalogstats/load_test.go b/internal/catalogstats/load_test.go new file mode 100644 index 000000000..9d3e8c272 --- /dev/null +++ b/internal/catalogstats/load_test.go @@ -0,0 +1,191 @@ +package catalogstats + +import ( + "context" + "errors" + "testing" + "time" + + "github.com/strelov1/freehire/internal/cache" +) + +// countingEstimator records how often the approximate path was taken. +type countingEstimator struct { + value int64 + err error + calls int +} + +func (e *countingEstimator) EstimateOpenJobs(context.Context) (int64, error) { + e.calls++ + return e.value, e.err +} + +// recordingCache captures what Store asked for, so the retention decision can be +// asserted rather than left in a constant nobody reads. +type recordingCache struct { + cache.Cache + key string + ttl time.Duration +} + +func (r *recordingCache) Set(ctx context.Context, key string, val []byte, ttl time.Duration) error { + r.key, r.ttl = key, ttl + return r.Cache.Set(ctx, key, val, ttl) +} + +// The snapshot must outlive the worker's schedule by a wide margin. A TTL near the cron +// interval means one skipped or slow run drops every surface back to the estimate; the +// whole point of publishing an exact figure is that a missed run degrades to +// stale-but-exact instead of fresh-but-wrong. +func TestStoreRetainsTheSnapshotBeyondTheWorkerSchedule(t *testing.T) { + rec := &recordingCache{Cache: cache.NewMemory()} + + if err := Store(context.Background(), rec, storedSnapshot()); err != nil { + t.Fatalf("Store: %v", err) + } + + // rollup-stats runs intra-day, every few hours. Anything under half a day would + // make a single missed run visible to users. + const workerSchedule = 12 * time.Hour + if rec.ttl <= workerSchedule { + t.Errorf("Store TTL = %s, want comfortably more than the %s worker schedule — "+ + "a skipped run would otherwise drop every surface back to the estimate", + rec.ttl, workerSchedule) + } + if rec.key != snapshotKey { + t.Errorf("Store wrote %q, want %q — readers look under one shared key", rec.key, snapshotKey) + } +} + +// brokenCache reports a backend failure on every operation. +type brokenCache struct{} + +func (brokenCache) Get(context.Context, string) ([]byte, bool, error) { + return nil, false, errors.New("backend unreachable") +} +func (brokenCache) Set(context.Context, string, []byte, time.Duration) error { + return errors.New("backend unreachable") +} + +func storedSnapshot() Snapshot { + return Snapshot{ + OpenJobs: 3_300_658, + Companies: 294_282, + Sources: Sources(), + ATSPlatforms: ATSPlatforms(), + TelegramChannels: 95, + ComputedAt: time.Unix(1_700_000_000, 0).UTC(), + } +} + +func TestLoadReturnsTheStoredSnapshot(t *testing.T) { + c := cache.NewMemory() + ctx := context.Background() + want := storedSnapshot() + + if err := Store(ctx, c, want); err != nil { + t.Fatalf("Store: %v", err) + } + + est := &countingEstimator{value: 999} + got := Load(ctx, c, est) + + if !got.Exact { + t.Error("Exact = false for a published snapshot") + } + if got.Snapshot != want { + t.Errorf("Snapshot = %+v, want %+v", got.Snapshot, want) + } + if est.calls != 0 { + t.Errorf("the estimator was called %d times despite a cache hit", est.calls) + } +} + +func TestLoadDegradesOnAnEmptyCache(t *testing.T) { + ctx := context.Background() + est := &countingEstimator{value: 3_150_000} + + got := Load(ctx, cache.NewMemory(), est) + + if got.Exact { + t.Error("Exact = true with nothing published — consumers cannot tell an estimate from a count") + } + if got.OpenJobs != 3_150_000 { + t.Errorf("OpenJobs = %d, want the estimate 3150000", got.OpenJobs) + } + if est.calls != 1 { + t.Errorf("estimator calls = %d, want 1", est.calls) + } + // Registry-derived figures cost nothing and need no cache, so a degraded read + // should still carry them rather than reporting a catalogue with no sources. + if got.Sources != Sources() || got.ATSPlatforms != ATSPlatforms() { + t.Errorf("Sources/ATSPlatforms = %d/%d on the degraded path, want %d/%d", + got.Sources, got.ATSPlatforms, Sources(), ATSPlatforms()) + } +} + +// Running the API without Redis is a supported deployment, not a misconfiguration. +func TestLoadDegradesWithNoCacheConfigured(t *testing.T) { + est := &countingEstimator{value: 3_150_000} + + got := Load(context.Background(), nil, est) + + if got.Exact { + t.Error("Exact = true with no cache configured") + } + if got.OpenJobs != 3_150_000 { + t.Errorf("OpenJobs = %d, want the estimate", got.OpenJobs) + } +} + +func TestLoadDegradesOnAnUnreachableCache(t *testing.T) { + est := &countingEstimator{value: 3_150_000} + + got := Load(context.Background(), brokenCache{}, est) + + if got.Exact { + t.Error("Exact = true against an unreachable cache") + } + if got.OpenJobs != 3_150_000 { + t.Errorf("OpenJobs = %d, want the estimate", got.OpenJobs) + } +} + +// Both fallbacks failing is a degraded read, not a failed one: the caller is serving a +// page, and a missing figure beats a 500. +func TestLoadSurvivesTheEstimatorFailing(t *testing.T) { + est := &countingEstimator{err: errors.New("database down")} + + got := Load(context.Background(), brokenCache{}, est) + + if got.Exact { + t.Error("Exact = true with neither a snapshot nor an estimate") + } + if got.OpenJobs != 0 { + t.Errorf("OpenJobs = %d, want 0 when no figure could be obtained", got.OpenJobs) + } +} + +// Load never recomputes, and that is enforced by its signature rather than by a test: +// it takes no ExactCounter, so no read path can reach the catalogue-wide scan even by +// mistake. A runtime assertion would be the weaker guarantee. + +func TestLoadTreatsAnUndecodableSnapshotAsAMiss(t *testing.T) { + c := cache.NewMemory() + ctx := context.Background() + + if err := c.Set(ctx, snapshotKey, []byte(`{"open_jobs": "not a number"}`), time.Hour); err != nil { + t.Fatalf("Set: %v", err) + } + + est := &countingEstimator{value: 3_150_000} + got := Load(ctx, c, est) + + if got.Exact { + t.Error("Exact = true for an undecodable payload — a half-filled snapshot would be published") + } + if got.OpenJobs != 3_150_000 { + t.Errorf("OpenJobs = %d, want the estimate", got.OpenJobs) + } +} diff --git a/internal/db/estimate_open_jobs_integration_test.go b/internal/db/estimate_open_jobs_integration_test.go new file mode 100644 index 000000000..5421ddb4e --- /dev/null +++ b/internal/db/estimate_open_jobs_integration_test.go @@ -0,0 +1,132 @@ +//go:build integration + +// Integration test for estimate_open_jobs(), the planner-backed approximate total the +// DB-backed /jobs list reports as meta.total. The function must estimate the same set +// the list paginates — open, not duplicate-suppressed, not private — rather than the +// wider "closed_at IS NULL" set, which counts suppressed reposts the list never shows. +// This is a plpgsql/planner behavior, verifiable only against a real Postgres. +// Run with: go test -tags=integration ./internal/db/ +package db + +import ( + "context" + "strconv" + "testing" + + "github.com/jackc/pgx/v5/pgxpool" +) + +// externalID mints a seeded job's external_id so a whole cohort is addressable by a +// LIKE prefix — the seeding below flips columns per cohort, not per row. +func externalID(cohort string, i int) string { + return cohort + ":" + strconv.Itoa(i) +} + +// countJobs runs an exact count under the given predicate, naming the two candidate +// sets the estimate could be describing. +func countJobs(t *testing.T, pool *pgxpool.Pool, where string) int64 { + t.Helper() + var n int64 + if err := pool.QueryRow(context.Background(), `SELECT COUNT(*) FROM jobs WHERE `+where).Scan(&n); err != nil { + t.Fatalf("count where %s: %v", where, err) + } + return n +} + +func TestEstimateOpenJobsExcludesSuppressedAndPrivate(t *testing.T) { + pool := startPostgres(t) + q := New(pool) + ctx := context.Background() + truncate(t, pool) + + // Enough rows that the planner has something to estimate from, and enough + // excluded rows that counting them would blow past the tolerance: 20 listed + // against 30 that must not be counted. + const ( + listed = 20 + closed = 10 + duplicate = 10 + private = 10 + ) + + var canonicalID int64 + for i := range listed { + job, err := ingestUpsert(ctx, q, ingestParams(externalID("open", i), "Go Engineer")) + if err != nil { + t.Fatalf("seed open job: %v", err) + } + if i == 0 { + canonicalID = job.ID // the row the suppressed copies point at + } + } + + for i := range closed { + if _, err := ingestUpsert(ctx, q, ingestParams(externalID("closed", i), "Go Engineer")); err != nil { + t.Fatalf("seed closed job: %v", err) + } + } + if _, err := pool.Exec(ctx, + `UPDATE jobs SET closed_at = now() WHERE external_id LIKE 'closed:%'`); err != nil { + t.Fatalf("close jobs: %v", err) + } + + for i := range duplicate { + if _, err := ingestUpsert(ctx, q, ingestParams(externalID("dup", i), "Go Engineer")); err != nil { + t.Fatalf("seed duplicate job: %v", err) + } + } + if _, err := pool.Exec(ctx, + `UPDATE jobs SET duplicate_of = $1 WHERE external_id LIKE 'dup:%'`, canonicalID); err != nil { + t.Fatalf("suppress duplicates: %v", err) + } + + for i := range private { + if _, err := ingestUpsert(ctx, q, ingestParams(externalID("priv", i), "Go Engineer")); err != nil { + t.Fatalf("seed private job: %v", err) + } + } + if _, err := pool.Exec(ctx, + `UPDATE jobs SET is_private = true WHERE external_id LIKE 'priv:%'`); err != nil { + t.Fatalf("mark private: %v", err) + } + + // The estimate reads planner statistics, so it is only meaningful once they + // reflect the rows just written. + if _, err := pool.Exec(ctx, `ANALYZE jobs`); err != nil { + t.Fatalf("analyze: %v", err) + } + + // The two sets the function could be describing: the one the list paginates, and + // the wider one it used to estimate. + paginated := countJobs(t, pool, "closed_at IS NULL AND duplicate_of IS NULL AND NOT is_private") + notClosed := countJobs(t, pool, "closed_at IS NULL") + if paginated != listed || notClosed != listed+duplicate+private { + t.Fatalf("seeding is wrong: paginated = %d (want %d), not-closed = %d (want %d)", + paginated, listed, notClosed, listed+duplicate+private) + } + + got, err := q.EstimateOpenJobs(ctx) + if err != nil { + t.Fatalf("EstimateOpenJobs: %v", err) + } + + // Assert which set is being estimated, not how accurate the planner is. The + // function returns an estimate by design, and on a table this small the planner's + // selectivity arithmetic carries real error — pinning the value to the exact count + // would test Postgres's statistics rather than this function's predicate. + distToPaginated := abs64(got - paginated) + distToNotClosed := abs64(got - notClosed) + if distToPaginated >= distToNotClosed { + t.Errorf("EstimateOpenJobs = %d sits nearer the %d not-closed rows than the %d rows the "+ + "list paginates — it is still estimating the wider set (%d duplicate-suppressed and "+ + "%d private rows must not be counted)", + got, notClosed, paginated, duplicate, private) + } +} + +func abs64(n int64) int64 { + if n < 0 { + return -n + } + return n +} diff --git a/internal/db/jobs.sql.go b/internal/db/jobs.sql.go index b30e4506a..21aa7850e 100644 --- a/internal/db/jobs.sql.go +++ b/internal/db/jobs.sql.go @@ -364,6 +364,37 @@ func (q *Queries) CompanyHasOtherJobs(ctx context.Context, arg CompanyHasOtherJo return carried, err } +const countCatalogueScale = `-- name: CountCatalogueScale :one +SELECT + COUNT(*)::bigint AS open_jobs, + COUNT(DISTINCT company_slug)::bigint AS companies +FROM jobs +WHERE closed_at IS NULL AND duplicate_of IS NULL AND NOT is_private +` + +type CountCatalogueScaleRow struct { + OpenJobs int64 `json:"open_jobs"` + Companies int64 `json:"companies"` +} + +// Exact open-job and company totals for the published catalogue-scale snapshot +// (internal/catalogstats). Deliberately the opposite trade to EstimateOpenJobs below: +// this is a full scan and belongs only in the scheduled rollup worker, never on a +// request path. +// +// Both figures come from ONE statement so they describe the same instant. Counting them +// separately would let an ingest land between the two reads and publish a company count +// for a catalogue the job count beside it no longer describes. +// +// The predicate is the one the public listings apply, so the totals describe exactly the +// set a visitor can page through. +func (q *Queries) CountCatalogueScale(ctx context.Context) (CountCatalogueScaleRow, error) { + row := q.db.QueryRow(ctx, countCatalogueScale) + var i CountCatalogueScaleRow + err := row.Scan(&i.OpenJobs, &i.Companies) + return i, err +} + const enqueueJobEnrichment = `-- name: EnqueueJobEnrichment :execrows INSERT INTO enrichment_outbox (job_id, target_version) SELECT id, $1::int diff --git a/internal/db/querier.go b/internal/db/querier.go index 82da61e95..8846e2be1 100644 --- a/internal/db/querier.go +++ b/internal/db/querier.go @@ -461,6 +461,18 @@ type Querier interface { // How many people a campaign would reach right now. Read before sending: a campaign // is irreversible and goes to everyone, so the number is worth seeing first. CountBroadcastCandidates(ctx context.Context, campaign string) (int64, error) + // Exact open-job and company totals for the published catalogue-scale snapshot + // (internal/catalogstats). Deliberately the opposite trade to EstimateOpenJobs below: + // this is a full scan and belongs only in the scheduled rollup worker, never on a + // request path. + // + // Both figures come from ONE statement so they describe the same instant. Counting them + // separately would let an ingest land between the two reads and publish a company count + // for a catalogue the job count beside it no longer describes. + // + // The predicate is the one the public listings apply, so the totals describe exactly the + // set a visitor can page through. + CountCatalogueScale(ctx context.Context) (CountCatalogueScaleRow, error) // Total companies matching the same optional name + facet filters as ListCompanies, // so search/filter pagination reports the filtered total. Keep this WHERE identical // to ListCompanies (including the job_count > 0 hiring scope). diff --git a/internal/db/queries/jobs.sql b/internal/db/queries/jobs.sql index dfae81f6b..6fd6301bf 100644 --- a/internal/db/queries/jobs.sql +++ b/internal/db/queries/jobs.sql @@ -162,6 +162,24 @@ SELECT similar_job_ids FROM jobs WHERE id = sqlc.arg(id)::bigint; +-- name: CountCatalogueScale :one +-- Exact open-job and company totals for the published catalogue-scale snapshot +-- (internal/catalogstats). Deliberately the opposite trade to EstimateOpenJobs below: +-- this is a full scan and belongs only in the scheduled rollup worker, never on a +-- request path. +-- +-- Both figures come from ONE statement so they describe the same instant. Counting them +-- separately would let an ingest land between the two reads and publish a company count +-- for a catalogue the job count beside it no longer describes. +-- +-- The predicate is the one the public listings apply, so the totals describe exactly the +-- set a visitor can page through. +SELECT + COUNT(*)::bigint AS open_jobs, + COUNT(DISTINCT company_slug)::bigint AS companies +FROM jobs +WHERE closed_at IS NULL AND duplicate_of IS NULL AND NOT is_private; + -- name: EstimateOpenJobs :one -- Fast approximate open-job total for the DB-backed /jobs list's meta.total. An -- exact count(*) over ~millions of open rows was a per-request full scan; the diff --git a/internal/handler/AGENTS.md b/internal/handler/AGENTS.md index e6c54959e..ea1b8babf 100644 --- a/internal/handler/AGENTS.md +++ b/internal/handler/AGENTS.md @@ -49,6 +49,17 @@ Fiber HTTP handlers: feature handler structs, route registration, auth surface, (`authH.*` config fields, `assistantH.realtime`, the `resumeH.llm`/`matchH.llm` bindings). Two same-typed clients go in a named struct (`assistantModels`), not as adjacent parameters — a swap there compiles. +- **Catalogue scale is read, never counted** (`stats.go` `CatalogScale`, `jobs.go` + `openJobTotal`). Every figure describing how big the catalogue is — + `GET /stats/catalog` and the jobs list's `meta.total` — comes from the snapshot + `cmd/rollup-stats` publishes (`internal/catalogstats`). Both call + `catalogstats.Load`, which takes no exact counter, so no request path can reach a + catalogue-wide scan even by mistake. Neither read can fail: a cold cache, an + unreachable Redis, a payload from an older build and no `cfg.Cache` at all all + degrade to the approximate estimate with `exact: false`. Add a new consumer by + calling `Load`, not by counting — and pass `exact` through to whatever renders it, + because a degraded snapshot zeroes the figures that exist only in the database and a + zero must not reach a page as if it were a measurement. - Tests construct the feature struct directly with fakes/stubs (e.g. `&trackingHandlers{tracking: ...}`) and mount routes on a bare `fiber.App`. - Central `handler.RenderError` (wired in `cmd/server` via `fiber.Config{ErrorHandler: handler.RenderError}`) renders JSON envelope: a `codedError`→its status with a `code` field, `*fiber.Error`→its code, `pgx.ErrNoRows`→404, FK-violation (SQLSTATE 23503)→404, `inbox.ErrNotFound`→404, `inbox.InvalidError`/`inbox.ErrSlugRequired`/`search.ErrBadQuery`→400, `inbox.ErrPendingSuggestion`→409, `context.Canceled`→499 (client closed request), everything else→500. diff --git a/internal/handler/handler.go b/internal/handler/handler.go index 3054f5575..a081a91d2 100644 --- a/internal/handler/handler.go +++ b/internal/handler/handler.go @@ -23,6 +23,7 @@ import ( "github.com/strelov1/freehire/internal/blobstore" "github.com/strelov1/freehire/internal/boardresolve" "github.com/strelov1/freehire/internal/browsertools" + "github.com/strelov1/freehire/internal/cache" "github.com/strelov1/freehire/internal/companyfeedback" "github.com/strelov1/freehire/internal/contribution" "github.com/strelov1/freehire/internal/credits" @@ -191,7 +192,12 @@ type Config struct { // Throttler backs every rate-limited route in the API (internal/ratelimit). // Required — there is no degraded/nil mode, unlike the optional dependencies // below. - Throttler ratelimit.Throttler + Throttler ratelimit.Throttler + // Cache holds values that are expensive to compute and identical for every + // caller — today the catalogue-scale snapshot (internal/catalogstats). Optional: + // nil degrades the figures it backs to their approximations, exactly as an + // unreachable backend does. + Cache cache.Cache FrontendOrigin string JWTSecret string JWTTTL time.Duration @@ -326,8 +332,8 @@ func Register(app *fiber.App, cfg Config) { // for the shapes that hide the posting behind a second id (see sources.PostingURLResolver). // Shared by /jobs/find and the link intake, which must agree on what a page is. postingURLs := sources.NewPostingURLResolver(ingestClient) - jobsH := newJobsHandlers(queries, moderationSvc, postingURLs) - statsH := newStatsHandlers(queries) + jobsH := newJobsHandlers(queries, moderationSvc, postingURLs, cfg.Cache) + statsH := newStatsHandlers(queries, cfg.Cache) votesH := newVoteHandlers(queries, cfg.Pool) communityH := newCommunityHandlers(queries) // Feedback reuses communityH's persona minting (via the communityPersonas diff --git a/internal/handler/jobs.go b/internal/handler/jobs.go index 52c42bbc1..7ee785aa7 100644 --- a/internal/handler/jobs.go +++ b/internal/handler/jobs.go @@ -1,11 +1,14 @@ package handler import ( + "context" "time" "github.com/gofiber/fiber/v2" "github.com/strelov1/freehire/internal/auth" + "github.com/strelov1/freehire/internal/cache" + "github.com/strelov1/freehire/internal/catalogstats" "github.com/strelov1/freehire/internal/db" "github.com/strelov1/freehire/internal/ghost" "github.com/strelov1/freehire/internal/jobview" @@ -24,10 +27,15 @@ type jobsHandlers struct { // offline rewrite, so a caller that hands over no client keeps every lookup that // does not need one. postings sources.PostingURLResolver + // cache and estimator resolve the list's meta.total: the published + // catalogue-scale snapshot when there is one, the approximate estimate when there + // is not. See openJobTotal. + cache cache.Cache + estimator catalogstats.Estimator } -func newJobsHandlers(queries *db.Queries, moderation *moderation.Service, postings sources.PostingURLResolver) *jobsHandlers { - return &jobsHandlers{queries: queries, moderation: moderation, postings: postings} +func newJobsHandlers(queries *db.Queries, moderation *moderation.Service, postings sources.PostingURLResolver, c cache.Cache) *jobsHandlers { + return &jobsHandlers{queries: queries, moderation: moderation, postings: postings, cache: c, estimator: queries} } func (h *jobsHandlers) register(api fiber.Router, mw middleware) { @@ -53,9 +61,9 @@ func (h *jobsHandlers) register(api fiber.Router, mw middleware) { // ListJobs returns a page of jobs using limit/offset pagination. Jobs are // served in the shared jobview wire shape (public_slug, no internal id) — the // same shape the detail and search endpoints use. The page rides the partial -// index jobs_open_created_idx (no full-table sort) and meta.total is an -// approximate planner estimate (EstimateOpenJobs), so neither query scans the -// whole open-job set at catalogue scale. +// index jobs_open_created_idx (no full-table sort) and meta.total comes from the +// precomputed catalogue-scale snapshot (see openJobTotal), so neither query scans +// the whole open-job set at catalogue scale. func (h *jobsHandlers) ListJobs(c *fiber.Ctx) error { limit, offset := pageParams(c) @@ -67,10 +75,7 @@ func (h *jobsHandlers) ListJobs(c *fiber.Ctx) error { return err } - total, err := h.queries.EstimateOpenJobs(c.Context()) - if err != nil { - return err - } + total := h.openJobTotal(c.Context()) views, err := jobview.FromRows(jobs) if err != nil { @@ -81,6 +86,16 @@ func (h *jobsHandlers) ListJobs(c *fiber.Ctx) error { return listResponse(c, views, total, limit, offset) } +// openJobTotal resolves the list's meta.total: the exact count from the published +// snapshot, or the approximate estimate when none is available. +// +// It cannot fail. The page of jobs is this endpoint's actual payload, and losing it to a +// 500 because a count was unavailable — which is what the previous EstimateOpenJobs call +// did — trades the whole response for one number. +func (h *jobsHandlers) openJobTotal(ctx context.Context) int64 { + return catalogstats.Load(ctx, h.cache, h.estimator).OpenJobs +} + // GetJob returns a single job addressed by its public slug. func (h *jobsHandlers) GetJob(c *fiber.Ctx) error { job, err := h.queries.GetJobBySlug(c.Context(), c.Params("slug")) diff --git a/internal/handler/jobs_total_test.go b/internal/handler/jobs_total_test.go new file mode 100644 index 000000000..08b4f3d17 --- /dev/null +++ b/internal/handler/jobs_total_test.go @@ -0,0 +1,45 @@ +package handler + +import ( + "context" + "errors" + "testing" + + "github.com/strelov1/freehire/internal/cache" + "github.com/strelov1/freehire/internal/catalogstats" +) + +// The list's meta.total is the most-quoted number freehire publishes — the /about strip, +// the /open page and every API consumer read it. It must be the exact published count +// when one exists, and must still answer when nothing does. +func TestOpenJobTotal_PrefersThePublishedSnapshot(t *testing.T) { + c := cache.NewMemory() + ctx := context.Background() + if err := catalogstats.Store(ctx, c, publishedSnapshot()); err != nil { + t.Fatalf("Store: %v", err) + } + + h := &jobsHandlers{cache: c, estimator: stubEstimator{value: 9_999_999}} + + if got := h.openJobTotal(ctx); got != 3_300_658 { + t.Errorf("openJobTotal = %d, want the snapshot's exact 3300658, not the estimate", got) + } +} + +func TestOpenJobTotal_FallsBackToTheEstimate(t *testing.T) { + h := &jobsHandlers{cache: cache.NewMemory(), estimator: stubEstimator{value: 3_150_000}} + + if got := h.openJobTotal(context.Background()); got != 3_150_000 { + t.Errorf("openJobTotal = %d, want the estimate 3150000", got) + } +} + +// Before this change a failing count returned 500 and took the whole page of jobs with +// it. The page is the endpoint's actual payload; a missing total is not worth losing it. +func TestOpenJobTotal_SurvivesTheEstimatorFailing(t *testing.T) { + h := &jobsHandlers{cache: cache.NewMemory(), estimator: stubEstimator{err: errors.New("database down")}} + + if got := h.openJobTotal(context.Background()); got != 0 { + t.Errorf("openJobTotal = %d, want 0 when no figure is obtainable", got) + } +} diff --git a/internal/handler/stats.go b/internal/handler/stats.go index 408bb3edc..da9136bf4 100644 --- a/internal/handler/stats.go +++ b/internal/handler/stats.go @@ -7,6 +7,8 @@ import ( "github.com/gofiber/fiber/v2" "github.com/jackc/pgx/v5/pgtype" + "github.com/strelov1/freehire/internal/cache" + "github.com/strelov1/freehire/internal/catalogstats" "github.com/strelov1/freehire/internal/db" ) @@ -16,16 +18,29 @@ import ( // aggregate-only reads — no record-level field or user identifier is exposed. type statsHandlers struct { queries *db.Queries + + // cache and estimator back the catalogue-scale read. The cache holds the snapshot + // the rollup worker publishes; the estimator is the approximate open-job count + // served when no snapshot is available. Neither is required — a nil cache degrades + // to the estimate, which is the same path a cold or unreachable cache takes. + cache cache.Cache + estimator catalogstats.Estimator } -func newStatsHandlers(queries *db.Queries) *statsHandlers { - return &statsHandlers{queries: queries} +func newStatsHandlers(queries *db.Queries, c cache.Cache) *statsHandlers { + return &statsHandlers{queries: queries, cache: c, estimator: queries} } func (h *statsHandlers) register(api fiber.Router) { // Public catalogue-activity time series (added vs. removed vacancies per period), // unauthenticated like the other public reads. Served from the job_daily_stats // rollup (cmd/rollup-stats); the /trends SPA page renders it as a bar chart. + // Public catalogue-scale figures, unauthenticated like the other public reads. + // One response carries every number a surface quotes about how big the catalogue + // is, so /about and /open render the same snapshot instead of taking their own + // totals at their own moments and disagreeing. + api.Get("/stats/catalog", h.CatalogScale) + api.Get("/stats/jobs-activity", h.JobsActivity) // Public member-growth time series (cumulative registrations per UTC day), @@ -178,6 +193,30 @@ func (h *statsHandlers) UserGrowth(c *fiber.Ctx) error { // connected, and searches saved. Aggregate-only — the query selects nothing but // integer totals, so no per-user field can leak. An empty database yields all // zeros (200). +// CatalogScale serves the catalogue-scale snapshot: how many open postings the +// catalogue holds, from how many companies, across how many sources, ATS platforms and +// Telegram channels. +// +// It never fails. A cold cache (before the first rollup run), an unreachable one, and a +// payload left by an older build all degrade to the approximate open-job count, and +// `exact` reports which of the two the caller holds — a transparency page showing a +// labelled estimate beats one showing a 500. +func (h *statsHandlers) CatalogScale(c *fiber.Ctx) error { + result := catalogstats.Load(c.Context(), h.cache, h.estimator) + + return c.JSON(fiber.Map{ + "data": fiber.Map{ + "open_jobs": result.OpenJobs, + "companies": result.Companies, + "sources": result.Sources, + "ats_platforms": result.ATSPlatforms, + "telegram_channels": result.TelegramChannels, + "computed_at": result.ComputedAt, + "exact": result.Exact, + }, + }) +} + func (h *statsHandlers) EngagementStats(c *fiber.Ctx) error { s, err := h.queries.GetEngagementStats(c.Context()) if err != nil { diff --git a/internal/handler/stats_catalog_test.go b/internal/handler/stats_catalog_test.go new file mode 100644 index 000000000..033fac8b2 --- /dev/null +++ b/internal/handler/stats_catalog_test.go @@ -0,0 +1,113 @@ +package handler + +import ( + "context" + "errors" + "testing" + "time" + + "github.com/gofiber/fiber/v2" + + "github.com/strelov1/freehire/internal/cache" + "github.com/strelov1/freehire/internal/catalogstats" +) + +type stubEstimator struct { + value int64 + err error +} + +func (s stubEstimator) EstimateOpenJobs(context.Context) (int64, error) { return s.value, s.err } + +func catalogApp(c cache.Cache, est catalogstats.Estimator) *fiber.App { + h := &statsHandlers{cache: c, estimator: est} + app := fiber.New(fiber.Config{ErrorHandler: RenderError}) + app.Get("/stats/catalog", h.CatalogScale) + return app +} + +func publishedSnapshot() catalogstats.Snapshot { + return catalogstats.Snapshot{ + OpenJobs: 3_300_658, + Companies: 294_282, + Sources: 227, + ATSPlatforms: 93, + TelegramChannels: 95, + ComputedAt: time.Unix(1_700_000_000, 0).UTC(), + } +} + +func decodeCatalog(t *testing.T, body map[string]any) map[string]any { + t.Helper() + data, ok := body["data"].(map[string]any) + if !ok { + t.Fatalf("response has no data object: %v", body) + } + return data +} + +func TestCatalogScale_ServesThePublishedSnapshot(t *testing.T) { + c := cache.NewMemory() + if err := catalogstats.Store(context.Background(), c, publishedSnapshot()); err != nil { + t.Fatalf("Store: %v", err) + } + + app := catalogApp(c, stubEstimator{value: 9_999_999}) + status, body := doGet(t, app, "/stats/catalog") + + if status != fiber.StatusOK { + t.Fatalf("status = %d, want 200", status) + } + data := decodeCatalog(t, body) + + for field, want := range map[string]float64{ + "open_jobs": 3_300_658, + "companies": 294_282, + "sources": 227, + "ats_platforms": 93, + "telegram_channels": 95, + } { + if got, _ := data[field].(float64); got != want { + t.Errorf("%s = %v, want %v", field, data[field], want) + } + } + if data["computed_at"] == nil { + t.Error("computed_at is absent — a consumer cannot tell how stale the snapshot is") + } + if exact, _ := data["exact"].(bool); !exact { + t.Error("exact = false for a published snapshot") + } +} + +// The endpoint must answer before the first worker run, and while Redis is down, with +// the approximate figure rather than an error — a transparency page that 500s is worse +// than one showing an estimate it labels as such. +func TestCatalogScale_DegradesRatherThanFailing(t *testing.T) { + app := catalogApp(cache.NewMemory(), stubEstimator{value: 3_150_000}) + status, body := doGet(t, app, "/stats/catalog") + + if status != fiber.StatusOK { + t.Fatalf("status = %d, want 200 on a cold cache", status) + } + data := decodeCatalog(t, body) + + if got, _ := data["open_jobs"].(float64); got != 3_150_000 { + t.Errorf("open_jobs = %v, want the estimate 3150000", data["open_jobs"]) + } + if exact, _ := data["exact"].(bool); exact { + t.Error("exact = true for a degraded read — consumers cannot tell an estimate from a count") + } + // The registry figures cost nothing and need no cache, so they must survive. + if got, _ := data["sources"].(float64); got <= 0 { + t.Errorf("sources = %v on the degraded path, want the registry-derived count", data["sources"]) + } +} + +func TestCatalogScale_SurvivesEverythingFailing(t *testing.T) { + app := catalogApp(cache.NewMemory(), stubEstimator{err: errors.New("database down")}) + status, _ := doGet(t, app, "/stats/catalog") + + if status != fiber.StatusOK { + t.Fatalf("status = %d, want 200 even with no figure available", status) + } +} diff --git a/migrations/0109_estimate_open_jobs_full_predicate.sql b/migrations/0109_estimate_open_jobs_full_predicate.sql new file mode 100644 index 000000000..d689de7e6 --- /dev/null +++ b/migrations/0109_estimate_open_jobs_full_predicate.sql @@ -0,0 +1,20 @@ +-- estimate_open_jobs() backs meta.total on the DB-backed /jobs list. It estimated +-- `closed_at IS NULL` alone, but the list it labels also applies +-- `duplicate_of IS NULL AND NOT is_private` — so every suppressed repost and every +-- private posting was counted in the total and absent from the results. On production +-- that gap, compounded by an inflated reltuples, published 5,226,661 against 3,300,658 +-- actual rows. +-- +-- The total stays an estimate: the planner answers from statistics, so this is still +-- O(1) and still not an exact count. It now estimates the right set. +CREATE OR REPLACE FUNCTION public.estimate_open_jobs() RETURNS bigint + LANGUAGE plpgsql + AS $$ +DECLARE + plan json; +BEGIN + EXECUTE 'EXPLAIN (FORMAT json) SELECT 1 FROM jobs WHERE closed_at IS NULL AND duplicate_of IS NULL AND NOT is_private' + INTO plan; + RETURN (plan -> 0 -> 'Plan' ->> 'Plan Rows')::bigint; +END; +$$; diff --git a/openspec/changes/consistent-catalog-counts/.openspec.yaml b/openspec/changes/consistent-catalog-counts/.openspec.yaml new file mode 100644 index 000000000..f161d5cc4 --- /dev/null +++ b/openspec/changes/consistent-catalog-counts/.openspec.yaml @@ -0,0 +1,2 @@ +schema: spec-driven +created: 2026-08-16 diff --git a/openspec/changes/consistent-catalog-counts/design.md b/openspec/changes/consistent-catalog-counts/design.md new file mode 100644 index 000000000..192078d10 --- /dev/null +++ b/openspec/changes/consistent-catalog-counts/design.md @@ -0,0 +1,188 @@ +## Context + +`GET /api/v1/jobs` fills `meta.total` from `estimate_open_jobs()`, a plpgsql +function added in `migrations/0001_init.sql` that runs `EXPLAIN (FORMAT json) +SELECT 1 FROM jobs WHERE closed_at IS NULL` and returns the planner's `Plan +Rows`. It was the right call at the time: an exact `count(*)` over millions of +open rows is a per-request sequential scan, and the estimate is O(1). + +Two things have since gone wrong with it, measured on production on 2026-08-16: + +| Figure | Value | +| --- | --- | +| `estimate_open_jobs()` (what the site shows) | 5,226,661 | +| Exact `COUNT(*)` with the listing predicate | 3,300,658 | +| Rows in `jobs` | 7,356,316 | +| `pg_class.reltuples` for `jobs` | 9,574,771 | + +The estimate omits `duplicate_of IS NULL AND NOT is_private`, which the list +itself applies, so it counts suppressed reposts. And it inherits `reltuples`, +which bloat has pushed 30% above the real row count and which only moves when +`ANALYZE` runs — so the published figure sits still and then jumps, rather than +tracking the catalogue. + +Consumers today: `GET /api/v1/jobs` (`meta.total`), `/about` +(`+page.server.ts`, one `listJobs(1,0)` plus one `listCompanies('',1,0)`), and +`/open` (the same two reads, inside a 60s module-level memo). `/open` +additionally hardcodes `ATS_PLATFORMS = 166` and `TELEGRAM_CHANNELS = 88` in +`+page.svelte`; the real figures are 227 registered adapters and 95 configured +channels. `HomeView.svelte` carries `3.4M+` / `200K+` API-down fallbacks and a +hardcoded `'166'`. + +Redis is already a runtime dependency: `cmd/server` builds a `*redis.Client` +from `cfg.RedisURL` and hands it to `ratelimit.NewRedisThrottler`. `miniredis` +is already a test dependency. `cmd/rollup-stats` already runs intra-day and +already recomputes rollups from `jobs` inside one transaction. + +## Goals / Non-Goals + +**Goals:** + +- One exact open-job figure, identical on every surface that quotes it. +- Stable between recomputations — no movement that isn't a real catalogue change. +- No per-request cost that grows with catalogue size. +- A cache abstraction good enough that the next consumer does not reach for + `*redis.Client` directly. +- Platform and channel counts derived from the backend, never a frontend + literal. + +**Non-Goals:** + +- Real-time accuracy. A snapshot hours old is fine; the catalogue moves well + under 1% per hour, and consistency is the actual ask. +- Making `/jobs/search`'s Meilisearch-estimated total exact. That is a different + number with a different meaning (matched hits, not catalogue size). +- Caching anything beyond catalogue scale in this change. `internal/cache` is + built to be reusable, but `/open`'s other five legs keep their existing + module-level memo. +- Fixing the table bloat that inflated `reltuples`. Worth doing, separately. + +## Decisions + +### Redis, not a Postgres table + +The snapshot is read on every `/jobs` request, which is the hottest public path +in the system. A `catalog_stats` table would survive a Redis flush and need no +TTL reasoning, but it puts a Postgres round-trip on that path to avoid a +Postgres round-trip, and Postgres is the resource already under I/O pressure on +this host. Redis is in the request path already for rate limiting. + +*Alternative considered:* in-process memoization in the Go server, no Redis at +all. Rejected because the count would then differ per process and reset on every +deploy — which is the inconsistency this change exists to remove. The frontend's +existing module-level memo has exactly that flaw. + +*Accepted cost:* a Redis flush or a restart without persistence drops the +snapshot until the next `rollup-stats` run, and the figure degrades to the +estimate meanwhile. The corrected estimate makes that degradation tolerable +rather than embarrassing, which is why fixing it is in scope. + +### `internal/cache` as a generic layer, not a single typed helper + +Building generic infrastructure ahead of need is the failure mode this codebase +avoids. It is not the case here: there are three present consumers — the open-job +count, the company count, and `/open`'s six-leg payload that today memoizes in a +single frontend process. A one-off `GetOpenJobCount` helper would be re-generalized +within the month. + +The layer stays deliberately thin — `Get`/`Set` over `[]byte` with a TTL, plus +free `GetJSON[T]`/`SetJSON` functions, because Go does not permit type +parameters on methods. It models the `ratelimit.Throttler` shape already in the +tree: the interface reports errors, and a single caller decides to fail open, so +every implementation gets that behaviour uniformly. Two implementations ship: +`RedisCache` and `Memory` (map plus mutex) for tests and for running without +Redis. + +### `internal/catalogstats` owns the figures; `internal/cache` only stores bytes + +The cache knows nothing about jobs. `catalogstats` owns the `Snapshot` type, +the exact queries, the registry and channel-config counts, the cache key, and +the fallback rule. That keeps the read path expressible as one call whose +signature says what it does, and keeps the invariant — *never recompute on a +request* — in one place rather than at each call site. + +`Load` returns the snapshot together with whether it is exact or degraded, so a +consumer can render `3,300,658` differently from `~3.3M`. The frontend need not +use that distinction on day one, but the API should not lie about which it is. + +### Write from `cmd/rollup-stats`, not a new worker + +`rollup-stats` already scans `jobs` on an intra-day cron and already exists as a +deployed systemd unit. Adding a snapshot write there costs one more query on a +run that is already doing heavier work, and costs nothing in ops surface — no +new unit file, no new timer, no new entry in the release script's worker list. + +*Alternative considered:* a dedicated `cmd/snapshot-stats` on a tighter (say +10-minute) schedule for fresher numbers. Rejected as unjustified: nothing about +this figure needs ten-minute freshness, and a new cron worker is real ops cost. +The seam is there if that changes — `catalogstats.Compute` is callable from +anywhere. + +### TTL longer than the cron interval + +The TTL is 24h against an intra-day schedule. Setting it near the cron interval +would mean a single skipped or slow run drops every surface back to the +estimate. Making it long means a failed cron degrades to *stale but exact*, +which is strictly better than *fresh but wrong*. `ComputedAt` travels in the +snapshot so staleness is observable rather than hidden. + +### Fix the estimate even though it becomes the fallback + +The estimate stays approximate — that is its job — but a fallback that is +systematically 58% high is a worse failure mode than one that is a few percent +off. Adding `duplicate_of IS NULL AND NOT is_private` to the `EXPLAIN` costs +nothing at runtime, since the planner still answers from statistics. This ships +as a new migration; the applied `0001_init.sql` is not edited. + +### One endpoint, not two list reads + +`GET /api/v1/stats/catalog` returns all four figures. `/about` and `/open` +switch to it. Beyond halving their request count, this is what makes the two +pages structurally incapable of disagreeing: they render one snapshot, not two +independently-taken estimates. + +## Risks / Trade-offs + +- **Redis becomes load-bearing for a public figure** → It is a soft dependency + by construction: a miss or an error is a miss, and the request completes with + the estimate. A test asserts the unreachable-backend path returns a normal + response, not a 5xx. +- **A stale snapshot silently misrepresents the catalogue** → `ComputedAt` ships + in the API response, so staleness is inspectable rather than invisible. The + 24h TTL bounds it; past that the figure degrades to the estimate rather than + going unboundedly stale. +- **The exact count is a sequential scan on an I/O-pressured host** → It runs in + a cron worker that is already scanning `jobs`, not on a request path, and once + per run rather than per request. Net request-path cost is negative: a Redis + `GET` replaces an `EXPLAIN`. +- **Two surfaces still read `/api/v1/companies` for other purposes** → Out of + scope. Only the catalogue-scale strip moves to the snapshot; the companies + listing keeps its own count semantics. +- **The published headline drops from 5.2M to 3.3M** → It is a correction, not a + regression, and the README and `docs/sources.md` have already been brought to + the true figures. Worth being deliberate about the timing given the Product + Hunt launch on 26 August 2026: better corrected before it than during it. + +## Migration Plan + +1. Ship the migration correcting `estimate_open_jobs()`. Safe alone: it only + improves the number every current consumer already reads. +2. Ship `internal/cache` and `internal/catalogstats` with the `rollup-stats` + write and the `/stats/catalog` endpoint. Until the first worker run the + endpoint reports a degraded snapshot, which is correct behaviour, not a + failure. +3. Run `rollup-stats` once by hand to populate the snapshot before the frontend + depends on it. +4. Ship the frontend switch to `/stats/catalog` and delete the constants. + +Rollback: reverting the frontend restores the list-read path; reverting the +backend leaves an unread Redis key that expires on its own. No schema rollback +is needed — the corrected `estimate_open_jobs()` is an improvement independent +of everything else. + +## Open Questions + +None blocking. One deliberately deferred: whether `/about` and `/open` should +visually distinguish an exact figure from a degraded one. The API carries the +distinction from day one; the frontend can adopt it whenever it is worth the +design. diff --git a/openspec/changes/consistent-catalog-counts/proposal.md b/openspec/changes/consistent-catalog-counts/proposal.md new file mode 100644 index 000000000..ce8e0342d --- /dev/null +++ b/openspec/changes/consistent-catalog-counts/proposal.md @@ -0,0 +1,86 @@ +## Why + +Every public surface that quotes catalogue scale is quoting a number that is +wrong by 58% and drifts on its own. `GET /api/v1/jobs` reports +`meta.total = 5,226,661`; the catalogue actually holds **3,300,658** open, +deduplicated, non-private postings. The `/about` landing strip, the `/open` +transparency page, and the API all read that figure, so the headline claim on +the marketing pages is inflated — and because the estimate rides on planner +statistics, it steps to a new wrong value every time autovacuum runs `ANALYZE`. + +Two independent defects produce it, both in `estimate_open_jobs()`: + +1. It estimates `WHERE closed_at IS NULL` only. The list it labels also applies + `duplicate_of IS NULL AND NOT is_private`, so every suppressed repost is + counted in the total but absent from the results. +2. The planner's row estimate derives from `reltuples`, which currently reads + 9,574,771 against 7,356,316 real rows. Table bloat inflates it, and the + figure only moves when statistics are refreshed — which is what makes it + jump rather than drift. + +A second, smaller instance of the same problem: `/about` and `/open` each issue +their own `limit=1` list read, so two adjacent pages can show two different +open-job counts from two estimates taken at different moments. + +## What Changes + +- Add `internal/cache`: a small best-effort key/value layer with TTL — a `Cache` + interface, a Redis-backed implementation, and an in-memory one for tests and + for a deployment without Redis. Failure handling follows the existing + `ratelimit.Throttler` precedent: the implementation reports the error, one + caller decides to fail open. +- Add `internal/catalogstats`: owns the catalogue-scale figures as a single + `Snapshot` (open jobs, companies, ATS platforms, Telegram channels, computed + at). `Compute` runs the exact counts; `Load` reads the cached snapshot and, on + a miss, degrades to the existing estimate. The exact count never runs on a + request path. +- Write the snapshot from `cmd/rollup-stats`, which already walks `jobs` on an + intra-day cron. No new worker, no new systemd unit. The cached snapshot + outlives a skipped run (24h TTL), so a missed cron degrades to a slightly + stale figure rather than back to a wrong one. +- Add `GET /api/v1/stats/catalog` returning the whole snapshot. `/about`, + `/open`, and `GET /api/v1/jobs`'s `meta.total` all read the same snapshot, so + every surface shows one mutually consistent set of numbers. +- Remove the hardcoded `ATS_PLATFORMS = 166` and `TELEGRAM_CHANNELS = 88` from + `web/src/routes/open/+page.svelte` — the backend already knows both from + `sources.Taxonomy()` and `sources/telegram.yml`. Refresh the API-unavailable + fallbacks in `HomeView.svelte` (`3.4M+` / `200K+`) to match reality. +- Correct `estimate_open_jobs()` to apply `duplicate_of IS NULL AND NOT + is_private`, via a new migration. It stays approximate, but as the degraded + path it should not be systematically wrong. + +## Capabilities + +### New Capabilities + +- `catalog-scale-snapshot`: the exact, periodically recomputed catalogue-scale + figures — how they are computed, cached, served over + `GET /api/v1/stats/catalog`, and what happens when the cache is cold. + +### Modified Capabilities + +- `job-search`: the requirement "DB-backed jobs list is index-served with an + approximate total" changes. `meta.total` becomes the exact cached snapshot + value when one is available, and falls back to the estimate only when it is + not. The prohibition on per-request work that scales with catalogue size is + unchanged and is what forces the cache. +- `open-transparency-page`: the catalogue-scale stat strip sources every figure + from the snapshot endpoint. The ATS-platform and Telegram-channel counts stop + being frontend constants. + +## Impact + +- **Schema:** one new migration amending `estimate_open_jobs()`. No table + changes. +- **New packages:** `internal/cache`, `internal/catalogstats`. +- **Modified:** `internal/handler` (new `/stats/catalog` route; the jobs list + total), `cmd/rollup-stats` (writes the snapshot), `cmd/server` (constructs the + cache, already builds the Redis client for rate limiting). +- **Frontend:** `web/src/routes/open/+page.svelte`, + `web/src/routes/open/+page.server.ts`, `web/src/routes/about/+page.server.ts`, + `web/src/lib/components/HomeView.svelte`. +- **Runtime dependency:** the API gains a soft dependency on Redis for this + figure. Redis being unreachable degrades the count to the estimate; it never + fails a request. +- **Docs:** `README.md` and `docs/sources.md` headline figures already corrected + to the true counts as part of the investigation that surfaced this. diff --git a/openspec/changes/consistent-catalog-counts/specs/catalog-scale-snapshot/spec.md b/openspec/changes/consistent-catalog-counts/specs/catalog-scale-snapshot/spec.md new file mode 100644 index 000000000..5d760249e --- /dev/null +++ b/openspec/changes/consistent-catalog-counts/specs/catalog-scale-snapshot/spec.md @@ -0,0 +1,98 @@ +## ADDED Requirements + +### Requirement: Catalogue scale is served from one periodically recomputed snapshot + +The system SHALL compute the public catalogue-scale figures — the open-job +count, the company count, the number of registered ATS platform adapters, and +the number of crawled Telegram channels — as a single `Snapshot` value carrying +the instant it was computed. + +The open-job and company counts SHALL be exact counts over the same predicate +the public listings apply (`closed_at IS NULL AND duplicate_of IS NULL AND NOT +is_private`), so the total a surface quotes describes the same set of postings a +visitor can page through. The ATS-platform count SHALL be derived from the +provider registry and the Telegram-channel count from the crawled channel +configuration, not maintained as a literal in either the backend or the +frontend. + +Every consumer SHALL read the same snapshot, so two surfaces rendered from the +same snapshot cannot disagree about the size of the catalogue. + +#### Scenario: Snapshot counts match the public listing predicate + +- **WHEN** the snapshot is computed against a catalogue containing open, + closed, duplicate-suppressed, and private postings +- **THEN** its open-job count includes only postings that are open, not + duplicate-suppressed, and not private — the same set `GET /api/v1/jobs` + paginates + +#### Scenario: Platform and channel counts are derived, not hardcoded + +- **WHEN** a new source adapter is registered or a new Telegram channel is added + to the crawled configuration +- **THEN** the next computed snapshot reflects the new count with no change to + any literal in the backend or the frontend + +### Requirement: The exact count never runs on a request path + +The system SHALL recompute the snapshot only from the scheduled rollup worker, +never during an HTTP request. A request SHALL NOT trigger a query whose cost +grows with catalogue size. + +The recomputed snapshot SHALL be published to a shared cache with a retention +window that outlives the worker's schedule, so a skipped or failed worker run +degrades the figure to a slightly stale exact value rather than to no value. + +#### Scenario: Serving a request never recomputes + +- **WHEN** any number of clients request a surface that quotes catalogue scale +- **THEN** no exact catalogue-wide count is executed as part of serving those + requests + +#### Scenario: A skipped worker run keeps the last snapshot + +- **WHEN** the rollup worker does not run for longer than its normal interval + but within the cache retention window +- **THEN** consumers continue to receive the last computed snapshot, and its + computed-at instant reports how stale it is + +### Requirement: A cold or unreachable cache degrades, never fails + +The system SHALL treat an absent or unreadable cached snapshot as a miss and +fall back to the existing approximate open-job estimate. A cache failure SHALL +NOT fail a request, and SHALL NOT be surfaced to the client as an error. + +Consumers SHALL be able to tell an exact snapshot from the degraded fallback, so +a surface can choose to present the figure differently when it is only an +estimate. + +#### Scenario: Cache is empty on a cold start + +- **WHEN** the cache holds no snapshot and a client requests the jobs list +- **THEN** `meta.total` carries the approximate estimate and the request + succeeds normally + +#### Scenario: Cache backend is unreachable + +- **WHEN** the cache backend cannot be reached +- **THEN** the request succeeds with the degraded figure rather than returning + an error + +### Requirement: The snapshot is exposed as a public endpoint + +The system SHALL serve the snapshot over an unauthenticated +`GET /api/v1/stats/catalog`, using the single-item envelope `{"data": ...}`. The +response SHALL carry the open-job count, the company count, the ATS-platform +count, the Telegram-channel count, and the instant the snapshot was computed. + +#### Scenario: Endpoint returns the whole snapshot + +- **WHEN** an anonymous client requests `GET /api/v1/stats/catalog` +- **THEN** the response is `{"data": {...}}` carrying all four figures and the + computed-at instant + +#### Scenario: One request replaces two list reads + +- **WHEN** a page needs both the open-job count and the company count +- **THEN** it obtains both from one `GET /api/v1/stats/catalog` response, and + the two figures come from the same snapshot diff --git a/openspec/changes/consistent-catalog-counts/specs/job-search/spec.md b/openspec/changes/consistent-catalog-counts/specs/job-search/spec.md new file mode 100644 index 000000000..d19e635ce --- /dev/null +++ b/openspec/changes/consistent-catalog-counts/specs/job-search/spec.md @@ -0,0 +1,50 @@ +## MODIFIED Requirements + +### Requirement: DB-backed jobs list is index-served with an approximate total + +The DB-backed `GET /api/v1/jobs` list endpoint SHALL return open jobs +(`closed_at IS NULL`) ordered newest-added first (`created_at` descending, `id` +descending) with `limit`/`offset` pagination, using the standard list envelope +`{"data": [...], "meta": {...}}`. The ordered page SHALL be served through a +partial index matching that order (no full-table sort at request time), so the +endpoint stays responsive at catalogue scale (millions of open jobs). + +The `meta.total` for this endpoint SHALL be the exact open-job count from the +current catalogue-scale snapshot when one is available, and the approximate +estimate only when it is not. The endpoint SHALL NOT run a query whose cost +grows linearly with the catalogue size on each request — which is precisely why +the exact count is read from a precomputed snapshot rather than counted per +request. + +Whichever figure is served, it SHALL describe the same set of postings the +endpoint paginates: open, not duplicate-suppressed, and not private. The +approximate fallback SHALL apply that full predicate, so it is an estimate of +the right set rather than an estimate of a larger one. + +#### Scenario: List returns a page ordered newest-added first + +- **WHEN** a client requests `GET /api/v1/jobs?limit=20&offset=0` +- **THEN** up to 20 open jobs are returned ordered by `created_at` descending + (ties broken by `id` descending), in the `{"data": [...], "meta": {...}}` + envelope + +#### Scenario: Meta carries the exact total when a snapshot is available + +- **WHEN** a catalogue-scale snapshot is available and a client requests + `GET /api/v1/jobs?limit=20&offset=0` +- **THEN** `meta` reports the applied `limit` and `offset` and a `total` equal + to the snapshot's exact open-job count + +#### Scenario: Meta falls back to an approximate total when no snapshot exists + +- **WHEN** no catalogue-scale snapshot is available and a client requests + `GET /api/v1/jobs?limit=20&offset=0` +- **THEN** `meta` reports the applied `limit` and `offset` and a `total` that is + an approximate open-job count, and the request succeeds + +#### Scenario: The approximate estimate describes the paginated set + +- **WHEN** the catalogue contains postings suppressed as duplicates or marked + private and the approximate fallback is served +- **THEN** those postings are excluded from the estimate, as they are from the + returned page diff --git a/openspec/changes/consistent-catalog-counts/specs/open-transparency-page/spec.md b/openspec/changes/consistent-catalog-counts/specs/open-transparency-page/spec.md new file mode 100644 index 000000000..e0fd4a9bf --- /dev/null +++ b/openspec/changes/consistent-catalog-counts/specs/open-transparency-page/spec.md @@ -0,0 +1,40 @@ +## MODIFIED Requirements + +### Requirement: Public transparency page +The system SHALL serve a public, unauthenticated `/open` page that renders live +freehire metrics server-side (SSR), covering catalogue scale, catalogue movement, +facet distributions, open-source stats, and member growth. + +Every figure in the catalogue-scale strip SHALL come from the catalogue-scale +snapshot endpoint. The page SHALL NOT carry the ATS-platform count or the +Telegram-channel count as frontend constants, and SHALL NOT read the open-job +and company counts as two separate list totals — a single snapshot response +supplies all four, so the strip cannot show figures taken at different moments. + +#### Scenario: Page is public and server-rendered +- **WHEN** an anonymous visitor opens `/open` +- **THEN** the page responds 200 with the metrics present in the initial server-rendered HTML (no client-only data fetch required to see the figures) + +#### Scenario: Catalogue scale section +- **WHEN** the page renders +- **THEN** it shows a stat-strip with the live open-job count, company count, the ATS-platform count, and the Telegram-channel count, all four taken from one catalogue-scale snapshot + +#### Scenario: Platform and channel counts track the backend +- **WHEN** a source adapter or a crawled Telegram channel is added or removed +- **THEN** the `/open` stat-strip reflects it on the next snapshot with no frontend change + +#### Scenario: Catalogue movement section +- **WHEN** the page renders +- **THEN** it shows the added-vs-removed activity over time, reusing the same chart as `/trends` fed by `/api/v1/stats/jobs-activity` + +#### Scenario: What's-inside section +- **WHEN** the page renders +- **THEN** it shows facet distributions (top countries, top skills, remote share, seniority split) derived from the precomputed `/api/v1/stats/facets` snapshot + +#### Scenario: Member-growth section +- **WHEN** the page renders +- **THEN** it shows a cumulative member-growth chart fed by `/api/v1/stats/user-growth` + +#### Scenario: Open-source section +- **WHEN** the page renders and the GitHub API is reachable +- **THEN** it shows the repository stars, forks, and contributor count, an MIT-license badge, and a contribute call to action diff --git a/openspec/changes/consistent-catalog-counts/tasks.md b/openspec/changes/consistent-catalog-counts/tasks.md new file mode 100644 index 000000000..6e9da30a2 --- /dev/null +++ b/openspec/changes/consistent-catalog-counts/tasks.md @@ -0,0 +1,126 @@ +## 1. Correct the estimate + +- [x] 1.1 Add `migrations/0109_estimate_open_jobs_full_predicate.sql` replacing + `estimate_open_jobs()` so its `EXPLAIN` applies the full listing predicate + (`closed_at IS NULL AND duplicate_of IS NULL AND NOT is_private`). Verify + the next free number at write time — `migrations/` already carries + duplicate numbers, so `ls migrations/ | tail` rather than assuming. + (Landed as 0109: main had moved past 0102 to 0108.) +- [x] 1.2 Add an integration test (`//go:build integration`, testdb) seeding + open, closed, duplicate-suppressed, and private jobs, asserting the + function's result excludes the last three. Assert *which set* is estimated + — the result must sit nearer the paginated count than the not-closed one — + rather than a tolerance around the exact count: on a small table the + planner's selectivity arithmetic carries error that is not a defect. + +## 2. `internal/cache` — the layer + +- [x] 2.1 Write the failing test for `Memory`: `Set` then `Get` round-trips, + a missing key reports `found == false` with no error, and an entry past + its TTL reports a miss. +- [x] 2.2 Define `Cache` (`Get(ctx, key) ([]byte, bool, error)`, + `Set(ctx, key, val []byte, ttl time.Duration) error`) and implement + `Memory` (map + mutex). Document on the interface that a returned error is + the caller's cue to treat the read as a miss — mirroring how + `ratelimit.Throttler` leaves the fail-open decision to one caller. +- [x] 2.3 Write the failing test for `RedisCache` against `miniredis` (already a + dependency): round-trip, miss, TTL honoured, and an error surfaced when + the backend is closed. +- [x] 2.4 Implement `RedisCache` over the existing `*redis.Client`. +- [x] 2.5 Write the failing test for `GetJSON[T]`/`SetJSON`: round-trip of a + struct, a miss on an absent key, and a decode failure reported as a miss + rather than a hard error (a stale incompatible payload must not wedge a + caller). +- [x] 2.6 Implement `GetJSON`/`SetJSON` as free generic functions — Go does not + allow type parameters on methods. + +## 3. `internal/catalogstats` — the figures + +- [x] 3.1 Write the failing unit test for the derived counts: the ATS-platform + count comes from `sources.Taxonomy()` and the Telegram-channel count from + the parsed channel config, so adding an entry moves the number with no + literal to edit. +- [x] 3.2 Define `Snapshot` (open jobs, companies, sources, ATS platforms, + Telegram channels, `ComputedAt`) and the derived-count half of `Compute`. + `Sources` was added on top of the planned fields: /open's "166 ATS + platforms" was really the whole registry mislabelled, and the honest ATS + count is 93 — so the strip leads with total reach (227) under an accurate + label and the narrower figure stays available in the API. +- [x] 3.3 Write the failing integration test for the exact counts: seed open, + closed, duplicate-suppressed and private jobs plus their companies, assert + `Compute` counts exactly the set `GET /api/v1/jobs` paginates. +- [x] 3.4 Add the sqlc query for the exact open-job and company counts and wire + it into `Compute` (`make sqlc`; edit `internal/db/queries/*.sql`, never the + generated code). +- [x] 3.5 Write the failing test for `Load`: a cached snapshot returns exact; an + empty cache and an erroring cache both return the estimate flagged as + degraded; neither returns an error to the caller. +- [x] 3.6 Implement `Load` and `Store`. `Load` must never invoke `Compute` — + made structural instead of asserted: `Load` takes no `ExactCounter`, so no + read path can reach the catalogue-wide scan even by mistake. A runtime + assertion would have been the weaker guarantee. + +## 4. Publish the snapshot + +- [x] 4.1 Write the failing test that `rollup-stats`'s snapshot step stores a + computed snapshot through the cache with the intended TTL. The TTL is + asserted in `catalogstats` against a recording cache — the retention + decision belongs next to the constant that encodes it, not in the worker. +- [x] 4.2 Call `catalogstats.Compute` + `Store` from `cmd/rollup-stats`, after + the existing rollups. A snapshot failure must not fail the run's exit code + — the rollups are the worker's primary job and already have their own + transaction semantics. +- [x] 4.3 Construct the cache in `cmd/rollup-stats` from `cfg.RedisURL` + (`cmd/server` already builds a client for rate limiting — follow that + wiring, do not duplicate parsing logic). + +## 5. Serve it + +- [x] 5.1 Write the failing handler test for `GET /api/v1/stats/catalog`: + `{"data": {...}}` carrying all four figures plus `computed_at`, and + unauthenticated access. +- [x] 5.2 Implement the handler and register the route alongside the existing + `/stats/*` routes in `internal/handler/stats.go`. +- [x] 5.3 Write the failing test that `GET /api/v1/jobs` `meta.total` prefers the + snapshot's exact count and falls back to the estimate when the cache is + empty or unreachable, returning 200 in both cases. +- [x] 5.4 Switch the jobs-list total to `catalogstats.Load`. +- [x] 5.5 Construct the cache in `cmd/server` from the `*redis.Client` it already + builds, and inject it into the handlers. + +## 6. Frontend + +- [x] 6.1 Add the `/api/v1/stats/catalog` client method (check whether the + contract is codegen'd — `cmd/gen-contracts` — and regenerate rather than + hand-writing the type if so). +- [x] 6.2 Switch `web/src/routes/about/+page.server.ts` to the single snapshot + call, replacing the two `limit=1` list reads. +- [x] 6.3 Switch `web/src/routes/open/+page.server.ts` to the snapshot call and + delete `ATS_PLATFORMS = 166` / `TELEGRAM_CHANNELS = 88` from + `+page.svelte`, reading both from the snapshot. +- [x] 6.4 Update the API-down fallbacks in `web/src/lib/components/HomeView.svelte` + (`3.4M+` → `3.3M+`, `200K+` → `290K+`) and take the ATS-platform figure + from the snapshot instead of the hardcoded `'166'`. +- [x] 6.5 Verify `/about` and `/open` render the same figures, headless, against + a running stack — the numbers agreeing is the whole point of the change, + so it needs looking at rather than asserting. Done against a local stack + seeded with 40 listed + 40 excluded postings: API, /about and /open all + reported 40/1/227/95. The degraded path was exercised too (cache flushed, + then Redis stopped): both endpoints stayed 200, and it caught a real bug — + /open rendered a static "290K+" companies figure when the backend had said + it could not measure one. Fixed by mapping database-only figures to null + instead of zero, so "not measured" stops looking like a value. + +## 7. Finish + +- [x] 7.1 `gofmt -w` the touched Go files, then `go vet ./...`, `go test ./...`, + and `go vet -tags=integration ./...`. +- [x] 7.2 Run the tagged suite for the packages whose behaviour changed + (`go test -tags=integration ./internal/db/ ./internal/handler/ + ./internal/catalogstats/`). +- [x] 7.3 Update `internal/handler/AGENTS.md` with the new route, and note the + snapshot's ownership and fallback rule wherever the module map wants it. + Recorded as a root AGENTS.md convention rather than a new per-package + AGENTS.md: the rule is cross-cutting (every scale figure reads one + snapshot, nothing counts on a request path) and `internal/catalogstats` is + too small to carry a module file of its own. diff --git a/web/src/lib/api.ts b/web/src/lib/api.ts index acfa4be1e..9747b5fbb 100644 --- a/web/src/lib/api.ts +++ b/web/src/lib/api.ts @@ -79,6 +79,7 @@ import type { ActivityGranularity, ActivityPoint, UserGrowthPoint, + CatalogScale, EngagementStats, IngestStatus, LocationPreferences, @@ -521,6 +522,17 @@ export function createApi( return res.data.facets ?? {}; } + /** How big the catalogue is, as one snapshot: open postings, companies, sources, + * ATS platforms and Telegram channels. Every surface quoting catalogue scale reads + * this one call, so /about and /open cannot show different numbers — which they + * could when each took its own list total at its own moment. + * + * Never fails: with no published snapshot the backend answers with an approximate + * open-job count and `exact: false`. Aggregate-only, unauthenticated. */ + async function catalogScale(): Promise { + return requestData(`/api/v1/stats/catalog`); + } + /** The public ingest-fleet status: a per-provider health rollup with a derived * operational/degraded/down verdict and an overall status. Sanitized * (no error text or board identifiers), aggregate-only, unauthenticated. */ @@ -2030,6 +2042,7 @@ export function createApi( userGrowth, engagementStats, statsFacets, + catalogScale, ingestStatus, listCompanies, getCompany, diff --git a/web/src/lib/catalogScale.test.ts b/web/src/lib/catalogScale.test.ts new file mode 100644 index 000000000..2a5f2d32f --- /dev/null +++ b/web/src/lib/catalogScale.test.ts @@ -0,0 +1,64 @@ +import { describe, expect, it } from 'vitest'; + +import { createApi } from './api'; +import type { CatalogScale } from './types'; + +/** A fetch that records the URL it was asked for and answers with a fixed envelope. */ +function snapshotFetch(seen: { url?: string }, snapshot: Partial): typeof fetch { + return ((url: string) => { + seen.url = url; + return Promise.resolve(new Response(JSON.stringify({ data: snapshot }), { status: 200 })); + }) as unknown as typeof fetch; +} + +describe('catalogScale', () => { + it('reads every scale figure from one request', async () => { + const seen: { url?: string } = {}; + const client = createApi( + snapshotFetch(seen, { + open_jobs: 3_300_658, + companies: 294_282, + sources: 227, + ats_platforms: 93, + telegram_channels: 95, + computed_at: '2026-08-16T10:00:00Z', + exact: true, + }), + '', + {}, + ); + + const scale = await client.catalogScale(); + + // One call, not two list reads — that is what stops /about and /open showing + // numbers measured at different moments. + expect(seen.url).toBe('/api/v1/stats/catalog'); + expect(scale.open_jobs).toBe(3_300_658); + expect(scale.companies).toBe(294_282); + expect(scale.sources).toBe(227); + expect(scale.telegram_channels).toBe(95); + expect(scale.exact).toBe(true); + }); + + // Before the first worker run, and whenever Redis is unreachable, the backend answers + // with an approximate job count and zeroes for the figures only the database holds. + // The client must surface that rather than smoothing it over — a page showing + // "0 companies" is worse than a page showing no companies stat at all. + it('passes a degraded snapshot through unchanged', async () => { + const client = createApi( + snapshotFetch( + {}, + { open_jobs: 3_150_000, companies: 0, sources: 227, telegram_channels: 0, exact: false }, + ), + '', + {}, + ); + + const scale = await client.catalogScale(); + + expect(scale.exact).toBe(false); + expect(scale.open_jobs).toBe(3_150_000); + expect(scale.companies).toBe(0); + expect(scale.sources).toBe(227); + }); +}); diff --git a/web/src/lib/components/HomeView.svelte b/web/src/lib/components/HomeView.svelte index ef2d69792..3e3fe95bf 100644 --- a/web/src/lib/components/HomeView.svelte +++ b/web/src/lib/components/HomeView.svelte @@ -5,9 +5,11 @@ import { NumberedGrid, SectionLabel } from '$lib/ui'; import { HOME_FAQ } from '$lib/homeFaq'; - // Live catalogue totals from the page's server load; either may be null on an - // API hiccup, so each has a static fallback (see `figures`). - const { stats }: { stats: { jobs: number | null; companies: number | null } } = $props(); + // Live catalogue figures from the page's server load, all from one snapshot. Any may + // be null when the API is unreachable, so each has a static fallback (see `figures`). + const { + stats, + }: { stats: { jobs: number | null; companies: number | null; sources: number | null } } = $props(); // Compact display for the live figures (2,939,967 → "2.9M+"). The fallbacks are // rounded DOWN from the last measured totals: the catalogue does not only grow @@ -16,13 +18,16 @@ const nf = new Intl.NumberFormat('en', { notation: 'compact', maximumFractionDigits: 1 }); const compact = (n: number | null, fallback: string) => (n == null ? fallback : `${nf.format(n)}+`); - // The under-the-fold stats strip: two live totals, then the two constants — - // ATS breadth and licensing — that don't need a query. The ATS count mirrors the - // /open stat-strip: registered adapters in internal/sources/source.go `All()`. + // The under-the-fold stats strip: three live figures plus the licensing constant. + // + // "sources" counts every registered adapter, not just ATS platforms. It was labelled + // "ATS platforms" while carrying the whole registry — and of today's 227 adapters, 104 + // are aggregators and 30 are single-company career feeds, so the label was counting + // things it did not describe. It is live now, so it also stops going stale. const figures = $derived([ - { value: compact(stats.jobs, '3.4M+'), label: 'open jobs' }, - { value: compact(stats.companies, '200K+'), label: 'companies' }, - { value: '166', label: 'ATS platforms' }, + { value: compact(stats.jobs, '3.3M+'), label: 'open jobs' }, + { value: compact(stats.companies, '290K+'), label: 'companies' }, + { value: stats.sources == null ? '220+' : String(stats.sources), label: 'sources' }, { value: '100%', label: 'open source' }, ]); diff --git a/web/src/lib/types.ts b/web/src/lib/types.ts index 5011d86cf..9ee4acd91 100644 --- a/web/src/lib/types.ts +++ b/web/src/lib/types.ts @@ -736,6 +736,27 @@ export interface EngagementStats { saved_searches: number; } +/** How big the catalogue is, as one snapshot. + * + * Every surface that quotes catalogue scale reads this, so two pages rendered from + * the same response cannot disagree. `open_jobs` and `companies` are exact counts over + * the set the public listings paginate; `sources`, `ats_platforms` and + * `telegram_channels` describe reach — what the crawler can read, whether or not each + * currently holds an open posting. + * + * `exact` is false when the backend had no published snapshot and fell back to an + * approximate open-job count. On that path only `open_jobs`, `sources` and + * `ats_platforms` are meaningful; the rest are zero. */ +export interface CatalogScale { + open_jobs: number; + companies: number; + sources: number; + ats_platforms: number; + telegram_channels: number; + computed_at: string; + exact: boolean; +} + /** The derived health verdict for a provider (and the fleet) on the public * /status page. */ export type HealthStatus = 'operational' | 'degraded' | 'down'; diff --git a/web/src/routes/about/+page.server.ts b/web/src/routes/about/+page.server.ts index a08defd8c..943b94a01 100644 --- a/web/src/routes/about/+page.server.ts +++ b/web/src/routes/about/+page.server.ts @@ -1,17 +1,27 @@ import { serverApi } from '$lib/server/api'; import type { PageServerLoad } from './$types'; -// Live catalogue scale for the landing stats strip. Both totals are a one-row -// list read (limit=1) — we only want `meta.total`, not the page. `allSettled` -// keeps the marketing page rendering even if the API hiccups: a failed leg -// yields `null`, and HomeView falls back to a static "+" figure. +// Live catalogue scale for the landing stats strip. One call: the backend publishes +// every scale figure as a single snapshot, so the numbers here and the ones on /open +// describe the same measurement rather than two list totals taken at two moments. +// +// The endpoint itself never fails — with no published snapshot it answers with an +// approximate open-job count. The catch covers the transport (the API unreachable), in +// which case HomeView falls back to its static "+" figures. export const load: PageServerLoad = async ({ fetch }) => { - const api = serverApi(fetch); - const [jobs, companies] = await Promise.allSettled([api.listJobs(1, 0), api.listCompanies('', 1, 0)]); - return { - stats: { - jobs: jobs.status === 'fulfilled' ? jobs.value.total ?? null : null, - companies: companies.status === 'fulfilled' ? companies.value.total ?? null : null, - }, - }; + try { + const scale = await serverApi(fetch).catalogScale(); + return { + stats: { + jobs: scale.open_jobs, + // Only the exact snapshot carries a company count; a degraded read reports + // zero, which is "not measured", not "none". Null instead, so HomeView shows + // its last-known figure rather than printing "0+ companies". + companies: scale.exact ? scale.companies : null, + sources: scale.sources, + }, + }; + } catch { + return { stats: { jobs: null, companies: null, sources: null } }; + } }; diff --git a/web/src/routes/open/+page.server.ts b/web/src/routes/open/+page.server.ts index 81a7dd3aa..feb3c2c8f 100644 --- a/web/src/routes/open/+page.server.ts +++ b/web/src/routes/open/+page.server.ts @@ -71,9 +71,11 @@ let pageCache: { at: number; data: OpenPayload } | null = null; async function buildPayload(fetchImpl: typeof fetch) { const api = serverApi(fetchImpl); - const [jobs, companies, activity, facets, growth, engagement, github] = await Promise.allSettled([ - api.listJobs(1, 0), - api.listCompanies('', 1, 0), + // One call for the whole scale strip instead of two list totals: the figures come + // from a single published snapshot, so this page and /about cannot show numbers + // measured at different moments. + const [scale, activity, facets, growth, engagement, github] = await Promise.allSettled([ + api.catalogScale(), api.jobsActivity('day'), api.statsFacets(), api.userGrowth(), @@ -84,10 +86,20 @@ async function buildPayload(fetchImpl: typeof fetch) { const value = (r: PromiseSettledResult): T | null => r.status === 'fulfilled' ? r.value : null; + const catalog = value(scale); + // A degraded snapshot carries the approximate job count and the registry figures; + // the counts that exist only in the database come back as zero. Map those to null + // rather than passing the zero on: "we could not measure this" and "we measured + // zero" must not look the same to a renderer, or a page ends up printing a figure + // nobody stands behind. + const dbOnly = (n: number | undefined) => (catalog?.exact && n != null ? n : null); + return { scale: { - jobs: value(jobs)?.total ?? null, - companies: value(companies)?.total ?? null, + jobs: catalog?.open_jobs ?? null, + companies: dbOnly(catalog?.companies), + sources: catalog?.sources ?? null, + telegramChannels: dbOnly(catalog?.telegram_channels), }, activity: value(activity) ?? [], facets: value(facets) ?? null, diff --git a/web/src/routes/open/+page.svelte b/web/src/routes/open/+page.svelte index 1790a36e2..672a6d076 100644 --- a/web/src/routes/open/+page.svelte +++ b/web/src/routes/open/+page.svelte @@ -32,17 +32,13 @@ ]) ); - // Repo constants — the crawler covers this many ATS platforms and Telegram - // channels. Not DB rows; they change only when adapters/channels are added (i.e. - // on deploy), mirroring the homepage stat-strip. Recount on change: - // ATS platforms → registered adapters in internal/sources/source.go `All()` - // Telegram channels → `- channel:` entries in sources/telegram.yml - const ATS_PLATFORMS = 166; - const TELEGRAM_CHANNELS = 88; - const nf = new Intl.NumberFormat('en'); const compactNf = new Intl.NumberFormat('en', { notation: 'compact', maximumFractionDigits: 1 }); - const fmt = (n: number | null, fallback: string) => (n == null ? fallback : compactNf.format(n)); + // A null fallback means "omit this stat entirely". This is the transparency page: a + // figure it cannot source should disappear from the strip rather than fall back to a + // number baked in at build time, which is exactly the habit this page exists to break. + const fmt = (n: number | null, fallback: F) => + n == null ? fallback : compactNf.format(n); const regionNames = new Intl.DisplayNames(['en'], { type: 'region' }); function countryName(code: string): string { @@ -92,12 +88,27 @@ return Math.round(((wm.remote ?? 0) / total) * 100); }); - const stats = $derived([ - { value: fmt(data.scale.jobs, '3.4M+'), label: 'open jobs', href: '/api/v1/jobs' }, - { value: fmt(data.scale.companies, '200K+'), label: 'companies', href: '/api/v1/companies' }, - { value: nf.format(ATS_PLATFORMS), label: 'ATS platforms', href: null }, - { value: nf.format(TELEGRAM_CHANNELS), label: 'Telegram channels', href: null }, - ]); + // Every figure here comes from the one catalogue-scale snapshot, so the strip cannot + // show numbers measured at different moments. The platform and channel counts used to + // be constants in this file and went stale between deploys; they are now derived from + // the source registry and the crawled channel config. + // + // A degraded snapshot (no published measurement yet, or Redis unreachable) carries + // only the approximate job count and the registry figures — the counts that exist + // solely in the database come back zero, and a zero is dropped rather than rendered as + // a real "0 companies". + const stats = $derived( + [ + { value: fmt(data.scale.jobs, null), label: 'open jobs', href: '/api/v1/jobs' }, + { value: fmt(data.scale.companies, null), label: 'companies', href: '/api/v1/companies' }, + { value: fmt(data.scale.sources, null), label: 'sources', href: '/api/v1/stats/catalog' }, + { + value: fmt(data.scale.telegramChannels, null), + label: 'Telegram channels', + href: '/api/v1/stats/catalog', + }, + ].filter((s) => s.value !== null), + ); const gh = $derived(data.github); const members = $derived.by(() => {