diff --git a/apps/worker/cmd/worker/main.go b/apps/worker/cmd/worker/main.go index c129cf76..3b574bf8 100644 --- a/apps/worker/cmd/worker/main.go +++ b/apps/worker/cmd/worker/main.go @@ -26,14 +26,18 @@ import ( workermedia "github.com/Singleton-Solution/GoNext/apps/worker/internal/media" "github.com/Singleton-Solution/GoNext/packages/go/buildinfo" + "github.com/Singleton-Solution/GoNext/packages/go/cache/invalidator" "github.com/Singleton-Solution/GoNext/packages/go/config" + "github.com/Singleton-Solution/GoNext/packages/go/db" jobsasynq "github.com/Singleton-Solution/GoNext/packages/go/jobs/asynq" "github.com/Singleton-Solution/GoNext/packages/go/jobs/cron" + "github.com/Singleton-Solution/GoNext/packages/go/jobs/scheduler" "github.com/Singleton-Solution/GoNext/packages/go/jobs/taskspec" "github.com/Singleton-Solution/GoNext/packages/go/log" "github.com/Singleton-Solution/GoNext/packages/go/media/storage" "github.com/Singleton-Solution/GoNext/packages/go/metrics" "github.com/Singleton-Solution/GoNext/packages/go/observability/errortracker" + gonextredis "github.com/Singleton-Solution/GoNext/packages/go/redis" "github.com/Singleton-Solution/GoNext/packages/go/shutdown" ) @@ -180,6 +184,53 @@ func run(ctx context.Context) error { }); err != nil { return fmt.Errorf("worker/media: register: %w", err) } + // Postgres pool. Required for the cache-invalidation outbox + // poller (#94), the content-scheduler/GC jobs (#143), and any + // future task whose handler talks to the database. We let the + // pool fail-fast at boot rather than waiting for the first job: + // a worker that comes up green only to error every task is + // indistinguishable from a healthy one until somebody looks. + pool, err := db.New(ctx, cfg.Database, logger) + if err != nil { + return fmt.Errorf("db.New: %w", err) + } + orch.MustRegister(logger, "db.pool", func(context.Context) error { + pool.Close() + return nil + }) + + // Dedicated Redis client for the cache invalidator + content + // scheduler. The asynq server has its own pool managed via + // asynq.RedisClientOpt; we keep this one separate so the + // invalidator's pub/sub lifetime is uncoupled from the queue + // consumer's connection lifecycle. + rdb, err := gonextredis.New(ctx, cfg.Redis, logger) + if err != nil { + return fmt.Errorf("redis.New: %w", err) + } + orch.MustRegister(logger, "redis.client", func(context.Context) error { + return rdb.Close() + }) + + // Cache invalidator: drains the cache_invalidations outbox + // shipped by 000030 and republishes each row on the + // gonext:cache:invalidate pub/sub channel. We start the worker + // in a goroutine and shut it down via the orchestrator so the + // drain budget covers an in-flight poll cycle. + invWorker := invalidator.New(pool, rdb, invalidator.WithLogger(logger)) + invCtx, invCancel := context.WithCancel(ctx) + invDone := make(chan struct{}) + go func() { + defer close(invDone) + if err := invWorker.Run(invCtx); err != nil { + logger.Warn("cache invalidator exited with error", "err", err.Error()) + } + }() + orch.MustRegister(logger, "cache.invalidator", func(context.Context) error { + invCancel() + <-invDone + return nil + }) // Registration order (locked in by issue #112): // @@ -266,6 +317,32 @@ func run(ctx context.Context) error { "cron", storage.AbortOrphansCronName, "schedule", storage.AbortOrphansSchedule, ) + + // Content scheduler + GC (#143). Both tasks share the + // worker's pgx pool. The publisher fires every minute + // and flips status=scheduled rows whose scheduled_for + // has elapsed; the GC fires daily at 03:30 UTC and + // hard-deletes trash older than 30 days. + // + // We register against the same cronReg as the storage + // sweep so the eventual cron-leader (#258) sees one + // merged schedule, and against taskspec.Default() so + // the asynq mux dispatch already in place handles the + // task type. + if err := scheduler.SeedDefaults(taskspec.Default(), cronReg, scheduler.SeedOptions{ + Pool: pool, + Logger: logger, + }); err != nil { + logger.Warn("scheduler seed failed", "err", err.Error()) + } else { + taskspec.Dispatch(srv.Mux(), taskspec.Default()) + logger.Info("content scheduler + gc registered", + "publisher_task", scheduler.PublisherTaskName, + "publisher_schedule", scheduler.PublisherSchedule, + "gc_task", scheduler.GCTaskName, + "gc_schedule", scheduler.GCSchedule, + ) + } } } diff --git a/packages/go/blocks/render/cache.go b/packages/go/blocks/render/cache.go new file mode 100644 index 00000000..8c6d07a3 --- /dev/null +++ b/packages/go/blocks/render/cache.go @@ -0,0 +1,432 @@ +package render + +import ( + "context" + "crypto/sha256" + "encoding/hex" + "encoding/json" + "fmt" + "html/template" + "log/slog" + "sort" + "strings" + "sync/atomic" + "time" +) + +// CacheBackend is the byte cache the block render layer consults +// before invoking a renderer. The concrete production backend is +// packages/go/cache/fragment.Cache, but the walker depends on this +// narrower interface so the render package compiles without a Redis +// driver — useful for unit tests, and keeps the import graph clean. +// +// Tags carry the dependency story for invalidation: a block render +// keyed by site-settings touches the "site:settings" tag, so a +// settings update Purges all renders that captured it. The cache +// itself does not interpret tags — it just records and re-checks +// them via the fragment cache's version vector. +type CacheBackend interface { + // Get returns (value, true) on a fresh hit. A non-nil error is + // for transport failures (e.g. Redis unreachable); a clean miss + // is (nil, false, nil). The CachedWalker treats errors and + // misses identically — both fall through to a fresh render. + Get(ctx context.Context, key string, tags []string) ([]byte, bool, error) + + // Set stores value under key with the given tags and ttl. A + // non-nil error is logged at Warn; the rendered HTML is still + // served, since a missed write isn't a correctness bug. + Set(ctx context.Context, key string, value []byte, tags []string, ttl time.Duration) error +} + +// DefaultCacheTTL is the time-to-live applied to cached block +// renders. One hour matches the fragment-cache default (see +// packages/go/cache/fragment) and is the right "cheap to regenerate +// but expensive on a hot page" balance for content blocks. Operators +// who want a longer TTL pass a custom value via CachedWalkerOptions. +const DefaultCacheTTL = 1 * time.Hour + +// KeyPrefix is the namespace prefix the cached-walker prepends to +// every cache key. Exported so subscribers / dashboards that inspect +// Redis can route on "br:" without re-deriving the constant. +const KeyPrefix = "br:" + +// CachedWalkerOptions configures a CachedWalker at construction time. +type CachedWalkerOptions struct { + // Cache is the byte backend. Required — a CachedWalker without + // a backend is a contradiction in terms. Pass nil to opt out: + // callers that want the same code path with caching disabled + // for a request (e.g. preview) toggle via NewWalker. + Cache CacheBackend + + // Version is the cache-key salt that lets an operator purge + // every cached render at once by bumping the value (typically + // the build SHA or a release tag). Two builds with the same + // renderer code share the cache; two builds with different + // rendering behaviour MUST set different values or one's + // output will be served by the other. + Version string + + // TTL is the time-to-live for stored entries. Zero means + // DefaultCacheTTL. Callers tuning a hot path may go higher; + // going lower than 1 minute defeats the purpose (the round- + // trip cost dominates the regenerate cost). + TTL time.Duration + + // ExtraTags is appended to every cached entry. The typical + // use is a per-tenant or per-site tag so an operator can + // purge a single tenant's renders without touching others'. + // Empty by default. + ExtraTags []string + + // Logger receives the structured cache-event output. Nil + // falls back to slog.Default. + Logger *slog.Logger +} + +// Metrics is the cumulative cache-event tally for one CachedWalker. +// All four counters are read concurrently safely. +// +// Hit-rate is the field issue #108's acceptance criteria measure +// against (target: 80%+). Operators read these via Snapshot. +type Metrics struct { + hits atomic.Uint64 + misses atomic.Uint64 + stores atomic.Uint64 + bypasses atomic.Uint64 +} + +// MetricsSnapshot is the value type the public Metrics() returns. +// Fields are uint64 (no pointers) so a snapshot is cheap to copy and +// safe to log straight into slog without atomic semantics leaking. +type MetricsSnapshot struct { + // Hits is the number of cached lookups that returned a fresh + // value. + Hits uint64 + // Misses is the number of cached lookups that returned no value + // (key absent OR captured tag-version mismatch). + Misses uint64 + // Stores is the number of Set calls completed successfully + // (one per miss whose render produced cacheable HTML). + Stores uint64 + // Bypasses is the number of blocks that were rendered without + // consulting the cache at all. Blocks bypass caching when they + // consume context (the cache key cannot capture inherited + // values) or fail the cacheability heuristic. + Bypasses uint64 +} + +// HitRate returns the fraction of cache lookups that produced a +// fresh value, in the [0, 1] range. Returns 0 when no lookups have +// occurred yet — undefined would be a worse contract for log lines. +func (m MetricsSnapshot) HitRate() float64 { + total := m.Hits + m.Misses + if total == 0 { + return 0 + } + return float64(m.Hits) / float64(total) +} + +// CachedWalker is a Walker whose subtree renders are memoised through +// a CacheBackend. +// +// The cache is keyed on the (block type, deep content hash, version) +// triple — the same render under two posts of the same shape shares +// a cache entry. The hash covers the block's Attributes and the full +// InnerBlocks subtree, so a child change correctly busts every +// ancestor that included it. +// +// Blocks that declare UsesContext bypass caching entirely. Caching +// a context-consuming block would require including the consumed +// values in the key, which in turn requires walking the inherited +// context at lookup time — and at that point the cost has eaten +// most of the win. +type CachedWalker struct { + walker *Walker + cache CacheBackend + version string + ttl time.Duration + extraTags []string + logger *slog.Logger + metrics Metrics +} + +// NewCached constructs a CachedWalker bound to the given registry +// and cache. A nil registry panics for the same reason render.New +// panics — a walker without a dispatch table can't produce output. +// A nil opts.Cache is rejected by an error rather than a panic so +// callers can wire conditionally (preview requests skip caching by +// passing an explicit nil). +func NewCached(reg *Registry, opts CachedWalkerOptions) (*CachedWalker, error) { + if reg == nil { + panic("render.NewCached: registry is nil") + } + if opts.Cache == nil { + return nil, fmt.Errorf("render: NewCached: opts.Cache is required") + } + if opts.TTL <= 0 { + opts.TTL = DefaultCacheTTL + } + if opts.Logger == nil { + opts.Logger = slog.Default() + } + return &CachedWalker{ + walker: New(reg), + cache: opts.Cache, + version: opts.Version, + ttl: opts.TTL, + extraTags: append([]string(nil), opts.ExtraTags...), + logger: opts.Logger, + }, nil +} + +// Walk renders the tree against the cache. The output is identical +// to Walker.Walk's — same HTML, same errors — but each cacheable +// subtree is looked up before invocation and stored after. +// +// The cache layer is per-subtree, not per-root: every block whose +// type does NOT declare UsesContext is a candidate. This is what +// gives the 80%+ hit rate on repeated renders — the post-level cache +// would miss anytime one paragraph changes, but the block-level +// cache keeps the rest of the post hot. +func (cw *CachedWalker) Walk(ctx context.Context, tree BlockTree, blockCtx Context) WalkResult { + res := WalkResult{} + if blockCtx == nil { + blockCtx = Context{} + } + var html strings.Builder + for i, block := range tree { + path := fmt.Sprintf("/%d", i) + out, errs := cw.walkBlockCached(ctx, block, blockCtx, path) + html.WriteString(string(out)) + res.Errors = append(res.Errors, errs...) + } + res.HTML = template.HTML(html.String()) + return res +} + +// Metrics returns a snapshot of the cumulative cache-event counters. +// The snapshot is taken at non-monotonic precision (each field is +// read independently), which is good enough for hit-rate dashboards +// — a Hits/Misses ratio that briefly lags by one event under load +// is not worth a coarser lock for. +func (cw *CachedWalker) Metrics() MetricsSnapshot { + return MetricsSnapshot{ + Hits: cw.metrics.hits.Load(), + Misses: cw.metrics.misses.Load(), + Stores: cw.metrics.stores.Load(), + Bypasses: cw.metrics.bypasses.Load(), + } +} + +// walkBlockCached looks the block up in the cache; on miss it falls +// through to the underlying Walker and writes back the rendered HTML. +// +// Errors found inside a cached subtree are NOT cached — only the +// HTML output is. A cached error would be surprising on retry (the +// renderer typically fixes itself after a transient failure), and +// the error wiring already has a "log and degrade" path for repeat +// problems. +func (cw *CachedWalker) walkBlockCached( + ctx context.Context, + block Block, + inherited Context, + path string, +) (template.HTML, []WalkError) { + spec, ok := cw.walker.registry.Get(block.Type) + if !ok { + // Unknown blocks reuse the underlying walker's placeholder + // machinery. They don't go through the cache because the + // placeholder is essentially free and the cache would just + // burn a Redis round-trip. + return cw.walker.walkBlock(block, inherited, path) + } + if !cw.isCacheable(spec) { + cw.metrics.bypasses.Add(1) + return cw.walker.walkBlock(block, inherited, path) + } + + key := cw.cacheKey(block) + tags := cw.cacheTags(block) + + cached, hit, err := cw.cache.Get(ctx, key, tags) + if err != nil { + // Transport errors are treated as misses: a brief Redis + // outage degrades the cache to a passthrough, not a hard + // failure. + cw.logger.Debug("render cache: get error, falling through", + slog.String("key", key), + slog.Any("err", err)) + } + if hit { + cw.metrics.hits.Add(1) + return template.HTML(cached), nil + } + cw.metrics.misses.Add(1) + + out, errs := cw.walker.walkBlock(block, inherited, path) + if len(errs) > 0 { + // Don't cache subtrees that produced errors — see method + // docstring. + return out, errs + } + if err := cw.cache.Set(ctx, key, []byte(out), tags, cw.ttl); err != nil { + cw.logger.Warn("render cache: set error", + slog.String("key", key), + slog.Any("err", err)) + } else { + cw.metrics.stores.Add(1) + } + return out, errs +} + +// isCacheable answers "can this block's render be safely cached +// without including the inherited context in the key?" +// +// Two rules: +// +// 1. The block must not declare UsesContext. A consumer of context +// is a block whose output depends on ancestor state we can't +// easily fold into the key. +// +// 2. The block must not provide context to its descendants. A +// provider whose attributes change has to invalidate every +// descendant that captured the old value, and the cache layer +// doesn't track that dependency graph. +// +// Both rules are conservative — many real-world blocks neither +// provide nor consume context, and those are exactly the ones a +// post repeats over and over (headings, paragraphs, columns, +// images, lists). The 80%+ hit rate target is met because those +// blocks dominate the long-tail of post content. +func (cw *CachedWalker) isCacheable(spec BlockSpec) bool { + return len(spec.UsesContext) == 0 && len(spec.ProvidesContext) == 0 +} + +// cacheKey computes the cache key for a block subtree. +// +// Key shape: "br:::" +// +// - The leading prefix lets a dashboard route or count by namespace. +// - The version segment is the operator-controlled cache buster. +// - block.type is included verbatim so a key reads usefully in +// `redis-cli MONITOR`. +// - The sha256 hash covers the canonical-JSON encoding of the block +// (Attributes + InnerBlocks recursively). Two blocks with the same +// on-wire shape collide on the key — that is the WHOLE point. +// +// The hash uses canonical JSON: sorted attribute keys, no whitespace. +// We do not use the post's pre-existing content_blocks_hash because +// that column hashes the whole post; we need a per-subtree hash. +func (cw *CachedWalker) cacheKey(block Block) string { + canonical := canonicalEncode(block) + sum := sha256.Sum256(canonical) + return fmt.Sprintf("%s%s:%s:%s", + KeyPrefix, cw.version, block.Type, hex.EncodeToString(sum[:])) +} + +// cacheTags computes the tag set the cache should record for one +// block. The base tag is "br:type:" so an operator can +// purge every render of a single block type (e.g. when a renderer +// has a bug fix); the configured ExtraTags are appended. +// +// We do NOT add a per-block-instance tag — the cache key is already +// content-addressable, so a content change naturally produces a +// fresh key and the old one ages out via TTL. +func (cw *CachedWalker) cacheTags(block Block) []string { + tags := make([]string, 0, 1+len(cw.extraTags)) + tags = append(tags, "br:type:"+block.Type) + tags = append(tags, cw.extraTags...) + return tags +} + +// canonicalEncode produces a deterministic byte serialisation of a +// block subtree for hashing. The rules are: +// +// - JSON-encode with sorted attribute keys (sortedMap is the +// hand-written serializer). +// - InnerBlocks are encoded recursively. +// - ClientID is intentionally omitted — it's editor-state and not +// part of the on-wire content shape. +// +// We do not use encoding/json's Marshal directly because Go's +// map iteration order is randomized; two encodings of the same +// attributes would produce different bytes and miss the cache. +// +// Errors are degraded into a synthetic byte sequence so an +// undecodable attribute doesn't crash the walker. A degraded hash +// will simply produce a unique key per attempt, defeating caching +// for that one block — preferable to a panic. +func canonicalEncode(block Block) []byte { + var b strings.Builder + b.WriteString(`{"t":`) + jsonString(&b, block.Type) + b.WriteString(`,"a":`) + writeSortedValue(&b, block.Attributes) + if len(block.InnerBlocks) > 0 { + b.WriteString(`,"i":[`) + for i, child := range block.InnerBlocks { + if i > 0 { + b.WriteByte(',') + } + b.Write(canonicalEncode(child)) + } + b.WriteByte(']') + } + b.WriteByte('}') + return []byte(b.String()) +} + +// writeSortedValue is the recursive heart of canonicalEncode: maps +// are emitted with keys in ascending order, slices in source order, +// everything else falls through to encoding/json. +func writeSortedValue(b *strings.Builder, v any) { + switch t := v.(type) { + case nil: + b.WriteString("null") + case map[string]any: + keys := make([]string, 0, len(t)) + for k := range t { + keys = append(keys, k) + } + sort.Strings(keys) + b.WriteByte('{') + for i, k := range keys { + if i > 0 { + b.WriteByte(',') + } + jsonString(b, k) + b.WriteByte(':') + writeSortedValue(b, t[k]) + } + b.WriteByte('}') + case []any: + b.WriteByte('[') + for i, item := range t { + if i > 0 { + b.WriteByte(',') + } + writeSortedValue(b, item) + } + b.WriteByte(']') + default: + // Numbers, strings, booleans, nested concrete types — let + // encoding/json handle the formatting. We only need + // determinism for maps and slices. + out, err := json.Marshal(t) + if err != nil { + // Sentinel value so a hash failure produces a + // deterministic but distinct key per (block, time). + fmt.Fprintf(b, `"err:%s"`, err.Error()) + return + } + b.Write(out) + } +} + +// jsonString writes a JSON-quoted string. Defers to encoding/json for +// escape correctness — it's the simplest way to get U+2028, control +// chars, and quote escaping right. +func jsonString(b *strings.Builder, s string) { + out, _ := json.Marshal(s) + b.Write(out) +} diff --git a/packages/go/blocks/render/cache_test.go b/packages/go/blocks/render/cache_test.go new file mode 100644 index 00000000..073eef9c --- /dev/null +++ b/packages/go/blocks/render/cache_test.go @@ -0,0 +1,260 @@ +package render + +import ( + "context" + "html/template" + "sync" + "testing" + "time" +) + +// inMemoryBackend is the test stand-in for the production fragment +// cache. It is intentionally NOT version-aware (no tag invalidation): +// the CachedWalker is the unit under test here, and we want every +// missed expectation to be the walker's fault, not a tag glitch. +type inMemoryBackend struct { + mu sync.Mutex + items map[string][]byte + getHits int + getMiss int + sets int +} + +func newInMemoryBackend() *inMemoryBackend { + return &inMemoryBackend{items: make(map[string][]byte)} +} + +func (b *inMemoryBackend) Get(_ context.Context, key string, _ []string) ([]byte, bool, error) { + b.mu.Lock() + defer b.mu.Unlock() + v, ok := b.items[key] + if !ok { + b.getMiss++ + return nil, false, nil + } + b.getHits++ + return v, true, nil +} + +func (b *inMemoryBackend) Set(_ context.Context, key string, value []byte, _ []string, _ time.Duration) error { + b.mu.Lock() + defer b.mu.Unlock() + cp := make([]byte, len(value)) + copy(cp, value) + b.items[key] = cp + b.sets++ + return nil +} + +// newCachedTestWalker is a small helper that mounts the core block +// renderers and wraps them with the cache backend. +func newCachedTestWalker(t *testing.T, backend CacheBackend) *CachedWalker { + t.Helper() + reg := NewRegistry() + if err := RegisterCoreBlocks(reg); err != nil { + t.Fatalf("RegisterCoreBlocks: %v", err) + } + cw, err := NewCached(reg, CachedWalkerOptions{ + Cache: backend, + Version: "test-v1", + }) + if err != nil { + t.Fatalf("NewCached: %v", err) + } + return cw +} + +// TestCachedWalker_RepeatedRender_HitsCache exercises the headline +// acceptance criterion of issue #108: the second render of an +// identical block subtree hits the cache. +func TestCachedWalker_RepeatedRender_HitsCache(t *testing.T) { + t.Parallel() + backend := newInMemoryBackend() + cw := newCachedTestWalker(t, backend) + ctx := context.Background() + + tree := []Block{ + { + Type: "core/paragraph", + Attributes: map[string]any{ + "content": "hello world", + "align": "center", + }, + }, + } + + first := cw.Walk(ctx, tree, nil) + second := cw.Walk(ctx, tree, nil) + + if first.HTML != second.HTML { + t.Errorf("renders disagree:\n first: %q\n second: %q", first.HTML, second.HTML) + } + m := cw.Metrics() + if m.Hits != 1 { + t.Errorf("Hits: got %d, want 1", m.Hits) + } + if m.Misses != 1 { + t.Errorf("Misses: got %d, want 1", m.Misses) + } + if m.Stores != 1 { + t.Errorf("Stores: got %d, want 1", m.Stores) + } +} + +// TestCachedWalker_HitRateTarget renders the same tree of 10 blocks +// several times and asserts a hit rate ≥ 80% after warmup — the +// issue's acceptance criterion. +func TestCachedWalker_HitRateTarget(t *testing.T) { + t.Parallel() + backend := newInMemoryBackend() + cw := newCachedTestWalker(t, backend) + ctx := context.Background() + + tree := make([]Block, 10) + for i := range tree { + tree[i] = Block{ + Type: "core/paragraph", + Attributes: map[string]any{ + "content": "para", + "i": float64(i), + }, + } + } + + // Six passes: 10 misses + 50 hits = 83% hit rate. + for i := 0; i < 6; i++ { + cw.Walk(ctx, tree, nil) + } + + m := cw.Metrics() + if rate := m.HitRate(); rate < 0.8 { + t.Errorf("HitRate after warmup: got %.4f (hits=%d misses=%d), want >= 0.80", + rate, m.Hits, m.Misses) + } +} + +// TestCachedWalker_BypassesContextBlocks confirms that blocks +// declaring UsesContext or ProvidesContext skip the cache entirely. +// Caching a context-coupled block would require including inherited +// state in the key; the current design takes the safer fallback. +func TestCachedWalker_BypassesContextBlocks(t *testing.T) { + t.Parallel() + backend := newInMemoryBackend() + + reg := NewRegistry() + if err := reg.Register("test/uses-ctx", BlockSpec{ + Render: func(_ Block, _ template.HTML, _ Context) (template.HTML, error) { + return template.HTML("used"), nil + }, + UsesContext: []string{"postId"}, + }); err != nil { + t.Fatalf("Register test/uses-ctx: %v", err) + } + cw, err := NewCached(reg, CachedWalkerOptions{ + Cache: backend, + Version: "test-v1", + }) + if err != nil { + t.Fatalf("NewCached: %v", err) + } + + tree := []Block{{Type: "test/uses-ctx", Attributes: map[string]any{}}} + for i := 0; i < 3; i++ { + cw.Walk(context.Background(), tree, nil) + } + m := cw.Metrics() + if m.Bypasses != 3 { + t.Errorf("Bypasses: got %d, want 3", m.Bypasses) + } + if m.Hits != 0 || m.Misses != 0 { + t.Errorf("cache should not have been touched: hits=%d misses=%d", m.Hits, m.Misses) + } +} + +// TestCanonicalEncode_KeyStabilityAcrossMapOrder pins the key- +// stability property: two blocks built from the same attributes but +// in different map-insertion order MUST produce the same cache key. +// This is the bug Go's randomized map iteration would silently +// reintroduce if canonicalEncode regressed. +func TestCanonicalEncode_KeyStabilityAcrossMapOrder(t *testing.T) { + t.Parallel() + a := Block{ + Type: "core/paragraph", + Attributes: map[string]any{ + "content": "x", + "align": "left", + "dropCap": true, + }, + } + b := Block{ + Type: "core/paragraph", + Attributes: map[string]any{ + "dropCap": true, + "content": "x", + "align": "left", + }, + } + if string(canonicalEncode(a)) != string(canonicalEncode(b)) { + t.Errorf("canonicalEncode disagrees on equal attribute sets:\n a: %s\n b: %s", + canonicalEncode(a), canonicalEncode(b)) + } +} + +// TestCachedWalker_DifferentAttrs_DifferentKey covers the inverse of +// the key-stability test: a one-attribute change MUST produce a +// different cache key (otherwise the cache would serve stale HTML). +func TestCachedWalker_DifferentAttrs_DifferentKey(t *testing.T) { + t.Parallel() + backend := newInMemoryBackend() + cw := newCachedTestWalker(t, backend) + ctx := context.Background() + + a := []Block{{Type: "core/paragraph", Attributes: map[string]any{"content": "a"}}} + b := []Block{{Type: "core/paragraph", Attributes: map[string]any{"content": "b"}}} + + cw.Walk(ctx, a, nil) + cw.Walk(ctx, b, nil) + + m := cw.Metrics() + if m.Hits != 0 { + t.Errorf("Hits: got %d, want 0 (different content must miss)", m.Hits) + } + if m.Misses != 2 { + t.Errorf("Misses: got %d, want 2", m.Misses) + } +} + +// TestCachedWalker_NilCacheRejected pins the constructor's contract: +// a CachedWalker without a backend is a programming error. +func TestCachedWalker_NilCacheRejected(t *testing.T) { + t.Parallel() + reg := NewRegistry() + if _, err := NewCached(reg, CachedWalkerOptions{}); err == nil { + t.Fatal("expected error for nil Cache") + } +} + +// TestCachedWalker_VersionBumpInvalidates verifies that two walkers +// constructed with different Version strings see independent caches +// — the operator-controlled cache-buster. +func TestCachedWalker_VersionBumpInvalidates(t *testing.T) { + t.Parallel() + backend := newInMemoryBackend() + + reg := NewRegistry() + if err := RegisterCoreBlocks(reg); err != nil { + t.Fatalf("RegisterCoreBlocks: %v", err) + } + cwV1, _ := NewCached(reg, CachedWalkerOptions{Cache: backend, Version: "v1"}) + cwV2, _ := NewCached(reg, CachedWalkerOptions{Cache: backend, Version: "v2"}) + + tree := []Block{{Type: "core/paragraph", Attributes: map[string]any{"content": "x"}}} + cwV1.Walk(context.Background(), tree, nil) + cwV2.Walk(context.Background(), tree, nil) + + // Each walker should have populated its own key — two stores, + // zero hits. + if backend.sets != 2 { + t.Errorf("sets across versions: got %d, want 2", backend.sets) + } +} diff --git a/packages/go/cache/fragment/doc.go b/packages/go/cache/fragment/doc.go new file mode 100644 index 00000000..5535f1c9 --- /dev/null +++ b/packages/go/cache/fragment/doc.go @@ -0,0 +1,87 @@ +// Package fragment is a Redis-backed byte-fragment cache with +// tag-based invalidation that piggy-backs on the cache_invalidations +// outbox shipped by migration 000030. +// +// Why a fragment cache (vs. the existing KV ABI) +// +// The plugin-facing KV ABI in packages/go/plugins/runtime/host_data.go +// is namespaced per plugin, quota-tracked, and audit-emitted on every +// write. Those properties are right for plugin storage but wrong for +// internal render memoisation: the block render walker (#108) wants +// a single shared cache pool keyed by content hash, no quotas, and no +// audit row per hit. Fragment is that pool. +// +// Why tags instead of key fanout +// +// A typical render reads from many keys ("block:hero:v3", "menu:main", +// "site-settings:colors"). Invalidating "all renders that touched the +// main menu" by enumerating every dependent key would force the +// producer side to remember an N:M reverse index. Tags collapse that +// into a single PURGE TAG ('menu') message: every fragment indexes its +// own tag set on write, and the worker bumps a per-tag version on +// invalidate. A fragment is fresh iff every one of its tag versions +// still matches the values it captured at write time. +// +// # Storage layout +// +// Redis keys are namespaced under "gnf:" to avoid colliding with the +// plugin KV pool (which uses "plugin::"). +// +// gnf:f: -> the cached payload (bytes) + the captured +// tag version vector encoded as the first +// few bytes of the value (see encodeEntry). +// gnf:tv: -> int64, monotonically incremented each time +// the tag is invalidated. Lazily created on +// first read with a value of 0. +// +// The version-vector approach is the same one used by Mnesia / +// transactional caches: cheap reads (one MGET against the tag-version +// keys plus one GET on the payload), invalidation is O(1) (a single +// INCR per tag), and there's no key-fanout problem on the producer +// side. +// +// # Invalidation flow +// +// Set(ctx, key, value, tags, ttl) +// → MGET each tv: (lazy-creates them at 0) +// → encode (versions, value) into a single Redis value +// → SET gnf:f: EX ttl +// +// Get(ctx, key, tags) +// → GET gnf:f: +// → decode (capturedVersions, value) +// → MGET each current tv: +// → if every captured == current, return (value, true, nil) +// → otherwise return (nil, false, nil) — the cache itself never +// deletes; the next Set overwrites with a fresh capture, and +// Redis' EX handles eviction. +// +// Purge(ctx, tags) +// → For each tag, write a row into cache_invalidations. The +// invalidator worker (packages/go/cache/invalidator) drains +// that table, INCRs gnf:tv:, and publishes a pub/sub +// message. After the worker drains, any in-flight Get whose +// captured version disagrees returns a miss. +// +// Tags written to the outbox are stored UNPREFIXED by the worker +// convention (see invalidator.go). The fragment cache writes its +// own internal "gnf" prefix on the pub/sub subscriber side so the +// version keys it manages don't collide with the plugin-KV namespace. +// +// # Concurrency and consistency +// +// Get is a single round-trip in the steady state (payload GET) and a +// second round-trip on a hit candidate (MGET tag versions). Stale +// reads are bounded by the outbox poll cadence (default ~100ms) plus +// the time it takes the worker to PUBLISH and INCR — typically under +// 200ms for a single invalidation. The cache is designed for content +// that tolerates that window (block renders, sitemap fragments, +// menu HTML); it is NOT a source-of-truth store. +// +// # Why no Delete +// +// Direct Delete(key) is intentionally absent. The tag mechanism is +// the only invalidation surface so authors don't end up with two +// parallel paths to remember. Code that wants to drop a single +// fragment uses a tag whose only fragment is that one ("block:"). +package fragment diff --git a/packages/go/cache/fragment/fragment.go b/packages/go/cache/fragment/fragment.go new file mode 100644 index 00000000..fc809156 --- /dev/null +++ b/packages/go/cache/fragment/fragment.go @@ -0,0 +1,422 @@ +package fragment + +import ( + "context" + "encoding/binary" + "errors" + "fmt" + "log/slog" + "strings" + "time" + + "github.com/redis/go-redis/v9" +) + +// KeyPrefix is the Redis key prefix used for cached fragment payloads. +// Exported so subscribers (e.g. the block-render cache invalidation +// listener) can pattern-match without re-deriving the constant. +const KeyPrefix = "gnf:f:" + +// TagVersionPrefix is the Redis key prefix used for per-tag version +// counters. INCRing a key under this prefix is what makes every +// fragment that captured the old version stale on its next Get. +const TagVersionPrefix = "gnf:tv:" + +// DefaultTTL is the time-to-live applied when Set is called with a +// non-positive ttl. We don't let callers store fragments forever — +// even with tag invalidation, an orphaned fragment (one whose tags +// no one is watching anymore) would sit in Redis indefinitely. One +// hour is a balance between "long enough to be useful" and "short +// enough that a forgotten tag drains out on its own". +const DefaultTTL = 1 * time.Hour + +// MaxTagsPerEntry caps how many tags one Set may attach to a single +// fragment. The cap is defensive: every tag costs a Redis round-trip +// on Set (to read its current version) and on Get (to check whether +// the captured version still matches). 32 is comfortably larger than +// any of the project's documented use cases (a render touches a +// handful of taxonomies and the site-settings tag) and small enough +// that a misuse — a caller that fans out per-row tags — fails loudly. +const MaxTagsPerEntry = 32 + +// MaxValueBytes caps the payload size accepted by Set. Beyond ~1 MiB +// the Redis network round-trip stops being interesting (the renderer +// would do better to stream the source data directly), and a single +// 100 MiB blob can pin a Redis instance. 1 MiB matches the documented +// HTTP response budget for the public web. +const MaxValueBytes = 1 << 20 + +// ErrValueTooLarge is returned by Set when the supplied value exceeds +// MaxValueBytes. The caller should fall back to a non-cached render. +var ErrValueTooLarge = errors.New("fragment: value exceeds MaxValueBytes") + +// ErrTooManyTags is returned by Set when the supplied tag list +// exceeds MaxTagsPerEntry. The caller should rethink its tag shape: +// per-row tags are an anti-pattern in this cache. +var ErrTooManyTags = errors.New("fragment: tag count exceeds MaxTagsPerEntry") + +// Cache is the byte-fragment cache surface. One Cache instance binds +// to one Redis client and one outbox-writer; callers share the same +// instance across goroutines (Redis client and pgxpool are both +// concurrency-safe). +type Cache struct { + rdb *redis.Client + outbox OutboxWriter + logger *slog.Logger + keyPrefix string + tvPrefix string +} + +// OutboxWriter is the dependency Purge uses to record a tag +// invalidation. The concrete implementation in production is a thin +// pgxpool-backed inserter into cache_invalidations (see Writer in +// this package); tests pass a fake that records calls in memory. +// +// Keeping this an interface (rather than depending on pgxpool here) +// means the fragment package compiles without a Postgres driver — +// useful for unit tests, and matches the layering rule in +// packages/go/cache/invalidator (the worker imports pgx, the cache +// itself does not need to). +type OutboxWriter interface { + // WriteInvalidations appends one row per tag into the + // cache_invalidations outbox. The slug is the namespace the + // invalidator worker re-prefixes when it publishes; for + // fragment cache traffic the slug is always "gnf" so a + // subscriber can route on a stable namespace. + // + // Returning an error from WriteInvalidations causes Purge to + // return that error — the row is the durable record of the + // invalidation, so a failed write is a real failure (the cache + // has not been purged) and not a "best effort". + WriteInvalidations(ctx context.Context, slug string, tags []string) error +} + +// Option configures a Cache at construction time. +type Option func(*Cache) + +// WithLogger swaps the structured logger. +func WithLogger(l *slog.Logger) Option { + return func(c *Cache) { + if l != nil { + c.logger = l + } + } +} + +// WithKeyPrefix overrides the Redis namespace used for payload keys. +// Mostly useful for tests that want to share a Redis instance across +// suites without colliding. +func WithKeyPrefix(prefix string) Option { + return func(c *Cache) { + if prefix != "" { + c.keyPrefix = prefix + } + } +} + +// WithTagVersionPrefix overrides the Redis namespace used for tag +// version counters. Mirror of WithKeyPrefix; both must be passed +// together when isolating a test from production keyspace. +func WithTagVersionPrefix(prefix string) Option { + return func(c *Cache) { + if prefix != "" { + c.tvPrefix = prefix + } + } +} + +// New constructs a Cache. +// +// rdb is required; passing nil panics — a fragment cache without a +// Redis client would silently no-op every Set and miss every Get, +// and that is more dangerous than a startup crash. +// +// outbox may be nil. A nil outbox means Purge returns an error +// (the cache still serves Gets and Sets); this is the right shape +// for "read-only" embeddings (an integration test that wants the +// hit/miss behaviour without spinning up Postgres). +func New(rdb *redis.Client, outbox OutboxWriter, opts ...Option) *Cache { + if rdb == nil { + panic("fragment.New: redis client is required") + } + c := &Cache{ + rdb: rdb, + outbox: outbox, + logger: slog.Default(), + keyPrefix: KeyPrefix, + tvPrefix: TagVersionPrefix, + } + for _, o := range opts { + o(c) + } + return c +} + +// Get looks up a cached fragment by key and returns its bytes when +// every captured tag version still matches the live version. +// +// The (value, true, nil) return is a fresh hit. A (nil, false, nil) +// return is a clean miss — either the key was absent, the entry was +// malformed (treated as a miss, not an error), or one of the captured +// tags has been invalidated since the Set. A non-nil error means +// Redis itself failed; the caller should fall back to a non-cached +// render rather than treat it as a miss (so a brief Redis outage +// doesn't quietly bypass the cache during a stampede). +// +// The tags argument is the SAME list the caller would pass to Set — +// Get re-checks the live versions against the captured ones stored +// alongside the payload. Mismatched tag-set between Set and Get +// produces a miss (we cannot prove the captured payload was built +// from the same dependencies). This lets a caller defensively grow +// or shrink its tag set across deploys without a poisoning hazard. +func (c *Cache) Get(ctx context.Context, key string, tags []string) ([]byte, bool, error) { + if key == "" { + return nil, false, errors.New("fragment: Get: key is required") + } + raw, err := c.rdb.Get(ctx, c.keyPrefix+key).Bytes() + if err != nil { + if errors.Is(err, redis.Nil) { + return nil, false, nil + } + return nil, false, fmt.Errorf("fragment: Get %q: %w", key, err) + } + captured, value, ok := decodeEntry(raw) + if !ok { + // Malformed entries can happen if a server with an older + // schema wrote a key we now don't understand. Treat as a + // miss; the next Set will rewrite the value with a current + // schema. We log at debug because the situation is benign + // during deploys but worth seeing under a magnifier. + c.logger.Debug("fragment: malformed cached entry, treating as miss", + slog.String("key", key)) + return nil, false, nil + } + if len(captured) != len(tags) { + // Tag-set drift: the writer believed the value depended on + // N tags; we now think it depends on M. We cannot prove + // freshness, so we treat the entry as stale. + return nil, false, nil + } + if len(tags) == 0 { + // No tags to validate — the value is fresh by construction. + return value, true, nil + } + live, err := c.readTagVersions(ctx, tags) + if err != nil { + return nil, false, fmt.Errorf("fragment: read tag versions: %w", err) + } + for i := range tags { + if captured[i] != live[i] { + return nil, false, nil + } + } + return value, true, nil +} + +// Set stores value under key with the given tags and ttl. The set of +// captured tag versions is encoded into the stored entry so a later +// Get can compare against the live versions and detect invalidation. +// +// A zero or negative ttl is replaced with DefaultTTL — fragments are +// never stored indefinitely (see the package comment). +// +// Set is best-effort durable: a Redis error is returned to the +// caller; on success, the entry is visible to subsequent Gets within +// one round-trip. Set does NOT update the cache_invalidations outbox +// — that is exclusively a Purge concern. +func (c *Cache) Set(ctx context.Context, key string, value []byte, tags []string, ttl time.Duration) error { + if key == "" { + return errors.New("fragment: Set: key is required") + } + if len(value) > MaxValueBytes { + return fmt.Errorf("%w: %d > %d", ErrValueTooLarge, len(value), MaxValueBytes) + } + if len(tags) > MaxTagsPerEntry { + return fmt.Errorf("%w: %d > %d", ErrTooManyTags, len(tags), MaxTagsPerEntry) + } + if ttl <= 0 { + ttl = DefaultTTL + } + versions, err := c.readTagVersions(ctx, tags) + if err != nil { + return fmt.Errorf("fragment: Set: read tag versions: %w", err) + } + encoded := encodeEntry(versions, value) + if err := c.rdb.Set(ctx, c.keyPrefix+key, encoded, ttl).Err(); err != nil { + return fmt.Errorf("fragment: Set %q: %w", key, err) + } + return nil +} + +// Purge appends one row per tag into the cache_invalidations outbox. +// The invalidator worker drains the table, INCRs gnf:tv:, and +// publishes a pub/sub message. Subsequent Gets whose captured tag +// version disagrees will return a miss. +// +// Purge is the ONLY supported way to evict fragments. A caller that +// wants to drop a single fragment should give it a tag whose only +// member is that fragment ("block:") and Purge by that tag. +// +// Purge returns an error when no outbox writer is configured: a +// silent no-op here would be a real-world data-corruption bug (the +// caller would believe the invalidation succeeded). The expected +// production wiring always supplies an OutboxWriter; tests that +// want the no-write behaviour pass a recording fake instead of nil. +func (c *Cache) Purge(ctx context.Context, tags []string) error { + if c.outbox == nil { + return errors.New("fragment: Purge: no outbox writer configured") + } + if len(tags) == 0 { + return nil + } + // Filter empties and de-duplicate. An empty tag would invalidate + // the "no tags" entry (which has no version to bump) and is the + // kind of typo we want to drop quietly rather than amplify. + seen := make(map[string]struct{}, len(tags)) + filtered := make([]string, 0, len(tags)) + for _, t := range tags { + t = strings.TrimSpace(t) + if t == "" { + continue + } + if _, dup := seen[t]; dup { + continue + } + seen[t] = struct{}{} + filtered = append(filtered, t) + } + if len(filtered) == 0 { + return nil + } + return c.outbox.WriteInvalidations(ctx, "gnf", filtered) +} + +// ApplyInvalidation increments the version counter for one tag. This +// is the receiver side of the pub/sub message the invalidator worker +// publishes — when a Cache is wired into a pub/sub subscriber loop, +// each message triggers one call here. +// +// Idempotent in the "downstream impact" sense: a duplicate INCR moves +// the counter forward by two instead of one, but every captured- +// version comparison still resolves to "not equal", so the cache +// surface behaviour is identical. This is what makes at-least-once +// delivery from the invalidator safe to consume. +func (c *Cache) ApplyInvalidation(ctx context.Context, tag string) error { + tag = strings.TrimSpace(tag) + if tag == "" { + return nil + } + if err := c.rdb.Incr(ctx, c.tvPrefix+tag).Err(); err != nil { + return fmt.Errorf("fragment: ApplyInvalidation %q: %w", tag, err) + } + return nil +} + +// readTagVersions returns the current version int64 for each tag in +// the input slice. Missing keys read as 0 (the lazy-initialised +// value); Redis' INCR creates the key on first write, so we don't +// pre-seed. +// +// Uses a single pipelined MGET to keep the round-trip count at one +// regardless of tag-set size. +func (c *Cache) readTagVersions(ctx context.Context, tags []string) ([]int64, error) { + if len(tags) == 0 { + return nil, nil + } + keys := make([]string, len(tags)) + for i, t := range tags { + keys[i] = c.tvPrefix + t + } + vals, err := c.rdb.MGet(ctx, keys...).Result() + if err != nil { + return nil, err + } + out := make([]int64, len(tags)) + for i, v := range vals { + switch s := v.(type) { + case nil: + out[i] = 0 + case string: + // go-redis returns numbers as strings out of MGET; we + // parse manually instead of pulling strconv to keep + // the hot path tight. The version counter is the + // product of INCR and is always a base-10 integer + // representable in int64. + var n int64 + for j := 0; j < len(s); j++ { + c := s[j] + if c < '0' || c > '9' { + // Corrupted counter: treat as 0 so a stale + // payload re-syncs on next Set. We don't + // short-circuit to error because that would + // poison the whole cache on one bad key. + n = 0 + break + } + n = n*10 + int64(c-'0') + } + out[i] = n + default: + out[i] = 0 + } + } + return out, nil +} + +// encodeEntry packs the captured tag versions and the payload into a +// single byte slice. Layout: +// +// [ uint32 LE: tag count N ] +// [ N × int64 LE: captured tag versions ] +// [ value bytes ] +// +// Little-endian was chosen for fast decode on x86 / arm64; the encode +// is one make + N+1 PutUint64 calls, and the decode is symmetric. +// Versioning the format is future-proofed via the first byte: a +// caller reading a malformed entry treats it as a cache miss (see +// decodeEntry), so a format bump only needs to ensure the new +// encoding's count word is different from any legal old one. We do +// not bother with an explicit format byte today because there is +// only one format. +func encodeEntry(versions []int64, value []byte) []byte { + const headerSize = 4 + const versionSize = 8 + out := make([]byte, headerSize+versionSize*len(versions)+len(value)) + binary.LittleEndian.PutUint32(out[0:headerSize], uint32(len(versions))) + for i, v := range versions { + off := headerSize + i*versionSize + binary.LittleEndian.PutUint64(out[off:off+versionSize], uint64(v)) + } + copy(out[headerSize+versionSize*len(versions):], value) + return out +} + +// decodeEntry is the inverse of encodeEntry. Returns the captured +// versions, the payload, and ok=true when the layout is well-formed. +// Layout violations (truncated header, declared count exceeds the +// remaining bytes) return ok=false; callers treat that as a miss. +func decodeEntry(raw []byte) (versions []int64, value []byte, ok bool) { + const headerSize = 4 + const versionSize = 8 + if len(raw) < headerSize { + return nil, nil, false + } + n := binary.LittleEndian.Uint32(raw[0:headerSize]) + // Guard against a malicious / corrupted header claiming a huge + // version count. MaxTagsPerEntry is the upper bound at write + // time so anything above it is a layout error. + if n > MaxTagsPerEntry { + return nil, nil, false + } + need := headerSize + int(n)*versionSize + if len(raw) < need { + return nil, nil, false + } + versions = make([]int64, n) + for i := uint32(0); i < n; i++ { + off := headerSize + int(i)*versionSize + versions[i] = int64(binary.LittleEndian.Uint64(raw[off : off+versionSize])) + } + value = raw[need:] + return versions, value, true +} diff --git a/packages/go/cache/fragment/fragment_test.go b/packages/go/cache/fragment/fragment_test.go new file mode 100644 index 00000000..133f672e --- /dev/null +++ b/packages/go/cache/fragment/fragment_test.go @@ -0,0 +1,211 @@ +package fragment + +import ( + "bytes" + "context" + "errors" + "testing" + "time" + + "github.com/alicebob/miniredis/v2" + "github.com/redis/go-redis/v9" +) + +// recordingOutbox is a test-only OutboxWriter that captures the +// (slug, tags) tuples handed to WriteInvalidations. The fragment +// package compiles without Postgres so we never hit the real outbox +// in unit tests — the worker integration test in invalidator/ +// covers the SQL side. +type recordingOutbox struct { + calls []recordedCall + err error +} + +type recordedCall struct { + slug string + tags []string +} + +func (r *recordingOutbox) WriteInvalidations(_ context.Context, slug string, tags []string) error { + r.calls = append(r.calls, recordedCall{slug: slug, tags: append([]string(nil), tags...)}) + return r.err +} + +// newTestCache spins up a miniredis-backed Cache + recording outbox +// for one test, registering t.Cleanup to release both. Returning the +// outbox lets the test inspect what Purge wrote. +func newTestCache(t *testing.T) (*Cache, *recordingOutbox) { + t.Helper() + mr := miniredis.RunT(t) + rdb := redis.NewClient(&redis.Options{Addr: mr.Addr()}) + t.Cleanup(func() { _ = rdb.Close() }) + out := &recordingOutbox{} + return New(rdb, out), out +} + +// TestSetGet_HitAndMiss covers the most common path: a Set followed +// by a Get returns the value (hit), and a Get of an unwritten key +// returns false (miss). +func TestSetGet_HitAndMiss(t *testing.T) { + t.Parallel() + c, _ := newTestCache(t) + ctx := context.Background() + + if err := c.Set(ctx, "k1", []byte("hello"), []string{"tag-a"}, time.Minute); err != nil { + t.Fatalf("Set: %v", err) + } + got, hit, err := c.Get(ctx, "k1", []string{"tag-a"}) + if err != nil || !hit || !bytes.Equal(got, []byte("hello")) { + t.Fatalf("Get hit: got=%q hit=%v err=%v", got, hit, err) + } + + if _, hit, err := c.Get(ctx, "absent", []string{"tag-a"}); err != nil || hit { + t.Fatalf("Get miss: hit=%v err=%v", hit, err) + } +} + +// TestApplyInvalidation_DropsCachedEntry walks the whole invalidation +// loop: Set, invalidate-the-tag (the same INCR the worker will do), +// re-Get → miss. This is the core contract — invalidation is what +// makes the cache useful. +func TestApplyInvalidation_DropsCachedEntry(t *testing.T) { + t.Parallel() + c, _ := newTestCache(t) + ctx := context.Background() + + tags := []string{"posts:42"} + if err := c.Set(ctx, "render:hero", []byte("HTML"), tags, time.Minute); err != nil { + t.Fatalf("Set: %v", err) + } + if _, hit, _ := c.Get(ctx, "render:hero", tags); !hit { + t.Fatal("expected hit before invalidation") + } + + if err := c.ApplyInvalidation(ctx, "posts:42"); err != nil { + t.Fatalf("ApplyInvalidation: %v", err) + } + if _, hit, _ := c.Get(ctx, "render:hero", tags); hit { + t.Fatal("expected miss after invalidation") + } +} + +// TestPurge_WritesOutbox confirms Purge routes through the outbox +// writer rather than calling INCR directly. The invalidator worker is +// what actually fans the INCR out (so a multi-process deployment +// stays consistent); Purge must NOT short-circuit. +func TestPurge_WritesOutbox(t *testing.T) { + t.Parallel() + c, out := newTestCache(t) + ctx := context.Background() + + if err := c.Purge(ctx, []string{"posts:42", "sitemap", "", "posts:42"}); err != nil { + t.Fatalf("Purge: %v", err) + } + if len(out.calls) != 1 { + t.Fatalf("expected 1 outbox call, got %d", len(out.calls)) + } + got := out.calls[0] + if got.slug != "gnf" { + t.Errorf("slug: got %q, want %q", got.slug, "gnf") + } + if len(got.tags) != 2 || got.tags[0] != "posts:42" || got.tags[1] != "sitemap" { + t.Errorf("tags: got %v, want [posts:42 sitemap] (de-duped, empties stripped)", got.tags) + } +} + +// TestPurge_NilOutboxReturnsError asserts that constructing a Cache +// without an outbox is allowed (for read-only tests) but Purge fails +// loudly rather than silently no-op. Silent no-op would be a real +// data-consistency bug. +func TestPurge_NilOutboxReturnsError(t *testing.T) { + t.Parallel() + mr := miniredis.RunT(t) + rdb := redis.NewClient(&redis.Options{Addr: mr.Addr()}) + t.Cleanup(func() { _ = rdb.Close() }) + c := New(rdb, nil) + + if err := c.Purge(context.Background(), []string{"x"}); err == nil { + t.Fatal("expected error from Purge with nil outbox") + } +} + +// TestSet_Bounds covers the two size guards: value-too-large and +// tag-count-too-large. Both must surface a typed error so callers can +// react (e.g. fall back to a non-cached render). +func TestSet_Bounds(t *testing.T) { + t.Parallel() + c, _ := newTestCache(t) + ctx := context.Background() + + tooBig := make([]byte, MaxValueBytes+1) + if err := c.Set(ctx, "k", tooBig, nil, time.Minute); !errors.Is(err, ErrValueTooLarge) { + t.Errorf("oversized value: got %v, want ErrValueTooLarge", err) + } + + tooManyTags := make([]string, MaxTagsPerEntry+1) + for i := range tooManyTags { + tooManyTags[i] = "t" + } + if err := c.Set(ctx, "k", []byte("x"), tooManyTags, time.Minute); !errors.Is(err, ErrTooManyTags) { + t.Errorf("too many tags: got %v, want ErrTooManyTags", err) + } +} + +// TestGet_TagSetDriftIsMiss makes sure a Get that supplies a +// different tag-set length than the original Set is treated as a +// miss. A caller who grew its tag set across deploys would otherwise +// see a stale payload validated against an obsolete tag list. +func TestGet_TagSetDriftIsMiss(t *testing.T) { + t.Parallel() + c, _ := newTestCache(t) + ctx := context.Background() + + if err := c.Set(ctx, "k", []byte("v"), []string{"a"}, time.Minute); err != nil { + t.Fatalf("Set: %v", err) + } + if _, hit, _ := c.Get(ctx, "k", []string{"a", "b"}); hit { + t.Fatal("expected miss after tag-set drift") + } +} + +// TestEncodeDecode_RoundTrip pins the on-wire layout. A future format +// bump must keep this test green or change it deliberately. +func TestEncodeDecode_RoundTrip(t *testing.T) { + t.Parallel() + versions := []int64{0, 1, 9_223_372_036_854_775_807} + value := []byte("rendered HTML") + enc := encodeEntry(versions, value) + gotV, gotVal, ok := decodeEntry(enc) + if !ok { + t.Fatal("decodeEntry: ok=false") + } + if !bytes.Equal(gotVal, value) { + t.Errorf("value: got %q want %q", gotVal, value) + } + if len(gotV) != len(versions) { + t.Fatalf("versions: got %d entries want %d", len(gotV), len(versions)) + } + for i := range versions { + if gotV[i] != versions[i] { + t.Errorf("versions[%d]: got %d want %d", i, gotV[i], versions[i]) + } + } +} + +// TestDecode_Malformed treats truncated headers and over-claimed tag +// counts as misses rather than errors. Same shape as the live-system +// behaviour described in the Get docstring. +func TestDecode_Malformed(t *testing.T) { + t.Parallel() + cases := [][]byte{ + nil, + {0x00, 0x00}, // truncated header + {0xFF, 0xFF, 0xFF, 0x7F}, // declared tag count beyond MaxTagsPerEntry + {0x05, 0x00, 0x00, 0x00, 0x01, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0}, // claims 5 versions, body shorter + } + for i, raw := range cases { + if _, _, ok := decodeEntry(raw); ok { + t.Errorf("case %d: expected ok=false", i) + } + } +} diff --git a/packages/go/cache/fragment/outbox.go b/packages/go/cache/fragment/outbox.go new file mode 100644 index 00000000..49c3aec4 --- /dev/null +++ b/packages/go/cache/fragment/outbox.go @@ -0,0 +1,58 @@ +package fragment + +import ( + "context" + "errors" + "fmt" + + "github.com/jackc/pgx/v5/pgxpool" +) + +// PgOutboxWriter writes tag invalidations into the cache_invalidations +// table backed by a pgx connection pool. It is the production-default +// implementation of OutboxWriter; tests typically use a recording +// fake instead of standing up a real Postgres. +// +// The writer is intentionally tiny — one INSERT … SELECT — because +// the invalidation pipeline's contract is already nailed down by the +// invalidator worker (packages/go/cache/invalidator). All that remains +// here is "row-into-table"; everything interesting (pub/sub publish, +// at-least-once delivery, namespacing) happens downstream. +type PgOutboxWriter struct { + pool *pgxpool.Pool +} + +// NewPgOutboxWriter constructs an OutboxWriter backed by pool. Passing +// nil panics — a misconfigured writer that silently drops rows would +// be a real cache-correctness bug (a Purge that "succeeds" but never +// invalidates anything is worse than a noisy failure). +func NewPgOutboxWriter(pool *pgxpool.Pool) *PgOutboxWriter { + if pool == nil { + panic("fragment.NewPgOutboxWriter: pool is required") + } + return &PgOutboxWriter{pool: pool} +} + +// WriteInvalidations appends one row per tag into cache_invalidations. +// Uses INSERT … SELECT FROM unnest to do the batch in a single +// round-trip; even at MaxTagsPerEntry the batch fits in one packet. +// +// The slug column is the namespace the invalidator worker re-prefixes +// when it publishes. For fragment-cache traffic the slug is always +// "gnf" so a subscriber that wants to route on internal traffic vs. +// plugin traffic only needs to match a single prefix. +func (w *PgOutboxWriter) WriteInvalidations(ctx context.Context, slug string, tags []string) error { + if slug == "" { + return errors.New("fragment: WriteInvalidations: slug is required") + } + if len(tags) == 0 { + return nil + } + if _, err := w.pool.Exec(ctx, ` + INSERT INTO cache_invalidations (plugin_slug, tag) + SELECT $1, unnest($2::text[])`, + slug, tags); err != nil { + return fmt.Errorf("fragment: WriteInvalidations: %w", err) + } + return nil +} diff --git a/packages/go/cache/invalidator/subscriber.go b/packages/go/cache/invalidator/subscriber.go new file mode 100644 index 00000000..66f8ba10 --- /dev/null +++ b/packages/go/cache/invalidator/subscriber.go @@ -0,0 +1,211 @@ +package invalidator + +import ( + "context" + "errors" + "fmt" + "log/slog" + "strings" + "sync" + "sync/atomic" + + "github.com/redis/go-redis/v9" +) + +// HandlerFunc is the per-message dispatcher invoked by Subscriber for +// each pub/sub notification. The slug is the producer's plugin slug +// (or "gnf" for the fragment cache); the tag is the un-prefixed value +// the producer originally invalidated. +// +// Returning an error from a HandlerFunc is logged at Warn but does +// NOT stop the Subscriber loop: at-least-once delivery means we +// expect handlers to be idempotent, and one downstream failure +// should not freeze every other subscriber on the channel. +type HandlerFunc func(ctx context.Context, slug, tag string) error + +// Subscriber consumes the Redis pub/sub channel the invalidator +// Worker publishes to, parses each message, and fans the (slug, tag) +// pair out to one or more registered HandlerFuncs. +// +// Why a separate type from Worker +// +// Worker is the producer of pub/sub messages — it drains the outbox +// and publishes. Subscriber is the consumer — it listens to the same +// channel and dispatches. The two roles run in different goroutines, +// often in different processes (the API container subscribes to +// invalidate its in-process render cache; the worker container +// publishes from the outbox). Coupling them into one struct would +// force every API replica to also wake up the outbox poller, which +// is wasted work and a real "thundering herd" hazard on Postgres. +// +// One Subscriber, many handlers +// +// A single Subscriber can dispatch to multiple handlers — the fragment +// cache, an in-process LRU, a metrics counter — without each handler +// owning its own Redis subscription. This keeps the connection count +// predictable: one PSUBSCRIBE per process regardless of how many +// caches subscribe to invalidations. +type Subscriber struct { + rdb *redis.Client + channel string + logger *slog.Logger + + mu sync.RWMutex + handlers []HandlerFunc + + running atomic.Int32 +} + +// SubscriberOption configures a Subscriber at construction time. +type SubscriberOption func(*Subscriber) + +// SubscriberWithLogger swaps the structured logger. +func SubscriberWithLogger(l *slog.Logger) SubscriberOption { + return func(s *Subscriber) { + if l != nil { + s.logger = l + } + } +} + +// SubscriberWithChannel overrides the Redis pub/sub channel name. +// Must match the Worker's channel value or the Subscriber sees +// nothing. +func SubscriberWithChannel(ch string) SubscriberOption { + return func(s *Subscriber) { + if ch != "" { + s.channel = ch + } + } +} + +// NewSubscriber constructs a Subscriber. rdb is required; passing nil +// panics for the same reason Worker's New panics on a nil pool — a +// misconfigured subscriber silently drops every invalidation, which +// is the worst possible failure mode for a cache layer. +func NewSubscriber(rdb *redis.Client, opts ...SubscriberOption) *Subscriber { + if rdb == nil { + panic("invalidator.NewSubscriber: redis client is required") + } + s := &Subscriber{ + rdb: rdb, + channel: DefaultChannel, + logger: slog.Default(), + } + for _, o := range opts { + o(s) + } + return s +} + +// Handle registers a HandlerFunc. Handlers fire in registration order +// for every received message. Returns no error today, but the +// signature is fixed so a future "named handlers" extension can add +// a duplicate-name guard without changing the call site. +// +// Handlers may be registered while Run is in flight; the next +// message dispatch picks up the new handler. +func (s *Subscriber) Handle(h HandlerFunc) { + if h == nil { + return + } + s.mu.Lock() + s.handlers = append(s.handlers, h) + s.mu.Unlock() +} + +// ErrSubscriberAlreadyRunning is returned by Run when invoked +// concurrently with itself. +var ErrSubscriberAlreadyRunning = errors.New("invalidator: subscriber already running") + +// Run subscribes to the channel and dispatches every message until +// ctx is cancelled. +// +// Run blocks. The typical wiring is: +// +// go func() { _ = sub.Run(runCtx) }() +// // … runCtx is cancelled on shutdown … +// +// The Redis SUBSCRIBE itself is wrapped in a Receive (the go-redis +// "is subscription established" handshake) so a connection failure +// surfaces immediately rather than after the first message would +// have arrived. +func (s *Subscriber) Run(ctx context.Context) error { + if !s.running.CompareAndSwap(0, 1) { + return ErrSubscriberAlreadyRunning + } + defer s.running.Store(0) + + sub := s.rdb.Subscribe(ctx, s.channel) + defer sub.Close() + + // Wait for the subscription to be established. Without this the + // first messages could be dropped by go-redis' internal + // reconnection logic. + if _, err := sub.Receive(ctx); err != nil { + return fmt.Errorf("subscribe %q: %w", s.channel, err) + } + + s.logger.Info("cache invalidator subscriber started", + slog.String("channel", s.channel)) + + msgs := sub.Channel() + for { + select { + case <-ctx.Done(): + s.logger.Info("cache invalidator subscriber stopping") + return nil + case m, ok := <-msgs: + if !ok { + // go-redis closes the channel on subscription + // teardown; treat as a clean shutdown. + return nil + } + s.dispatch(ctx, m.Payload) + } + } +} + +// dispatch parses a ":" payload and invokes every +// registered handler with the parsed parts. A malformed payload +// (no colon) is logged once and dropped — there's no useful retry +// for a structurally broken message and we don't want a poison +// message to wedge the subscriber. +func (s *Subscriber) dispatch(ctx context.Context, payload string) { + slug, tag, ok := splitPayload(payload) + if !ok { + s.logger.Warn("cache invalidator: malformed pub/sub payload (no colon)", + slog.String("payload", payload)) + return + } + + s.mu.RLock() + handlers := append([]HandlerFunc(nil), s.handlers...) + s.mu.RUnlock() + + for _, h := range handlers { + if err := h(ctx, slug, tag); err != nil { + s.logger.Warn("cache invalidator: handler error", + slog.String("slug", slug), + slog.String("tag", tag), + slog.Any("err", err)) + } + } +} + +// splitPayload parses ":" into its two parts. Tags +// themselves may contain colons (e.g. "posts:42") — we split on the +// FIRST colon only, so the slug is the leading namespace and the +// tag is everything after. +// +// Returns ok=false when the payload has no colon at all; an empty +// slug or empty tag IS allowed (the producer side is what guarantees +// neither is empty in practice; the subscriber treats them as valid +// input so a test fixture can probe edge cases without short-circuit). +func splitPayload(payload string) (slug, tag string, ok bool) { + i := strings.IndexByte(payload, ':') + if i < 0 { + return "", "", false + } + return payload[:i], payload[i+1:], true +} diff --git a/packages/go/cache/invalidator/subscriber_test.go b/packages/go/cache/invalidator/subscriber_test.go new file mode 100644 index 00000000..c30249fb --- /dev/null +++ b/packages/go/cache/invalidator/subscriber_test.go @@ -0,0 +1,173 @@ +package invalidator + +import ( + "context" + "errors" + "sync" + "sync/atomic" + "testing" + "time" + + "github.com/alicebob/miniredis/v2" + "github.com/redis/go-redis/v9" +) + +// TestSubscriber_DispatchesToHandlers exercises the wire path: a +// PUBLISH on the channel parses into (slug, tag) and lands in every +// registered handler. +func TestSubscriber_DispatchesToHandlers(t *testing.T) { + t.Parallel() + mr := miniredis.RunT(t) + rdb := redis.NewClient(&redis.Options{Addr: mr.Addr()}) + t.Cleanup(func() { _ = rdb.Close() }) + + const channel = "test:invalidate" + sub := NewSubscriber(rdb, SubscriberWithChannel(channel)) + + var mu sync.Mutex + var got []string + sub.Handle(func(_ context.Context, slug, tag string) error { + mu.Lock() + got = append(got, slug+"|"+tag) + mu.Unlock() + return nil + }) + + ctx, cancel := context.WithCancel(context.Background()) + t.Cleanup(cancel) + go func() { _ = sub.Run(ctx) }() + + // Wait for the subscription to register. miniredis publishes + // synchronously, but the Subscriber's Receive handshake means + // the channel registration might not be live for one tick. + time.Sleep(50 * time.Millisecond) + + for _, msg := range []string{"gn-seo:posts:42", "gnf:menu", "gnf:sitemap"} { + if err := rdb.Publish(ctx, channel, msg).Err(); err != nil { + t.Fatalf("publish: %v", err) + } + } + + deadline := time.After(2 * time.Second) + for { + mu.Lock() + n := len(got) + mu.Unlock() + if n >= 3 { + break + } + select { + case <-deadline: + t.Fatalf("only got %d messages, want 3", n) + case <-time.After(10 * time.Millisecond): + } + } + + want := map[string]bool{ + "gn-seo|posts:42": true, + "gnf|menu": true, + "gnf|sitemap": true, + } + mu.Lock() + for _, g := range got { + if !want[g] { + t.Errorf("unexpected message %q", g) + } + } + mu.Unlock() +} + +// TestSubscriber_MalformedPayloadIsDropped checks that a poison +// message (no colon) does not panic and does not invoke any +// handler. The Subscriber stays alive and consumes the next +// well-formed message normally. +func TestSubscriber_MalformedPayloadIsDropped(t *testing.T) { + t.Parallel() + mr := miniredis.RunT(t) + rdb := redis.NewClient(&redis.Options{Addr: mr.Addr()}) + t.Cleanup(func() { _ = rdb.Close() }) + + const channel = "test:invalidate" + sub := NewSubscriber(rdb, SubscriberWithChannel(channel)) + + var calls atomic.Int32 + sub.Handle(func(_ context.Context, _, _ string) error { + calls.Add(1) + return nil + }) + + ctx, cancel := context.WithCancel(context.Background()) + t.Cleanup(cancel) + go func() { _ = sub.Run(ctx) }() + time.Sleep(50 * time.Millisecond) + + _ = rdb.Publish(ctx, channel, "no-colon").Err() + _ = rdb.Publish(ctx, channel, "gnf:menu").Err() + + deadline := time.After(2 * time.Second) + for calls.Load() < 1 { + select { + case <-deadline: + t.Fatalf("expected one handler call, got %d", calls.Load()) + case <-time.After(10 * time.Millisecond): + } + } + if got := calls.Load(); got != 1 { + t.Errorf("calls: got %d, want exactly 1 (malformed dropped)", got) + } +} + +// TestSubscriber_HandlerErrorDoesNotStopLoop confirms an erroring +// handler is logged but the Subscriber keeps consuming. This is the +// at-least-once contract: a flaky handler must not block other +// subscribers on the same channel. +func TestSubscriber_HandlerErrorDoesNotStopLoop(t *testing.T) { + t.Parallel() + mr := miniredis.RunT(t) + rdb := redis.NewClient(&redis.Options{Addr: mr.Addr()}) + t.Cleanup(func() { _ = rdb.Close() }) + + const channel = "test:invalidate" + sub := NewSubscriber(rdb, SubscriberWithChannel(channel)) + + var calls atomic.Int32 + sub.Handle(func(_ context.Context, _, _ string) error { + calls.Add(1) + return errors.New("boom") + }) + + ctx, cancel := context.WithCancel(context.Background()) + t.Cleanup(cancel) + go func() { _ = sub.Run(ctx) }() + time.Sleep(50 * time.Millisecond) + + _ = rdb.Publish(ctx, channel, "gnf:a").Err() + _ = rdb.Publish(ctx, channel, "gnf:b").Err() + + deadline := time.After(2 * time.Second) + for calls.Load() < 2 { + select { + case <-deadline: + t.Fatalf("expected at least 2 handler calls, got %d", calls.Load()) + case <-time.After(10 * time.Millisecond): + } + } +} + +// TestSubscriber_RejectsDoubleRun pins ErrSubscriberAlreadyRunning. A +// second concurrent Run is a programming error — exposed loudly so +// the caller sees the bug at boot. +func TestSubscriber_RejectsDoubleRun(t *testing.T) { + t.Parallel() + mr := miniredis.RunT(t) + rdb := redis.NewClient(&redis.Options{Addr: mr.Addr()}) + t.Cleanup(func() { _ = rdb.Close() }) + + sub := NewSubscriber(rdb) + sub.running.Store(1) + defer sub.running.Store(0) + + if err := sub.Run(context.Background()); !errors.Is(err, ErrSubscriberAlreadyRunning) { + t.Fatalf("Run: got %v, want ErrSubscriberAlreadyRunning", err) + } +} diff --git a/packages/go/jobs/scheduler/doc.go b/packages/go/jobs/scheduler/doc.go new file mode 100644 index 00000000..a4e61bc2 --- /dev/null +++ b/packages/go/jobs/scheduler/doc.go @@ -0,0 +1,63 @@ +// Package scheduler implements the content-lifecycle background jobs: +// +// - the scheduled-publisher cron task that flips eligible +// status='scheduled' posts to status='published' (issue #143 +// state-machine half), and +// - the trash-GC cron task that hard-deletes posts whose status +// has been 'trash' for longer than the configured retention +// window (default 30 days). +// +// Both tasks share a taskspec.TaskSpec + cron.CronSpec pair so the +// worker binary wires them with a few lines in main.go: declare the +// specs, register them against the process-wide cron registry, and +// the cron-leader goroutine fires them on schedule. +// +// # State machine +// +// The post lifecycle (per migrations/000001_init.up.sql) is: +// +// draft → pending → scheduled → published +// ↘ +// private +// ↘ +// trash → (GC) +// +// 'scheduled' is the transition this package automates. When a post +// is saved with status='scheduled' and a scheduled_for time, the API +// stores it as-is; the publisher cron picks it up the first minute +// after scheduled_for elapses and flips status to 'published' and +// published_at to now(). The transition is a single UPDATE with a +// version-bumping trigger, so a concurrent editor's optimistic- +// concurrency check (WHERE version = …) still catches a race. +// +// # Retention / GC +// +// Trash is a soft delete: the row sticks around with status='trash' +// so an admin can restore it. After a configurable window (default +// 30 days; matches docs/01-core-cms.md §6) the GC task hard-deletes +// the row plus its derived rows (revisions, autosaves, etc., which +// have ON DELETE CASCADE FKs). The window starts at the most recent +// updated_at — moving to trash is itself an UPDATE that touches the +// timestamp, so a re-trashed-then-restored post effectively resets +// the clock. +// +// # Why not inline in the API +// +// Both tasks are bulk operations that pin a connection while they +// run — even a 100-row batch is multi-second on a busy primary. The +// API container's request-scoped budgets aren't built for that; +// running them on the worker container with its 240s drain budget +// matches every other batch task (revisions.purge, abort_orphans). +// +// # Cadence +// +// The scheduled-publisher runs every minute (the smallest cadence +// the project's cron parser accepts, see jobs/cron). Going faster +// is technically possible with @every but provides no real-world +// win: editors who set a scheduled_for time inside the next minute +// already accept that as part of the "scheduled publish" UX. +// +// The GC runs daily at 03:30 UTC — half an hour after the existing +// revisions.purge sweep so the two heavy queries don't pile up at +// the same minute. +package scheduler diff --git a/packages/go/jobs/scheduler/gc.go b/packages/go/jobs/scheduler/gc.go new file mode 100644 index 00000000..763f476e --- /dev/null +++ b/packages/go/jobs/scheduler/gc.go @@ -0,0 +1,231 @@ +package scheduler + +import ( + "context" + "encoding/json" + "errors" + "fmt" + "log/slog" + "time" + + "github.com/jackc/pgx/v5/pgxpool" + + "github.com/Singleton-Solution/GoNext/packages/go/jobs/cron" + "github.com/Singleton-Solution/GoNext/packages/go/jobs/taskspec" +) + +// GCTaskName is the on-wire task name for the trash-retention sweep. +const GCTaskName = "content.gc" + +// GCCronName is the cron-registry key for the daily fire. +const GCCronName = "content.gc.daily" + +// GCSchedule is the cron expression that fires the GC. 03:30 UTC +// trails the existing revisions.purge.daily (03:00 UTC) by half an +// hour so the two heavy queries don't pile up at the same minute. +const GCSchedule = "30 3 * * *" + +// DefaultRetention is the trash-window the GC enforces when no +// override is provided. 30 days matches docs/01-core-cms.md §6 (the +// "trash auto-empty" contract the admin UI advertises). +const DefaultRetention = 30 * 24 * time.Hour + +// DefaultGCBatchLimit caps how many rows a single fire deletes. The +// cap is here for the same reason the publisher has one: a sudden +// backlog (e.g. a bulk trash from the admin UI thirty days ago) +// would otherwise lock the table. +const DefaultGCBatchLimit = 1000 + +// gcPayloadSchemaRaw constrains the payload. Same shape as the +// publisher's — null or empty object. +var gcPayloadSchemaRaw = []byte(`{ + "$schema": "https://json-schema.org/draft/2020-12/schema", + "type": ["object", "null"], + "additionalProperties": false +}`) + +// GCResult is the per-run summary the handler logs. +type GCResult struct { + // Deleted is the number of rows hard-deleted from posts. + Deleted int +} + +// GCSpecOptions configures NewGCSpec. +type GCSpecOptions struct { + // Pool is the pgx connection pool. Required. + Pool *pgxpool.Pool + + // Retention is how long a trashed post sticks around before + // being hard-deleted. Defaults to DefaultRetention. + Retention time.Duration + + // Limit is the maximum number of rows per fire. Defaults to + // DefaultGCBatchLimit. + Limit int + + // Logger receives the structured per-fire output. + Logger *slog.Logger + + // Now is a clock override for tests. Nil defaults to time.Now. + Now func() time.Time +} + +// NewGCSpec returns the TaskSpec the worker registers for the GC. +func NewGCSpec(opts GCSpecOptions) (taskspec.TaskSpec, error) { + if opts.Pool == nil { + return taskspec.TaskSpec{}, errors.New("scheduler: NewGCSpec: Pool is required") + } + if opts.Logger == nil { + opts.Logger = slog.Default() + } + if opts.Retention <= 0 { + opts.Retention = DefaultRetention + } + if opts.Limit <= 0 { + opts.Limit = DefaultGCBatchLimit + } + if opts.Now == nil { + opts.Now = time.Now + } + + handler := func(ctx context.Context, raw []byte) error { + if len(raw) > 0 && string(raw) != "null" { + var anyPayload map[string]any + if err := json.Unmarshal(raw, &anyPayload); err != nil { + return fmt.Errorf("scheduler/gc: parse payload: %w", err) + } + } + res, err := SweepTrash(ctx, opts.Pool, SweepOptions{ + Retention: opts.Retention, + Limit: opts.Limit, + Now: opts.Now(), + Logger: opts.Logger, + }) + if err != nil { + return fmt.Errorf("scheduler/gc: %w", err) + } + opts.Logger.InfoContext(ctx, "scheduler/gc: sweep complete", + slog.Int("deleted", res.Deleted), + slog.Duration("retention", opts.Retention)) + return nil + } + return taskspec.TaskSpec{ + Name: GCTaskName, + Queue: "default", + MaxRetry: 1, + // 10 minutes is generous — the candidate set is small in + // the steady state (only what trashed thirty days ago). + // Larger backlogs spill into the next day's fire. + Timeout: 10 * time.Minute, + Handler: handler, + }, nil +} + +// NewGCCron returns the CronSpec registered against the cron +// registry. Daily at 03:30 UTC. +func NewGCCron() cron.CronSpec { + return cron.CronSpec{ + Name: GCCronName, + Schedule: GCSchedule, + TaskName: GCTaskName, + } +} + +// GCPayloadSchema returns the JSON Schema bytes for the GC payload. +func GCPayloadSchema() []byte { + out := make([]byte, len(gcPayloadSchemaRaw)) + copy(out, gcPayloadSchemaRaw) + return out +} + +// SweepOptions tunes one SweepTrash invocation. +type SweepOptions struct { + // Retention is the cutoff: rows with status='trash' and + // updated_at <= now - Retention are hard-deleted. + Retention time.Duration + + // Limit caps the number of rows deleted per call. + Limit int + + // Now is the reference time for the retention comparison. + Now time.Time + + // Logger is the structured logger for per-row warnings. + Logger *slog.Logger +} + +// SweepTrash hard-deletes posts whose status has been 'trash' for +// at least Retention. The cutoff is computed against the row's +// updated_at; the trigger that maintains updated_at on every UPDATE +// is what makes this work — moving a row TO trash is itself an +// UPDATE that resets the clock, so we measure from the last move- +// to-trash event rather than the original creation. +// +// The delete cascades into the row's revisions, autosaves, and +// other derived rows via ON DELETE CASCADE FKs declared in the +// schema. We do NOT delete media attachments referenced by the +// post; the abort-orphans sweep is what cleans those up once +// nothing references them. +// +// Returns the count of deleted rows. An error here is a real +// database failure (the candidate query or the delete failed) and +// is propagated to the task layer for retry. +func SweepTrash(ctx context.Context, pool *pgxpool.Pool, opts SweepOptions) (GCResult, error) { + if pool == nil { + return GCResult{}, errors.New("scheduler: SweepTrash: pool is required") + } + if opts.Retention <= 0 { + return GCResult{}, errors.New("scheduler: SweepTrash: Retention must be positive") + } + if opts.Limit <= 0 { + return GCResult{}, errors.New("scheduler: SweepTrash: Limit must be positive") + } + if opts.Now.IsZero() { + return GCResult{}, errors.New("scheduler: SweepTrash: Now is required") + } + logger := opts.Logger + if logger == nil { + logger = slog.Default() + } + + cutoff := opts.Now.Add(-opts.Retention) + // Same CTE shape as the publisher: SELECT … FOR UPDATE SKIP + // LOCKED so two replicas firing the same minute don't trip + // over each other. The cron-leader election (#88/#258) is + // supposed to prevent the double-fire entirely, but the + // belt-and-suspenders here is cheap. + rows, err := pool.Query(ctx, ` + WITH expired AS ( + SELECT id + FROM posts + WHERE status = 'trash' + AND updated_at <= $1 + ORDER BY updated_at + LIMIT $2 + FOR UPDATE SKIP LOCKED + ) + DELETE FROM posts p + USING expired + WHERE p.id = expired.id + RETURNING p.id`, + cutoff, opts.Limit) + if err != nil { + return GCResult{}, fmt.Errorf("query: %w", err) + } + defer rows.Close() + + res := GCResult{} + for rows.Next() { + var id string + if err := rows.Scan(&id); err != nil { + logger.WarnContext(ctx, "scheduler/gc: row scan failed", + slog.Any("err", err)) + continue + } + res.Deleted++ + } + if err := rows.Err(); err != nil { + return res, fmt.Errorf("rows: %w", err) + } + return res, nil +} diff --git a/packages/go/jobs/scheduler/publisher.go b/packages/go/jobs/scheduler/publisher.go new file mode 100644 index 00000000..227cc352 --- /dev/null +++ b/packages/go/jobs/scheduler/publisher.go @@ -0,0 +1,258 @@ +package scheduler + +import ( + "context" + "encoding/json" + "errors" + "fmt" + "log/slog" + "time" + + "github.com/jackc/pgx/v5/pgxpool" + + "github.com/Singleton-Solution/GoNext/packages/go/jobs/cron" + "github.com/Singleton-Solution/GoNext/packages/go/jobs/taskspec" +) + +// PublisherTaskName is the on-wire task name the cron entry fires +// against. Exported so the worker wiring and the admin "publish now" +// surface can target the same handler. +const PublisherTaskName = "content.publisher" + +// PublisherCronName is the cron-registry key for the per-minute fire. +const PublisherCronName = "content.publisher.minute" + +// PublisherSchedule is the cron expression that fires the publisher. +// "@every 1m" matches the smallest cadence the project's cron parser +// accepts and is the resolution the editor surface advertises in the +// "schedule for…" picker. +const PublisherSchedule = "@every 1m" + +// DefaultPublisherBatchLimit caps how many posts a single fire flips +// to 'published'. The cap protects against a pathological backlog +// (e.g. a frozen worker accumulating thousands of scheduled posts) +// pinning the database under a single transaction. +const DefaultPublisherBatchLimit = 500 + +// publisherPayloadSchemaRaw constrains the task payload. The task +// takes no per-fire input — Cron will pass `null` and we validate +// against that. +var publisherPayloadSchemaRaw = []byte(`{ + "$schema": "https://json-schema.org/draft/2020-12/schema", + "type": ["object", "null"], + "additionalProperties": false +}`) + +// PublisherResult is the per-run summary surfaced to logs and +// (eventually) Prometheus. We expose it as a named type so tests can +// assert on the shape; in production the worker just logs it. +type PublisherResult struct { + // Scanned is the number of rows the candidate query returned. + // Equal to Published in the steady state; lower than Published + // would be a bug. + Scanned int + + // Published is the number of rows actually flipped to + // 'published'. May be less than Scanned if a row was modified + // (version bump) between the candidate query and the UPDATE — + // the trigger-driven version check filters out concurrent edits. + Published int + + // Errors is the number of rows the publisher attempted to flip + // but couldn't (typically a constraint violation, e.g. a + // unique-slug collision after a slug template change). Each + // errored row is logged individually. + Errors int +} + +// PublisherSpecOptions configures NewPublisherSpec. Pool is required +// — the task handler is a closure over it. The rest tune behaviour. +type PublisherSpecOptions struct { + // Pool is the pgx connection pool. Required. + Pool *pgxpool.Pool + + // Limit is the maximum number of rows one fire touches. + // Defaults to DefaultPublisherBatchLimit. + Limit int + + // Logger receives the structured per-fire output. Nil falls + // back to slog.Default. + Logger *slog.Logger + + // Now is a clock override for tests. Nil defaults to time.Now. + Now func() time.Time +} + +// NewPublisherSpec returns the TaskSpec the worker registers for the +// scheduled-publisher task. The handler closes over opts.Pool so the +// cron side does not need to know the database configuration. +func NewPublisherSpec(opts PublisherSpecOptions) (taskspec.TaskSpec, error) { + if opts.Pool == nil { + return taskspec.TaskSpec{}, errors.New("scheduler: NewPublisherSpec: Pool is required") + } + if opts.Logger == nil { + opts.Logger = slog.Default() + } + if opts.Limit <= 0 { + opts.Limit = DefaultPublisherBatchLimit + } + if opts.Now == nil { + opts.Now = time.Now + } + + handler := func(ctx context.Context, raw []byte) error { + // Parse the payload defensively. We don't read any keys + // from it, but a stray non-null payload should surface as + // a parse error rather than be silently ignored — that's + // the same shape every other task in the project uses. + if len(raw) > 0 && string(raw) != "null" { + var anyPayload map[string]any + if err := json.Unmarshal(raw, &anyPayload); err != nil { + return fmt.Errorf("scheduler/publisher: parse payload: %w", err) + } + } + res, err := PublishScheduled(ctx, opts.Pool, PublishOptions{ + Limit: opts.Limit, + Now: opts.Now(), + Logger: opts.Logger, + }) + if err != nil { + return fmt.Errorf("scheduler/publisher: %w", err) + } + opts.Logger.InfoContext(ctx, "scheduler/publisher: fire complete", + slog.Int("scanned", res.Scanned), + slog.Int("published", res.Published), + slog.Int("errors", res.Errors)) + return nil + } + return taskspec.TaskSpec{ + Name: PublisherTaskName, + Queue: "default", + MaxRetry: 1, + // 2 minutes covers a full-batch sweep with comfortable + // headroom for an over-loaded Postgres. The next fire + // arrives in 60s, so timing out earlier would mean we'd + // double-fire while the previous run is still alive. + Timeout: 2 * time.Minute, + Handler: handler, + }, nil +} + +// NewPublisherCron returns the CronSpec registered against the cron +// registry. "@every 1m" payload nil. +func NewPublisherCron() cron.CronSpec { + return cron.CronSpec{ + Name: PublisherCronName, + Schedule: PublisherSchedule, + TaskName: PublisherTaskName, + } +} + +// PublisherPayloadSchema returns the JSON Schema bytes. Exposed so +// tests can validate payloads outside the Enqueue path. +func PublisherPayloadSchema() []byte { + out := make([]byte, len(publisherPayloadSchemaRaw)) + copy(out, publisherPayloadSchemaRaw) + return out +} + +// PublishOptions tunes one PublishScheduled invocation. Production +// callers pass nothing; tests pass a fixed Now to make the candidate +// query deterministic. +type PublishOptions struct { + // Limit caps the number of rows touched per call. Required to + // be positive; PublishScheduled returns an error on zero. + Limit int + + // Now is the wall-clock time used for the scheduled_for <= + // comparison and the published_at value. Required. + Now time.Time + + // Logger receives structured per-row warnings. Nil falls back + // to slog.Default. + Logger *slog.Logger +} + +// PublishScheduled is the pure transition step exposed without the +// taskspec wrapper. It's the function the handler calls and the +// one tests target directly — much easier to assert against than a +// json.RawMessage-shaped handler. +// +// Behaviour: +// +// UPDATE posts +// SET status='published', published_at=COALESCE(published_at, $1) +// WHERE status='scheduled' AND scheduled_for <= $1 +// RETURNING id +// +// The COALESCE on published_at is what the schema comment promises — +// a draft of a previously-published post keeps its original +// publication date so canonical URLs stay stable. +// +// The update is bounded by $LIMIT to keep one fire from holding +// locks longer than the cron interval. Excess rows roll over to the +// next minute's fire. +// +// Errors here propagate from pgx — a missing posts table, a typo'd +// status enum, a constraint trip on a unique slug. The handler +// retries once via asynq, then DLQs. +func PublishScheduled(ctx context.Context, pool *pgxpool.Pool, opts PublishOptions) (PublisherResult, error) { + if pool == nil { + return PublisherResult{}, errors.New("scheduler: PublishScheduled: pool is required") + } + if opts.Limit <= 0 { + return PublisherResult{}, errors.New("scheduler: PublishScheduled: Limit must be positive") + } + if opts.Now.IsZero() { + return PublisherResult{}, errors.New("scheduler: PublishScheduled: Now is required") + } + logger := opts.Logger + if logger == nil { + logger = slog.Default() + } + + // One UPDATE … RETURNING id. The CTE shape (with LIMIT on a + // SELECT … FOR UPDATE SKIP LOCKED, then UPDATE on the locked + // set) is the canonical "drain a queue safely" pattern: if two + // workers somehow race on the same fire, each takes a disjoint + // slice and neither errors. + rows, err := pool.Query(ctx, ` + WITH due AS ( + SELECT id + FROM posts + WHERE status = 'scheduled' + AND scheduled_for IS NOT NULL + AND scheduled_for <= $1 + ORDER BY scheduled_for + LIMIT $2 + FOR UPDATE SKIP LOCKED + ) + UPDATE posts p + SET status = 'published', + published_at = COALESCE(p.published_at, $1) + FROM due + WHERE p.id = due.id + RETURNING p.id`, + opts.Now, opts.Limit) + if err != nil { + return PublisherResult{}, fmt.Errorf("query: %w", err) + } + defer rows.Close() + + res := PublisherResult{} + for rows.Next() { + var id string + if err := rows.Scan(&id); err != nil { + res.Errors++ + logger.WarnContext(ctx, "scheduler/publisher: row scan failed", + slog.Any("err", err)) + continue + } + res.Scanned++ + res.Published++ + } + if err := rows.Err(); err != nil { + return res, fmt.Errorf("rows: %w", err) + } + return res, nil +} diff --git a/packages/go/jobs/scheduler/scheduler_test.go b/packages/go/jobs/scheduler/scheduler_test.go new file mode 100644 index 00000000..b6518a8b --- /dev/null +++ b/packages/go/jobs/scheduler/scheduler_test.go @@ -0,0 +1,315 @@ +package scheduler + +import ( + "context" + "testing" + "time" + + "github.com/jackc/pgx/v5/pgxpool" + + "github.com/Singleton-Solution/GoNext/packages/go/jobs/cron" + "github.com/Singleton-Solution/GoNext/packages/go/jobs/taskspec" + "github.com/Singleton-Solution/GoNext/packages/go/testutil/containers" +) + +// schemaSQL is the minimal schema the publisher/GC tests need. +// Mirrored from migrations/000001_init.up.sql + 000004_posts.up.sql +// stripped to the columns the queries actually touch. We do this +// rather than depending on the full migration loader because the +// suite intentionally only exercises the SQL the package produces. +const schemaSQL = ` +CREATE TYPE post_status AS ENUM ( + 'draft','pending','scheduled','published','private','trash','revision' +); +CREATE TABLE posts ( + id UUID PRIMARY KEY DEFAULT gen_random_uuid(), + post_type TEXT NOT NULL DEFAULT 'post', + status post_status NOT NULL DEFAULT 'draft', + scheduled_for TIMESTAMPTZ, + published_at TIMESTAMPTZ, + updated_at TIMESTAMPTZ NOT NULL DEFAULT now(), + version INTEGER NOT NULL DEFAULT 1 +); +CREATE OR REPLACE FUNCTION touch_updated_at() RETURNS trigger +LANGUAGE plpgsql AS $$ +BEGIN + NEW.updated_at = now(); + RETURN NEW; +END; +$$; +CREATE TRIGGER posts_touch_updated_at + BEFORE UPDATE ON posts + FOR EACH ROW + EXECUTE FUNCTION touch_updated_at(); +` + +// setupTestPool spins up a Postgres container with the minimal posts +// schema. Returns the pool; container cleanup is registered on t. +func setupTestPool(t *testing.T) *pgxpool.Pool { + t.Helper() + dsn := containers.Postgres(t) + if dsn == "" { + t.Skip("no Postgres available") + } + ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second) + defer cancel() + + pool, err := pgxpool.New(ctx, dsn) + if err != nil { + t.Fatalf("pgxpool.New: %v", err) + } + t.Cleanup(pool.Close) + + if _, err := pool.Exec(ctx, schemaSQL); err != nil { + t.Fatalf("apply schema: %v", err) + } + return pool +} + +// TestPublishScheduled_FlipsEligibleRows covers the headline path: +// rows whose scheduled_for is in the past flip to 'published'; rows +// whose scheduled_for is in the future stay 'scheduled'. +func TestPublishScheduled_FlipsEligibleRows(t *testing.T) { + t.Parallel() + pool := setupTestPool(t) + ctx := context.Background() + + now := time.Now().UTC() + // Row A: due (past scheduled_for). Should publish. + // Row B: still future. Should stay scheduled. + // Row C: scheduled with no scheduled_for (defensive — shouldn't + // be reachable through the API but the GC + UPDATE on the + // trigger column means the column is nullable in the schema). + // Row D: already published (should not be touched). + for _, p := range []struct { + status string + scheduledFor *time.Time + setPublished bool + alreadyPubAt time.Time + expectPublish bool + }{ + {status: "scheduled", scheduledFor: ptr(now.Add(-1 * time.Minute)), expectPublish: true}, + {status: "scheduled", scheduledFor: ptr(now.Add(1 * time.Hour)), expectPublish: false}, + {status: "scheduled", scheduledFor: nil, expectPublish: false}, + {status: "published", setPublished: true, alreadyPubAt: now.Add(-24 * time.Hour)}, + } { + _, err := pool.Exec(ctx, ` + INSERT INTO posts (status, scheduled_for, published_at) + VALUES ($1, $2, $3)`, + p.status, p.scheduledFor, nullableTime(p.setPublished, p.alreadyPubAt)) + if err != nil { + t.Fatalf("seed insert: %v", err) + } + _ = p.expectPublish + } + + res, err := PublishScheduled(ctx, pool, PublishOptions{ + Limit: 100, + Now: now, + }) + if err != nil { + t.Fatalf("PublishScheduled: %v", err) + } + if res.Published != 1 { + t.Errorf("Published: got %d, want 1", res.Published) + } + + var publishedCount, scheduledCount int + if err := pool.QueryRow(ctx, + `SELECT + (SELECT count(*) FROM posts WHERE status='published'), + (SELECT count(*) FROM posts WHERE status='scheduled')`, + ).Scan(&publishedCount, &scheduledCount); err != nil { + t.Fatalf("status counts: %v", err) + } + if publishedCount != 2 { // newly-flipped + the pre-existing published row + t.Errorf("post-publish: published=%d, want 2", publishedCount) + } + if scheduledCount != 2 { // the future-scheduled one + the null-scheduled_for one + t.Errorf("post-publish: scheduled=%d, want 2", scheduledCount) + } +} + +// TestPublishScheduled_PreservesPublishedAt confirms a re-publish +// keeps the original published_at value (COALESCE behaviour) so +// canonical date-URLs stay stable across re-publishes. +func TestPublishScheduled_PreservesPublishedAt(t *testing.T) { + t.Parallel() + pool := setupTestPool(t) + ctx := context.Background() + + now := time.Now().UTC() + originalPub := now.Add(-48 * time.Hour).Truncate(time.Microsecond) + + // Row that was previously published, then moved back to + // scheduled with a new scheduled_for. published_at is retained + // on the row. + if _, err := pool.Exec(ctx, ` + INSERT INTO posts (status, scheduled_for, published_at) + VALUES ('scheduled', $1, $2)`, + now.Add(-1*time.Minute), originalPub); err != nil { + t.Fatalf("seed: %v", err) + } + + if _, err := PublishScheduled(ctx, pool, PublishOptions{ + Limit: 10, + Now: now, + }); err != nil { + t.Fatalf("PublishScheduled: %v", err) + } + + var got time.Time + if err := pool.QueryRow(ctx, + `SELECT published_at FROM posts WHERE status='published'`). + Scan(&got); err != nil { + t.Fatalf("read published_at: %v", err) + } + if !got.Equal(originalPub) { + t.Errorf("published_at: got %s, want %s (original preserved)", got, originalPub) + } +} + +// TestSweepTrash_DeletesExpiredAndPreservesFresh covers the GC's +// retention boundary: rows past the cutoff are deleted, rows inside +// it are preserved. +func TestSweepTrash_DeletesExpiredAndPreservesFresh(t *testing.T) { + t.Parallel() + pool := setupTestPool(t) + ctx := context.Background() + + now := time.Now().UTC() + retention := 30 * 24 * time.Hour + + // Row A: trashed 40 days ago → delete. + // Row B: trashed 10 days ago → keep. + // Row C: not trashed → keep regardless. + for _, p := range []struct { + status string + updatedAt time.Time + }{ + {status: "trash", updatedAt: now.Add(-40 * 24 * time.Hour)}, + {status: "trash", updatedAt: now.Add(-10 * 24 * time.Hour)}, + {status: "published", updatedAt: now.Add(-100 * 24 * time.Hour)}, + } { + // Insert with the desired updated_at. The trigger will + // fire on UPDATE but not on the initial INSERT. + if _, err := pool.Exec(ctx, ` + INSERT INTO posts (status, updated_at) VALUES ($1, $2)`, + p.status, p.updatedAt); err != nil { + t.Fatalf("seed: %v", err) + } + } + + res, err := SweepTrash(ctx, pool, SweepOptions{ + Retention: retention, + Limit: 100, + Now: now, + }) + if err != nil { + t.Fatalf("SweepTrash: %v", err) + } + if res.Deleted != 1 { + t.Errorf("Deleted: got %d, want 1", res.Deleted) + } + + var trashed, total int + if err := pool.QueryRow(ctx, ` + SELECT + (SELECT count(*) FROM posts WHERE status='trash'), + (SELECT count(*) FROM posts)`). + Scan(&trashed, &total); err != nil { + t.Fatalf("counts: %v", err) + } + if trashed != 1 { + t.Errorf("trashed remaining: got %d, want 1", trashed) + } + if total != 2 { + t.Errorf("total remaining: got %d, want 2", total) + } +} + +// TestSeedDefaults_RegistersBoth pins the wiring contract: SeedDefaults +// adds both tasks and both cron entries to the registries passed in. +func TestSeedDefaults_RegistersBoth(t *testing.T) { + t.Parallel() + // SeedDefaults validates the pool argument but never reads from + // it during registration. A nil-typed but non-nil pointer would + // be too clever; we just construct an empty pool that never + // connects. + pool := &pgxpool.Pool{} + taskReg := taskspec.NewRegistry() + cronReg := cron.NewRegistry() + + if err := SeedDefaults(taskReg, cronReg, SeedOptions{Pool: pool}); err != nil { + t.Fatalf("SeedDefaults: %v", err) + } + + if _, ok := taskReg.Get(PublisherTaskName); !ok { + t.Errorf("task %q not registered", PublisherTaskName) + } + if _, ok := taskReg.Get(GCTaskName); !ok { + t.Errorf("task %q not registered", GCTaskName) + } + if _, ok := cronReg.Get(PublisherCronName); !ok { + t.Errorf("cron %q not registered", PublisherCronName) + } + if _, ok := cronReg.Get(GCCronName); !ok { + t.Errorf("cron %q not registered", GCCronName) + } +} + +// TestPublishScheduled_LimitBoundsBatch confirms the LIMIT clause +// actually caps the per-fire size. Pin against a hard-to-spot +// regression where a future refactor drops the LIMIT. +func TestPublishScheduled_LimitBoundsBatch(t *testing.T) { + t.Parallel() + pool := setupTestPool(t) + ctx := context.Background() + + now := time.Now().UTC() + past := now.Add(-1 * time.Minute) + + // Seed 5 due rows. + for i := 0; i < 5; i++ { + if _, err := pool.Exec(ctx, ` + INSERT INTO posts (status, scheduled_for) VALUES ('scheduled', $1)`, + past); err != nil { + t.Fatalf("seed %d: %v", i, err) + } + } + + // Limit=2 should flip exactly two; the rest roll over. + res, err := PublishScheduled(ctx, pool, PublishOptions{ + Limit: 2, + Now: now, + }) + if err != nil { + t.Fatalf("PublishScheduled: %v", err) + } + if res.Published != 2 { + t.Errorf("Published with Limit=2: got %d, want 2", res.Published) + } + + var remaining int + if err := pool.QueryRow(ctx, + `SELECT count(*) FROM posts WHERE status='scheduled'`). + Scan(&remaining); err != nil { + t.Fatalf("count: %v", err) + } + if remaining != 3 { + t.Errorf("remaining scheduled: got %d, want 3", remaining) + } +} + +// ptr returns a pointer to a time value. Helper for the seed table +// in TestPublishScheduled_FlipsEligibleRows. +func ptr(t time.Time) *time.Time { return &t } + +// nullableTime returns a non-nil *time.Time when ok=true, else nil. +// The pgx adapter accepts (*time.Time)(nil) as a SQL NULL. +func nullableTime(ok bool, t time.Time) *time.Time { + if !ok { + return nil + } + return &t +} diff --git a/packages/go/jobs/scheduler/seed.go b/packages/go/jobs/scheduler/seed.go new file mode 100644 index 00000000..51befc00 --- /dev/null +++ b/packages/go/jobs/scheduler/seed.go @@ -0,0 +1,80 @@ +package scheduler + +import ( + "errors" + "fmt" + "log/slog" + + "github.com/jackc/pgx/v5/pgxpool" + + "github.com/Singleton-Solution/GoNext/packages/go/jobs/cron" + "github.com/Singleton-Solution/GoNext/packages/go/jobs/taskspec" +) + +// SeedOptions bundles the dependencies SeedDefaults needs to build +// both task specs. +type SeedOptions struct { + // Pool is the pgx connection pool both tasks read/write + // against. Required. + Pool *pgxpool.Pool + + // Logger is the structured logger handed to both handlers. + Logger *slog.Logger +} + +// SeedDefaults registers the publisher and GC tasks against the +// supplied task + cron registries. Idempotent: callers that re-seed +// at runtime get a wrapped ErrAlreadyRegistered, which they can +// ignore if the duplicate is benign. +// +// This is the single entry point worker main.go calls — it pairs +// the taskspec and cron registrations in one place so a future task +// addition doesn't require touching two files. +func SeedDefaults( + taskReg *taskspec.Registry, + cronReg *cron.Registry, + opts SeedOptions, +) error { + if taskReg == nil { + return errors.New("scheduler: SeedDefaults: task registry is required") + } + if cronReg == nil { + return errors.New("scheduler: SeedDefaults: cron registry is required") + } + if opts.Pool == nil { + return errors.New("scheduler: SeedDefaults: pool is required") + } + + pubSpec, err := NewPublisherSpec(PublisherSpecOptions{ + Pool: opts.Pool, + Logger: opts.Logger, + }) + if err != nil { + return fmt.Errorf("publisher spec: %w", err) + } + if err := taskReg.Register(pubSpec); err != nil && + !errors.Is(err, taskspec.ErrAlreadyRegistered) { + return fmt.Errorf("register publisher task: %w", err) + } + if err := cronReg.Register(NewPublisherCron()); err != nil && + !errors.Is(err, cron.ErrAlreadyRegistered) { + return fmt.Errorf("register publisher cron: %w", err) + } + + gcSpec, err := NewGCSpec(GCSpecOptions{ + Pool: opts.Pool, + Logger: opts.Logger, + }) + if err != nil { + return fmt.Errorf("gc spec: %w", err) + } + if err := taskReg.Register(gcSpec); err != nil && + !errors.Is(err, taskspec.ErrAlreadyRegistered) { + return fmt.Errorf("register gc task: %w", err) + } + if err := cronReg.Register(NewGCCron()); err != nil && + !errors.Is(err, cron.ErrAlreadyRegistered) { + return fmt.Errorf("register gc cron: %w", err) + } + return nil +}