tapir/cmd/tapir/main.go

// Command tapir is the service entrypoint and CLI. Subcommands are dispatched
// off os.Args[1]; each lives in its own file (list.go, show.go, …). It serves
// the Stage-0 demo:
//
//	tapir auth   mint a YouTube refresh token interactively (one-time setup)
//	tapir run    detect new videos, summarize them, deliver to the store
//	tapir list   list stored summaries, recent first
//	tapir show   show one stored summary in full
//
// Standalone-vs-homelab is a wiring choice (ADR-003): every dependency is
// resolved from config (internal/config) and the SecretStore port, so live
// credentials plug in at runtime without code changes.
package main

import (
	"context"
	"errors"
	"fmt"
	"log/slog"
	"net/http"
	"os"
	"os/signal"
	"sync"
	"time"

	"gitea.d-ma.be/mathias/tapir/internal/adapters/secrets"
	"gitea.d-ma.be/mathias/tapir/internal/adapters/store"
	"gitea.d-ma.be/mathias/tapir/internal/adapters/youtube"
	"gitea.d-ma.be/mathias/tapir/internal/auth"
	"gitea.d-ma.be/mathias/tapir/internal/config"
	"gitea.d-ma.be/mathias/tapir/internal/metrics"
	"gitea.d-ma.be/mathias/tapir/internal/runner"
	"gitea.d-ma.be/mathias/tapir/internal/web"
	"gitea.d-ma.be/mathias/tapir/internal/web/oidc"
)

func main() {
	if len(os.Args) < 2 {
		usage()
		os.Exit(2)
	}

	log := slog.New(slog.NewTextHandler(os.Stderr, nil))
	ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt)
	defer stop()

	var err error
	switch os.Args[1] {
	case "list":
		err = runList(ctx, os.Args[2:])
	case "show":
		err = runShow(ctx, os.Args[2:])
	case "auth":
		err = cmdAuth(ctx, log)
	case "run":
		err = cmdRun(ctx, log)
	case "serve":
		err = cmdServe(ctx, log)
	case "report":
		err = runReport(ctx, os.Args[2:])
	default:
		usage()
		os.Exit(2)
	}

	if err != nil {
		log.Error("command failed", "command", os.Args[1], "err", err)
		os.Exit(1)
	}
}

func usage() {
	fmt.Fprint(os.Stderr, `tapir — summarize new videos from your YouTube subscriptions

usage:
  tapir auth              one-time: authorize YouTube and store a refresh token
  tapir run               detect new videos, summarize, deliver to your store
  tapir serve             run the web UI (read summaries, record watch/skip/save)
  tapir list [-limit N]   list stored summaries, recent first
  tapir show <video-id>   show one summary in full
  tapir report            Stage-0 usage gate: per-user distinct active weeks

configuration is via TAPIR_* environment variables (see .env.example).
`)
}

// cmdAuth runs the interactive OAuth flow and persists the refresh token.
func cmdAuth(ctx context.Context, log *slog.Logger) error {
	cfg, err := config.Load()
	if err != nil {
		return err
	}
	if err := cfg.ValidateForAuth(); err != nil {
		return err
	}

	secretStore := secrets.NewFileStore(cfg.SecretsFile)
	authCfg := auth.Config{
		ClientID:     cfg.YTClientID,
		ClientSecret: cfg.YTClientSecret,
		RedirectURL:  "http://" + cfg.OAuthRedirectAddr + "/callback",
		TokenRef:     cfg.YTTokenRef,
	}
	log.Info("starting youtube authorization", "redirect", authCfg.RedirectURL, "token_ref", cfg.YTTokenRef)
	return auth.Run(ctx, authCfg, secretStore, os.Stdout)
}

