tapir/cmd/tapir/scheduler.go

package main

import (
	"context"
	"fmt"
	"log/slog"
	"time"

	"gitea.d-ma.be/mathias/tapir/internal/adapters/store"
	"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"},
		MinVideoSeconds:    cfg.MinVideoSeconds,
	}, secretStore)

	engine := usecase.NewEngine(src, buildSummarizer(cfg), st)
	// Share the transcript cache (ADR-021) on the scheduler path too — without
	// this every scheduled pass re-fetches transcripts it already had, burning the
	// scarce per-IP caption budget (ADR-014) and starving other users. The web
	// "Summarize now" path already sets this; the scheduler omitting it was a bug.
	engine.Transcripts = st

	return runner.New(src, st, engine, userID, log,
		runner.WithBackoff(cfg.FetchBackoff),
		runner.WithAutoWindow(cfg.AutoSummarizeWindow)), nil
}

// userLister enumerates every registered user and reports a user's video
// connections. *store.Store satisfies it via ListAllUsers + ConnectionsForUser.
// A small local interface keeps the scheduler testable with a fake.
type userLister interface {
	ListAllUsers(ctx context.Context) ([]store.UserIdentity, error)
	ConnectionsForUser(ctx context.Context, userID string) ([]store.Connection, 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,
	pass int,
	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{}
	}

	// Keep only users with a video connection. A pass for a connectionless user
	// (e.g. a stale Dex-era orphan identity) only tries to resolve a token that
	// was never minted, logging a spurious "ref not found" every tick. Filtering
	// here — BEFORE rotation — also keeps fairness honest: rotation is over the
	// users that actually consume the caption budget, so a dead identity can't eat
	// a rotation slot and skew the lead share.
	var connected []store.UserIdentity
	for _, u := range users {
		if ctx.Err() != nil {
			return runner.Stats{} // shutting down
		}
		conns, err := lister.ConnectionsForUser(ctx, u.UserID)
		if err != nil {
			log.Warn("scheduler: list connections failed", "user", u.UserID, "err", err)
			continue
		}
		if len(conns) == 0 {
			log.Debug("scheduler: skipping user with no video connections", "user", u.UserID)
			continue
		}
		connected = append(connected, u)
	}

	// Rotate who goes first each pass. Caption fetches share one per-egress-IP
	// rate budget (ADR-014); whoever runs first each pass spends the pre-throttle
	// window, so a FIXED order permanently starves whoever is last (a new pilot
	// user got 0 fetches for 12h while the first-listed user got all of them).
	// Rotation over the connected set gives each real user the lead in turn.
	connected = rotateUsers(connected, pass)

	log.Info("scheduler: starting discovery pass", "users", len(connected))
	var total runner.Stats
	for _, u := range connected {
		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_too_old", total.SkippedTooOld,
		"skipped_rate_limited", total.SkippedRateLimited,
		"channel_unavailable", total.ChannelUnavailable, "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
	}

	pass := 0
	runDiscoveryPass(ctx, pass, lister, runUser, log)

	ticker := time.NewTicker(interval)
	defer ticker.Stop()
	for {
		select {
		case <-ctx.Done():
			return
		case <-ticker.C:
			pass++
			runDiscoveryPass(ctx, pass, lister, runUser, log)
		}
	}
}

// rotateUsers left-rotates users by pass positions so a different user leads each
// pass. With n users, user i leads on every pass where pass ≡ i (mod n). A pass
// offset that is negative or exceeds n is normalised. Order within the rotation
// is otherwise preserved, so the set of users run is unchanged — only who is
// first (and thus wins the scarce caption-fetch budget) rotates.
func rotateUsers(users []store.UserIdentity, pass int) []store.UserIdentity {
	n := len(users)
	if n <= 1 {
		return users
	}
	off := ((pass % n) + n) % n
	if off == 0 {
		return users
	}
	out := make([]store.UserIdentity, 0, n)
	out = append(out, users[off:]...)
	out = append(out, users[:off]...)
	return out
}

// 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,
		SkippedTooOld:      a.SkippedTooOld + b.SkippedTooOld,
		SkippedRateLimited: a.SkippedRateLimited + b.SkippedRateLimited,
		ChannelUnavailable: a.ChannelUnavailable + b.ChannelUnavailable,
		Errors:             a.Errors + b.Errors,
	}
}