feat(cmd): shared buildProcessor + wire immediate web summarization
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) <noreply@anthropic.com>
This commit is contained in:
+24
-21
@@ -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(),
|
||||
|
||||
@@ -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
|
||||
}
|
||||
@@ -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)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user