tapir/internal/adapters/youtube/youtube.go

// 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"`
}