feat(serve): in-process scheduled discovery for all users (ADR-018)

Stage 0's "returns and reads in >=2 weeks" gate can't be met while discovery is host-side manual (`tapir run`): a newly onboarded user sees an empty list and never comes back. Make Tapir watch on its own. cmdServe launches a background goroutine (when TAPIR_DISCOVERY_INTERVAL > 0) that runs a discovery pass for ALL users on that cadence: enumerate via the un-RLS'd ListAllUsers, then run each user's pass through the EXISTING runner.Runner — the only new code is the per-user loop, not a new scheduler. Run-once-on-startup then ticked; ctx-cancelled on SIGTERM; per-user failures (including buildUserRunner errors) are logged and skipped so one bad user never aborts the rest. interval <= 0 disables it entirely (dev/tests). buildUserRunner binds each runner to that user's own YouTube refresh token (web.YouTubeTokenRef) — the Stage-1 per-tenant ref — reusing buildProcessor's engine wiring. SetFetchRate is also wired in cmdServe so the click-path shares the gate. SINGLE-REPLICA is now load-bearing: the loop lives in the web process, so >1 replica double-runs discovery (429s + duplicate work). Documented in cmdServe and warned at startup. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
2026-06-05 23:39:39 +02:00
parent 149ec2adae
commit f6623afd41
3 changed files with 312 additions and 0 deletions
@@ -165,6 +165,11 @@ func cmdServe(ctx context.Context, log *slog.Logger) error {
 	}
 	defer st.Close()

+	// Process-wide caption-fetch rate gate (ADR-014 item 2): the web click-path
+	// and the scheduled-discovery runners share one per-egress-IP limiter so they
+	// cannot collectively trip 429s. Must be set before either path fetches.
+	youtube.SetFetchRate(cfg.FetchRate)
+
 	// Auth seam (handlers depend on web.Auth only). With Dex configured
 	// (TAPIR_OIDC_ISSUER set) serve uses real OIDC login — any Dex subject may
 	// authenticate, then registers a tapir user (ADR-012); otherwise it falls
@@ -234,6 +239,29 @@ func cmdServe(ctx context.Context, log *slog.Logger) error {
 		log.Info("web summarization is queue-only (incomplete engine config)")
 	}

