// 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 ( "context" "errors" "fmt" "log/slog" "os" "time" "gitea.d-ma.be/mathias/tapir/internal/domain" "gitea.d-ma.be/mathias/tapir/internal/ports" "gitea.d-ma.be/mathias/tapir/internal/usecase" ) // 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 } // 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 } // 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) *Runner { if log == nil { log = slog.Default() } return &Runner{src: src, store: store, engine: engine, userID: userID, log: log} } // 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 Errors int } // RunOnce performs a single pass over the user's subscriptions. Per-item errors // are logged and collected (one bad video or channel does not abort the pass) // and returned joined alongside the Stats gathered. func (r *Runner) RunOnce(ctx context.Context) (Stats, error) { var ( stats Stats errs []error ) // 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 // a full pass under the radar. 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 // (the original behavior). Manual = still discover/persist videos so the user // sees them, but only summarize the ones explicitly queued via the web UI // (summarize_requested). The queued set is loaded once per pass, like seen. auto, err := r.store.GetAutoSummarize(ctx, r.userID) if err != nil { return stats, fmt.Errorf("runner: load summarize mode: %w", err) } var requested map[string]bool if !auto { requested, err = r.store.RequestedVideoIDs(ctx, r.userID) if err != nil { return stats, fmt.Errorf("runner: load requested videos: %w", err) } } subs, err := r.src.ListSubscriptions(ctx, r.userID) if err != nil { return stats, fmt.Errorf("runner: list subscriptions: %w", err) } for _, sub := range subs { vids, err := r.src.NewVideos(ctx, sub) if err != nil { 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 // keep the dedup/FK key consistent with config 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 // also guard against the same video within this pass // Manual mode: skip summarization for videos the user has not queued. // Discovery already happened (UpsertVideo above), so the new video is // visible in the list; it just isn't summarized until requested. if !auto && !requested[id] { stats.SkippedManual++ continue } if fetchDelay > 0 { time.Sleep(fetchDelay) } res, err := r.engine.ProcessNewVideo(ctx, v) if err != nil { errs = append(errs, fmt.Errorf("process %q: %w", v.ProviderVideoID, err)) stats.Errors++ continue } switch { case res.Skipped: stats.SkippedNoText++ r.log.Info("skipped video (no transcript)", "video", v.ProviderVideoID, "title", v.Title) case res.Summary != nil: stats.Summarized++ // In manual mode the video was processed because it was queued; // clear the flag so it is not re-summarized and the UI drops the // "Queued" chip. (Auto mode never sets the flag.) if !auto { if err := r.store.ClearSummarizeRequested(ctx, r.userID, id); err != nil { errs = append(errs, fmt.Errorf("clear summarize flag %q: %w", v.ProviderVideoID, err)) stats.Errors++ } } r.log.Info("summarized video", "video", v.ProviderVideoID, "title", 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, "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() } } }