// 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" "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, and read the already-summarized set. *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) } // 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 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 ) seen, err := r.store.SeenVideoIDs(ctx, r.userID) if err != nil { return stats, fmt.Errorf("runner: load seen 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 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++ 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, "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() } } }