// Package usecase holds Tapir's application core: the engine that detects new // videos, resolves transcripts, summarizes, and delivers to sinks. It depends // only on internal/ports and internal/domain. // // This is a SCAFFOLD. Methods return ErrNotImplemented so the acceptance tests // in test/acceptance fail RED. Implementing them to make those tests pass is // the first build task (see the build issue). Behaviour is specified in // docs/use-cases/*.feature. package usecase import ( "context" "errors" "fmt" "sync" "gitea.d-ma.be/mathias/tapir/internal/domain" "gitea.d-ma.be/mathias/tapir/internal/ports" ) // ErrNotImplemented marks scaffold methods awaiting implementation. var ErrNotImplemented = errors.New("not implemented") // Engine is the provider- and sink-agnostic core. 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, processed: make(map[string]bool)} } // ProcessResult reports what happened for one video. type ProcessResult struct { Video domain.Video Skipped bool Reason string // set when Skipped (e.g. "no transcript") Summary *domain.Summary // nil when Skipped } // ProcessNewVideo runs the core use case for a single video: // 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) if err != nil { return ProcessResult{Video: v}, fmt.Errorf("fetch transcript: %w", err) } if !t.HasText() { // No usable transcript: record the skip, produce no summary, deliver nothing // (captions-first, ADR-007; the watcher uses this to avoid reprocessing). return ProcessResult{Video: v, Skipped: true, Reason: "no transcript"}, nil } sum, err := e.AI.Summarize(ctx, v, t) if err != nil { return ProcessResult{Video: v}, fmt.Errorf("summarize: %w", err) } // Sinks fail independently: a failing sink must not abort the others, and // successful deliveries are not dropped. Collect every error, return them joined. var errs []error for _, sink := range e.Sinks { if err := sink.Deliver(ctx, sum); err != nil { errs = append(errs, fmt.Errorf("deliver to %s: %w", sink.Name(), err)) } } 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 }