From c40b46b66182d3c2901b1cae14afe64e99ba735f Mon Sep 17 00:00:00 2001 From: Mathias Date: Tue, 2 Jun 2026 20:55:49 +0200 Subject: [PATCH 1/7] fix(llm): send generous max_tokens on every request The copied OpenAI-compatible client sent no max_tokens. Thinking models (qwen3, deepseek-r1) spend their budget on the reasoning trace and return EMPTY content when max_tokens is unset, which the summarizer treats as an error. ADR-004 says change Tapir's copy rather than the hyperguild upstream, so set a generous default (8192) leaving room for both reasoning and output. Co-Authored-By: Claude Opus 4.8 (1M context) --- internal/adapters/llm/client.go | 12 ++++++++++++ internal/adapters/llm/client_test.go | 21 +++++++++++++++++++++ 2 files changed, 33 insertions(+) diff --git a/internal/adapters/llm/client.go b/internal/adapters/llm/client.go index 0c84a36..8e40af7 100644 --- a/internal/adapters/llm/client.go +++ b/internal/adapters/llm/client.go @@ -17,11 +17,20 @@ import ( "time" ) +// defaultMaxTokens is sent on every request. Tapir CHANGES this from the +// hyperguild copy (ADR-004 says change the copy, not the upstream): thinking +// models (qwen3, deepseek-r1) spend their budget on reasoning and return EMPTY +// content when max_tokens is unset or too low. A generous ceiling leaves room +// for both the reasoning trace and the actual summary. See +// docs/homelab-integration.md. +const defaultMaxTokens = 8192 + // Client calls an OpenAI-compatible chat completions endpoint. type Client struct { baseURL string apiKey string model string + maxTokens int httpClient *http.Client } @@ -31,6 +40,7 @@ func New(baseURL, apiKey, model string, timeout time.Duration) *Client { baseURL: strings.TrimRight(baseURL, "/"), apiKey: apiKey, model: model, + maxTokens: defaultMaxTokens, httpClient: &http.Client{Timeout: timeout}, } } @@ -39,6 +49,7 @@ type chatRequest struct { Model string `json:"model"` Messages []message `json:"messages"` Temperature float64 `json:"temperature"` + MaxTokens int `json:"max_tokens,omitempty"` } type message struct { @@ -62,6 +73,7 @@ func (c *Client) Complete(ctx context.Context, system, user string) (string, err {Role: "user", Content: user}, }, Temperature: 0.2, + MaxTokens: c.maxTokens, } b, err := json.Marshal(body) if err != nil { diff --git a/internal/adapters/llm/client_test.go b/internal/adapters/llm/client_test.go index 742bc96..1f0dc91 100644 --- a/internal/adapters/llm/client_test.go +++ b/internal/adapters/llm/client_test.go @@ -43,6 +43,27 @@ func TestClient_Complete(t *testing.T) { } } +// TestClient_SendsMaxTokens guards Tapir's ADR-004 change to the copied client: +// it MUST send a positive max_tokens, or thinking models return empty content. +func TestClient_SendsMaxTokens(t *testing.T) { + var body chatRequest + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + _ = json.NewDecoder(r.Body).Decode(&body) + _ = json.NewEncoder(w).Encode(map[string]any{ + "choices": []map[string]any{{"message": map[string]any{"content": "ok"}}}, + }) + })) + defer srv.Close() + + c := New(srv.URL, "", "test-model", 10*time.Second) + if _, err := c.Complete(context.Background(), "sys", "user"); err != nil { + t.Fatalf("Complete: %v", err) + } + if body.MaxTokens <= 0 { + t.Errorf("max_tokens = %d, want > 0 (thinking models return empty content without it)", body.MaxTokens) + } +} + func TestClient_ReturnsErrorOnNon200(t *testing.T) { srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { http.Error(w, "overloaded", http.StatusServiceUnavailable) From c424d88c95c7796c150ceef3e8d2f24d4e58cafb Mon Sep 17 00:00:00 2001 From: Mathias Date: Tue, 2 Jun 2026 20:57:07 +0200 Subject: [PATCH 2/7] feat(config): typed env-driven configuration MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Parse TAPIR_* env into a typed Config with homelab defaults (gateway URL, summarizer model, token ref, redirect addr). Secrets (gateway key, OAuth client secret) come from env only; the refresh token never lives here — it is addressed by an opaque ref behind the SecretStore port. Per-command validation (ValidateForAuth/ValidateForRun) so each command demands only what it needs. Co-Authored-By: Claude Opus 4.8 (1M context) --- internal/config/config.go | 170 +++++++++++++++++++++++++++++++++ internal/config/config_test.go | 112 ++++++++++++++++++++++ 2 files changed, 282 insertions(+) create mode 100644 internal/config/config.go create mode 100644 internal/config/config_test.go diff --git a/internal/config/config.go b/internal/config/config.go new file mode 100644 index 0000000..1d58cd6 --- /dev/null +++ b/internal/config/config.go @@ -0,0 +1,170 @@ +// Package config parses Tapir's runtime configuration from environment +// variables into a typed struct. No secrets are baked into code: the gateway +// key and OAuth client secret are read from the environment (later resolved via +// op/ESO), and the OAuth refresh token is never held here — it lives behind the +// SecretStore port, addressed by the opaque TokenRef. +// +// Defaults target the homelab snapshot in docs/homelab-integration.md; every +// value is overridable so the same binary runs standalone or in-cluster. +package config + +import ( + "fmt" + "os" + "path/filepath" + "sort" + "strings" + "time" +) + +// Config is Tapir's fully-resolved runtime configuration. +type Config struct { + // UserID is the Tapir user the run operates as. Stage 0 has exactly one. + // It must be a UUID: it keys the UUID user_id/video_id columns (data-model). + UserID string + + // GatewayURL is the OpenAI-compatible LiteLLM base URL (".../v1"). + GatewayURL string + // GatewayKey authorizes the gateway. Read from env, never committed. + GatewayKey string + // SummarizerModel is the alias in host/name form, e.g. "koala/phi4-mini". + SummarizerModel string + // SummarizerTimeout bounds a single completion call. Thinking models are + // slow, so the default is generous. + SummarizerTimeout time.Duration + + // DBDSN is the Postgres DSN for the store sink. + DBDSN string + + // YouTube OAuth app credentials (the registered client), read from env. + YTClientID string + YTClientSecret string + // YTTokenRef is the opaque SecretStore reference under which the YouTube + // refresh token is persisted/resolved. Not the token itself. + YTTokenRef string + + // SecretsFile is the path to the local file-backed SecretStore (0600). A + // Stage-0 stand-in for op/ESO, swappable behind the SecretStore port. + SecretsFile string + + // OAuthRedirectAddr is the host:port the `auth` command's local listener + // binds for the OAuth redirect, e.g. "localhost:8080". + OAuthRedirectAddr string + + // PollInterval, when > 0, makes `run` loop on that cadence; 0 means run once. + PollInterval time.Duration +} + +// Defaults (see docs/homelab-integration.md). All overridable via env. +const ( + defaultGatewayURL = "http://koala:30401/v1" + defaultSummarizerModel = "koala/phi4-mini" + defaultSummarizerTimeout = 5 * time.Minute + defaultYTTokenRef = "youtube/refresh_token" + defaultOAuthRedirectAddr = "localhost:8080" +) + +// Load reads the environment into a Config, applying defaults. It does not +// validate that required fields are present — call ValidateForAuth or +// ValidateForRun for the command being run, so each command demands only what +// it needs. +func Load() (Config, error) { + c := Config{ + UserID: os.Getenv("TAPIR_USER_ID"), + GatewayURL: envOr("TAPIR_GATEWAY_URL", defaultGatewayURL), + GatewayKey: os.Getenv("TAPIR_GATEWAY_KEY"), + SummarizerModel: envOr("TAPIR_SUMMARIZER_MODEL", defaultSummarizerModel), + DBDSN: os.Getenv("TAPIR_DB_DSN"), + YTClientID: os.Getenv("TAPIR_YT_CLIENT_ID"), + YTClientSecret: os.Getenv("TAPIR_YT_CLIENT_SECRET"), + YTTokenRef: envOr("TAPIR_YT_TOKEN_REF", defaultYTTokenRef), + SecretsFile: envOr("TAPIR_SECRETS_FILE", defaultSecretsFile()), + OAuthRedirectAddr: envOr("TAPIR_OAUTH_REDIRECT_ADDR", defaultOAuthRedirectAddr), + } + + timeout, err := durationOr("TAPIR_SUMMARIZER_TIMEOUT", defaultSummarizerTimeout) + if err != nil { + return Config{}, err + } + c.SummarizerTimeout = timeout + + interval, err := durationOr("TAPIR_POLL_INTERVAL", 0) + if err != nil { + return Config{}, err + } + c.PollInterval = interval + + return c, nil +} + +// ValidateForAuth checks the fields the `auth` command needs: the YouTube OAuth +// app credentials, a place to persist the token, and the redirect listener. +func (c Config) ValidateForAuth() error { + return c.require(map[string]string{ + "TAPIR_YT_CLIENT_ID": c.YTClientID, + "TAPIR_YT_CLIENT_SECRET": c.YTClientSecret, + "TAPIR_YT_TOKEN_REF": c.YTTokenRef, + "TAPIR_SECRETS_FILE": c.SecretsFile, + }) +} + +// ValidateForRun checks the fields the `run` command needs end to end. +func (c Config) ValidateForRun() error { + return c.require(map[string]string{ + "TAPIR_USER_ID": c.UserID, + "TAPIR_GATEWAY_URL": c.GatewayURL, + "TAPIR_SUMMARIZER_MODEL": c.SummarizerModel, + "TAPIR_DB_DSN": c.DBDSN, + "TAPIR_YT_CLIENT_ID": c.YTClientID, + "TAPIR_YT_CLIENT_SECRET": c.YTClientSecret, + "TAPIR_YT_TOKEN_REF": c.YTTokenRef, + "TAPIR_SECRETS_FILE": c.SecretsFile, + }) +} + +func (c Config) require(fields map[string]string) error { + var missing []string + for name, val := range fields { + if strings.TrimSpace(val) == "" { + missing = append(missing, name) + } + } + if len(missing) > 0 { + return fmt.Errorf("missing required config: %s", strings.Join(sortedMissing(missing), ", ")) + } + return nil +} + +func sortedMissing(xs []string) []string { + sort.Strings(xs) + return xs +} + +func envOr(key, fallback string) string { + if v := os.Getenv(key); v != "" { + return v + } + return fallback +} + +func durationOr(key string, fallback time.Duration) (time.Duration, error) { + v := os.Getenv(key) + if v == "" { + return fallback, nil + } + d, err := time.ParseDuration(v) + if err != nil { + return 0, fmt.Errorf("config: %s=%q: %w", key, v, err) + } + return d, nil +} + +// defaultSecretsFile resolves to /tapir/secrets.json, falling +// back to a cwd-relative path when the config dir is unavailable. +func defaultSecretsFile() string { + dir, err := os.UserConfigDir() + if err != nil { + return "tapir-secrets.json" + } + return filepath.Join(dir, "tapir", "secrets.json") +} diff --git a/internal/config/config_test.go b/internal/config/config_test.go new file mode 100644 index 0000000..c8fce0d --- /dev/null +++ b/internal/config/config_test.go @@ -0,0 +1,112 @@ +package config + +import ( + "strings" + "testing" + "time" +) + +// setEnv sets env vars for the test and clears them afterward, so cases don't +// leak into one another. t.Setenv handles restoration. +func setEnv(t *testing.T, kv map[string]string) { + t.Helper() + for k, v := range kv { + t.Setenv(k, v) + } +} + +func TestLoad_AppliesDefaults(t *testing.T) { + // Clear the ones with defaults so we observe the fallback, not the host env. + setEnv(t, map[string]string{ + "TAPIR_GATEWAY_URL": "", + "TAPIR_SUMMARIZER_MODEL": "", + "TAPIR_YT_TOKEN_REF": "", + "TAPIR_OAUTH_REDIRECT_ADDR": "", + "TAPIR_SUMMARIZER_TIMEOUT": "", + "TAPIR_POLL_INTERVAL": "", + }) + + c, err := Load() + if err != nil { + t.Fatalf("Load: %v", err) + } + if c.GatewayURL != defaultGatewayURL { + t.Errorf("GatewayURL = %q, want default %q", c.GatewayURL, defaultGatewayURL) + } + if c.SummarizerModel != defaultSummarizerModel { + t.Errorf("SummarizerModel = %q, want default %q", c.SummarizerModel, defaultSummarizerModel) + } + if c.YTTokenRef != defaultYTTokenRef { + t.Errorf("YTTokenRef = %q, want default %q", c.YTTokenRef, defaultYTTokenRef) + } + if c.SummarizerTimeout != defaultSummarizerTimeout { + t.Errorf("SummarizerTimeout = %v, want default %v", c.SummarizerTimeout, defaultSummarizerTimeout) + } + if c.PollInterval != 0 { + t.Errorf("PollInterval = %v, want 0 (run once)", c.PollInterval) + } +} + +func TestLoad_ParsesValues(t *testing.T) { + setEnv(t, map[string]string{ + "TAPIR_USER_ID": "11111111-1111-1111-1111-111111111111", + "TAPIR_GATEWAY_URL": "http://example/v1", + "TAPIR_GATEWAY_KEY": "sk-test", + "TAPIR_SUMMARIZER_MODEL": "iguana/deepseek-r1-14b", + "TAPIR_SUMMARIZER_TIMEOUT": "90s", + "TAPIR_DB_DSN": "postgres://x", + "TAPIR_POLL_INTERVAL": "10m", + }) + + c, err := Load() + if err != nil { + t.Fatalf("Load: %v", err) + } + if c.UserID != "11111111-1111-1111-1111-111111111111" { + t.Errorf("UserID = %q", c.UserID) + } + if c.GatewayKey != "sk-test" { + t.Errorf("GatewayKey = %q", c.GatewayKey) + } + if c.SummarizerModel != "iguana/deepseek-r1-14b" { + t.Errorf("SummarizerModel = %q", c.SummarizerModel) + } + if c.SummarizerTimeout != 90*time.Second { + t.Errorf("SummarizerTimeout = %v, want 90s", c.SummarizerTimeout) + } + if c.PollInterval != 10*time.Minute { + t.Errorf("PollInterval = %v, want 10m", c.PollInterval) + } +} + +func TestLoad_RejectsBadDuration(t *testing.T) { + setEnv(t, map[string]string{"TAPIR_SUMMARIZER_TIMEOUT": "not-a-duration"}) + if _, err := Load(); err == nil { + t.Fatal("want error on unparsable duration, got nil") + } +} + +func TestValidateForRun_ReportsMissing(t *testing.T) { + c := Config{} // nothing set + err := c.ValidateForRun() + if err == nil { + t.Fatal("want error when required run fields are missing") + } + for _, want := range []string{"TAPIR_USER_ID", "TAPIR_DB_DSN", "TAPIR_YT_CLIENT_ID"} { + if !strings.Contains(err.Error(), want) { + t.Errorf("error %q must name missing field %q", err, want) + } + } +} + +func TestValidateForAuth_PassesWhenComplete(t *testing.T) { + c := Config{ + YTClientID: "id", + YTClientSecret: "secret", + YTTokenRef: "youtube/refresh_token", + SecretsFile: "/tmp/secrets.json", + } + if err := c.ValidateForAuth(); err != nil { + t.Errorf("ValidateForAuth: unexpected error %v", err) + } +} From c93b433aafe681196f53d67123c81c7ca956fca4 Mon Sep 17 00:00:00 2001 From: Mathias Date: Tue, 2 Jun 2026 20:57:44 +0200 Subject: [PATCH 3/7] feat(secrets): file-backed SecretStore for Stage-0 Implements ports.SecretStore over a 0600 JSON file as a stand-in for op/ESO so the demo runs without live op. Put persists atomically (temp + rename) and merges; Get returns ErrNotFound for unknown refs so a missing token fails loud. Behind the port, so swapping to op/ESO later is wiring, not code (ADR-002). Co-Authored-By: Claude Opus 4.8 (1M context) --- internal/adapters/secrets/file.go | 107 +++++++++++++++++++++++++ internal/adapters/secrets/file_test.go | 72 +++++++++++++++++ 2 files changed, 179 insertions(+) create mode 100644 internal/adapters/secrets/file.go create mode 100644 internal/adapters/secrets/file_test.go diff --git a/internal/adapters/secrets/file.go b/internal/adapters/secrets/file.go new file mode 100644 index 0000000..1b90124 --- /dev/null +++ b/internal/adapters/secrets/file.go @@ -0,0 +1,107 @@ +// Package secrets provides a local file-backed implementation of the +// ports.SecretStore port. It is a Stage-0 stand-in for op/ESO: secret material +// (the YouTube OAuth refresh token) is kept in a 0600 JSON file rather than the +// vault, so the demo runs without live op. Because every consumer depends on +// the SecretStore port, swapping this for an op/ESO-backed store later is a +// wiring change, not a code change (ADR-002, docs/homelab-integration.md). +// +// Secret values are never logged. Get returns an error for an unknown ref so a +// missing token surfaces loudly rather than as an empty string. +package secrets + +import ( + "context" + "encoding/json" + "errors" + "fmt" + "os" + "path/filepath" + "sync" + + "gitea.d-ma.be/mathias/tapir/internal/ports" +) + +// FileStore is a SecretStore backed by a single 0600 JSON file mapping opaque +// refs to secret values. Safe for concurrent use within one process. +type FileStore struct { + path string + mu sync.RWMutex +} + +// Static check: FileStore satisfies the read side of the port. +var _ ports.SecretStore = (*FileStore)(nil) + +// ErrNotFound is returned by Get when no secret is stored under the ref. +var ErrNotFound = errors.New("secrets: ref not found") + +// NewFileStore returns a store backed by path. The file need not exist yet; it +// is created on the first Put. +func NewFileStore(path string) *FileStore { + return &FileStore{path: path} +} + +// Get resolves a ref to its secret value. It returns ErrNotFound if the file or +// the ref is absent. +func (s *FileStore) Get(_ context.Context, ref string) (string, error) { + s.mu.RLock() + defer s.mu.RUnlock() + + m, err := s.load() + if err != nil { + return "", err + } + v, ok := m[ref] + if !ok { + return "", fmt.Errorf("%w: %q", ErrNotFound, ref) + } + return v, nil +} + +// Put stores value under ref, persisting the file with 0600 permissions. It +// merges into any existing entries and writes atomically (temp file + rename). +func (s *FileStore) Put(ref, value string) error { + s.mu.Lock() + defer s.mu.Unlock() + + m, err := s.load() + if err != nil && !errors.Is(err, os.ErrNotExist) { + return err + } + if m == nil { + m = make(map[string]string) + } + m[ref] = value + + if err := os.MkdirAll(filepath.Dir(s.path), 0o700); err != nil { + return fmt.Errorf("secrets: create dir: %w", err) + } + b, err := json.Marshal(m) + if err != nil { + return fmt.Errorf("secrets: marshal: %w", err) + } + tmp := s.path + ".tmp" + if err := os.WriteFile(tmp, b, 0o600); err != nil { + return fmt.Errorf("secrets: write temp: %w", err) + } + if err := os.Rename(tmp, s.path); err != nil { + return fmt.Errorf("secrets: rename: %w", err) + } + return nil +} + +// load reads the backing file. A missing file yields an empty map (not an +// error) for Get's caller, except Put distinguishes os.ErrNotExist. +func (s *FileStore) load() (map[string]string, error) { + b, err := os.ReadFile(s.path) + if err != nil { + if errors.Is(err, os.ErrNotExist) { + return map[string]string{}, nil + } + return nil, fmt.Errorf("secrets: read %s: %w", s.path, err) + } + var m map[string]string + if err := json.Unmarshal(b, &m); err != nil { + return nil, fmt.Errorf("secrets: parse %s: %w", s.path, err) + } + return m, nil +} diff --git a/internal/adapters/secrets/file_test.go b/internal/adapters/secrets/file_test.go new file mode 100644 index 0000000..1233383 --- /dev/null +++ b/internal/adapters/secrets/file_test.go @@ -0,0 +1,72 @@ +package secrets_test + +import ( + "context" + "errors" + "os" + "path/filepath" + "testing" + + "gitea.d-ma.be/mathias/tapir/internal/adapters/secrets" +) + +func TestPutThenGet(t *testing.T) { + path := filepath.Join(t.TempDir(), "nested", "secrets.json") + s := secrets.NewFileStore(path) + + if err := s.Put("youtube/refresh_token", "rt-123"); err != nil { + t.Fatalf("Put: %v", err) + } + got, err := s.Get(context.Background(), "youtube/refresh_token") + if err != nil { + t.Fatalf("Get: %v", err) + } + if got != "rt-123" { + t.Errorf("Get = %q, want %q", got, "rt-123") + } +} + +func TestGetUnknownRef(t *testing.T) { + s := secrets.NewFileStore(filepath.Join(t.TempDir(), "secrets.json")) + _, err := s.Get(context.Background(), "missing") + if !errors.Is(err, secrets.ErrNotFound) { + t.Errorf("err = %v, want ErrNotFound", err) + } +} + +func TestPutIsOwnerOnly(t *testing.T) { + path := filepath.Join(t.TempDir(), "secrets.json") + s := secrets.NewFileStore(path) + if err := s.Put("k", "v"); err != nil { + t.Fatalf("Put: %v", err) + } + info, err := os.Stat(path) + if err != nil { + t.Fatalf("Stat: %v", err) + } + if perm := info.Mode().Perm(); perm != 0o600 { + t.Errorf("file perm = %o, want 0600 (token must not be world-readable)", perm) + } +} + +func TestPutMergesEntries(t *testing.T) { + path := filepath.Join(t.TempDir(), "secrets.json") + s := secrets.NewFileStore(path) + if err := s.Put("a", "1"); err != nil { + t.Fatalf("Put a: %v", err) + } + if err := s.Put("b", "2"); err != nil { + t.Fatalf("Put b: %v", err) + } + // Re-open from disk to prove persistence, not in-memory state. + s2 := secrets.NewFileStore(path) + for k, want := range map[string]string{"a": "1", "b": "2"} { + got, err := s2.Get(context.Background(), k) + if err != nil { + t.Fatalf("Get %q: %v", k, err) + } + if got != want { + t.Errorf("Get %q = %q, want %q", k, got, want) + } + } +} From 460f4bb1dee7347c40e53dca18edac789529d2a7 Mon Sep 17 00:00:00 2001 From: Mathias Date: Tue, 2 Jun 2026 20:59:16 +0200 Subject: [PATCH 4/7] feat(auth): interactive YouTube OAuth code flow `tapir auth` mints a refresh token for the single Stage-0 user: bind a local redirect listener, print the consent URL (offline access + forced consent so Google returns a refresh token), validate the state param, exchange the code, and persist the refresh token through the SecretStore port. Written fresh on x/oauth2 (ADR-006). Token is never logged or returned. Tests cover exchange, missing-refresh-token rejection, and the full listener flow with httptest. Co-Authored-By: Claude Opus 4.8 (1M context) --- internal/auth/auth.go | 176 ++++++++++++++++++++++++++++++++++ internal/auth/auth_test.go | 188 +++++++++++++++++++++++++++++++++++++ 2 files changed, 364 insertions(+) create mode 100644 internal/auth/auth.go create mode 100644 internal/auth/auth_test.go diff --git a/internal/auth/auth.go b/internal/auth/auth.go new file mode 100644 index 0000000..cc00b19 --- /dev/null +++ b/internal/auth/auth.go @@ -0,0 +1,176 @@ +// Package auth implements the interactive OAuth 2.0 authorization-code flow used +// by `tapir auth` to mint a YouTube refresh token for the single Stage-0 user. +// It is written fresh on golang.org/x/oauth2 (ADR-006): the ingestion repo's +// oauth package is inbound MCP server auth and unrelated to this outbound +// provider flow. +// +// The minted refresh token is persisted through the SecretStore port (the file +// store at Stage 0) and is never logged or returned to the caller. Only the +// opaque token ref crosses package boundaries afterward. +package auth + +import ( + "context" + "crypto/rand" + "encoding/hex" + "fmt" + "io" + "net" + "net/http" + "net/url" + + "golang.org/x/oauth2" +) + +// GoogleEndpoint is Google's OAuth2 endpoint, inlined to avoid the heavy +// golang.org/x/oauth2/google dependency for two URLs (mirrors the youtube +// adapter's choice). +var GoogleEndpoint = oauth2.Endpoint{ + AuthURL: "https://accounts.google.com/o/oauth2/auth", + TokenURL: "https://oauth2.googleapis.com/token", +} + +// DefaultScopes request read access to subscriptions/search and the force-ssl +// scope the Data API captions endpoints require. +var DefaultScopes = []string{ + "https://www.googleapis.com/auth/youtube.readonly", + "https://www.googleapis.com/auth/youtube.force-ssl", +} + +// TokenWriter persists a secret value under an opaque ref. *secrets.FileStore +// satisfies it; tests use a fake. (The read side is ports.SecretStore.) +type TokenWriter interface { + Put(ref, value string) error +} + +// Config wires the flow. Endpoint defaults to GoogleEndpoint when zero, so tests +// can point it at an httptest token server. +type Config struct { + ClientID string + ClientSecret string + RedirectURL string // e.g. "http://localhost:8080/callback" + Scopes []string + TokenRef string // SecretStore ref to persist the refresh token under + Endpoint oauth2.Endpoint +} + +func oauthConfig(c Config) *oauth2.Config { + ep := c.Endpoint + if ep == (oauth2.Endpoint{}) { + ep = GoogleEndpoint + } + scopes := c.Scopes + if len(scopes) == 0 { + scopes = DefaultScopes + } + return &oauth2.Config{ + ClientID: c.ClientID, + ClientSecret: c.ClientSecret, + RedirectURL: c.RedirectURL, + Scopes: scopes, + Endpoint: ep, + } +} + +// Exchange swaps an authorization code for a token and persists the refresh +// token through the writer. It errors if the provider returned no refresh token +// (e.g. consent was not forced with offline access), since without one the +// token is useless for the unattended run loop. The token is never logged. +func Exchange(ctx context.Context, c Config, secrets TokenWriter, code string) error { + conf := oauthConfig(c) + tok, err := conf.Exchange(ctx, code) + if err != nil { + return fmt.Errorf("auth: exchange code: %w", err) + } + if tok.RefreshToken == "" { + return fmt.Errorf("auth: provider returned no refresh token (re-consent with offline access)") + } + if err := secrets.Put(c.TokenRef, tok.RefreshToken); err != nil { + return fmt.Errorf("auth: persist refresh token: %w", err) + } + return nil +} + +// Run performs the full interactive flow: it binds a local listener on the +// redirect URL's host, prints the consent URL to out, waits for the provider's +// redirect (validating the state parameter), then exchanges the captured code +// and persists the refresh token. It blocks until the callback arrives, ctx is +// cancelled, or the listener fails. +func Run(ctx context.Context, c Config, secrets TokenWriter, out io.Writer) error { + conf := oauthConfig(c) + + u, err := url.Parse(c.RedirectURL) + if err != nil { + return fmt.Errorf("auth: parse redirect url %q: %w", c.RedirectURL, err) + } + ln, err := net.Listen("tcp", u.Host) + if err != nil { + return fmt.Errorf("auth: bind redirect listener on %q: %w", u.Host, err) + } + + state, err := randomState() + if err != nil { + return err + } + + type result struct { + code string + err error + } + resCh := make(chan result, 1) + + mux := http.NewServeMux() + mux.HandleFunc(u.Path, func(w http.ResponseWriter, r *http.Request) { + q := r.URL.Query() + if e := q.Get("error"); e != "" { + http.Error(w, "authorization failed: "+e, http.StatusBadRequest) + resCh <- result{err: fmt.Errorf("auth: provider returned error %q", e)} + return + } + if q.Get("state") != state { + http.Error(w, "state mismatch", http.StatusBadRequest) + resCh <- result{err: fmt.Errorf("auth: state mismatch (possible CSRF)")} + return + } + code := q.Get("code") + if code == "" { + http.Error(w, "missing code", http.StatusBadRequest) + resCh <- result{err: fmt.Errorf("auth: redirect carried no code")} + return + } + _, _ = io.WriteString(w, "Tapir: authorization received. You can close this tab.") + resCh <- result{code: code} + }) + + srv := &http.Server{Handler: mux} + go func() { _ = srv.Serve(ln) }() + defer func() { _ = srv.Shutdown(context.Background()) }() + + // AccessTypeOffline + ApprovalForce make Google return a refresh token even + // on re-authorization; without them a repeat consent yields only an access + // token and Exchange would reject it. + authURL := conf.AuthCodeURL(state, oauth2.AccessTypeOffline, oauth2.ApprovalForce) + fmt.Fprintf(out, "Open this URL to authorize Tapir, then return here:\n\n%s\n\n", authURL) + + select { + case <-ctx.Done(): + return ctx.Err() + case res := <-resCh: + if res.err != nil { + return res.err + } + if err := Exchange(ctx, c, secrets, res.code); err != nil { + return err + } + fmt.Fprintln(out, "Refresh token stored. You can now run `tapir run`.") + return nil + } +} + +func randomState() (string, error) { + b := make([]byte, 16) + if _, err := rand.Read(b); err != nil { + return "", fmt.Errorf("auth: generate state: %w", err) + } + return hex.EncodeToString(b), nil +} diff --git a/internal/auth/auth_test.go b/internal/auth/auth_test.go new file mode 100644 index 0000000..2008872 --- /dev/null +++ b/internal/auth/auth_test.go @@ -0,0 +1,188 @@ +package auth_test + +import ( + "context" + "net" + "net/http" + "net/http/httptest" + "regexp" + "sync" + "testing" + "time" + + "golang.org/x/oauth2" + + "gitea.d-ma.be/mathias/tapir/internal/auth" +) + +// fakeWriter is a TokenWriter capturing the persisted (ref, value). +type fakeWriter struct { + mu sync.Mutex + ref, val string + calls int +} + +func (w *fakeWriter) Put(ref, value string) error { + w.mu.Lock() + defer w.mu.Unlock() + w.ref, w.val, w.calls = ref, value, w.calls+1 + return nil +} + +// tokenServer fakes Google's token endpoint, returning the given JSON body for +// any POST. No live Google contact. +func tokenServer(t *testing.T, body string) *httptest.Server { + t.Helper() + return httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.Header().Set("Content-Type", "application/json") + _, _ = w.Write([]byte(body)) + })) +} + +func cfg(srvURL string) auth.Config { + return auth.Config{ + ClientID: "cid", + ClientSecret: "csecret", + RedirectURL: "http://localhost:18099/callback", + TokenRef: "youtube/refresh_token", + Endpoint: oauth2.Endpoint{AuthURL: srvURL + "/auth", TokenURL: srvURL + "/token"}, + } +} + +func TestExchange_PersistsRefreshToken(t *testing.T) { + srv := tokenServer(t, `{"access_token":"at","refresh_token":"rt-secret","token_type":"Bearer","expires_in":3600}`) + defer srv.Close() + + w := &fakeWriter{} + if err := auth.Exchange(context.Background(), cfg(srv.URL), w, "the-code"); err != nil { + t.Fatalf("Exchange: %v", err) + } + if w.ref != "youtube/refresh_token" { + t.Errorf("persisted ref = %q", w.ref) + } + if w.val != "rt-secret" { + t.Errorf("persisted token = %q, want rt-secret", w.val) + } +} + +func TestExchange_RejectsMissingRefreshToken(t *testing.T) { + srv := tokenServer(t, `{"access_token":"at","token_type":"Bearer","expires_in":3600}`) + defer srv.Close() + + w := &fakeWriter{} + err := auth.Exchange(context.Background(), cfg(srv.URL), w, "the-code") + if err == nil { + t.Fatal("want error when no refresh token returned") + } + if w.calls != 0 { + t.Errorf("nothing should be persisted on failure; Put called %d times", w.calls) + } +} + +// stateRe pulls the CSRF state out of the printed consent URL. +var stateRe = regexp.MustCompile(`[?&]state=([a-f0-9]+)`) + +// urlWriter forwards each Write to a channel so the test can read the consent +// URL Run prints before it blocks on the redirect. +type urlWriter struct{ ch chan string } + +func (w urlWriter) Write(p []byte) (int, error) { + w.ch <- string(p) + return len(p), nil +} + +func TestRun_FullFlow(t *testing.T) { + srv := tokenServer(t, `{"access_token":"at","refresh_token":"rt-secret","token_type":"Bearer","expires_in":3600}`) + defer srv.Close() + + // Ensure the fixed redirect port is free before binding. + if ln, err := net.Listen("tcp", "localhost:18099"); err == nil { + _ = ln.Close() + } else { + t.Skipf("redirect port 18099 unavailable: %v", err) + } + + w := &fakeWriter{} + out := urlWriter{ch: make(chan string, 4)} + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + + runErr := make(chan error, 1) + go func() { runErr <- auth.Run(ctx, cfg(srv.URL), w, out) }() + + // First message carries the consent URL with the state param. + var state string + select { + case msg := <-out.ch: + m := stateRe.FindStringSubmatch(msg) + if m == nil { + t.Fatalf("no state in consent message: %q", msg) + } + state = m[1] + case <-time.After(3 * time.Second): + t.Fatal("timed out waiting for consent URL") + } + + // Simulate the browser hitting the local redirect with code + state. + resp, err := http.Get("http://localhost:18099/callback?state=" + state + "&code=the-code") + if err != nil { + t.Fatalf("callback GET: %v", err) + } + _ = resp.Body.Close() + + select { + case err := <-runErr: + if err != nil { + t.Fatalf("Run: %v", err) + } + case <-time.After(3 * time.Second): + t.Fatal("Run did not complete after callback") + } + + if w.val != "rt-secret" { + t.Errorf("persisted token = %q, want rt-secret", w.val) + } +} + +func TestRun_RejectsStateMismatch(t *testing.T) { + srv := tokenServer(t, `{"access_token":"at","refresh_token":"rt","token_type":"Bearer","expires_in":3600}`) + defer srv.Close() + + if ln, err := net.Listen("tcp", "localhost:18099"); err == nil { + _ = ln.Close() + } else { + t.Skipf("redirect port 18099 unavailable: %v", err) + } + + w := &fakeWriter{} + out := urlWriter{ch: make(chan string, 4)} + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + + runErr := make(chan error, 1) + go func() { runErr <- auth.Run(ctx, cfg(srv.URL), w, out) }() + + select { + case <-out.ch: // drain consent URL + case <-time.After(3 * time.Second): + t.Fatal("timed out waiting for consent URL") + } + + resp, err := http.Get("http://localhost:18099/callback?state=WRONG&code=the-code") + if err != nil { + t.Fatalf("callback GET: %v", err) + } + _ = resp.Body.Close() + + select { + case err := <-runErr: + if err == nil { + t.Fatal("want error on state mismatch") + } + case <-time.After(3 * time.Second): + t.Fatal("Run did not return after bad callback") + } + if w.calls != 0 { + t.Errorf("no token should be persisted on state mismatch; Put called %d times", w.calls) + } +} From 6beb779df68b22a8fc502068b683c9c653669f9f Mon Sep 17 00:00:00 2001 From: Mathias Date: Tue, 2 Jun 2026 21:00:04 +0200 Subject: [PATCH 5/7] feat(store): UpsertVideo to persist video metadata The Sink port carries only a Summary, so video title/url/published_at would never reach the store. UpsertVideo (new file, store.go untouched) persists them and returns the durable videos.id UUID, idempotent on (user_id, provider, provider_video_id). The run loop uses that id as v.ID, so it equals summaries.video_id and SeenVideoIDs dedup survives restarts. subscription_id stays NULL: the YouTube resource id is not a UUID (Stage 0). Co-Authored-By: Claude Opus 4.8 (1M context) --- internal/adapters/store/videos.go | 77 ++++++++++++++++++++++++ internal/adapters/store/videos_test.go | 83 ++++++++++++++++++++++++++ 2 files changed, 160 insertions(+) create mode 100644 internal/adapters/store/videos.go create mode 100644 internal/adapters/store/videos_test.go diff --git a/internal/adapters/store/videos.go b/internal/adapters/store/videos.go new file mode 100644 index 0000000..64e63a5 --- /dev/null +++ b/internal/adapters/store/videos.go @@ -0,0 +1,77 @@ +package store + +import ( + "context" + "fmt" + "time" + + "gitea.d-ma.be/mathias/tapir/internal/domain" +) + +// UpsertVideo persists a video's metadata and returns its durable store id (the +// videos.id UUID). It is idempotent on (user_id, provider, provider_video_id): +// the same provider video for a user always resolves to the same row and the +// same returned id, so the run loop can use that id as the stable dedup key +// across restarts (it matches summaries.video_id once a summary exists). +// +// This lives in a separate file from store.go on purpose: the Sink port only +// carries a domain.Summary (no title/channel), so video metadata is persisted +// here, out of the delivery path, to keep the reader's rows readable. +// +// subscription_id is intentionally left NULL at Stage 0: the YouTube +// Subscription.ID is a provider resource id, not the UUID that column expects, +// and the subscriptions table is not part of this slice (data-model.md). +func (s *Store) UpsertVideo(ctx context.Context, v domain.Video) (string, error) { + if v.UserID == "" { + return "", fmt.Errorf("store: upsert video: empty user id") + } + if v.ProviderVideoID == "" { + return "", fmt.Errorf("store: upsert video: empty provider video id") + } + + tx, err := s.pool.Begin(ctx) + if err != nil { + return "", fmt.Errorf("store: begin: %w", err) + } + defer tx.Rollback(ctx) //nolint:errcheck // no-op after Commit + + // Ensure the owning user exists (FK target) — same as the Deliver path. + if _, err := tx.Exec(ctx, + `INSERT INTO users (id) VALUES ($1) ON CONFLICT (id) DO NOTHING`, + v.UserID); err != nil { + return "", fmt.Errorf("store: upsert user: %w", err) + } + + provider := string(v.Provider) + if provider == "" { + provider = string(domain.ProviderYouTube) + } + + var id string + if err := tx.QueryRow(ctx, + `INSERT INTO videos (user_id, provider, provider_video_id, title, url, published_at) + VALUES ($1, $2, $3, $4, $5, $6) + ON CONFLICT (user_id, provider, provider_video_id) DO UPDATE SET + title = EXCLUDED.title, + url = EXCLUDED.url, + published_at = EXCLUDED.published_at + RETURNING id`, + v.UserID, provider, v.ProviderVideoID, v.Title, v.URL, nullTime(v.PublishedAt), + ).Scan(&id); err != nil { + return "", fmt.Errorf("store: upsert video: %w", err) + } + + if err := tx.Commit(ctx); err != nil { + return "", fmt.Errorf("store: commit: %w", err) + } + return id, nil +} + +// nullTime maps the zero time to NULL so an unknown published_at is stored as +// SQL NULL rather than year 0001. +func nullTime(t time.Time) *time.Time { + if t.IsZero() { + return nil + } + return &t +} diff --git a/internal/adapters/store/videos_test.go b/internal/adapters/store/videos_test.go new file mode 100644 index 0000000..a30d383 --- /dev/null +++ b/internal/adapters/store/videos_test.go @@ -0,0 +1,83 @@ +package store_test + +import ( + "context" + "testing" + "time" + + "github.com/stretchr/testify/require" + + "gitea.d-ma.be/mathias/tapir/internal/domain" +) + +func ytVideo(userID, provVideoID, title string) domain.Video { + return domain.Video{ + UserID: userID, + Provider: domain.ProviderYouTube, + ProviderVideoID: provVideoID, + Title: title, + URL: "https://www.youtube.com/watch?v=" + provVideoID, + PublishedAt: time.Date(2026, 6, 1, 12, 0, 0, 0, time.UTC), + // SubscriptionID is a provider resource id, not a UUID — must not be + // written to the UUID column. Set it to prove UpsertVideo ignores it. + SubscriptionID: "yt-subscription-resource-id", + } +} + +func TestUpsertVideo_ReturnsStableID(t *testing.T) { + ctx := context.Background() + s := newStore(t) + resetDB(t, rawPool(t)) + + id1, err := s.UpsertVideo(ctx, ytVideo(userA, "dQw4w9WgXcQ", "first title")) + require.NoError(t, err) + require.NotEmpty(t, id1) + + // Same (user, provider, provider_video_id) -> same row, same id, updated meta. + id2, err := s.UpsertVideo(ctx, ytVideo(userA, "dQw4w9WgXcQ", "updated title")) + require.NoError(t, err) + require.Equal(t, id1, id2, "idempotent upsert must return the same durable id") + + p := rawPool(t) + var ( + title string + count int + ) + require.NoError(t, p.QueryRow(ctx, + `SELECT title FROM videos WHERE id = $1`, id1).Scan(&title)) + require.Equal(t, "updated title", title, "second upsert must update metadata in place") + + require.NoError(t, p.QueryRow(ctx, + `SELECT count(*) FROM videos WHERE user_id = $1`, userA).Scan(&count)) + require.Equal(t, 1, count, "must not duplicate the row") +} + +func TestUpsertVideo_IDMatchesSummaryDedup(t *testing.T) { + ctx := context.Background() + s := newStore(t) + resetDB(t, rawPool(t)) + + // Upsert assigns the durable video id; a summary delivered under that id + // must then show up in SeenVideoIDs — this is the cross-restart dedup chain. + id, err := s.UpsertVideo(ctx, ytVideo(userA, "abc123", "t")) + require.NoError(t, err) + + require.NoError(t, s.Deliver(ctx, summary(userA, id, "the summary"))) + + seen, err := s.SeenVideoIDs(ctx, userA) + require.NoError(t, err) + require.True(t, seen[id], "the upserted video id must match the summary dedup key") +} + +func TestUpsertVideo_PerUserIsolation(t *testing.T) { + ctx := context.Background() + s := newStore(t) + resetDB(t, rawPool(t)) + + idA, err := s.UpsertVideo(ctx, ytVideo(userA, "same-provider-id", "a")) + require.NoError(t, err) + idB, err := s.UpsertVideo(ctx, ytVideo(userB, "same-provider-id", "b")) + require.NoError(t, err) + + require.NotEqual(t, idA, idB, "same provider video for two users must be two distinct rows") +} From 5645c2c012bbe798fde20bc78f27b402f17ae044 Mon Sep 17 00:00:00 2001 From: Mathias Date: Tue, 2 Jun 2026 21:01:23 +0200 Subject: [PATCH 6/7] feat(runner): end-to-end run loop with durable dedup MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit RunOnce walks the user's subscriptions, upserts each candidate video (assigning its durable store id), skips videos already summarized via the store's SeenVideoIDs (cross-restart dedup the engine's in-memory map can't provide), and processes the rest through the engine. Loop adds an optional poll cadence; per-item errors are collected, not fatal. Tested with fakes — no live deps. Co-Authored-By: Claude Opus 4.8 (1M context) --- internal/runner/runner.go | 159 ++++++++++++++++++++++++++++++++ internal/runner/runner_test.go | 161 +++++++++++++++++++++++++++++++++ 2 files changed, 320 insertions(+) create mode 100644 internal/runner/runner.go create mode 100644 internal/runner/runner_test.go diff --git a/internal/runner/runner.go b/internal/runner/runner.go new file mode 100644 index 0000000..7df7d31 --- /dev/null +++ b/internal/runner/runner.go @@ -0,0 +1,159 @@ +// Package runner wires the engine to the durable store for the `tapir run` +// command. It owns the cross-restart dedup the engine core deliberately does +// not: the engine's in-memory processed map is process-lifetime only, so this +// loads the store's SeenVideoIDs and skips videos already summarized in a prior +// run. It also assigns each video its durable store id (UpsertVideo) before +// processing, so the summary's video_id equals the dedup key. +// +// It depends on small local interfaces (VideoStore, Processor), not concrete +// types, so the loop is tested with fakes — no live YouTube, gateway, or PG. +package runner + +import ( + "context" + "errors" + "fmt" + "log/slog" + "time" + + "gitea.d-ma.be/mathias/tapir/internal/domain" + "gitea.d-ma.be/mathias/tapir/internal/ports" + "gitea.d-ma.be/mathias/tapir/internal/usecase" +) + +// VideoStore is the durable persistence the run loop needs: assign a stable id + +// metadata, and read the already-summarized set. *store.Store satisfies it. +type VideoStore interface { + UpsertVideo(ctx context.Context, v domain.Video) (string, error) + SeenVideoIDs(ctx context.Context, userID string) (map[string]bool, error) +} + +// Processor runs the core use case for a single video. *usecase.Engine +// satisfies it. +type Processor interface { + ProcessNewVideo(ctx context.Context, v domain.Video) (usecase.ProcessResult, error) +} + +// Runner walks a user's subscriptions, persists each candidate video, skips the +// ones already summarized (durably), and processes the rest through the engine. +type Runner struct { + src ports.VideoSource + store VideoStore + engine Processor + userID string + log *slog.Logger +} + +// New builds a Runner. A nil logger falls back to slog.Default. +func New(src ports.VideoSource, store VideoStore, engine Processor, userID string, log *slog.Logger) *Runner { + if log == nil { + log = slog.Default() + } + return &Runner{src: src, store: store, engine: engine, userID: userID, log: log} +} + +// Stats summarizes one RunOnce pass. +type Stats struct { + Candidates int + Summarized int + SkippedSeen int + SkippedNoText int + Errors int +} + +// RunOnce performs a single pass over the user's subscriptions. Per-item errors +// are logged and collected (one bad video or channel does not abort the pass) +// and returned joined alongside the Stats gathered. +func (r *Runner) RunOnce(ctx context.Context) (Stats, error) { + var ( + stats Stats + errs []error + ) + + seen, err := r.store.SeenVideoIDs(ctx, r.userID) + if err != nil { + return stats, fmt.Errorf("runner: load seen videos: %w", err) + } + + subs, err := r.src.ListSubscriptions(ctx, r.userID) + if err != nil { + return stats, fmt.Errorf("runner: list subscriptions: %w", err) + } + + for _, sub := range subs { + vids, err := r.src.NewVideos(ctx, sub) + if err != nil { + errs = append(errs, fmt.Errorf("new videos for %q: %w", sub.ChannelTitle, err)) + stats.Errors++ + continue + } + for _, v := range vids { + stats.Candidates++ + v.UserID = r.userID // keep the dedup/FK key consistent with config + + id, err := r.store.UpsertVideo(ctx, v) + if err != nil { + errs = append(errs, fmt.Errorf("upsert video %q: %w", v.ProviderVideoID, err)) + stats.Errors++ + continue + } + v.ID = id + + if seen[id] { + stats.SkippedSeen++ + continue + } + seen[id] = true // also guard against the same video within this pass + + res, err := r.engine.ProcessNewVideo(ctx, v) + if err != nil { + errs = append(errs, fmt.Errorf("process %q: %w", v.ProviderVideoID, err)) + stats.Errors++ + continue + } + switch { + case res.Skipped: + stats.SkippedNoText++ + r.log.Info("skipped video (no transcript)", "video", v.ProviderVideoID, "title", v.Title) + case res.Summary != nil: + stats.Summarized++ + r.log.Info("summarized video", "video", v.ProviderVideoID, "title", v.Title, + "provider", res.Summary.AIProvider, "model", res.Summary.AIModel) + } + } + } + + return stats, errors.Join(errs...) +} + +// Loop runs RunOnce immediately, then on every interval tick until ctx is +// cancelled. A zero or negative interval means a single pass (no loop). Per-pass +// errors are logged, not fatal, so a transient failure doesn't kill the watcher. +func (r *Runner) Loop(ctx context.Context, interval time.Duration) error { + runPass := func() { + stats, err := r.RunOnce(ctx) + r.log.Info("run pass complete", + "candidates", stats.Candidates, "summarized", stats.Summarized, + "skipped_seen", stats.SkippedSeen, "skipped_no_text", stats.SkippedNoText, + "errors", stats.Errors) + if err != nil { + r.log.Warn("run pass had errors", "err", err) + } + } + + runPass() + if interval <= 0 { + return nil + } + + ticker := time.NewTicker(interval) + defer ticker.Stop() + for { + select { + case <-ctx.Done(): + return ctx.Err() + case <-ticker.C: + runPass() + } + } +} diff --git a/internal/runner/runner_test.go b/internal/runner/runner_test.go new file mode 100644 index 0000000..5bacdce --- /dev/null +++ b/internal/runner/runner_test.go @@ -0,0 +1,161 @@ +package runner_test + +import ( + "context" + "io" + "log/slog" + "testing" + + "github.com/stretchr/testify/require" + + "gitea.d-ma.be/mathias/tapir/internal/domain" + "gitea.d-ma.be/mathias/tapir/internal/runner" + "gitea.d-ma.be/mathias/tapir/internal/usecase" +) + +const testUser = "11111111-1111-1111-1111-111111111111" + +// --- fakes ----------------------------------------------------------------- + +type fakeSource struct { + subs []domain.Subscription + videos map[string][]domain.Video // keyed by channel id + transcripts map[string]domain.Transcript +} + +func (f *fakeSource) ListSubscriptions(_ context.Context, _ string) ([]domain.Subscription, error) { + return f.subs, nil +} + +func (f *fakeSource) NewVideos(_ context.Context, sub domain.Subscription) ([]domain.Video, error) { + return f.videos[sub.ChannelID], nil +} + +func (f *fakeSource) FetchTranscript(_ context.Context, v domain.Video) (domain.Transcript, error) { + if t, ok := f.transcripts[v.ProviderVideoID]; ok { + return t, nil + } + return domain.Transcript{VideoID: v.ID, UserID: v.UserID, Source: domain.SourceCaptions, Content: "default transcript text"}, nil +} + +// fakeStore assigns deterministic ids ("id-"+provider video id) so a pre-seeded +// seen set lines up with UpsertVideo output, modelling cross-restart dedup. +type fakeStore struct { + seen map[string]bool + upserted []domain.Video +} + +func (f *fakeStore) UpsertVideo(_ context.Context, v domain.Video) (string, error) { + f.upserted = append(f.upserted, v) + return "id-" + v.ProviderVideoID, nil +} + +func (f *fakeStore) SeenVideoIDs(_ context.Context, _ string) (map[string]bool, error) { + cp := make(map[string]bool, len(f.seen)) + for k, v := range f.seen { + cp[k] = v + } + return cp, nil +} + +type fakeSummarizer struct{} + +func (fakeSummarizer) Summarize(_ context.Context, v domain.Video, _ domain.Transcript) (domain.Summary, error) { + return domain.Summary{UserID: v.UserID, VideoID: v.ID, Summary: "s", AIProvider: "local", AIModel: "koala/phi4-mini"}, nil +} + +type recordingSink struct{ delivered []domain.Summary } + +func (s *recordingSink) Name() string { return "store" } +func (s *recordingSink) Deliver(_ context.Context, sum domain.Summary) error { + s.delivered = append(s.delivered, sum) + return nil +} + +func sub(channelID, title string) domain.Subscription { + return domain.Subscription{UserID: testUser, ChannelID: channelID, ChannelTitle: title, Active: true} +} + +func vid(provID, title string) domain.Video { + return domain.Video{UserID: testUser, Provider: domain.ProviderYouTube, ProviderVideoID: provID, Title: title} +} + +func quietLogger() *slog.Logger { + return slog.New(slog.NewTextHandler(io.Discard, nil)) +} + +// --- tests ----------------------------------------------------------------- + +func TestRunOnce_SummarizesNewVideos(t *testing.T) { + src := &fakeSource{ + subs: []domain.Subscription{sub("chan1", "Channel One")}, + videos: map[string][]domain.Video{"chan1": {vid("v1", "Video 1"), vid("v2", "Video 2")}}, + } + st := &fakeStore{seen: map[string]bool{}} + sink := &recordingSink{} + eng := usecase.NewEngine(src, fakeSummarizer{}, sink) + r := runner.New(src, st, eng, testUser, quietLogger()) + + stats, err := r.RunOnce(context.Background()) + require.NoError(t, err) + require.Equal(t, 2, stats.Candidates) + require.Equal(t, 2, stats.Summarized) + require.Equal(t, 0, stats.SkippedSeen) + require.Len(t, sink.delivered, 2) + + // Each delivered summary must carry the durable store id as its video id. + require.Equal(t, "id-v1", sink.delivered[0].VideoID) + require.Equal(t, "id-v2", sink.delivered[1].VideoID) +} + +func TestRunOnce_SkipsAlreadySummarized(t *testing.T) { + src := &fakeSource{ + subs: []domain.Subscription{sub("chan1", "Channel One")}, + videos: map[string][]domain.Video{"chan1": {vid("v1", "Video 1"), vid("v2", "Video 2")}}, + } + // v1 was summarized in a prior run (durable seen set). + st := &fakeStore{seen: map[string]bool{"id-v1": true}} + sink := &recordingSink{} + eng := usecase.NewEngine(src, fakeSummarizer{}, sink) + r := runner.New(src, st, eng, testUser, quietLogger()) + + stats, err := r.RunOnce(context.Background()) + require.NoError(t, err) + require.Equal(t, 1, stats.SkippedSeen) + require.Equal(t, 1, stats.Summarized) + require.Len(t, sink.delivered, 1) + require.Equal(t, "id-v2", sink.delivered[0].VideoID, "only the unseen video is summarized") +} + +func TestRunOnce_SkipsVideosWithoutTranscript(t *testing.T) { + src := &fakeSource{ + subs: []domain.Subscription{sub("chan1", "Channel One")}, + videos: map[string][]domain.Video{"chan1": {vid("v1", "Video 1")}}, + transcripts: map[string]domain.Transcript{"v1": {Source: domain.SourceNone}}, + } + st := &fakeStore{seen: map[string]bool{}} + sink := &recordingSink{} + eng := usecase.NewEngine(src, fakeSummarizer{}, sink) + r := runner.New(src, st, eng, testUser, quietLogger()) + + stats, err := r.RunOnce(context.Background()) + require.NoError(t, err) + require.Equal(t, 1, stats.SkippedNoText) + require.Equal(t, 0, stats.Summarized) + require.Empty(t, sink.delivered, "no summary delivered when there is no transcript") +} + +func TestRunOnce_UpsertsEveryCandidate(t *testing.T) { + src := &fakeSource{ + subs: []domain.Subscription{sub("chan1", "Channel One")}, + videos: map[string][]domain.Video{"chan1": {vid("v1", "Video 1"), vid("v2", "Video 2")}}, + } + // Even an already-seen video gets upserted so its metadata stays fresh. + st := &fakeStore{seen: map[string]bool{"id-v1": true}} + eng := usecase.NewEngine(src, fakeSummarizer{}, &recordingSink{}) + r := runner.New(src, st, eng, testUser, quietLogger()) + + _, err := r.RunOnce(context.Background()) + require.NoError(t, err) + require.Len(t, st.upserted, 2, "every candidate is upserted, including seen ones") +} From 61796db16b66e7cf993d0218b0ff3a49146bc6b7 Mon Sep 17 00:00:00 2001 From: Mathias Date: Tue, 2 Jun 2026 21:03:21 +0200 Subject: [PATCH 7/7] feat(cmd): wire auth/run dispatcher + demo docs MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit main.go dispatches `tapir auth` (interactive OAuth → persist refresh token via SecretStore) and `tapir run` (wire YouTube source + local summarizer + store sink, build engine, run the dedup-aware loop). Config-driven so live creds plug in at demo time; SIGINT stops the loop cleanly. Block kept minimal so Worker E's list/show cases union cleanly at merge. Add .env.example documenting every TAPIR_* var and a README demo runbook. Pin the summarizer alias-as-config decision and record the max_tokens fix in docs/homelab-integration.md (clears two `confirm` items). Co-Authored-By: Claude Opus 4.8 (1M context) --- README.md | 33 ++++++++++ cmd/tapir/main.go | 128 ++++++++++++++++++++++++++++++++++-- docs/homelab-integration.md | 21 +++--- internal/auth/auth.go | 4 +- 4 files changed, 171 insertions(+), 15 deletions(-) diff --git a/README.md b/README.md index 18de2a2..d33b62b 100644 --- a/README.md +++ b/README.md @@ -32,6 +32,39 @@ intent is version-controlled and the build has something to be checked against. - **Trunk-Based Development.** Commit directly to `main`, one logical change per commit, every commit deployable (see ADR-009). CI is the quality gate. +## Running the Stage-0 demo + +Tapir runs on **your own** YouTube account: authorize once, then run the +watch→summarize→deliver loop. All configuration is via `TAPIR_*` environment +variables — copy [`.env.example`](.env.example) to `.env` and fill it in (no +secrets are committed; at demo time source them from op, e.g. `op run -- ...`). + +```sh +# 1. configure (UUID user id, gateway URL+key, Postgres DSN, YouTube OAuth app, +# summarizer model). See .env.example for every variable. +cp .env.example .env && $EDITOR .env +set -a && . ./.env && set +a # export them into the shell + +go build -o bin/tapir ./cmd/tapir + +# 2. one-time: authorize YouTube. Opens a consent URL, captures the redirect on +# TAPIR_OAUTH_REDIRECT_ADDR, and stores the refresh token via the SecretStore +# (a 0600 file at Stage 0). The token is never logged. +./bin/tapir auth + +# 3. run: detect new videos across your subscriptions, summarize, deliver to the +# store. Unset TAPIR_POLL_INTERVAL = single pass; set it (e.g. 15m) to loop. +./bin/tapir run +``` + +Live prerequisites at demo time: the LiteLLM gateway reachable +(`TAPIR_GATEWAY_URL` + a valid key — resolve from op, the documented +`sk-local-123` is stale), a Postgres DSN (`TAPIR_DB_DSN`, migrations apply on +first connect), and a registered YouTube OAuth client whose authorized redirect +URI matches `TAPIR_OAUTH_REDIRECT_ADDR`. The summarizer model +(`TAPIR_SUMMARIZER_MODEL`, default `koala/phi4-mini`) is overridable; pick the +final alias when the gateway is reachable (see `docs/homelab-integration.md`). + ## Conventions Reuses homelab conventions: Go, Dex for identity, ESO + 1Password for secrets, Postgres for diff --git a/cmd/tapir/main.go b/cmd/tapir/main.go index dd1e7a6..098e04e 100644 --- a/cmd/tapir/main.go +++ b/cmd/tapir/main.go @@ -1,10 +1,128 @@ -// Command tapir is the service entrypoint. Scaffold: it identifies itself so -// the CI smoke test has something to grep for, and exits. Wiring the HTTP -// server, watcher, adapters, and config is part of the build. +// Command tapir is the service entrypoint. It dispatches the Stage-0 demo +// subcommands: +// +// tapir auth mint a YouTube refresh token interactively (one-time setup) +// tapir run detect new videos, summarize them, deliver to the store +// +// 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 "fmt" +import ( + "context" + "fmt" + "log/slog" + "os" + "os/signal" + + "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" +) func main() { - fmt.Println("tapir: scaffold — not yet implemented") + 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 "auth": + err = cmdAuth(ctx, log) + case "run": + err = cmdRun(ctx, log) + 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 + +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() + + 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) + + engine := usecase.NewEngine(src, sum, st) + r := runner.New(src, st, engine, cfg.UserID, log) + + log.Info("starting run", "user", cfg.UserID, "model", cfg.SummarizerModel, + "gateway", cfg.GatewayURL, "poll_interval", cfg.PollInterval) + return r.Loop(ctx, cfg.PollInterval) } diff --git a/docs/homelab-integration.md b/docs/homelab-integration.md index 7407519..d596d3d 100644 --- a/docs/homelab-integration.md +++ b/docs/homelab-integration.md @@ -18,16 +18,21 @@ it** — endpoints and aliases drift, and this file is a snapshot (2026-06-02), Resolve the live key from the vault when wiring; `sk-local-123` is no longer valid. `confirm` partially resolved. - **Model alias format:** `host/name`, e.g. `koala/qwen3-coder-30b`, `koala/phi4-mini`, `iguana/devstral`, `iguana/deepseek-r1-14b`. **Not** the `ollama/` prefix form. -- **Which alias for summarization:** NOT yet decided. `confirm`. Tapir summarizes transcript - text, so a capable general/instruct model on koala or iguana is the candidate — pick during the - build and record the choice (an ADR if it's load-bearing). Do not assume a coder alias is right - for prose summarization. The summarizer adapter does **not** hardcode an alias: it is config, - env `TAPIR_SUMMARIZER_MODEL` (format `host/name`, e.g. `iguana/deepseek-r1-14b`). +- **Which alias for summarization:** alias-as-config, **`confirm` resolved** (2026-06-02, Worker F). + The alias is never hardcoded: it is `TAPIR_SUMMARIZER_MODEL` (format `host/name`), wired through + the summarizer's `Endpoint.Model`. **Default: `koala/phi4-mini`** — a non-thinking instruct model + chosen for safety: it cannot fall into the empty-content trap below, so the demo summarizes even + if no one tunes it. It is provisional and overridable; **final live alias selection happens at + demo time when the gateway is reachable**, where a more capable model (e.g. + `iguana/deepseek-r1-14b`) is preferred for summary quality if its latency/output is acceptable. + The `max_tokens` fix below means thinking models no longer return empty content, so they are now + viable choices, not blocked ones. Do not assume a coder alias is right for prose. - **Thinking models need an explicit `max_tokens`.** qwen3 / deepseek-r1 spend the budget on reasoning and return **empty content** if `max_tokens` is too low (or unset). The summarizer's - parser treats an empty summary as an error for exactly this reason. When the alias resolves to a - thinking model, add a generous `max_tokens` to the copied `llm.Client` request (it currently - sends none — change Tapir's copy per ADR-004), or pick a non-thinking instruct model. + parser treats an empty summary as an error for exactly this reason. **Done (2026-06-02, Worker F):** + the copied `llm.Client` now sends a generous `max_tokens` (8192) on every request per ADR-004, so + thinking models no longer return empty content. A non-thinking instruct model remains the safe + default (`koala/phi4-mini`), but thinking aliases are now viable. This maps directly onto the copied `llm` package: `Client` is the OpenAI-compatible caller, `Router.Primary` points at this gateway with a chosen alias, `Router.Fallback` is the user's BYO. diff --git a/internal/auth/auth.go b/internal/auth/auth.go index cc00b19..8929dcd 100644 --- a/internal/auth/auth.go +++ b/internal/auth/auth.go @@ -150,7 +150,7 @@ func Run(ctx context.Context, c Config, secrets TokenWriter, out io.Writer) erro // on re-authorization; without them a repeat consent yields only an access // token and Exchange would reject it. authURL := conf.AuthCodeURL(state, oauth2.AccessTypeOffline, oauth2.ApprovalForce) - fmt.Fprintf(out, "Open this URL to authorize Tapir, then return here:\n\n%s\n\n", authURL) + _, _ = fmt.Fprintf(out, "Open this URL to authorize Tapir, then return here:\n\n%s\n\n", authURL) select { case <-ctx.Done(): @@ -162,7 +162,7 @@ func Run(ctx context.Context, c Config, secrets TokenWriter, out io.Writer) erro if err := Exchange(ctx, c, secrets, res.code); err != nil { return err } - fmt.Fprintln(out, "Refresh token stored. You can now run `tapir run`.") + _, _ = fmt.Fprintln(out, "Refresh token stored. You can now run `tapir run`.") return nil } }