package runner_test import ( "context" "io" "log/slog" "testing" "time" "github.com/stretchr/testify/require" "gitea.d-ma.be/mathias/tapir/internal/domain" "gitea.d-ma.be/mathias/tapir/internal/runner" "gitea.d-ma.be/mathias/tapir/internal/usecase" ) const testUser = "11111111-1111-1111-1111-111111111111" // --- fakes ----------------------------------------------------------------- type fakeSource struct { subs []domain.Subscription videos map[string][]domain.Video // keyed by channel id transcripts map[string]domain.Transcript } func (f *fakeSource) ListSubscriptions(_ context.Context, _ string) ([]domain.Subscription, error) { return f.subs, nil } func (f *fakeSource) NewVideos(_ context.Context, sub domain.Subscription) ([]domain.Video, error) { return f.videos[sub.ChannelID], nil } func (f *fakeSource) FetchTranscript(_ context.Context, v domain.Video) (domain.Transcript, error) { if t, ok := f.transcripts[v.ProviderVideoID]; ok { return t, nil } return domain.Transcript{VideoID: v.ID, UserID: v.UserID, Source: domain.SourceCaptions, Content: "default transcript text"}, nil } // fakeStore assigns deterministic ids ("id-"+provider video id) so a pre-seeded // seen set lines up with UpsertVideo output, modelling cross-restart dedup. // auto controls the summarization mode; requested is the manual-mode queue keyed // by store id; cleared records the ids whose queue flag the runner reset. type fakeStore struct { seen map[string]bool upserted []domain.Video auto bool requested map[string]bool cleared []string rateLimited map[string]time.Time // id -> when 429'd (seeds the backoff window) statuses map[string]string // id -> last SetTranscriptStatus value } func (f *fakeStore) UpsertVideo(_ context.Context, v domain.Video) (string, error) { f.upserted = append(f.upserted, v) return "id-" + v.ProviderVideoID, nil } func (f *fakeStore) SeenVideoIDs(_ context.Context, _ string) (map[string]bool, error) { cp := make(map[string]bool, len(f.seen)) for k, v := range f.seen { cp[k] = v } return cp, nil } func (f *fakeStore) GetAutoSummarize(_ context.Context, _ string) (bool, error) { return f.auto, nil } func (f *fakeStore) RequestedVideoIDs(_ context.Context, _ string) (map[string]bool, error) { cp := make(map[string]bool, len(f.requested)) for k, v := range f.requested { cp[k] = v } return cp, nil } func (f *fakeStore) ClearSummarizeRequested(_ context.Context, _, videoID string) error { f.cleared = append(f.cleared, videoID) return nil } func (f *fakeStore) RateLimitedVideoIDs(_ context.Context, _ string) (map[string]time.Time, error) { cp := make(map[string]time.Time, len(f.rateLimited)) for k, v := range f.rateLimited { cp[k] = v } return cp, nil } func (f *fakeStore) UpsertChannelError(_ context.Context, _, _, _ string) error { return nil } func (f *fakeStore) SetTranscriptStatus(_ context.Context, _, videoID, status string) error { if f.statuses == nil { f.statuses = map[string]string{} } f.statuses[videoID] = status return nil } type fakeSummarizer struct{} func (fakeSummarizer) Summarize(_ context.Context, v domain.Video, _ domain.Transcript) (domain.Summary, error) { return domain.Summary{UserID: v.UserID, VideoID: v.ID, Summary: "s", AIProvider: "local", AIModel: "koala/phi4-mini"}, nil } type recordingSink struct{ delivered []domain.Summary } func (s *recordingSink) Name() string { return "store" } func (s *recordingSink) Deliver(_ context.Context, sum domain.Summary) error { s.delivered = append(s.delivered, sum) return nil } func sub(channelID, title string) domain.Subscription { return domain.Subscription{UserID: testUser, ChannelID: channelID, ChannelTitle: title, Active: true} } func vid(provID, title string) domain.Video { return domain.Video{UserID: testUser, Provider: domain.ProviderYouTube, ProviderVideoID: provID, Title: title} } func vidAt(provID, title string, publishedAt time.Time) domain.Video { v := vid(provID, title) v.PublishedAt = publishedAt return v } func quietLogger() *slog.Logger { return slog.New(slog.NewTextHandler(io.Discard, nil)) } // --- tests ----------------------------------------------------------------- func TestRunOnce_SummarizesNewVideos(t *testing.T) { src := &fakeSource{ subs: []domain.Subscription{sub("chan1", "Channel One")}, videos: map[string][]domain.Video{"chan1": {vid("v1", "Video 1"), vid("v2", "Video 2")}}, } st := &fakeStore{seen: map[string]bool{}, auto: true} sink := &recordingSink{} eng := usecase.NewEngine(src, fakeSummarizer{}, sink) r := runner.New(src, st, eng, testUser, quietLogger()) stats, err := r.RunOnce(context.Background()) require.NoError(t, err) require.Equal(t, 2, stats.Candidates) require.Equal(t, 2, stats.Summarized) require.Equal(t, 0, stats.SkippedSeen) require.Len(t, sink.delivered, 2) // Each delivered summary must carry the durable store id as its video id. require.Equal(t, "id-v1", sink.delivered[0].VideoID) require.Equal(t, "id-v2", sink.delivered[1].VideoID) } func TestRunOnce_SkipsAlreadySummarized(t *testing.T) { src := &fakeSource{ subs: []domain.Subscription{sub("chan1", "Channel One")}, videos: map[string][]domain.Video{"chan1": {vid("v1", "Video 1"), vid("v2", "Video 2")}}, } // v1 was summarized in a prior run (durable seen set). st := &fakeStore{seen: map[string]bool{"id-v1": true}, auto: true} sink := &recordingSink{} eng := usecase.NewEngine(src, fakeSummarizer{}, sink) r := runner.New(src, st, eng, testUser, quietLogger()) stats, err := r.RunOnce(context.Background()) require.NoError(t, err) require.Equal(t, 1, stats.SkippedSeen) require.Equal(t, 1, stats.Summarized) require.Len(t, sink.delivered, 1) require.Equal(t, "id-v2", sink.delivered[0].VideoID, "only the unseen video is summarized") } func TestRunOnce_SkipsVideosWithoutTranscript(t *testing.T) { src := &fakeSource{ subs: []domain.Subscription{sub("chan1", "Channel One")}, videos: map[string][]domain.Video{"chan1": {vid("v1", "Video 1")}}, transcripts: map[string]domain.Transcript{"v1": {Source: domain.SourceNone}}, } st := &fakeStore{seen: map[string]bool{}, auto: true} sink := &recordingSink{} eng := usecase.NewEngine(src, fakeSummarizer{}, sink) r := runner.New(src, st, eng, testUser, quietLogger()) stats, err := r.RunOnce(context.Background()) require.NoError(t, err) require.Equal(t, 1, stats.SkippedNoText) require.Equal(t, 0, stats.Summarized) require.Empty(t, sink.delivered, "no summary delivered when there is no transcript") } func TestRunOnce_ManualMode_SkipsUnrequested(t *testing.T) { src := &fakeSource{ subs: []domain.Subscription{sub("chan1", "Channel One")}, videos: map[string][]domain.Video{"chan1": {vid("v1", "Video 1"), vid("v2", "Video 2")}}, } // Manual mode, nothing queued: discover (upsert) but summarize nothing. st := &fakeStore{seen: map[string]bool{}, auto: false, requested: map[string]bool{}} sink := &recordingSink{} eng := usecase.NewEngine(src, fakeSummarizer{}, sink) r := runner.New(src, st, eng, testUser, quietLogger()) stats, err := r.RunOnce(context.Background()) require.NoError(t, err) require.Equal(t, 2, stats.Candidates) require.Equal(t, 2, stats.SkippedManual, "manual mode skips unqueued videos") require.Equal(t, 0, stats.Summarized) require.Empty(t, sink.delivered, "no summary in manual mode without a request") require.Len(t, st.upserted, 2, "discovery still persists every candidate") } func TestRunOnce_ManualMode_ProcessesRequested(t *testing.T) { src := &fakeSource{ subs: []domain.Subscription{sub("chan1", "Channel One")}, videos: map[string][]domain.Video{"chan1": {vid("v1", "Video 1"), vid("v2", "Video 2")}}, } // Manual mode, v1 queued (by store id). Only v1 is summarized; its flag clears. st := &fakeStore{seen: map[string]bool{}, auto: false, requested: map[string]bool{"id-v1": true}} sink := &recordingSink{} eng := usecase.NewEngine(src, fakeSummarizer{}, sink) r := runner.New(src, st, eng, testUser, quietLogger()) stats, err := r.RunOnce(context.Background()) require.NoError(t, err) require.Equal(t, 1, stats.Summarized, "only the queued video is summarized") require.Equal(t, 1, stats.SkippedManual, "the unqueued video is skipped") require.Len(t, sink.delivered, 1) require.Equal(t, "id-v1", sink.delivered[0].VideoID) require.Equal(t, []string{"id-v1"}, st.cleared, "the queue flag is cleared after summarizing") } // noFetchSource fails the test if a transcript fetch happens — used to prove the // runner skips a rate-limited video before touching the caption endpoint. type noFetchSource struct{ *fakeSource } func (noFetchSource) FetchTranscript(context.Context, domain.Video) (domain.Transcript, error) { panic("FetchTranscript must not be called for a rate-limited video within the backoff window") } func TestRunOnce_SkipsRateLimitedWithinBackoff(t *testing.T) { base := time.Date(2026, 6, 3, 12, 0, 0, 0, time.UTC) src := &fakeSource{ subs: []domain.Subscription{sub("chan1", "Channel One")}, videos: map[string][]domain.Video{"chan1": {vid("v1", "Video 1")}}, } // v1 was rate-limited 5m ago; backoff is 1h, so it is still inside the window. st := &fakeStore{ seen: map[string]bool{}, auto: true, rateLimited: map[string]time.Time{"id-v1": base.Add(-5 * time.Minute)}, } eng := usecase.NewEngine(noFetchSource{src}, fakeSummarizer{}, &recordingSink{}) r := runner.New(noFetchSource{src}, st, eng, testUser, quietLogger(), runner.WithBackoff(time.Hour), runner.WithClock(func() time.Time { return base })) stats, err := r.RunOnce(context.Background()) require.NoError(t, err) require.Equal(t, 1, stats.SkippedRateLimited, "still throttled -> skipped") require.Equal(t, 0, stats.Summarized) require.Empty(t, st.statuses, "no status write: the engine was never invoked") } func TestRunOnce_RetriesRateLimitedAfterBackoff(t *testing.T) { base := time.Date(2026, 6, 3, 12, 0, 0, 0, time.UTC) src := &fakeSource{ subs: []domain.Subscription{sub("chan1", "Channel One")}, videos: map[string][]domain.Video{"chan1": {vid("v1", "Video 1")}}, } // v1 was rate-limited 2h ago; backoff is 1h, so the window has expired. st := &fakeStore{ seen: map[string]bool{}, auto: true, rateLimited: map[string]time.Time{"id-v1": base.Add(-2 * time.Hour)}, } sink := &recordingSink{} eng := usecase.NewEngine(src, fakeSummarizer{}, sink) r := runner.New(src, st, eng, testUser, quietLogger(), runner.WithBackoff(time.Hour), runner.WithClock(func() time.Time { return base })) stats, err := r.RunOnce(context.Background()) require.NoError(t, err) require.Equal(t, 0, stats.SkippedRateLimited, "window expired -> not skipped") require.Equal(t, 1, stats.Summarized, "the video is retried and summarized") require.Len(t, sink.delivered, 1) require.Equal(t, "fetched", st.statuses["id-v1"], "status advances to fetched on success") } func TestRunOnce_UpsertsEveryCandidate(t *testing.T) { src := &fakeSource{ subs: []domain.Subscription{sub("chan1", "Channel One")}, videos: map[string][]domain.Video{"chan1": {vid("v1", "Video 1"), vid("v2", "Video 2")}}, } // Even an already-seen video gets upserted so its metadata stays fresh. st := &fakeStore{seen: map[string]bool{"id-v1": true}, auto: true} eng := usecase.NewEngine(src, fakeSummarizer{}, &recordingSink{}) r := runner.New(src, st, eng, testUser, quietLogger()) _, err := r.RunOnce(context.Background()) require.NoError(t, err) require.Len(t, st.upserted, 2, "every candidate is upserted, including seen ones") } // TestRunOnce_NewestFirstOrdering asserts that within a pass, candidates are // processed newest-first (published_at DESC, NULLS LAST) across all channels, // and that the set of processed videos is identical to what per-channel inline // processing would produce (only the order differs). // // Fixture: two channels, four videos with mixed published_at (one NULL). // // Per-channel (before): chanA=[v-old, v-mid], chanB=[v-new, v-null] // → [v-old, v-mid, v-new, v-null] // Newest-first (after): [v-new, v-mid, v-old, v-null] func TestRunOnce_NewestFirstOrdering(t *testing.T) { old := time.Date(2024, 1, 1, 0, 0, 0, 0, time.UTC) mid := time.Date(2024, 6, 1, 0, 0, 0, 0, time.UTC) newt := time.Date(2024, 12, 1, 0, 0, 0, 0, time.UTC) // zero time = NULL published_at (schema 001: nullable) src := &fakeSource{ subs: []domain.Subscription{ sub("chanA", "Channel A"), sub("chanB", "Channel B"), }, videos: map[string][]domain.Video{ "chanA": { vidAt("v-old", "Old Video", old), vidAt("v-mid", "Mid Video", mid), }, "chanB": { vidAt("v-new", "New Video", newt), vidAt("v-null", "No Date Video", time.Time{}), // NULL }, }, } st := &fakeStore{seen: map[string]bool{}, auto: true} sink := &recordingSink{} eng := usecase.NewEngine(src, fakeSummarizer{}, sink) r := runner.New(src, st, eng, testUser, quietLogger()) stats, err := r.RunOnce(context.Background()) require.NoError(t, err) // Same set: all 4 candidates processed regardless of order. require.Equal(t, 4, stats.Candidates) require.Equal(t, 4, stats.Summarized, "same set of videos processed as per-channel order") require.Len(t, sink.delivered, 4) // Build video-id → delivery-position map. order := make(map[string]int, len(sink.delivered)) for i, s := range sink.delivered { order[s.VideoID] = i t.Logf("position %d: %s", i, s.VideoID) } require.Less(t, order["id-v-new"], order["id-v-mid"], "newest (Dec) before mid (Jun)") require.Less(t, order["id-v-mid"], order["id-v-old"], "mid (Jun) before old (Jan)") require.Less(t, order["id-v-old"], order["id-v-null"], "dated before NULL (NULLS LAST)") } // TestRunOnce_NewestFirstNullsOnly asserts that when all candidates have NULL // published_at, discovery order (stable) is preserved as the tiebreak. func TestRunOnce_NewestFirstNullsOnly(t *testing.T) { src := &fakeSource{ subs: []domain.Subscription{ sub("chanA", "Channel A"), sub("chanB", "Channel B"), }, videos: map[string][]domain.Video{ "chanA": {vidAt("v1", "V1", time.Time{}), vidAt("v2", "V2", time.Time{})}, "chanB": {vidAt("v3", "V3", time.Time{})}, }, } st := &fakeStore{seen: map[string]bool{}, auto: true} sink := &recordingSink{} eng := usecase.NewEngine(src, fakeSummarizer{}, sink) r := runner.New(src, st, eng, testUser, quietLogger()) stats, err := r.RunOnce(context.Background()) require.NoError(t, err) require.Equal(t, 3, stats.Summarized, "all null-date videos processed") // Discovery order: chanA[v1, v2], chanB[v3] → [v1, v2, v3]. // All have NULL published_at so the sort is stable; discovery order must hold. require.Equal(t, "id-v1", sink.delivered[0].VideoID) require.Equal(t, "id-v2", sink.delivered[1].VideoID) require.Equal(t, "id-v3", sink.delivered[2].VideoID) }