From 8c6c7ca947d766eada0d75f272895050bfa093f0 Mon Sep 17 00:00:00 2001 From: Mathias Date: Wed, 3 Jun 2026 19:50:06 +0200 Subject: [PATCH] feat(cmd): shared buildProcessor + wire immediate web summarization MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Extract the engine wiring (YouTube source, AI-router summarizer, store sink) into buildProcessor, shared by cmdRun and cmdServe. It returns (nil, nil) — not an error — on incomplete config, which is the queue-only fallback for serve. engineProcessor adapts the engine to web.Processor: load the video row, run the engine, clear the manual queue flag on a produced summary (mirrors the runner). cmdServe wires it onto web.App.Processor; cmdRun reuses buildProcessor so the wiring is no longer duplicated. Co-Authored-By: Claude Opus 4.8 (1M context) --- cmd/tapir/main.go | 45 ++++++++++--------- cmd/tapir/processor.go | 86 +++++++++++++++++++++++++++++++++++++ cmd/tapir/processor_test.go | 47 ++++++++++++++++++++ 3 files changed, 157 insertions(+), 21 deletions(-) create mode 100644 cmd/tapir/processor.go create mode 100644 cmd/tapir/processor_test.go diff --git a/cmd/tapir/main.go b/cmd/tapir/main.go index f1db179..f8c2384 100644 --- a/cmd/tapir/main.go +++ b/cmd/tapir/main.go @@ -22,15 +22,11 @@ import ( "os/signal" "time" - "gitea.d-ma.be/mathias/tapir/internal/adapters/llm" "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/summarizer" - "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/runner" - "gitea.d-ma.be/mathias/tapir/internal/usecase" "gitea.d-ma.be/mathias/tapir/internal/web" "gitea.d-ma.be/mathias/tapir/internal/web/oidc" ) @@ -120,24 +116,16 @@ func cmdRun(ctx context.Context, log *slog.Logger) error { } defer st.Close() - secretStore := secrets.NewFileStore(cfg.SecretsFile) - src := youtube.New(youtube.Config{ - ClientID: cfg.YTClientID, - ClientSecret: cfg.YTClientSecret, - TokenSecretRef: cfg.YTTokenRef, - PreferredLanguages: []string{"en"}, - }, secretStore) - - // Local Primary only; no BYO fallback for the demo (fallback nil). - primary := summarizer.Endpoint{ - Client: llm.New(cfg.GatewayURL, cfg.GatewayKey, cfg.SummarizerModel, cfg.SummarizerTimeout), - Provider: "local", - Model: cfg.SummarizerModel, + // 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 } - sum := summarizer.New(primary, nil) - - engine := usecase.NewEngine(src, sum, st) - r := runner.New(src, st, engine, cfg.UserID, log) + if engine == nil { + return fmt.Errorf("run: incomplete summarization config (gateway, youtube credentials, secrets file)") + } + r := runner.New(engine.Source, st, engine, cfg.UserID, log) log.Info("starting run", "user", cfg.UserID, "model", cfg.SummarizerModel, "gateway", cfg.GatewayURL, "poll_interval", cfg.PollInterval) @@ -204,6 +192,21 @@ func cmdServe(ctx context.Context, log *slog.Logger) error { log.Info("web youtube connect 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)") + } + srv := &http.Server{ Addr: cfg.HTTPAddr, Handler: app.Router(), diff --git a/cmd/tapir/processor.go b/cmd/tapir/processor.go new file mode 100644 index 0000000..7e979cd --- /dev/null +++ b/cmd/tapir/processor.go @@ -0,0 +1,86 @@ +package main + +import ( + "context" + "fmt" + + "gitea.d-ma.be/mathias/tapir/internal/adapters/llm" + "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/summarizer" + "gitea.d-ma.be/mathias/tapir/internal/adapters/youtube" + "gitea.d-ma.be/mathias/tapir/internal/config" + "gitea.d-ma.be/mathias/tapir/internal/domain" + "gitea.d-ma.be/mathias/tapir/internal/usecase" +) + +// 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. +func buildProcessor(cfg config.Config, st *store.Store) (*usecase.Engine, error) { + if cfg.GatewayURL == "" || cfg.YTClientID == "" || cfg.YTClientSecret == "" || cfg.SecretsFile == "" { + return nil, nil + } + + secretStore := secrets.NewFileStore(cfg.SecretsFile) + src := youtube.New(youtube.Config{ + ClientID: cfg.YTClientID, + ClientSecret: cfg.YTClientSecret, + TokenSecretRef: cfg.YTTokenRef, + PreferredLanguages: []string{"en"}, + }, secretStore) + + // Local Primary only; no BYO fallback for the demo (fallback nil). + primary := summarizer.Endpoint{ + Client: llm.New(cfg.GatewayURL, cfg.GatewayKey, cfg.SummarizerModel, cfg.SummarizerTimeout), + Provider: "local", + Model: cfg.SummarizerModel, + } + sum := summarizer.New(primary, nil) + + return usecase.NewEngine(src, sum, st), 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 { + 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) + } + if res.Summary != nil { + if err := p.store.ClearSummarizeRequested(ctx, userID, videoID); err != nil { + return fmt.Errorf("clear summarize flag %q: %w", videoID, err) + } + } + return nil +} diff --git a/cmd/tapir/processor_test.go b/cmd/tapir/processor_test.go new file mode 100644 index 0000000..ccb2c6b --- /dev/null +++ b/cmd/tapir/processor_test.go @@ -0,0 +1,47 @@ +package main + +import ( + "testing" + + "gitea.d-ma.be/mathias/tapir/internal/config" +) + +// TestBuildProcessorNilOnIncompleteConfig asserts the queue-only fallback: when a +// required input is missing, buildProcessor returns (nil, nil) — never an error — +// so the web UI degrades to queue-only instead of failing to start. +func TestBuildProcessorNilOnIncompleteConfig(t *testing.T) { + // A complete config (the fields buildProcessor gates on). The store is nil: + // buildProcessor must not touch it on the incomplete paths, and the complete + // path only stores the pointer (no connection), so nil is fine for this test. + complete := config.Config{ + GatewayURL: "http://gw/v1", + YTClientID: "id", + YTClientSecret: "secret", + SecretsFile: "/tmp/secrets.json", + } + + tests := []struct { + name string + mutate func(config.Config) config.Config + wantNil bool + }{ + {"complete", func(c config.Config) config.Config { return c }, false}, + {"no gateway url", func(c config.Config) config.Config { c.GatewayURL = ""; return c }, true}, + {"no yt client id", func(c config.Config) config.Config { c.YTClientID = ""; return c }, true}, + {"no yt client secret", func(c config.Config) config.Config { c.YTClientSecret = ""; return c }, true}, + {"no secrets file", func(c config.Config) config.Config { c.SecretsFile = ""; return c }, true}, + {"empty config", func(config.Config) config.Config { return config.Config{} }, true}, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + engine, err := buildProcessor(tt.mutate(complete), nil) + if err != nil { + t.Fatalf("buildProcessor returned an error, want nil: %v", err) + } + if (engine == nil) != tt.wantNil { + t.Fatalf("engine == nil is %v, want %v", engine == nil, tt.wantNil) + } + }) + } +}