tapir/cmd/tapir/processor.go

package main

import (
	"context"
	"fmt"
	"strings"

	"git.d-ma.be/mathias/tapir/internal/adapters/chat"
	"git.d-ma.be/mathias/tapir/internal/adapters/llm"
	"git.d-ma.be/mathias/tapir/internal/adapters/secrets"
	"git.d-ma.be/mathias/tapir/internal/adapters/store"
	"git.d-ma.be/mathias/tapir/internal/adapters/summarizer"
	"git.d-ma.be/mathias/tapir/internal/adapters/youtube"
	"git.d-ma.be/mathias/tapir/internal/config"
	"git.d-ma.be/mathias/tapir/internal/domain"
	"git.d-ma.be/mathias/tapir/internal/metrics"
	"git.d-ma.be/mathias/tapir/internal/ports"
	"git.d-ma.be/mathias/tapir/internal/usecase"
	"git.d-ma.be/mathias/tapir/internal/web"
)

// videoFetcher adapts the YouTube adapter to web.VideoFetcher for the paste flow
// (Feature 2). It builds a per-user adapter bound to that user's token ref and
// resolves a single video's metadata via the Data API — ungated; only the later
// transcript fetch goes through globalFetchGate.
type videoFetcher struct {
	cfg     config.Config
	secrets ports.SecretStore
}

func (f videoFetcher) FetchVideo(ctx context.Context, userID, videoID string) (domain.Video, error) {
	a := youtube.New(youtube.Config{
		ClientID:       f.cfg.YTClientID,
		ClientSecret:   f.cfg.YTClientSecret,
		TokenSecretRef: web.YouTubeTokenRef(userID),
	}, f.secrets)
	return a.VideoByID(ctx, userID, videoID)
}

// buildProcessor wires the summarization engine — YouTube source (captions-first),
// AI-router summarizer, store sink — shared by `tapir run` and the web
// "Summarize now" path so the wiring lives in one place. It returns (nil, nil) —
// not an error — when the config cannot support live summarization (no gateway
// URL, no YouTube client credentials, or no secrets file). That nil is the
// queue-only fallback: the web UI keeps working (the button just queues) and
// `tapir run` reports the gap via its own ValidateForRun. Missing engine config
// is never an error here.
// buildSummarizer wires the summarization endpoint chain (ADR-022) shared by the
// web "Summarize now" path and the scheduler's per-user runners. The chain is:
// primary (local, fast) → local fallback → cloud fallback (worst case). Each
// endpoint reaches the same LiteLLM gateway with a different model alias — the
// gateway fronts both llama-swap and berget — so a fallback is just a different
// alias, not a second client config. Empty model entries are skipped, so a
// client deployment can set the cloud fallback empty to keep content local.
func buildSummarizer(cfg config.Config) *summarizer.Summarizer {
	mk := summarizerEndpoint(cfg)
	eps := []summarizer.Endpoint{mk(cfg.SummarizerModel)}
	if cfg.FallbackModel != "" && cfg.FallbackModel != cfg.SummarizerModel {
		eps = append(eps, mk(cfg.FallbackModel))
	}
	if cfg.CloudFallbackModel != "" && cfg.CloudFallbackModel != cfg.SummarizerModel {
		eps = append(eps, mk(cfg.CloudFallbackModel))
	}
	return summarizer.NewChain(eps, cfg.MaxTranscriptChars)
}

// summarizerEndpoint returns a constructor for a chain endpoint over the one
// LiteLLM gateway, varying only the model alias (the gateway fronts both
// llama-swap and berget). Shared by the standard and burst chains.
func summarizerEndpoint(cfg config.Config) func(model string) summarizer.Endpoint {
	return func(model string) summarizer.Endpoint {
		return summarizer.Endpoint{
			Client:   llm.New(cfg.GatewayURL, cfg.GatewayKey, model, cfg.SummarizerTimeout, llm.WithMaxTokens(cfg.SummaryMaxTokens), llm.WithUsageHook(metrics.RecordTokens)),
			Provider: providerOf(model),
			Model:    model,
		}
	}
}

