tapir/internal/runner/runner.go

// 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"

	"gitea.d-ma.be/mathias/tapir/internal/domain"
	"gitea.d-ma.be/mathias/tapir/internal/ports"
	"gitea.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
	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
}

// 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
}

// 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 } }

// 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
	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)
		}
	}

	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
			}

			// 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, 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++
			}
			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++
			}
			// 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,
			"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()
		}
	}
}