// cmdRun wires the adapters and engine, then runs the watch→summarize→deliver
// loop. With TAPIR_POLL_INTERVAL unset it performs a single pass.
func cmdRun(ctx context.Context, log *slog.Logger) error {
	cfg, err := config.Load()
	if err != nil {
		return err
	}
	if err := cfg.ValidateForRun(); err != nil {
		return err
	}

	st, err := store.New(ctx, cfg.DBDSN)
	if err != nil {
		return err
	}
	defer st.Close()

	// Same wiring the web serve path uses (buildProcessor). ValidateForRun above
	// already required the engine's inputs, so a nil here is a genuine config gap.
	engine, err := buildProcessor(cfg, st)
	if err != nil {
		return err
	}
	if engine == nil {
		return fmt.Errorf("run: incomplete summarization config (gateway, youtube credentials, secrets file)")
	}
	// Process-wide caption-fetch rate gate (ADR-014 item 2): the batch path shares
	// the same per-egress-IP limiter as the web click-path.
	youtube.SetFetchRate(cfg.FetchRate)

	r := runner.New(engine.Source, st, engine, cfg.UserID, log,
		runner.WithBackoff(cfg.FetchBackoff),
		runner.WithAutoWindow(cfg.AutoSummarizeWindow),
		runner.WithCaptionMemory(cfg.ChannelCaptionlessThreshold, cfg.ChannelCaptionlessWindow))

	log.Info("starting run", "user", cfg.UserID, "model", cfg.SummarizerModel,
		"gateway", cfg.GatewayURL, "poll_interval", cfg.PollInterval, "fetch_backoff", cfg.FetchBackoff,
		"fetch_rate", cfg.FetchRate, "auto_window", cfg.AutoSummarizeWindow)
	return r.Loop(ctx, cfg.PollInterval)
}