// burstChainModels is the ordered, deduped model list for the onboarding burst
// (ADR-028): the stronger onboard model leads, then the standard ADR-022 chain
// (primary → local fallback → cloud) follows as resilience. Empty entries are
// dropped and duplicates collapsed, so the NDA lever (empty cloud fallback) keeps
// the burst chain fully local exactly as the standard chain does.
func burstChainModels(cfg config.Config) []string {
	var models []string
	add := func(m string) {
		if m == "" {
			return
		}
		for _, e := range models {
			if e == m {
				return
			}
		}
		models = append(models, m)
	}
	add(cfg.OnboardSummarizerModel)
	add(cfg.SummarizerModel)
	add(cfg.FallbackModel)
	add(cfg.CloudFallbackModel)
	return models
}

// buildBurstSummarizer builds the onboarding-burst summarizer chain (ADR-028):
// the onboard model first, then the standard chain as fallback, deduped.
func buildBurstSummarizer(cfg config.Config) *summarizer.Summarizer {
	mk := summarizerEndpoint(cfg)
	var eps []summarizer.Endpoint
	for _, m := range burstChainModels(cfg) {
		eps = append(eps, mk(m))
	}
	return summarizer.NewChain(eps, cfg.MaxTranscriptChars)
}

// chatModels is the ordered, local-first set of models offered in the chat
// switcher (ADR-027), reusing the ADR-022 chain: primary → local fallback →
// cloud. Empty entries are dropped and duplicates collapsed, so a client/NDA
// deployment that sets the cloud fallback empty simply has no external model in
// the switcher — the same local-first lever the summarizer honours.
func chatModels(cfg config.Config) []string {
	var models []string
	add := func(m string) {
		if m == "" {
			return
		}
		for _, e := range models {
			if e == m {
				return
			}
		}
		models = append(models, m)
	}
	add(cfg.SummarizerModel)
	add(cfg.FallbackModel)
	add(cfg.CloudFallbackModel)
	return models
}

// uploadChatModels is the chat switcher for a PRIVATE uploaded recording: the same
// ordered set as chatModels, local models only. The whole transcript (up to 90,000
// characters) is sent to the one chosen, so a cloud model the video chat offers must not
// be offered here; with none local, none is offered (ADR-035 review).
func uploadChatModels(cfg config.Config) []string {
	var local []string
	for _, m := range chatModels(cfg) {
		if config.IsLocalModel(m) {
			local = append(local, m)
		}
	}
	return local
}

// buildChat wires the per-video chat service (ADR-027): a Completer factory over
// the SAME LiteLLM gateway the summarizer uses (a different alias per model, not a
// second client config) and the same transcript-truncation budget. It returns nil
// when no gateway is configured — chat is simply not mounted, the read path is
// unaffected. It deliberately takes NO YouTube source: chat is stored-only.
func buildChat(cfg config.Config) *chat.Service {
	if cfg.GatewayURL == "" {
		return nil
	}
	models := chatModels(cfg)
	if len(models) == 0 {
		return nil
	}
	newClient := func(model string) chat.Completer {
		return llm.New(cfg.GatewayURL, cfg.GatewayKey, model, cfg.SummarizerTimeout, llm.WithMaxTokens(cfg.SummaryMaxTokens), llm.WithUsageHook(metrics.RecordTokens))
	}
	return chat.New(newClient, models, cfg.MaxTranscriptChars)
}

// providerOf maps a model alias to the domain AIProvider recorded on summaries.
// A "berget/" alias is an external provider; everything else is the local stack.
func providerOf(model string) string {
	if strings.HasPrefix(model, "berget/") {
		return "berget"
	}
	return "local"
}

