// Package runner wires the engine to the durable store for the `tapir run` // command. It owns the cross-restart dedup the engine core deliberately does // not: the engine's in-memory processed map is process-lifetime only, so this // loads the store's SeenVideoIDs and skips videos already summarized in a prior // run. It also assigns each video its durable store id (UpsertVideo) before // processing, so the summary's video_id equals the dedup key. // // It depends on small local interfaces (VideoStore, Processor), not concrete // types, so the loop is tested with fakes — no live YouTube, gateway, or PG. package runner import ( "cmp" "context" "errors" "fmt" "log/slog" "os" "slices" "time" "git.d-ma.be/mathias/tapir/internal/domain" "git.d-ma.be/mathias/tapir/internal/ports" "git.d-ma.be/mathias/tapir/internal/usecase" ) // passCandidate is a video that passed all pre-filters (seen/manual/backoff) // and is queued for transcript fetch + summarization in this pass. type passCandidate struct { v domain.Video channelID string // owning channel — keys the caption-availability memory (ADR-024) pos int // discovery position — used as a stable tiebreak when published_at ties } // compareNewestFirst orders candidates by published_at descending, NULLS LAST, // with pos ascending as a stable tiebreak. Videos with a zero published_at // (schema 001: nullable) sort after all dated videos regardless of pos. func compareNewestFirst(a, b passCandidate) int { aNull := a.v.PublishedAt.IsZero() bNull := b.v.PublishedAt.IsZero() switch { case aNull && bNull: return cmp.Compare(a.pos, b.pos) case aNull: return 1 // a is null → after b case bNull: return -1 // b is null → after a } if !a.v.PublishedAt.Equal(b.v.PublishedAt) { if a.v.PublishedAt.After(b.v.PublishedAt) { return -1 // newer first } return 1 } return cmp.Compare(a.pos, b.pos) // same timestamp: preserve discovery order } // VideoStore is the durable persistence the run loop needs: assign a stable id + // metadata, read the already-summarized set, and (for manual summarization mode) // read the user's mode + queued videos and clear a video's queue flag once it has // been summarized. *store.Store satisfies it. type VideoStore interface { UpsertVideo(ctx context.Context, v domain.Video) (string, error) SeenVideoIDs(ctx context.Context, userID string) (map[string]bool, error) GetAutoSummarize(ctx context.Context, userID string) (bool, error) RequestedVideoIDs(ctx context.Context, userID string) (map[string]bool, error) ClearSummarizeRequested(ctx context.Context, userID, videoID string) error // RateLimitedVideoIDs maps the user's still-throttled videos to when they were // rate-limited, so the loop can back off without re-hitting the caption endpoint. RateLimitedVideoIDs(ctx context.Context, userID string) (map[string]time.Time, error) // SetTranscriptStatus records the outcome of a transcript attempt: "none", // "rate_limited" (stamps the backoff clock), or "fetched". SetTranscriptStatus(ctx context.Context, userID, videoID, status string) error // UpsertChannelError records a channel that returned HTTP 404 (deleted/private). // Called when NewVideos returns domain.ErrChannelUnavailable; best-effort, errors // are logged and never abort the pass. UpsertChannelError(ctx context.Context, userID, channelID, channelTitle string) error // CaptionlessChannels returns channel ids currently suppressed because their // recent videos all yielded no captions (ADR-024). The loop skips caption // fetches for these channels' (non-requested) videos. CaptionlessChannels(ctx context.Context, userID string) (map[string]bool, error) // RecordChannelCaptionOutcome updates a channel's caption memory after a fetch: // hadCaptions resets it, otherwise the no-caption streak grows and the channel // is suppressed for window once it reaches threshold. A no-op when threshold<=0. RecordChannelCaptionOutcome(ctx context.Context, userID, channelID string, hadCaptions bool, threshold int, window time.Duration) error } // Processor runs the core use case for a single video. *usecase.Engine // satisfies it. type Processor interface { ProcessNewVideo(ctx context.Context, v domain.Video) (usecase.ProcessResult, error) } // Runner walks a user's subscriptions, persists each candidate video, skips the // ones already summarized (durably), and processes the rest through the engine. type Runner struct { src ports.VideoSource store VideoStore engine Processor userID string log *slog.Logger backoff time.Duration // rate-limit retry window; 0 = always retry autoWindow time.Duration // recency bound for auto-summarize; 0 = no bound now func() time.Time // injectable clock (tests); defaults to time.Now captionThreshold int // consecutive no-caption results before a channel is suppressed; 0 = feature off captionWindow time.Duration // how long a caption-less channel stays suppressed before re-probe } // Option configures a Runner at construction. Variadic so existing call sites // stay valid as new knobs (backoff, clock) are added. type Option func(*Runner) // WithBackoff sets the rate-limit retry window. A video that returned HTTP 429 is // skipped (no caption fetch) until this much time has passed; 0 = always retry. func WithBackoff(d time.Duration) Option { return func(r *Runner) { r.backoff = d } } // WithClock overrides the clock used for backoff comparisons. Tests inject a // fixed time; production leaves the time.Now default. func WithClock(now func() time.Time) Option { return func(r *Runner) { r.now = now } } // WithAutoWindow bounds auto-summarization to videos published within d of now. // In auto mode a video older than d is discovered and listed but not summarized // automatically — it waits for an explicit manual request — so a large // back-catalogue does not self-inflict 429s. An explicitly requested video // bypasses the bound. 0 (the default) disables it (summarize every unseen video). func WithAutoWindow(d time.Duration) Option { return func(r *Runner) { r.autoWindow = d } } // WithCaptionMemory enables per-channel caption-availability suppression // (ADR-024): after threshold consecutive no-caption results a channel's videos // are skipped (no caption fetch) for window, then one is re-probed. threshold<=0 // (the default) disables the feature entirely. func WithCaptionMemory(threshold int, window time.Duration) Option { return func(r *Runner) { r.captionThreshold = threshold; r.captionWindow = window } } // New builds a Runner. A nil logger falls back to slog.Default. func New(src ports.VideoSource, store VideoStore, engine Processor, userID string, log *slog.Logger, opts ...Option) *Runner { if log == nil { log = slog.Default() } r := &Runner{src: src, store: store, engine: engine, userID: userID, log: log, now: time.Now} for _, opt := range opts { opt(r) } if r.now == nil { r.now = time.Now } return r } // Stats summarizes one RunOnce pass. type Stats struct { Candidates int Summarized int SkippedSeen int SkippedNoText int SkippedManual int // discovered but not queued, in manual mode SkippedTooOld int // auto mode: published outside the recency window (not requested) SkippedRateLimited int // 429'd previously and still inside the backoff window SkippedNoCaptionChannel int // channel suppressed as caption-less (ADR-024) Errors int ChannelUnavailable int // channels that returned HTTP 404 (deleted/private) } // tooOld reports whether a video published at publishedAt falls outside the // auto-summarize recency window. A zero window disables the bound, and a zero // publishedAt (undated video) is never aged out — it cannot be dated, so it is // processed rather than silently stranded. func (r *Runner) tooOld(publishedAt time.Time) bool { if r.autoWindow <= 0 || publishedAt.IsZero() { return false } return r.now().Sub(publishedAt) > r.autoWindow } // RunOnce performs a single pass over the user's subscriptions in three phases: // // 1. Discovery: walk all channels, persist each candidate video (UpsertVideo), // apply pre-filters (seen/manual/backoff) — same as before. // 2. Sort: order the surviving candidates newest-first (published_at DESC, NULLS // LAST) so new users get summaries of their most recent, relevant videos first; // the back-catalogue fills in behind across subsequent passes. // 3. Process: feed candidates to the engine in sorted order through the shared // globalFetchGate — the gate is unchanged and still governs honest rate pacing. // // All existing behaviour is preserved: per-item failure isolation, the rate-limit // backoff skip, manual mode, channel-unavailable handling, and stats accounting. // Only the processing order changes within a pass. func (r *Runner) RunOnce(ctx context.Context) (Stats, error) { var ( stats Stats errs []error candidates []passCandidate pos int ) // Throttle transcript fetches: unauthenticated caption scraping gets // soft-throttled by YouTube under heavy back-to-back volume (captionTracks // silently stripped from the player response). A small per-video delay keeps // the fetch rate polite. TAPIR_FETCH_DELAY (Go duration), 0 = off. fetchDelay, _ := time.ParseDuration(os.Getenv("TAPIR_FETCH_DELAY")) seen, err := r.store.SeenVideoIDs(ctx, r.userID) if err != nil { return stats, fmt.Errorf("runner: load seen videos: %w", err) } // Summarization mode (per-user, ADR-012). Auto = summarize every unseen video. // Manual = discover/persist videos (visible in list) but only summarize ones // explicitly queued via the web UI (summarize_requested). auto, err := r.store.GetAutoSummarize(ctx, r.userID) if err != nil { return stats, fmt.Errorf("runner: load summarize mode: %w", err) } // requested is needed in manual mode (the queue) and in auto mode when a // recency window is active (an explicit request bypasses the bound). var requested map[string]bool if !auto || r.autoWindow > 0 { requested, err = r.store.RequestedVideoIDs(ctx, r.userID) if err != nil { return stats, fmt.Errorf("runner: load requested videos: %w", err) } } // Rate-limit backoff: videos that 429'd on a prior pass, mapped to when. // Within the backoff window they are skipped before any caption fetch. // Disabled when backoff <= 0 ("always retry"). var rateLimited map[string]time.Time if r.backoff > 0 { rateLimited, err = r.store.RateLimitedVideoIDs(ctx, r.userID) if err != nil { return stats, fmt.Errorf("runner: load rate-limited videos: %w", err) } } // Per-channel caption memory (ADR-024): channels whose recent videos all // yielded no captions are suppressed so their new videos don't burn the scarce // fetch budget. Loaded only when the feature is enabled (threshold > 0). var captionless map[string]bool if r.captionThreshold > 0 { captionless, err = r.store.CaptionlessChannels(ctx, r.userID) if err != nil { return stats, fmt.Errorf("runner: load caption-less channels: %w", err) } } subs, err := r.src.ListSubscriptions(ctx, r.userID) if err != nil { return stats, fmt.Errorf("runner: list subscriptions: %w", err) } // ── Phase 1: discover, persist, filter ─────────────────────────────────── // Walk all channels. Persist every video (UpsertVideo) so it appears in the // list regardless of whether it will be summarized this pass. Apply pre-filters // and collect surviving candidates with their discovery position. for _, sub := range subs { vids, err := r.src.NewVideos(ctx, sub) if err != nil { var unavail *domain.ErrChannelUnavailable if errors.As(err, &unavail) { stats.ChannelUnavailable++ r.log.Warn("channel unavailable (playlist 404)", "channel", sub.ChannelTitle, "channel_id", sub.ChannelID) if storeErr := r.store.UpsertChannelError(ctx, r.userID, unavail.ChannelID, unavail.ChannelTitle); storeErr != nil { r.log.Warn("failed to store channel error", "err", storeErr) } } else { errs = append(errs, fmt.Errorf("new videos for %q: %w", sub.ChannelTitle, err)) stats.Errors++ } continue } for _, v := range vids { stats.Candidates++ v.UserID = r.userID id, err := r.store.UpsertVideo(ctx, v) if err != nil { errs = append(errs, fmt.Errorf("upsert video %q: %w", v.ProviderVideoID, err)) stats.Errors++ continue } v.ID = id if seen[id] { stats.SkippedSeen++ continue } seen[id] = true // guard against duplicates within this pass // Manual mode: skip summarization for videos the user has not queued. // Discovery already happened (UpsertVideo above), so the video is // visible in the list; it just isn't summarized until requested. if !auto && !requested[id] { stats.SkippedManual++ continue } // Recency bound (auto mode): summarize only recent videos automatically; // older ones are discovered + listed (UpsertVideo above) but wait for an // explicit manual request, so a large back-catalogue does not self-inflict // 429s against the caption rate gate. A requested video bypasses the bound. if auto && !requested[id] && r.tooOld(v.PublishedAt) { stats.SkippedTooOld++ continue } // Caption-less channel (ADR-024): its recent videos all returned no // captions, so skip the fetch entirely. The video is still listed // (UpsertVideo above); an explicit manual request bypasses the skip. if !requested[id] && captionless[sub.ChannelID] { stats.SkippedNoCaptionChannel++ continue } // Still inside the rate-limit backoff window: skip without fetching. if at, ok := rateLimited[id]; ok && r.now().Sub(at) < r.backoff { stats.SkippedRateLimited++ r.log.Info("skipped video (rate-limited, backing off)", "video", v.ProviderVideoID, "title", v.Title) continue } candidates = append(candidates, passCandidate{v: v, channelID: sub.ChannelID, pos: pos}) pos++ } } // ── Phase 2: sort newest-first, NULLS LAST ──────────────────────────────── // Within this pass, process the newest videos first so a new user gets // summaries of their most recent content quickly; the back-catalogue fills in // behind across subsequent passes. Both this background batch and the foreground // "Try now" button honour the shared globalFetchGate — ordering is onboarding // prioritisation, not rate-limit evasion. slices.SortStableFunc(candidates, compareNewestFirst) // ── Phase 3: process in sorted order ───────────────────────────────────── for _, c := range candidates { if fetchDelay > 0 { time.Sleep(fetchDelay) } res, err := r.engine.ProcessNewVideo(ctx, c.v) if err != nil { errs = append(errs, fmt.Errorf("process %q: %w", c.v.ProviderVideoID, err)) stats.Errors++ continue } id := c.v.ID switch { case res.Skipped && res.TranscriptSource == string(domain.SourceRateLimited): // Fresh 429: stamp the backoff clock so the next pass skips it. stats.SkippedRateLimited++ if err := r.store.SetTranscriptStatus(ctx, r.userID, id, "rate_limited"); err != nil { errs = append(errs, fmt.Errorf("set rate_limited status %q: %w", c.v.ProviderVideoID, err)) stats.Errors++ } r.log.Info("skipped video (rate-limited)", "video", c.v.ProviderVideoID, "title", c.v.Title) case res.Skipped: stats.SkippedNoText++ if err := r.store.SetTranscriptStatus(ctx, r.userID, id, "none"); err != nil { errs = append(errs, fmt.Errorf("set none status %q: %w", c.v.ProviderVideoID, err)) stats.Errors++ } // No captions: grow this channel's no-caption streak (ADR-024). if err := r.store.RecordChannelCaptionOutcome(ctx, r.userID, c.channelID, false, r.captionThreshold, r.captionWindow); err != nil { errs = append(errs, fmt.Errorf("record no-caption %q: %w", c.v.ProviderVideoID, err)) stats.Errors++ } r.log.Info("skipped video (no transcript)", "video", c.v.ProviderVideoID, "title", c.v.Title) case res.Summary != nil: stats.Summarized++ if err := r.store.SetTranscriptStatus(ctx, r.userID, id, "fetched"); err != nil { errs = append(errs, fmt.Errorf("set fetched status %q: %w", c.v.ProviderVideoID, err)) stats.Errors++ } // Captions present: reset this channel's caption memory (ADR-024). if err := r.store.RecordChannelCaptionOutcome(ctx, r.userID, c.channelID, true, r.captionThreshold, r.captionWindow); err != nil { errs = append(errs, fmt.Errorf("record has-caption %q: %w", c.v.ProviderVideoID, err)) stats.Errors++ } // In manual mode the video was explicitly queued; clear the flag so // it is not re-summarized and the UI drops the "Queued" chip. if !auto { if err := r.store.ClearSummarizeRequested(ctx, r.userID, id); err != nil { errs = append(errs, fmt.Errorf("clear summarize flag %q: %w", c.v.ProviderVideoID, err)) stats.Errors++ } } r.log.Info("summarized video", "video", c.v.ProviderVideoID, "title", c.v.Title, "provider", res.Summary.AIProvider, "model", res.Summary.AIModel) } } return stats, errors.Join(errs...) } // Loop runs RunOnce immediately, then on every interval tick until ctx is // cancelled. A zero or negative interval means a single pass (no loop). Per-pass // errors are logged, not fatal, so a transient failure doesn't kill the watcher. func (r *Runner) Loop(ctx context.Context, interval time.Duration) error { runPass := func() { stats, err := r.RunOnce(ctx) r.log.Info("run pass complete", "candidates", stats.Candidates, "summarized", stats.Summarized, "skipped_seen", stats.SkippedSeen, "skipped_no_text", stats.SkippedNoText, "skipped_manual", stats.SkippedManual, "skipped_too_old", stats.SkippedTooOld, "skipped_rate_limited", stats.SkippedRateLimited, "skipped_no_caption_channel", stats.SkippedNoCaptionChannel, "channel_unavailable", stats.ChannelUnavailable, "errors", stats.Errors) if err != nil { r.log.Warn("run pass had errors", "err", err) } } runPass() if interval <= 0 { return nil } ticker := time.NewTicker(interval) defer ticker.Stop() for { select { case <-ctx.Done(): return ctx.Err() case <-ticker.C: runPass() } } }