// 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" "git.d-ma.be/mathias/tapir/internal/domain" "git.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 // MinVideoSeconds drops videos shorter than this from discovery (Shorts/clips, // ADR-023). NewVideos enriches candidates with a single cheap videos.list call // (contentDetails.duration + snippet.liveBroadcastContent) and filters before // returning, so the scarce caption-fetch budget is never spent on them. Live // and upcoming broadcasts are dropped too. Zero disables the filter. MinVideoSeconds int // BaseURL overrides the Data API root. Empty means defaultBaseURL. BaseURL string // PlayerBaseURL overrides the InnerTube / watch-page host used for // unauthenticated transcript acquisition (ADR-010). Empty means // defaultPlayerBaseURL. Tests point it at an httptest server. PlayerBaseURL string } const defaultMaxVideos = 10 // Adapter implements ports.VideoSource for YouTube. type Adapter struct { cfg Config secrets ports.SecretStore baseURL string playerBaseURL 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 } player := cfg.PlayerBaseURL if player == "" { player = defaultPlayerBaseURL } return &Adapter{ cfg: cfg, secrets: secrets, baseURL: strings.TrimRight(base, "/"), playerBaseURL: strings.TrimRight(player, "/"), } } // 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. // // Discovery uses playlistItems.list against the channel's "uploads" playlist // (1 quota unit/call) instead of search.list (100 units): with ~143 channels a // single search-based pass exhausted the entire 10,000 unit/day cap. For a // standard channel id "UCxxxx" the uploads playlist is "UUxxxx" — derived with // zero API cost. Non-standard ids that don't start with "UC" fall back to // channels.list (1 unit) to read contentDetails.relatedPlaylists.uploads. // Dropping search.list also removes the accountDelegationForbidden failure mode // that endpoint exhibited for one channel. 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 } playlistID, ok := uploadsPlaylistID(sub.ChannelID) if !ok { // Non-standard channel id: resolve the uploads playlist explicitly. playlistID, err = a.resolveUploadsPlaylist(ctx, client, sub.ChannelID) if err != nil { return nil, fmt.Errorf("new videos for channel %q: %w", sub.ChannelID, err) } } q := url.Values{ "part": {"snippet"}, "playlistId": {playlistID}, "maxResults": {fmt.Sprintf("%d", max)}, } var resp playlistItemListResponse if err := a.getJSON(ctx, client, "/playlistItems", q, &resp); err != nil { if isHTTP404(err) { return nil, &domain.ErrChannelUnavailable{ChannelID: sub.ChannelID, ChannelTitle: sub.ChannelTitle} } 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 { vid := item.Snippet.ResourceID.VideoID if vid == "" { continue // defensive: skip items without a resolvable video id } videos = append(videos, domain.Video{ UserID: sub.UserID, SubscriptionID: sub.ID, Provider: domain.ProviderYouTube, ProviderVideoID: vid, Title: item.Snippet.Title, ChannelTitle: sub.ChannelTitle, URL: "https://www.youtube.com/watch?v=" + vid, PublishedAt: item.Snippet.PublishedAt, }) if len(videos) >= max { break } } // Drop Shorts/sub-minute clips and live/upcoming broadcasts before they ever // reach the rate-limited caption path (ADR-023). One cheap videos.list call // (quota API, not the timedtext throttle) supplies duration + live status. return a.filterLowValue(ctx, client, videos), nil } // filterLowValue removes videos shorter than cfg.MinVideoSeconds and any live or // upcoming broadcast, using a single videos.list lookup for duration + // liveBroadcastContent. The filter is best-effort: if MinVideoSeconds is 0 (off) // or the lookup fails, the input is returned unfiltered — discovery must not break // because a metadata call hiccuped; the worst case is the pre-ADR-023 behaviour. func (a *Adapter) filterLowValue(ctx context.Context, client *http.Client, videos []domain.Video) []domain.Video { if a.cfg.MinVideoSeconds <= 0 || len(videos) == 0 { return videos } ids := make([]string, 0, len(videos)) for _, v := range videos { ids = append(ids, v.ProviderVideoID) } q := url.Values{ "part": {"contentDetails,snippet"}, "id": {strings.Join(ids, ",")}, } var resp videoListResponse if err := a.getJSON(ctx, client, "/videos", q, &resp); err != nil { // Degrade open: keep the candidates rather than lose discovery. return videos } type meta struct { seconds int live string } byID := make(map[string]meta, len(resp.Items)) for _, it := range resp.Items { byID[it.ID] = meta{seconds: parseISO8601Seconds(it.ContentDetails.Duration), live: it.Snippet.LiveBroadcastContent} } kept := videos[:0] for _, v := range videos { m, ok := byID[v.ProviderVideoID] if !ok { kept = append(kept, v) // unknown metadata: keep, let the fetch decide continue } if m.live != "" && m.live != "none" { continue // live or upcoming broadcast } if m.seconds > 0 && m.seconds < a.cfg.MinVideoSeconds { continue // Short / sub-threshold clip } // Carry the duration we already fetched onto the kept video so the store // can persist it (ADR-028) — the burst's length-aware selection depends on // it. Discarding it here was the gap the onboarding investigation found. v.DurationSeconds = m.seconds kept = append(kept, v) } return kept } // parseISO8601Seconds parses an ISO 8601 duration as returned by the YouTube Data // API (e.g. "PT1H2M3S", "PT45S", "PT3M") into seconds. Only the hour/minute/second // components YouTube emits are handled; an unparseable or zero value returns 0, // which the caller treats as "unknown" (not filtered on duration). func parseISO8601Seconds(d string) int { if !strings.HasPrefix(d, "PT") { return 0 } d = d[2:] total, num := 0, 0 seen := false for _, r := range d { switch { case r >= '0' && r <= '9': num = num*10 + int(r-'0') seen = true case r == 'H': total += num * 3600 num, seen = 0, false case r == 'M': total += num * 60 num, seen = 0, false case r == 'S': total += num num, seen = 0, false default: return 0 // unexpected component (days/weeks) — treat as unknown } } if seen { return 0 // trailing digits without a unit: malformed } return total } // VideoByID fetches a single video's metadata (videos.list, snippet) for an // arbitrary video id — including channels the user does not follow (paste-a-URL, // Feature 2). This is a Data API call (1 quota unit), NOT the rate-limited // caption path, so it is not gated: only the later transcript fetch goes through // globalFetchGate. UserID is set on the result and SubscriptionID is left empty // (a pasted video has no subscription parent). Returns ErrVideoNotFound when the // id resolves to no video. func (a *Adapter) VideoByID(ctx context.Context, userID, videoID string) (domain.Video, error) { client, err := a.httpClient(ctx, a.cfg.TokenSecretRef) if err != nil { return domain.Video{}, err } q := url.Values{"part": {"snippet"}, "id": {videoID}} var resp videoListResponse if err := a.getJSON(ctx, client, "/videos", q, &resp); err != nil { return domain.Video{}, fmt.Errorf("video by id %q: %w", videoID, err) } if len(resp.Items) == 0 { return domain.Video{}, fmt.Errorf("video %q: %w", videoID, domain.ErrVideoNotFound) } it := resp.Items[0] return domain.Video{ UserID: userID, Provider: domain.ProviderYouTube, ProviderVideoID: videoID, Title: it.Snippet.Title, ChannelTitle: it.Snippet.ChannelTitle, URL: "https://www.youtube.com/watch?v=" + videoID, PublishedAt: it.Snippet.PublishedAt, }, nil } // uploadsPlaylistID derives a channel's uploads playlist id at zero API cost: // a standard channel id "UCxxxx" maps to uploads playlist "UUxxxx". Returns // ok=false for ids that don't follow this convention (caller falls back to // channels.list). func uploadsPlaylistID(channelID string) (string, bool) { if !strings.HasPrefix(channelID, "UC") { return "", false } return "UU" + channelID[2:], true } // resolveUploadsPlaylist reads contentDetails.relatedPlaylists.uploads via // channels.list (1 quota unit) for channels whose id can't be mapped UC->UU. func (a *Adapter) resolveUploadsPlaylist(ctx context.Context, client *http.Client, channelID string) (string, error) { q := url.Values{ "part": {"contentDetails"}, "id": {channelID}, } var resp channelListResponse if err := a.getJSON(ctx, client, "/channels", q, &resp); err != nil { return "", fmt.Errorf("resolve uploads playlist: %w", err) } if len(resp.Items) == 0 || resp.Items[0].ContentDetails.RelatedPlaylists.Uploads == "" { return "", fmt.Errorf("no uploads playlist for channel %q", channelID) } return resp.Items[0].ContentDetails.RelatedPlaylists.Uploads, nil } // isHTTP404 reports whether err came from a YouTube API call that returned HTTP 404. // getRaw encodes the status as "youtube api : status 404: ...". func isHTTP404(err error) bool { return err != nil && strings.Contains(err.Error(), "status 404") } // 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 playlistItemListResponse struct { Items []struct { Snippet struct { Title string `json:"title"` PublishedAt time.Time `json:"publishedAt"` ResourceID struct { VideoID string `json:"videoId"` } `json:"resourceId"` } `json:"snippet"` } `json:"items"` } type videoListResponse struct { Items []struct { ID string `json:"id"` Snippet struct { Title string `json:"title"` ChannelTitle string `json:"channelTitle"` PublishedAt time.Time `json:"publishedAt"` LiveBroadcastContent string `json:"liveBroadcastContent"` } `json:"snippet"` ContentDetails struct { Duration string `json:"duration"` // ISO 8601, e.g. "PT1M30S" } `json:"contentDetails"` } `json:"items"` } type channelListResponse struct { Items []struct { ContentDetails struct { RelatedPlaylists struct { Uploads string `json:"uploads"` } `json:"relatedPlaylists"` } `json:"contentDetails"` } `json:"items"` }