func buildProcessor(cfg config.Config, st *store.Store) (*usecase.Engine, error) {
	if cfg.GatewayURL == "" || cfg.YTClientID == "" || cfg.YTClientSecret == "" || cfg.SecretsFile == "" {
		return nil, nil
	}

	src := newYouTubeSource(cfg, secrets.NewFileStore(cfg.SecretsFile))
	sum := buildSummarizer(cfg)

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

// newYouTubeSource builds the captions-first VideoSource shared by the standard
// and burst processors — same per-process YouTube credentials and ADR-023 Shorts
// filter; only the summarizer chain differs between them.
func newYouTubeSource(cfg config.Config, secretStore ports.SecretStore) ports.VideoSource {
	return youtube.New(youtube.Config{
		ClientID:           cfg.YTClientID,
		ClientSecret:       cfg.YTClientSecret,
		TokenSecretRef:     cfg.YTTokenRef,
		PreferredLanguages: []string{"en"},
		MinVideoSeconds:    cfg.MinVideoSeconds,
	}, secretStore)
}

// buildBurstProcessor wires a processor whose summarizer leads with the stronger
// onboard model (ADR-028), used only by the connect-time burst over the SAME
// store / transcript cache / sink — a wiring choice; the engine and ports are
// unchanged. Returns (nil, nil) — the collapse lever — when the onboard model is
// empty or equal to the primary (the burst then reuses the shared processor), or
// when the engine config is incomplete (queue-only, same as buildProcessor).
func buildBurstProcessor(cfg config.Config, st *store.Store) (*usecase.Engine, error) {
	if cfg.OnboardSummarizerModel == "" || cfg.OnboardSummarizerModel == cfg.SummarizerModel {
		return nil, nil
	}
	if cfg.GatewayURL == "" || cfg.YTClientID == "" || cfg.YTClientSecret == "" || cfg.SecretsFile == "" {
		return nil, nil
	}
	src := newYouTubeSource(cfg, secrets.NewFileStore(cfg.SecretsFile))
	eng := usecase.NewEngine(src, buildBurstSummarizer(cfg), st)
	eng.Transcripts = st
	return eng, nil
}

// engineProcessor adapts the engine (which works in terms of a domain.Video) to
// the web.Processor port (which works in terms of a stored video id): it loads the
// video row, runs the engine, and — on a produced summary — clears the manual
// queue flag, mirroring the runner so the video is not re-summarized on the next
// `tapir run` and the UI drops the "Queued" chip. A skip (no transcript) leaves
// the flag set so a later run can retry.
type engineProcessor struct {
	engine *usecase.Engine
	store  *store.Store
}

func (p *engineProcessor) ProcessVideo(ctx context.Context, userID, videoID string) error {
	// This is the user-initiated (foreground) path — a click on "Summarize",
	// "Try now", or a pasted URL. Mark the context so the caption gate gives it
	// priority over the background sweep (ADR-026, Pillar A).
	ctx = youtube.ForegroundContext(ctx)

	row, err := p.store.GetVideoRow(ctx, userID, videoID)
	if err != nil {
		return fmt.Errorf("load video %q: %w", videoID, err)
	}

	v := domain.Video{
		ID:              row.VideoID,
		UserID:          userID,
		Provider:        domain.Provider(row.Channel),
		ProviderVideoID: row.ProviderVideoID,
		Title:           row.Title,
		URL:             row.URL,
		PublishedAt:     row.PublishedAt,
	}

	res, err := p.engine.ProcessNewVideo(ctx, v)
	if err != nil {
		return fmt.Errorf("process video %q: %w", videoID, err)
	}

	// Record the outcome so the status endpoint can show honest state (ADR-025):
	// a 429'd or caption-less click used to leave transcript_status unset, so the
	// poll silently reverted to the "Summarize" button. Mirror the runner: stamp
	// rate_limited / none / fetched. A rate-limited video keeps its requested flag
	// so the background sweep retries it; none and fetched are terminal here.
	switch {
	case res.Skipped && res.TranscriptSource == string(domain.SourceRateLimited):
		if err := p.store.SetTranscriptStatus(ctx, userID, videoID, "rate_limited"); err != nil {
			return fmt.Errorf("set rate_limited status %q: %w", videoID, err)
		}
	case res.Skipped:
		if err := p.store.SetTranscriptStatus(ctx, userID, videoID, "none"); err != nil {
			return fmt.Errorf("set none status %q: %w", videoID, err)
		}
		if err := p.store.ClearSummarizeRequested(ctx, userID, videoID); err != nil {
			return fmt.Errorf("clear summarize flag %q: %w", videoID, err)
		}
	case res.Summary != nil:
		if err := p.store.SetTranscriptStatus(ctx, userID, videoID, "fetched"); err != nil {
			return fmt.Errorf("set fetched status %q: %w", videoID, err)
		}
		if err := p.store.ClearSummarizeRequested(ctx, userID, videoID); err != nil {
			return fmt.Errorf("clear summarize flag %q: %w", videoID, err)
		}
	}
	return nil
}