feat(usecase): read stored transcript before fetching (ADR-021)

The engine now resolves transcripts store-first: a stored transcript — including a stored SourceNone — is summarized without touching YouTube, so re-analysis never re-fetches. On a miss it fetches through the source (caption call still gated, ADR-014) and persists the terminal outcome for the next analysis by any user. A transient SourceRateLimited is surfaced to the runner for per-user backoff but never cached, so persistence can never mask a 429 as a permanent "no transcript". The TranscriptStore is optional (nil → fetch every time), keeping the pure-core and scaffold wiring valid. cmd/tapir wires the store as both summary sink and transcript cache, so `tapir run` and the web summarize path (incl. paste + onboarding) all share the dedup. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
2026-06-09 23:35:04 +02:00
parent cb6917ca59
commit 5c70408e75
3 changed files with 235 additions and 3 deletions
@@ -27,6 +27,13 @@ type Engine struct {
 	AI     ports.Summarizer
 	Sinks  []ports.Sink

+	// Transcripts, when set, is the shared transcript cache (ADR-021): the engine
+	// reads it before any caption fetch and writes resolved transcripts back, so
+	// re-analysis — the same user re-summarizing, or a second user with the same
+	// video — never re-touches YouTube (ADR-010/014). Optional: nil disables
+	// persistence (fetch every time), keeping the pure-core/scaffold wiring valid.
+	Transcripts ports.TranscriptStore
+
 	// 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,
@@ -57,9 +64,9 @@ type ProcessResult struct {
 // resolve transcript -> (summarize -> deliver) | skip.
 // See docs/use-cases/summarize_new_video.feature.
 func (e *Engine) ProcessNewVideo(ctx context.Context, v domain.Video) (ProcessResult, error) {
-	t, err := e.Source.FetchTranscript(ctx, v)
+	t, err := e.resolveTranscript(ctx, v)
 	if err != nil {
-		return ProcessResult{Video: v}, fmt.Errorf("fetch transcript: %w", err)
+		return ProcessResult{Video: v}, err
 	}
 	if !t.HasText() {
 		// No usable transcript: record the skip, produce no summary, deliver nothing
@@ -86,6 +93,39 @@ func (e *Engine) ProcessNewVideo(ctx context.Context, v domain.Video) (ProcessRe
 	return ProcessResult{Video: v, Summary: &sum, TranscriptSource: string(t.Source)}, errors.Join(errs...)
 }

+// resolveTranscript returns v's transcript, reading the shared store first
+// (ADR-021): a stored transcript — including a stored SourceNone (captions
+// permanently absent) — is returned without touching YouTube, so re-analysis
+// never re-fetches. On a store miss it fetches through the source (which gates
+// the caption call, ADR-014) and persists the terminal outcome so the next
+// analysis, for any user, reads from the store. A transient SourceRateLimited is
+// returned to the caller (the runner stamps a per-user backoff) but never stored,
+// so persistence can never mask a 429 as a permanent "no transcript". When no
+// TranscriptStore is wired the engine simply fetches every time.
+func (e *Engine) resolveTranscript(ctx context.Context, v domain.Video) (domain.Transcript, error) {
+	if e.Transcripts != nil {
+		stored, ok, err := e.Transcripts.GetTranscript(ctx, string(v.Provider), v.ProviderVideoID)
+		if err != nil {
+			return domain.Transcript{}, fmt.Errorf("get stored transcript: %w", err)
+		}
+		if ok {
+			return stored, nil
+		}
+	}
+
+	t, err := e.Source.FetchTranscript(ctx, v)
+	if err != nil {
+		return domain.Transcript{}, fmt.Errorf("fetch transcript: %w", err)
+	}
+
+	if e.Transcripts != nil && t.Source != domain.SourceRateLimited {
+		if err := e.Transcripts.SaveTranscript(ctx, string(v.Provider), v.ProviderVideoID, t); err != nil {
+			return domain.Transcript{}, fmt.Errorf("save transcript: %w", err)
+		}
+	}
+	return t, nil
+}
+
 // 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
@@ -0,0 +1,187 @@
+package usecase
+
+import (
+	"context"
+	"testing"
+
+	"gitea.d-ma.be/mathias/tapir/internal/domain"
+)
+
+// These tests pin the ADR-021 read-stored-first behaviour at the engine core:
+// a stored transcript is summarized without re-touching the source, a miss
+// fetches once and persists, and a transient rate-limit is never cached.
+
+type recordingSource struct {
+	transcript domain.Transcript
+	fetchCalls int
+}
+
+func (s *recordingSource) ListSubscriptions(context.Context, string) ([]domain.Subscription, error) {
+	return nil, nil
+}
+
+func (s *recordingSource) NewVideos(context.Context, domain.Subscription) ([]domain.Video, error) {
+	return nil, nil
+}
+
+func (s *recordingSource) FetchTranscript(context.Context, domain.Video) (domain.Transcript, error) {
+	s.fetchCalls++
+	return s.transcript, nil
+}
+
+type fakeTranscriptStore struct {
+	stored map[string]domain.Transcript
+	saves  int
+}
+
+func newFakeTranscriptStore() *fakeTranscriptStore {
+	return &fakeTranscriptStore{stored: make(map[string]domain.Transcript)}
+}
+
+func (f *fakeTranscriptStore) key(provider, id string) string { return provider + "|" + id }
+
+func (f *fakeTranscriptStore) GetTranscript(_ context.Context, provider, id string) (domain.Transcript, bool, error) {
+	t, ok := f.stored[f.key(provider, id)]
+	return t, ok, nil
+}
+
+func (f *fakeTranscriptStore) SaveTranscript(_ context.Context, provider, id string, t domain.Transcript) error {
+	f.saves++
+	f.stored[f.key(provider, id)] = t
+	return nil
+}
+
+type countingSummarizer struct{ calls int }
+
+func (c *countingSummarizer) Summarize(_ context.Context, v domain.Video, _ domain.Transcript) (domain.Summary, error) {
+	c.calls++
+	return domain.Summary{VideoID: v.ID, UserID: v.UserID, Summary: "s", AIProvider: "local"}, nil
+}
+
+type nopSink struct{}
+
+func (nopSink) Name() string                                  { return "nop" }
+func (nopSink) Deliver(context.Context, domain.Summary) error { return nil }
+
+func testVideo() domain.Video {
+	return domain.Video{ID: "v1", UserID: "u1", Provider: domain.ProviderYouTube, ProviderVideoID: "yt1"}
+}
+
+func TestProcessNewVideo_StoredTranscriptSkipsFetch(t *testing.T) {
+	src := &recordingSource{}
+	ts := newFakeTranscriptStore()
+	ts.stored[ts.key("youtube", "yt1")] = domain.Transcript{Source: domain.SourceCaptions, Content: "stored words"}
+	sum := &countingSummarizer{}
+	eng := NewEngine(src, sum, nopSink{})
+	eng.Transcripts = ts
+
+	res, err := eng.ProcessNewVideo(context.Background(), testVideo())
+	if err != nil {
+		t.Fatalf("ProcessNewVideo: %v", err)
+	}
+	if src.fetchCalls != 0 {
+		t.Fatalf("stored transcript must not re-fetch from source; got %d fetches", src.fetchCalls)
+	}
+	if ts.saves != 0 {
+		t.Fatalf("a store hit must not re-save; got %d saves", ts.saves)
+	}
+	if sum.calls != 1 || res.Summary == nil {
+		t.Fatalf("expected a summary from the stored transcript; calls=%d summary=%v", sum.calls, res.Summary)
+	}
+}
+
+func TestProcessNewVideo_StoreMissFetchesAndPersists(t *testing.T) {
+	src := &recordingSource{transcript: domain.Transcript{Source: domain.SourceCaptions, Language: "en", Content: "fetched words"}}
+	ts := newFakeTranscriptStore()
+	sum := &countingSummarizer{}
+	eng := NewEngine(src, sum, nopSink{})
+	eng.Transcripts = ts
+
+	if _, err := eng.ProcessNewVideo(context.Background(), testVideo()); err != nil {
+		t.Fatalf("ProcessNewVideo: %v", err)
+	}
+	if src.fetchCalls != 1 {
+		t.Fatalf("a store miss must fetch exactly once; got %d", src.fetchCalls)
+	}
+	if ts.saves != 1 {
+		t.Fatalf("a fetched transcript must be persisted; got %d saves", ts.saves)
+	}
+	got, ok, _ := ts.GetTranscript(context.Background(), "youtube", "yt1")
+	if !ok || got.Content != "fetched words" {
+		t.Fatalf("persisted transcript not readable back: ok=%v content=%q", ok, got.Content)
+	}
+}
+
+// The second summarize of the same video reads the persisted transcript and does
+// NOT re-fetch — the primary ADR-021 win, proven end to end at the engine.
+func TestProcessNewVideo_SecondSummarizeDoesNotRefetch(t *testing.T) {
+	src := &recordingSource{transcript: domain.Transcript{Source: domain.SourceCaptions, Content: "words"}}
+	ts := newFakeTranscriptStore()
+	eng := NewEngine(src, &countingSummarizer{}, nopSink{})
+	eng.Transcripts = ts
+
+	for i := 0; i < 2; i++ {
+		if _, err := eng.ProcessNewVideo(context.Background(), testVideo()); err != nil {
+			t.Fatalf("pass %d: %v", i, err)
+		}
+	}
+	if src.fetchCalls != 1 {
+		t.Fatalf("the second summarize must reuse the stored transcript; got %d fetches", src.fetchCalls)
+	}
+}
+
+// A stored "no captions" outcome short-circuits before both fetch and summarize.
+func TestProcessNewVideo_StoredNoneSkipsFetchAndSummarize(t *testing.T) {
+	src := &recordingSource{}
+	ts := newFakeTranscriptStore()
+	ts.stored[ts.key("youtube", "yt1")] = domain.Transcript{Source: domain.SourceNone}
+	sum := &countingSummarizer{}
+	eng := NewEngine(src, sum, nopSink{})
+	eng.Transcripts = ts
+
+	res, err := eng.ProcessNewVideo(context.Background(), testVideo())
+	if err != nil {
+		t.Fatalf("ProcessNewVideo: %v", err)
+	}
+	if !res.Skipped {
+		t.Fatal("a stored SourceNone must skip")
+	}
+	if src.fetchCalls != 0 || sum.calls != 0 {
+		t.Fatalf("stored none must neither fetch nor summarize; fetches=%d calls=%d", src.fetchCalls, sum.calls)
+	}
+}
+
+// A transient 429 is surfaced (so the runner backs off per-user) but never cached
+// as a shared terminal state — otherwise it would mask a rate-limit as permanent.
+func TestProcessNewVideo_RateLimitedIsNotPersisted(t *testing.T) {
+	src := &recordingSource{transcript: domain.Transcript{Source: domain.SourceRateLimited}}
+	ts := newFakeTranscriptStore()
+	eng := NewEngine(src, &countingSummarizer{}, nopSink{})
+	eng.Transcripts = ts
+
+	res, err := eng.ProcessNewVideo(context.Background(), testVideo())
+	if err != nil {
+		t.Fatalf("ProcessNewVideo: %v", err)
+	}
+	if !res.Skipped || res.TranscriptSource != string(domain.SourceRateLimited) {
+		t.Fatalf("expected a rate-limited skip; skipped=%v source=%q", res.Skipped, res.TranscriptSource)
+	}
+	if ts.saves != 0 {
+		t.Fatalf("a transient rate-limit must not be persisted; got %d saves", ts.saves)
+	}
+}
+
+// With no TranscriptStore wired the engine fetches every time (back-compat).
+func TestProcessNewVideo_NilStoreFetchesEveryTime(t *testing.T) {
+	src := &recordingSource{transcript: domain.Transcript{Source: domain.SourceCaptions, Content: "words"}}
+	eng := NewEngine(src, &countingSummarizer{}, nopSink{})
+
+	for i := 0; i < 2; i++ {
+		if _, err := eng.ProcessNewVideo(context.Background(), testVideo()); err != nil {
+			t.Fatalf("pass %d: %v", i, err)
+		}
+	}
+	if src.fetchCalls != 2 {
+		t.Fatalf("nil store must fetch every time; got %d", src.fetchCalls)
+	}
+}