feat(usecase): add subscription-driven watch loop with per-video dedup

ProcessNewVideos walks a user's subscriptions and processes each newly
seen video. Two remaining .feature scenarios are now covered: a channel
the user is not subscribed to is never surfaced (so never processed), and
a video already processed in this engine's lifetime is not summarized
twice. Dedup is in-memory and per-user; durable cross-restart dedup stays
the store's concern (no new port), per docs/data-model.md.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

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

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