// cmdServe runs the Stage-1 web UI: the summary reader over the existing store
// (ADR-003 — a new transport, not new core). Auth (web.Auth) gates access; the
// registration gate resolves the authenticated subject to a tapir user_id and
// scopes every store access by it (ADR-012). With Dex configured, real OIDC login
// is used; otherwise StubAuth (dev only). The store doubles as the Identity port.
func cmdServe(ctx context.Context, log *slog.Logger) error {
	cfg, err := config.Load()
	if err != nil {
		return err
	}
	if err := cfg.ValidateForServe(); err != nil {
		return err
	}

	st, err := store.New(ctx, cfg.DBDSN)
	if err != nil {
		return err
	}
	defer st.Close()

	// Process-wide caption-fetch rate gate (ADR-014 item 2): the web click-path
	// and the scheduled-discovery runners share one per-egress-IP limiter so they
	// cannot collectively trip 429s. Must be set before either path fetches.
	youtube.SetFetchRate(cfg.FetchRate)

	// Auth seam (handlers depend on web.Auth only). With Dex configured
	// (TAPIR_OIDC_ISSUER set) serve uses real OIDC login — any Dex subject may
	// authenticate, then registers a tapir user (ADR-012); otherwise it falls
	// back to the allow-all StubAuth for local dev — never expose StubAuth publicly.
	var authn web.Auth
	if cfg.DexConfigured() {
		authn, err = oidc.New(ctx, oidc.Config{
			Issuer:        cfg.OIDCIssuer,
			ClientID:      cfg.DexClientID,
			ClientSecret:  cfg.DexClientSecret,
			RedirectURL:   cfg.OIDCRedirectURL,
			SessionSecret: cfg.SessionSecret,
		})
		if err != nil {
			return fmt.Errorf("dex oidc: %w", err)
		}
		log.Info("web auth: dex oidc", "issuer", cfg.OIDCIssuer)
	} else {
		authn = web.StubAuth{U: web.User{Subject: cfg.UserID}}
		log.Warn("web auth: STUB allow-all (no TAPIR_OIDC_ISSUER) — local dev only, do not expose")
	}

	// The file-backed SecretStore is shared by the connect flow (writes tokens)
	// and account management (deletes them on disconnect / delete-account).
	secretStore := secrets.NewFileStore(cfg.SecretsFile)
	app := &web.App{Store: st, Identity: st, Auth: authn, Secrets: secretStore, Log: log, RecencyWindow: cfg.AutoSummarizeWindow}

	// Per-video deeper-dive chat over the STORED transcript (ADR-027). Enabled
	// whenever a gateway is configured — it needs no YouTube credentials because it
	// never fetches. Guarded so a typed-nil never lands in the interface field
	// (which would mount the routes over a nil backend).
	if c := buildChat(cfg); c != nil {
		app.Chat = c
		log.Info("web chat enabled (stored-transcript only)", "models", chatModels(cfg))
	}

	// User onboarding is handled by the IdP (Authentik invite flow), not Tapir —
	// the Dex local-password provisioning path was removed (ADR-019). An
	// authenticated subject with no Tapir user is routed to /register.

	// Web-initiated YouTube connect (ADR-006). Mounted only when the OAuth client
	// credentials are present; the refresh token persists through the SecretStore
	// under a per-user ref (web.YouTubeTokenRef). Live connect also needs the
	// callback URL registered in the Google OAuth client's authorized redirects.
	if cfg.YTClientID != "" && cfg.YTClientSecret != "" {
		app.Connect = web.NewConnectHandler(auth.Config{
			ClientID:     cfg.YTClientID,
			ClientSecret: cfg.YTClientSecret,
			RedirectURL:  cfg.YTConnectRedirectURL,
		}, secretStore, st, log)
		// Paste-a-URL (Feature 2): same YouTube credentials, per-user adapter built
		// per request. Mounting the /paste route keys off app.Fetcher being set.
		app.Fetcher = videoFetcher{cfg: cfg, secrets: secretStore}
		log.Info("web youtube connect + paste enabled", "redirect", cfg.YTConnectRedirectURL)
	}

	// Immediate summarization for the web "Summarize" button. When the engine can
	// be built (gateway + YouTube credentials + secrets present), a click runs the
	// summary now in the background; otherwise the button stays queue-only and the
	// next `tapir run` does the work (buildProcessor returns nil — never an error).
	engine, err := buildProcessor(cfg, st)
	if err != nil {
		return err
	}
	if engine != nil {
		app.Processor = &engineProcessor{engine: engine, store: st}
		log.Info("web immediate summarization enabled", "model", cfg.SummarizerModel)
	} else {
		log.Info("web summarization is queue-only (incomplete engine config)")
	}

	// In-process scheduled discovery (ADR-018): when enabled, a background
	// goroutine runs a discovery pass for ALL users on TAPIR_DISCOVERY_INTERVAL,
	// reusing the per-user runner.Runner. Cancelled by the same ctx as the server.
	//
	// SINGLE-REPLICA ASSUMPTION (load-bearing): this loop lives in the web process.
	// Running serve at >1 replica would make every replica fetch every user in
	// parallel — duplicate work and self-inflicted 429s. replicas: 1 is required in
	// the deployment manifest; scaling up needs a CronJob or leader election first.
	if cfg.DiscoveryInterval > 0 {
		log.Info("scheduled discovery enabled", "interval", cfg.DiscoveryInterval, "fetch_rate", cfg.FetchRate)
		log.Warn("scheduled discovery assumes a SINGLE replica — running serve at >1 replica double-runs discovery (ADR-018)")
		rawRunUser := func(ctx context.Context, userID string) (runner.Stats, error) {
			r, err := buildUserRunner(cfg, st, secretStore, userID, log)
			if err != nil {
				return runner.Stats{}, err
			}
			return r.RunOnce(ctx)
		}
		// One lock shared by the scheduler and connect-triggered passes (#6) so
		// they never fetch concurrently — the single-fetcher invariant (ADR-018).
		runUser := serialize(&sync.Mutex{}, rawRunUser)
		// Onboarding burst (Feature 1, refined by ADR-028): after the connect-triggered
		// discovery pass, summarize up to OnboardSummarizeCount of the user's newest
		// LIKELY-GOOD unsummarized videos so a fresh account gets a strong first
		// session. Selection avoids known-junk (Shorts/over-long/livestream VODs via
		// the persisted duration); the burst leads its chain with the stronger onboard
		// model. Hard cap; explicit, so it bypasses the recency window — but every
		// fetch still goes through globalFetchGate. No-op when disabled (count 0) or
		// queue-only (no processor).
		//
		// burstProcessor leads with the stronger model (ADR-028); it collapses onto the
		// shared Processor when the onboard model is empty/equal-to-primary or the
		// engine config is incomplete.
		burstProcessor := app.Processor
		if burstEngine, berr := buildBurstProcessor(cfg, st); berr != nil {
			return berr
		} else if burstEngine != nil {
			burstProcessor = &engineProcessor{engine: burstEngine, store: st}
			log.Info("onboarding burst uses a stronger model", "onboard_model", cfg.OnboardSummarizerModel)
		}
		onboard := func(ctx context.Context, userID string) {
			if cfg.OnboardSummarizeCount <= 0 || burstProcessor == nil {
				return
			}
			ids, err := st.OnboardBurstVideoIDs(ctx, userID, cfg.OnboardSummarizeCount, cfg.MinVideoSeconds, cfg.OnboardMaxVideoSeconds)
			if err != nil {
				log.Warn("onboarding: list burst candidates", "user", userID, "err", err)
				return
			}
			for _, id := range ids {
				if err := burstProcessor.ProcessVideo(ctx, userID, id); err != nil {
					log.Warn("onboarding: summarize", "user", userID, "video", id, "err", err)
				}
			}
			if len(ids) > 0 {
				log.Info("onboarding burst complete", "user", userID, "summarized", len(ids), "cap", cfg.OnboardSummarizeCount)
			}
		}
		if app.Connect != nil {
			app.Connect.Discovery = &discoveryTrigger{ctx: ctx, run: runUser, onboard: onboard, log: log}
			log.Info("connect-triggered discovery enabled", "onboard_cap", cfg.OnboardSummarizeCount)
		}
		go runScheduler(ctx, cfg.DiscoveryInterval, st, runUser, log)
	} else {
		log.Info("scheduled discovery disabled (TAPIR_DISCOVERY_INTERVAL unset or 0)")
	}

	srv := &http.Server{
		Addr:              cfg.HTTPAddr,
		Handler:           metrics.HTTPMiddleware(app.Router()),
		ReadHeaderTimeout: 10 * time.Second,
	}

	// Prometheus /metrics on a SEPARATE port (ADR-030) — never on the public app
	// mux, so a scrape is in-cluster only. Empty TAPIR_METRICS_ADDR disables it.
	var metricsSrv *http.Server
	if cfg.MetricsAddr != "" {
		mmux := http.NewServeMux()
		mmux.Handle("GET /metrics", metrics.Handler())
		metricsSrv = &http.Server{Addr: cfg.MetricsAddr, Handler: mmux, ReadHeaderTimeout: 10 * time.Second}
		go func() {
			log.Info("serving metrics", "addr", cfg.MetricsAddr)
			if err := metricsSrv.ListenAndServe(); err != nil && !errors.Is(err, http.ErrServerClosed) {
				log.Error("metrics server", "err", err)
			}
		}()
	}

	// Graceful shutdown on signal: stop accepting, drain in-flight requests.
	go func() {
		<-ctx.Done()
		shutdownCtx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
		defer cancel()
		_ = srv.Shutdown(shutdownCtx)
		if metricsSrv != nil {
			_ = metricsSrv.Shutdown(shutdownCtx)
		}
	}()

	log.Info("serving web ui", "addr", cfg.HTTPAddr, "user", cfg.UserID)
	if err := srv.ListenAndServe(); err != nil && !errors.Is(err, http.ErrServerClosed) {
		return fmt.Errorf("serve: %w", err)
	}
	return nil
}