tapir/internal/usecase/engine.go

// 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")
	// TranscriptSource is how the transcript resolved (or that there was none):
	// the domain.TranscriptSource value as a string. The runner reads it to tell a
	// permanent absence (SourceNone) from a transient 429 (SourceRateLimited) and
	// persist the right transcript_status. Empty when a fetch error short-circuits.
	TranscriptSource string
	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).
		// Surface the source so the runner separates SourceNone (permanent) from
		// SourceRateLimited (retry after a backoff window).
		return ProcessResult{Video: v, Skipped: true, Reason: "no transcript", TranscriptSource: string(t.Source)}, 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, TranscriptSource: string(t.Source)}, 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
}