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

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

	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

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