feat(adapters): add captions-first YouTube VideoSource
Implements ports.VideoSource against the YouTube Data API v3: ListSubscriptions (paginated), NewVideos (recent per channel), and captions-first FetchTranscript — an absent caption track yields domain.SourceNone (not an error) per ADR-007, with no audio download or speech-to-text. OAuth is written fresh on golang.org/x/oauth2 (ADR-006, distinct from ingestion's inbound MCP auth); the Google token endpoint is inlined to avoid the heavy x/oauth2/google dep. The per-connection refresh token is resolved through the SecretStore port from an opaque TokenSecretRef and is never stored on the adapter or logged. Unit-tested against an httptest server + fake SecretStore (no live googleapis egress): subscriptions list/pagination, new-video detection, captions present -> Source set, captions absent -> SourceNone no error, and secret-ref resolution failure surfacing as an error. oauth2 pinned to v0.30.0 to keep the go directive at 1.23.x (koala runner), not the v0.36 line that requires a newer toolchain. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
Claude Opus 4.8
@@ -0,0 +1,275 @@
|
||||
// Package youtube is a ports.VideoSource adapter backed by the YouTube Data API
|
||||
// v3. It lists a user's subscriptions, detects recent videos per subscription,
|
||||
// and resolves transcripts captions-first (ADR-007): when no usable caption
|
||||
// track exists it returns a domain.Transcript with Source == SourceNone, never
|
||||
// an error, and it never downloads audio or runs speech-to-text.
|
||||
//
|
||||
// OAuth is written fresh against golang.org/x/oauth2 (ADR-006 — the ingestion
|
||||
// repo's oauth package is inbound MCP auth, unrelated to outbound provider
|
||||
// OAuth). The per-user refresh token is resolved through the SecretStore port
|
||||
// from an opaque reference; the token material is never stored on the adapter,
|
||||
// logged, or returned.
|
||||
//
|
||||
// Live runtime egress: www.googleapis.com (Data API) and oauth2.googleapis.com
|
||||
// (token refresh). Tests inject an httptest transport and a fake SecretStore so
|
||||
// neither host is contacted.
|
||||
package youtube
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"io"
|
||||
"net/http"
|
||||
"net/url"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
"golang.org/x/oauth2"
|
||||
|
||||
"gitea.d-ma.be/mathias/tapir/internal/domain"
|
||||
"gitea.d-ma.be/mathias/tapir/internal/ports"
|
||||
)
|
||||
|
||||
// defaultBaseURL is the YouTube Data API v3 root. Overridable via Config.BaseURL
|
||||
// (tests point it at an httptest server).
|
||||
const defaultBaseURL = "https://www.googleapis.com/youtube/v3"
|
||||
|
||||
// googleEndpoint is the OAuth2 token endpoint for Google. Defined inline to avoid
|
||||
// pulling the heavy golang.org/x/oauth2/google dependency for a single URL.
|
||||
var googleEndpoint = oauth2.Endpoint{
|
||||
AuthURL: "https://accounts.google.com/o/oauth2/auth",
|
||||
TokenURL: "https://oauth2.googleapis.com/token",
|
||||
}
|
||||
|
||||
// Config wires the adapter. ClientID/ClientSecret are the registered app
|
||||
// credentials. TokenSecretRef is the opaque SecretStore reference that resolves
|
||||
// to the connection's OAuth refresh token — the exact ref/vault-item naming for
|
||||
// Tapir is still `confirm` (see docs/homelab-integration.md); the adapter treats
|
||||
// it as an opaque parameter and never assumes a scheme.
|
||||
//
|
||||
// At Stage 0 there is a single connection, so one TokenSecretRef/ConnectionID
|
||||
// covers every call. Mapping a Subscription back to its connection's ref is a
|
||||
// Stage 1 concern (per-user isolation, data-model.md) and is deliberately not
|
||||
// modelled here.
|
||||
type Config struct {
|
||||
ClientID string
|
||||
ClientSecret string
|
||||
TokenSecretRef string
|
||||
ConnectionID string
|
||||
|
||||
// PreferredLanguages orders caption-track selection (e.g. {"en"}). The first
|
||||
// track matching a preferred language wins; otherwise the first track is used.
|
||||
PreferredLanguages []string
|
||||
|
||||
// MaxVideosPerSubscription caps how many recent videos NewVideos returns per
|
||||
// poll. Zero means defaultMaxVideos.
|
||||
MaxVideosPerSubscription int
|
||||
|
||||
// BaseURL overrides the Data API root. Empty means defaultBaseURL.
|
||||
BaseURL string
|
||||
}
|
||||
|
||||
const defaultMaxVideos = 10
|
||||
|
||||
// Adapter implements ports.VideoSource for YouTube.
|
||||
type Adapter struct {
|
||||
cfg Config
|
||||
secrets ports.SecretStore
|
||||
baseURL string
|
||||
|
||||
// transport, when non-nil, replaces the live OAuth2 transport for API calls.
|
||||
// Production leaves it nil and an oauth2-authorized client is built per call.
|
||||
// Tests inject an httptest transport so no Google host is contacted while the
|
||||
// SecretStore resolution path still runs.
|
||||
transport http.RoundTripper
|
||||
}
|
||||
|
||||
// New builds a YouTube adapter. secrets must be non-nil: every authorized call
|
||||
// resolves the refresh token through it by reference.
|
||||
func New(cfg Config, secrets ports.SecretStore) *Adapter {
|
||||
base := cfg.BaseURL
|
||||
if base == "" {
|
||||
base = defaultBaseURL
|
||||
}
|
||||
return &Adapter{
|
||||
cfg: cfg,
|
||||
secrets: secrets,
|
||||
baseURL: strings.TrimRight(base, "/"),
|
||||
}
|
||||
}
|
||||
|
||||
// httpClient returns an HTTP client authorized for the connection's token ref.
|
||||
// The refresh token is resolved through SecretStore on every call and is never
|
||||
// retained or logged; oauth2 exchanges it for a short-lived access token against
|
||||
// googleEndpoint.TokenURL.
|
||||
func (a *Adapter) httpClient(ctx context.Context, tokenRef string) (*http.Client, error) {
|
||||
refresh, err := a.secrets.Get(ctx, tokenRef)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("resolve youtube oauth token: %w", err)
|
||||
}
|
||||
if a.transport != nil {
|
||||
// Test seam: the SecretStore lookup above has run (proving by-reference
|
||||
// resolution); skip the live token exchange and use the injected transport.
|
||||
return &http.Client{Transport: a.transport}, nil
|
||||
}
|
||||
conf := &oauth2.Config{
|
||||
ClientID: a.cfg.ClientID,
|
||||
ClientSecret: a.cfg.ClientSecret,
|
||||
Endpoint: googleEndpoint,
|
||||
}
|
||||
ts := conf.TokenSource(ctx, &oauth2.Token{RefreshToken: refresh})
|
||||
return oauth2.NewClient(ctx, ts), nil
|
||||
}
|
||||
|
||||
// ListSubscriptions returns the channels the user subscribes to, paging through
|
||||
// the Data API until exhausted.
|
||||
func (a *Adapter) ListSubscriptions(ctx context.Context, userID string) ([]domain.Subscription, error) {
|
||||
client, err := a.httpClient(ctx, a.cfg.TokenSecretRef)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
var subs []domain.Subscription
|
||||
pageToken := ""
|
||||
for {
|
||||
q := url.Values{
|
||||
"part": {"snippet"},
|
||||
"mine": {"true"},
|
||||
"maxResults": {"50"},
|
||||
}
|
||||
if pageToken != "" {
|
||||
q.Set("pageToken", pageToken)
|
||||
}
|
||||
|
||||
var resp subscriptionListResponse
|
||||
if err := a.getJSON(ctx, client, "/subscriptions", q, &resp); err != nil {
|
||||
return nil, fmt.Errorf("list subscriptions: %w", err)
|
||||
}
|
||||
for _, item := range resp.Items {
|
||||
subs = append(subs, domain.Subscription{
|
||||
ID: item.ID,
|
||||
UserID: userID,
|
||||
ConnectionID: a.cfg.ConnectionID,
|
||||
ChannelID: item.Snippet.ResourceID.ChannelID,
|
||||
ChannelTitle: item.Snippet.Title,
|
||||
Active: true,
|
||||
})
|
||||
}
|
||||
if resp.NextPageToken == "" {
|
||||
break
|
||||
}
|
||||
pageToken = resp.NextPageToken
|
||||
}
|
||||
return subs, nil
|
||||
}
|
||||
|
||||
// NewVideos returns the most recent videos for a subscription's channel, newest
|
||||
// first. The engine dedups within its lifetime and the store dedups durably
|
||||
// (data-model.md), so the adapter returns recent candidates rather than tracking
|
||||
// "seen" state itself.
|
||||
func (a *Adapter) NewVideos(ctx context.Context, sub domain.Subscription) ([]domain.Video, error) {
|
||||
client, err := a.httpClient(ctx, a.cfg.TokenSecretRef)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
max := a.cfg.MaxVideosPerSubscription
|
||||
if max <= 0 {
|
||||
max = defaultMaxVideos
|
||||
}
|
||||
|
||||
q := url.Values{
|
||||
"part": {"snippet"},
|
||||
"channelId": {sub.ChannelID},
|
||||
"order": {"date"},
|
||||
"type": {"video"},
|
||||
"maxResults": {fmt.Sprintf("%d", max)},
|
||||
}
|
||||
|
||||
var resp searchListResponse
|
||||
if err := a.getJSON(ctx, client, "/search", q, &resp); err != nil {
|
||||
return nil, fmt.Errorf("new videos for channel %q: %w", sub.ChannelID, err)
|
||||
}
|
||||
|
||||
videos := make([]domain.Video, 0, len(resp.Items))
|
||||
for _, item := range resp.Items {
|
||||
if item.ID.VideoID == "" {
|
||||
continue // non-video result; type=video should prevent this, be defensive
|
||||
}
|
||||
videos = append(videos, domain.Video{
|
||||
UserID: sub.UserID,
|
||||
SubscriptionID: sub.ID,
|
||||
Provider: domain.ProviderYouTube,
|
||||
ProviderVideoID: item.ID.VideoID,
|
||||
Title: item.Snippet.Title,
|
||||
URL: "https://www.youtube.com/watch?v=" + item.ID.VideoID,
|
||||
PublishedAt: item.Snippet.PublishedAt,
|
||||
})
|
||||
}
|
||||
return videos, nil
|
||||
}
|
||||
|
||||
// getJSON issues a GET and decodes a JSON body into out. A non-200 status is an
|
||||
// error carrying a bounded slice of the response body for diagnosis.
|
||||
func (a *Adapter) getJSON(ctx context.Context, client *http.Client, path string, q url.Values, out any) error {
|
||||
body, err := a.getRaw(ctx, client, path, q)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if err := json.Unmarshal(body, out); err != nil {
|
||||
return fmt.Errorf("decode %s response: %w", path, err)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// getRaw issues a GET and returns the raw response body (used for caption
|
||||
// downloads, which are not JSON).
|
||||
func (a *Adapter) getRaw(ctx context.Context, client *http.Client, path string, q url.Values) ([]byte, error) {
|
||||
u := a.baseURL + path
|
||||
if len(q) > 0 {
|
||||
u += "?" + q.Encode()
|
||||
}
|
||||
req, err := http.NewRequestWithContext(ctx, http.MethodGet, u, nil)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("build request %s: %w", path, err)
|
||||
}
|
||||
resp, err := client.Do(req)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("request %s: %w", path, err)
|
||||
}
|
||||
defer func() { _ = resp.Body.Close() }()
|
||||
|
||||
if resp.StatusCode != http.StatusOK {
|
||||
snippet, _ := io.ReadAll(io.LimitReader(resp.Body, 2048))
|
||||
return nil, fmt.Errorf("youtube api %s: status %d: %s", path, resp.StatusCode, strings.TrimSpace(string(snippet)))
|
||||
}
|
||||
return io.ReadAll(resp.Body)
|
||||
}
|
||||
|
||||
// --- Data API response shapes (only the fields used) -----------------------
|
||||
|
||||
type subscriptionListResponse struct {
|
||||
NextPageToken string `json:"nextPageToken"`
|
||||
Items []struct {
|
||||
ID string `json:"id"`
|
||||
Snippet struct {
|
||||
Title string `json:"title"`
|
||||
ResourceID struct {
|
||||
ChannelID string `json:"channelId"`
|
||||
} `json:"resourceId"`
|
||||
} `json:"snippet"`
|
||||
} `json:"items"`
|
||||
}
|
||||
|
||||
type searchListResponse struct {
|
||||
Items []struct {
|
||||
ID struct {
|
||||
VideoID string `json:"videoId"`
|
||||
} `json:"id"`
|
||||
Snippet struct {
|
||||
Title string `json:"title"`
|
||||
PublishedAt time.Time `json:"publishedAt"`
|
||||
} `json:"snippet"`
|
||||
} `json:"items"`
|
||||
}
|
||||
Reference in New Issue
Block a user