From 5c70408e75b64b02c9ef5e4a111dc63011bb08a5 Mon Sep 17 00:00:00 2001 From: Mathias Date: Tue, 9 Jun 2026 23:35:04 +0200 Subject: [PATCH] feat(usecase): read stored transcript before fetching (ADR-021) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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) --- cmd/tapir/processor.go | 7 +- internal/usecase/engine.go | 44 ++++- internal/usecase/engine_transcript_test.go | 187 +++++++++++++++++++++ 3 files changed, 235 insertions(+), 3 deletions(-) create mode 100644 internal/usecase/engine_transcript_test.go diff --git a/cmd/tapir/processor.go b/cmd/tapir/processor.go index f4dadea..5a2ba5a 100644 --- a/cmd/tapir/processor.go +++ b/cmd/tapir/processor.go @@ -63,7 +63,12 @@ func buildProcessor(cfg config.Config, st *store.Store) (*usecase.Engine, error) } sum := summarizer.New(primary, nil) - return usecase.NewEngine(src, sum, st), nil + // The store is both the summary sink and the shared transcript cache (ADR-021): + // the engine reads stored transcripts before any caption fetch and writes + // resolved ones back, so re-analysis never re-touches YouTube. + eng := usecase.NewEngine(src, sum, st) + eng.Transcripts = st + return eng, nil } // engineProcessor adapts the engine (which works in terms of a domain.Video) to diff --git a/internal/usecase/engine.go b/internal/usecase/engine.go index 2dafddf..8c7379a 100644 --- a/internal/usecase/engine.go +++ b/internal/usecase/engine.go @@ -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 diff --git a/internal/usecase/engine_transcript_test.go b/internal/usecase/engine_transcript_test.go new file mode 100644 index 0000000..c1fb58f --- /dev/null +++ b/internal/usecase/engine_transcript_test.go @@ -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) + } +}