+	// In-process scheduled discovery (ADR-018): when enabled, a background
+	// goroutine runs a discovery pass for ALL users on TAPIR_DISCOVERY_INTERVAL,
+	// reusing the per-user runner.Runner. Cancelled by the same ctx as the server.
+	//
+	// SINGLE-REPLICA ASSUMPTION (load-bearing): this loop lives in the web process.
+	// Running serve at >1 replica would make every replica fetch every user in
+	// parallel — duplicate work and self-inflicted 429s. replicas: 1 is required in
+	// the deployment manifest; scaling up needs a CronJob or leader election first.
+	if cfg.DiscoveryInterval > 0 {
+		log.Info("scheduled discovery enabled", "interval", cfg.DiscoveryInterval, "fetch_rate", cfg.FetchRate)
+		log.Warn("scheduled discovery assumes a SINGLE replica — running serve at >1 replica double-runs discovery (ADR-018)")
+		runUser := func(ctx context.Context, userID string) (runner.Stats, error) {
+			r, err := buildUserRunner(cfg, st, secretStore, userID, log)
+			if err != nil {
+				return runner.Stats{}, err
+			}
+			return r.RunOnce(ctx)
+		}
+		go runScheduler(ctx, cfg.DiscoveryInterval, st, runUser, log)
+	} else {
+		log.Info("scheduled discovery disabled (TAPIR_DISCOVERY_INTERVAL unset or 0)")
+	}
+
 	srv := &http.Server{
 		Addr:              cfg.HTTPAddr,
 		Handler:           app.Router(),
@@ -0,0 +1,135 @@
+package main
+
+import (
+	"context"
+	"fmt"
+	"log/slog"
+	"time"
+
+	"gitea.d-ma.be/mathias/tapir/internal/adapters/llm"
+	"gitea.d-ma.be/mathias/tapir/internal/adapters/store"
+	"gitea.d-ma.be/mathias/tapir/internal/adapters/summarizer"
+	"gitea.d-ma.be/mathias/tapir/internal/adapters/youtube"
+	"gitea.d-ma.be/mathias/tapir/internal/config"
+	"gitea.d-ma.be/mathias/tapir/internal/ports"
+	"gitea.d-ma.be/mathias/tapir/internal/runner"
+	"gitea.d-ma.be/mathias/tapir/internal/usecase"
+	"gitea.d-ma.be/mathias/tapir/internal/web"
+)
+
+// buildUserRunner constructs a runner.Runner for one user, reusing the same
+// engine wiring as buildProcessor but bound to that user's own YouTube refresh
+// token (web.YouTubeTokenRef(userID)) — the Stage-1 per-tenant ref, not the
+// Stage-0 single ref. It returns an error (not nil) when the global config can't
+// support live summarization (gateway, YouTube client creds, secrets file), so
+// the scheduler can skip that user gracefully. A user who simply hasn't connected
+// YouTube yet builds fine here; their token ref fails to resolve at RunOnce time,
+// surfacing as a per-user error the scheduler logs and skips.
+func buildUserRunner(cfg config.Config, st *store.Store, secretStore ports.SecretStore, userID string, log *slog.Logger) (*runner.Runner, error) {
+	if cfg.GatewayURL == "" || cfg.YTClientID == "" || cfg.YTClientSecret == "" || cfg.SecretsFile == "" {
+		return nil, fmt.Errorf("buildUserRunner: incomplete summarization config (gateway, youtube credentials, secrets file)")
+	}
+
+	src := youtube.New(youtube.Config{
+		ClientID:           cfg.YTClientID,
+		ClientSecret:       cfg.YTClientSecret,
+		TokenSecretRef:     web.YouTubeTokenRef(userID),
+		PreferredLanguages: []string{"en"},
+	}, secretStore)
+
+	primary := summarizer.Endpoint{
+		Client:   llm.New(cfg.GatewayURL, cfg.GatewayKey, cfg.SummarizerModel, cfg.SummarizerTimeout),
+		Provider: "local",
+		Model:    cfg.SummarizerModel,
+	}
+	engine := usecase.NewEngine(src, summarizer.New(primary, nil), st)
+
+	return runner.New(src, st, engine, userID, log, runner.WithBackoff(cfg.FetchBackoff)), nil
+}
+
+// userLister enumerates every registered user. *store.Store satisfies it via
+// ListAllUsers. A small local interface keeps the scheduler testable with a fake.
+type userLister interface {
+	ListAllUsers(ctx context.Context) ([]store.UserIdentity, error)
+}
+
+// runDiscoveryPass runs one discovery pass for every user. runUser performs a
+// single user's pass (production: build a runner and RunOnce). Per-user failures
+// — including a buildUserRunner error or a RunOnce error — are logged and skipped
+// so one bad user, channel, or video never aborts the others (ADR-018 failure
+// isolation). Returns the stats summed across users.
+func runDiscoveryPass(
+	ctx context.Context,
+	lister userLister,
+	runUser func(context.Context, string) (runner.Stats, error),
+	log *slog.Logger,
+) runner.Stats {
+	users, err := lister.ListAllUsers(ctx)
+	if err != nil {
+		log.Error("scheduler: list users failed", "err", err)
+		return runner.Stats{}
+	}
+
+	log.Info("scheduler: starting discovery pass", "users", len(users))
+	var total runner.Stats
+	for _, u := range users {
+		if ctx.Err() != nil {
+			break // shutting down: stop enumerating
+		}
+		stats, err := runUser(ctx, u.UserID)
+		total = sumStats(total, stats)
+		if err != nil {
+			log.Warn("scheduler: user discovery pass had errors", "user", u.UserID, "err", err)
+		}
+	}
+	log.Info("scheduler: pass complete",
+		"candidates", total.Candidates, "summarized", total.Summarized,
+		"skipped_seen", total.SkippedSeen, "skipped_no_text", total.SkippedNoText,
+		"skipped_manual", total.SkippedManual, "skipped_rate_limited", total.SkippedRateLimited,
+		"errors", total.Errors)
+	return total
+}
+
+// runScheduler runs a discovery pass on startup, then once every interval until
+// ctx is cancelled (pod SIGTERM exits the loop cleanly). A non-positive interval
+// disables scheduling entirely (no startup pass) so dev/tests never auto-fetch.
+// It reuses the existing runner.Runner via runUser — the only new behaviour over
+// runner.Loop is iterating all users per tick (ADR-018).
+func runScheduler(
+	ctx context.Context,
+	interval time.Duration,
+	lister userLister,
+	runUser func(context.Context, string) (runner.Stats, error),
+	log *slog.Logger,
+) {
+	if interval <= 0 {
+		return // disabled
+	}
+
+	runDiscoveryPass(ctx, lister, runUser, log)
+
+	ticker := time.NewTicker(interval)
+	defer ticker.Stop()
+	for {
+		select {
+		case <-ctx.Done():
+			return
+		case <-ticker.C:
+			runDiscoveryPass(ctx, lister, runUser, log)
+		}
+	}
+}
+
+// sumStats adds two passes' stats field-wise, so runDiscoveryPass can report a
+// per-tick aggregate across all users.
+func sumStats(a, b runner.Stats) runner.Stats {
+	return runner.Stats{
+		Candidates:         a.Candidates + b.Candidates,
+		Summarized:         a.Summarized + b.Summarized,
+		SkippedSeen:        a.SkippedSeen + b.SkippedSeen,
+		SkippedNoText:      a.SkippedNoText + b.SkippedNoText,
+		SkippedManual:      a.SkippedManual + b.SkippedManual,
+		SkippedRateLimited: a.SkippedRateLimited + b.SkippedRateLimited,
+		Errors:             a.Errors + b.Errors,
+	}
+}
@@ -0,0 +1,149 @@
+package main
+
+import (
+	"context"
+	"errors"
+	"io"
+	"log/slog"
+	"sync"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/require"
+
+	"gitea.d-ma.be/mathias/tapir/internal/adapters/store"
+	"gitea.d-ma.be/mathias/tapir/internal/runner"
+)
+
+func quietLog() *slog.Logger {
+	return slog.New(slog.NewTextHandler(io.Discard, nil))
+}
+
+// fakeLister returns a fixed user set (or an error) for the scheduler under test.
+type fakeLister struct {
+	users []store.UserIdentity
+	err   error
+}
+
+func (f fakeLister) ListAllUsers(context.Context) ([]store.UserIdentity, error) {
+	return f.users, f.err
+}
+
+// countingRunUser records how many passes each user got, optionally failing for
+// specific users, under a mutex so it is safe across the scheduler goroutine.
+type countingRunUser struct {
+	mu      sync.Mutex
+	calls   map[string]int
+	failFor map[string]bool
+}
+
+func newCountingRunUser(failFor ...string) *countingRunUser {
+	c := &countingRunUser{calls: map[string]int{}, failFor: map[string]bool{}}
+	for _, u := range failFor {
+		c.failFor[u] = true
+	}
+	return c
+}
+
+func (c *countingRunUser) run(_ context.Context, userID string) (runner.Stats, error) {
+	c.mu.Lock()
+	defer c.mu.Unlock()
+	c.calls[userID]++
+	if c.failFor[userID] {
+		return runner.Stats{Errors: 1}, errors.New("boom")
+	}
+	return runner.Stats{Summarized: 1}, nil
+}
+
+func (c *countingRunUser) count(userID string) int {
+	c.mu.Lock()
+	defer c.mu.Unlock()
+	return c.calls[userID]
+}
+
+func (c *countingRunUser) total() int {
+	c.mu.Lock()
+	defer c.mu.Unlock()
+	n := 0
+	for _, v := range c.calls {
+		n += v
+	}
+	return n
+}
+
+func usersN(ids ...string) []store.UserIdentity {
+	out := make([]store.UserIdentity, len(ids))
+	for i, id := range ids {
+		out[i] = store.UserIdentity{UserID: id, DexSubject: "dex|" + id}
+	}
+	return out
+}
+
+func TestDiscoveryPassRunsEveryUserOnce(t *testing.T) {
+	lister := fakeLister{users: usersN("a", "b", "c")}
+	rc := newCountingRunUser()
+
+	stats := runDiscoveryPass(context.Background(), lister, rc.run, quietLog())
+
+	require.Equal(t, 1, rc.count("a"))
+	require.Equal(t, 1, rc.count("b"))
+	require.Equal(t, 1, rc.count("c"))
+	require.Equal(t, 3, stats.Summarized, "stats are summed across users")
+}
+
+func TestDiscoveryPassOneUserFailureDoesNotStopOthers(t *testing.T) {
+	lister := fakeLister{users: usersN("a", "b", "c")}
+	rc := newCountingRunUser("b") // user b's pass errors
+
+	stats := runDiscoveryPass(context.Background(), lister, rc.run, quietLog())
+
+	require.Equal(t, 1, rc.count("a"))
+	require.Equal(t, 1, rc.count("b"))
+	require.Equal(t, 1, rc.count("c"), "a failing user must not abort the rest")
+	require.Equal(t, 2, stats.Summarized) // a + c
+	require.Equal(t, 1, stats.Errors)     // b
+}
+
+func TestDiscoveryPassListerErrorIsContained(t *testing.T) {
+	lister := fakeLister{err: errors.New("db down")}
+	rc := newCountingRunUser()
+
+	stats := runDiscoveryPass(context.Background(), lister, rc.run, quietLog())
+
+	require.Equal(t, 0, rc.total(), "no users enumerated → no passes")
+	require.Equal(t, runner.Stats{}, stats)
+}
+
+func TestSchedulerIntervalZeroDisablesEntirely(t *testing.T) {
+	lister := fakeLister{users: usersN("a", "b")}
+	rc := newCountingRunUser()
+
+	runScheduler(context.Background(), 0, lister, rc.run, quietLog())
+
+	require.Equal(t, 0, rc.total(), "interval 0 must not run even a startup pass")
+}
+
+func TestSchedulerRunsStartupPassThenStopsOnCancel(t *testing.T) {
+	lister := fakeLister{users: usersN("a", "b", "c")}
+	rc := newCountingRunUser()
+
+	ctx, cancel := context.WithCancel(context.Background())
+	done := make(chan struct{})
+	go func() {
+		// A long interval so only the startup pass runs before we cancel.
+		runScheduler(ctx, time.Hour, lister, rc.run, quietLog())
+		close(done)
+	}()
+
+	// The startup pass is synchronous at the top of runScheduler; once all three
+	// users have a pass it has completed and the loop is parked on the ticker.
+	require.Eventually(t, func() bool { return rc.total() == 3 }, time.Second, 5*time.Millisecond)
+
+	cancel()
+	select {
+	case <-done:
+	case <-time.After(time.Second):
+		t.Fatal("scheduler did not exit after ctx cancel")
+	}
+	require.Equal(t, 3, rc.total(), "no extra passes fired between startup and cancel")
+}