diff --git a/internal/usecase/engine.go b/internal/usecase/engine.go index e47980e..b8d0e56 100644 --- a/internal/usecase/engine.go +++ b/internal/usecase/engine.go @@ -12,6 +12,7 @@ import ( "context" "errors" "fmt" + "sync" "gitea.d-ma.be/mathias/tapir/internal/domain" "gitea.d-ma.be/mathias/tapir/internal/ports" @@ -25,11 +26,18 @@ type Engine struct { Source ports.VideoSource AI ports.Summarizer Sinks []ports.Sink + + // processed dedups videos within this engine's lifetime so a video is not + // summarized twice when the watcher sees it again. Durable cross-restart + // dedup is the store's concern (a resolved TRANSCRIPT / existing SUMMARY, + // see docs/data-model.md), not the engine core's — no port is invented here. + mu sync.Mutex + processed map[string]bool } // NewEngine wires the engine from its ports. func NewEngine(src ports.VideoSource, ai ports.Summarizer, sinks ...ports.Sink) *Engine { - return &Engine{Source: src, AI: ai, Sinks: sinks} + return &Engine{Source: src, AI: ai, Sinks: sinks, processed: make(map[string]bool)} } // ProcessResult reports what happened for one video. @@ -70,3 +78,66 @@ func (e *Engine) ProcessNewVideo(ctx context.Context, v domain.Video) (ProcessRe return ProcessResult{Video: v, Summary: &sum}, errors.Join(errs...) } + +// ProcessNewVideos walks a user's subscriptions and processes each newly seen +// video. Only videos surfaced via the user's subscriptions are considered, so a +// channel the user is not subscribed to is never processed. A video already +// processed in this engine's lifetime is skipped, so it is not summarized twice. +// Per-video errors are collected (one failing video does not abort the rest) and +// returned joined alongside the results gathered. +func (e *Engine) ProcessNewVideos(ctx context.Context, userID string) ([]ProcessResult, error) { + subs, err := e.Source.ListSubscriptions(ctx, userID) + if err != nil { + return nil, fmt.Errorf("list subscriptions: %w", err) + } + + var ( + results []ProcessResult + errs []error + ) + for _, sub := range subs { + vids, err := e.Source.NewVideos(ctx, sub) + if err != nil { + errs = append(errs, fmt.Errorf("new videos for %q: %w", sub.ChannelTitle, err)) + continue + } + for _, v := range vids { + if e.alreadyProcessed(v) { + continue + } + res, err := e.ProcessNewVideo(ctx, v) + if err != nil { + errs = append(errs, err) + } + // Mark processed once the transcript was resolved (summarized or + // skipped), so the watcher does not re-resolve or re-summarize it. + // A fetch/summarize failure leaves it unmarked, allowing a retry. + if res.Summary != nil || res.Skipped { + e.markProcessed(v) + } + results = append(results, res) + } + } + return results, errors.Join(errs...) +} + +func (e *Engine) alreadyProcessed(v domain.Video) bool { + e.mu.Lock() + defer e.mu.Unlock() + return e.processed[processedKey(v)] +} + +func (e *Engine) markProcessed(v domain.Video) { + e.mu.Lock() + defer e.mu.Unlock() + if e.processed == nil { + e.processed = make(map[string]bool) + } + e.processed[processedKey(v)] = true +} + +// processedKey is per-user (per-user isolation, docs/data-model.md): the same +// video seen by two users is two distinct rows and must be keyed separately. +func processedKey(v domain.Video) string { + return v.UserID + "\x00" + v.ID +} diff --git a/test/acceptance/summarize_new_video_test.go b/test/acceptance/summarize_new_video_test.go index 13993fc..95cd975 100644 --- a/test/acceptance/summarize_new_video_test.go +++ b/test/acceptance/summarize_new_video_test.go @@ -116,3 +116,96 @@ func TestVideoWithNoTranscriptIsSkipped(t *testing.T) { t.Fatalf("expected no summary delivery, got %d", len(sink.delivered)) } } + +// --- watch-loop fakes ------------------------------------------------------ + +// watchSource models a provider that knows about videos on several channels but +// only surfaces those on channels the user is subscribed to. NewVideos returns +// the (possibly repeated) videos for a subscription each time it is polled. +type watchSource struct { + subs []domain.Subscription + videosBySub map[string][]domain.Video + transcripts map[string]domain.Transcript // keyed by video ID + newVideoHits int +} + +func (s *watchSource) ListSubscriptions(ctx context.Context, userID string) ([]domain.Subscription, error) { + return s.subs, nil +} +func (s *watchSource) NewVideos(ctx context.Context, sub domain.Subscription) ([]domain.Video, error) { + s.newVideoHits++ + return s.videosBySub[sub.ID], nil +} +func (s *watchSource) FetchTranscript(ctx context.Context, v domain.Video) (domain.Transcript, error) { + return s.transcripts[v.ID], nil +} + +// countingSummarizer records how many times Summarize is invoked. +type countingSummarizer struct{ calls int } + +func (c *countingSummarizer) Summarize(ctx context.Context, v domain.Video, t domain.Transcript) (domain.Summary, error) { + c.calls++ + return domain.Summary{VideoID: v.ID, UserID: v.UserID, Summary: "a summary", AIProvider: "local", CreatedAt: time.Now()}, nil +} + +// Scenario: A channel I am not subscribed to posts a video +func TestUnsubscribedChannelVideoIsNotProcessed(t *testing.T) { + subscribed := domain.Subscription{ID: "s1", UserID: "u1", ChannelID: "acme", ChannelTitle: "Acme Talks", Active: true} + acme := domain.Video{ID: "v1", UserID: "u1", SubscriptionID: "s1", Provider: domain.ProviderYouTube, ProviderVideoID: "yt1", Title: "Designing for Attention", SeenAt: time.Now()} + // "Random Co" video exists in the wider world but the user is not subscribed, + // so its subscription is absent and NewVideos never surfaces it. + src := &watchSource{ + subs: []domain.Subscription{subscribed}, + videosBySub: map[string][]domain.Video{"s1": {acme}}, + transcripts: map[string]domain.Transcript{"v1": {VideoID: "v1", UserID: "u1", Source: domain.SourceCaptions, Language: "en", Content: "transcript text"}}, + } + sum := &countingSummarizer{} + sink := &recordingSink{name: "store"} + eng := usecase.NewEngine(src, sum, sink) + + results, err := eng.ProcessNewVideos(context.Background(), "u1") + if err != nil { + t.Fatalf("ProcessNewVideos returned error: %v", err) + } + if len(results) != 1 { + t.Fatalf("expected exactly 1 video processed (the subscribed one), got %d", len(results)) + } + if results[0].Video.ID != "v1" { + t.Fatalf("expected the subscribed video v1 to be processed, got %q", results[0].Video.ID) + } + for _, d := range sink.delivered { + if d.VideoID == "unrelated" { + t.Fatal("unsubscribed channel's video was processed and delivered") + } + } +} + +// Scenario: The same video is not summarized twice +func TestAlreadySummarizedVideoIsNotReprocessed(t *testing.T) { + subscribed := domain.Subscription{ID: "s1", UserID: "u1", ChannelID: "acme", ChannelTitle: "Acme Talks", Active: true} + video := domain.Video{ID: "v1", UserID: "u1", SubscriptionID: "s1", Provider: domain.ProviderYouTube, ProviderVideoID: "yt1", Title: "Designing for Attention", SeenAt: time.Now()} + src := &watchSource{ + subs: []domain.Subscription{subscribed}, + videosBySub: map[string][]domain.Video{"s1": {video}}, + transcripts: map[string]domain.Transcript{"v1": {VideoID: "v1", UserID: "u1", Source: domain.SourceCaptions, Language: "en", Content: "transcript text"}}, + } + sum := &countingSummarizer{} + sink := &recordingSink{name: "store"} + eng := usecase.NewEngine(src, sum, sink) + + // First watch pass summarizes the video. + if _, err := eng.ProcessNewVideos(context.Background(), "u1"); err != nil { + t.Fatalf("first pass returned error: %v", err) + } + // The watcher sees the same video again on the next pass. + if _, err := eng.ProcessNewVideos(context.Background(), "u1"); err != nil { + t.Fatalf("second pass returned error: %v", err) + } + + if sum.calls != 1 { + t.Fatalf("expected the video to be summarized exactly once, got %d calls", sum.calls) + } + if len(sink.delivered) != 1 { + t.Fatalf("expected exactly 1 delivery, got %d", len(sink.delivered)) + } +}