diff --git a/apps/web/src/app/api/stream/egress/start/route.ts b/apps/web/src/app/api/stream/egress/start/route.ts index d32d805..f38bf2f 100644 --- a/apps/web/src/app/api/stream/egress/start/route.ts +++ b/apps/web/src/app/api/stream/egress/start/route.ts @@ -3,6 +3,7 @@ import { createClient as createSupabaseAdminClient } from '@supabase/supabase-js import { createClient, getAuthenticatedUser } from '@/lib/supabase/server'; import { successResponse, errorResponse, handleApiError } from '@/lib/api'; import { getEgressClient } from '@/lib/livekit-egress'; +import { autoTransitionYouTube } from '@/lib/youtubeAutoLive'; import type { Database, Session } from '@pairux/shared-types'; /** @@ -122,6 +123,13 @@ export async function POST(request: Request) { { layout: 'speaker', encodingOptions } ); + // If streaming to YouTube and the host has connected their YouTube account, + // auto-transition the broadcast to live so it never sits on "Preparing". + // Fire-and-forget — it polls for ~90s and must not delay/fail egress start. + if (rtmpUrls.some((u) => /(^|\.)youtube\.com\//i.test(u))) { + void autoTransitionYouTube(user.id); + } + // egressIds[] kept for clients that track the full set; egressId for older ones. return successResponse({ egressIds: [info.egressId], egressId: info.egressId }); } catch (error) { diff --git a/apps/web/src/app/api/youtube/auth/route.ts b/apps/web/src/app/api/youtube/auth/route.ts new file mode 100644 index 0000000..f04f612 --- /dev/null +++ b/apps/web/src/app/api/youtube/auth/route.ts @@ -0,0 +1,30 @@ +import { NextResponse } from 'next/server'; +import { randomBytes } from 'node:crypto'; +import { createClient, getAuthenticatedUser } from '@/lib/supabase/server'; +import { buildAuthUrl, youtubeOAuthConfigured } from '@/lib/youtube'; + +/** Start the YouTube OAuth flow: set a CSRF state cookie, redirect to Google. */ +export async function GET(request: Request) { + const appBase = process.env.NEXT_PUBLIC_APP_URL ?? new URL(request.url).origin; + + if (!youtubeOAuthConfigured()) { + return NextResponse.redirect(new URL('/settings?youtube=unconfigured', appBase)); + } + + const supabase = await createClient(); + const { user } = await getAuthenticatedUser(supabase); + if (!user) { + return NextResponse.redirect(new URL('/login', appBase)); + } + + const state = randomBytes(16).toString('hex'); + const res = NextResponse.redirect(buildAuthUrl(state)); + res.cookies.set('yt_oauth_state', state, { + httpOnly: true, + secure: true, + sameSite: 'lax', + path: '/', + maxAge: 600, + }); + return res; +} diff --git a/apps/web/src/app/api/youtube/callback/route.ts b/apps/web/src/app/api/youtube/callback/route.ts new file mode 100644 index 0000000..70bf4f5 --- /dev/null +++ b/apps/web/src/app/api/youtube/callback/route.ts @@ -0,0 +1,52 @@ +import { NextResponse } from 'next/server'; +import { cookies } from 'next/headers'; +import { createClient, getAuthenticatedUser } from '@/lib/supabase/server'; +import { serviceClient } from '@/lib/supabase/service'; +import { exchangeCodeForTokens } from '@/lib/youtube'; + +/** Google redirects here with ?code. Verify state, store the refresh token. */ +export async function GET(request: Request) { + const url = new URL(request.url); + const appBase = process.env.NEXT_PUBLIC_APP_URL ?? url.origin; + const back = (status: string) => + NextResponse.redirect(new URL(`/settings?youtube=${status}`, appBase)); + + const code = url.searchParams.get('code'); + const state = url.searchParams.get('state'); + if (url.searchParams.get('error') || !code) return back('error'); + + const cookieStore = await cookies(); + const expectedState = cookieStore.get('yt_oauth_state')?.value; + if (!state || !expectedState || state !== expectedState) return back('error'); + + const supabase = await createClient(); + const { user } = await getAuthenticatedUser(supabase); + if (!user) return NextResponse.redirect(new URL('/login', appBase)); + + try { + const tokens = await exchangeCodeForTokens(code); + if (!tokens.refresh_token) { + // Google only returns a refresh token on first consent. Re-consent forces + // it (prompt=consent), but guard anyway. + return back('noref'); + } + + // youtube_credentials isn't in the generated Database types yet (migration + // ships in this PR), so the typed client infers `never` for the payload. + await serviceClient() + .from('youtube_credentials') + .upsert({ + user_id: user.id, + refresh_token: tokens.refresh_token, + scope: tokens.scope ?? null, + updated_at: new Date().toISOString(), + } as never); + + const res = back('connected'); + res.cookies.delete('yt_oauth_state'); + return res; + } catch (err) { + console.error('[youtube/callback] failed:', err); + return back('error'); + } +} diff --git a/apps/web/src/app/api/youtube/disconnect/route.ts b/apps/web/src/app/api/youtube/disconnect/route.ts new file mode 100644 index 0000000..458fcbf --- /dev/null +++ b/apps/web/src/app/api/youtube/disconnect/route.ts @@ -0,0 +1,13 @@ +import { createClient, getAuthenticatedUser } from '@/lib/supabase/server'; +import { serviceClient } from '@/lib/supabase/service'; +import { successResponse, errorResponse } from '@/lib/api'; + +/** Remove the current user's stored YouTube credentials. */ +export async function POST() { + const supabase = await createClient(); + const { user } = await getAuthenticatedUser(supabase); + if (!user) return errorResponse('Authentication required', 401); + + await serviceClient().from('youtube_credentials').delete().eq('user_id', user.id); + return successResponse({ disconnected: true }); +} diff --git a/apps/web/src/app/api/youtube/status/route.ts b/apps/web/src/app/api/youtube/status/route.ts new file mode 100644 index 0000000..7227310 --- /dev/null +++ b/apps/web/src/app/api/youtube/status/route.ts @@ -0,0 +1,23 @@ +import { createClient, getAuthenticatedUser } from '@/lib/supabase/server'; +import { serviceClient } from '@/lib/supabase/service'; +import { successResponse, errorResponse } from '@/lib/api'; +import { youtubeOAuthConfigured } from '@/lib/youtube'; + +/** Whether the current user has connected YouTube (token never exposed). */ +export async function GET() { + const supabase = await createClient(); + const { user } = await getAuthenticatedUser(supabase); + if (!user) return errorResponse('Authentication required', 401); + + const { data } = await serviceClient() + .from('youtube_credentials') + .select('user_id, channel_title') + .eq('user_id', user.id) + .maybeSingle(); + + return successResponse({ + configured: youtubeOAuthConfigured(), + connected: Boolean(data), + channelTitle: (data as { channel_title?: string } | null)?.channel_title ?? null, + }); +} diff --git a/apps/web/src/app/settings/settings-content.tsx b/apps/web/src/app/settings/settings-content.tsx index 3423f03..78e5b23 100644 --- a/apps/web/src/app/settings/settings-content.tsx +++ b/apps/web/src/app/settings/settings-content.tsx @@ -6,6 +6,7 @@ import { User, Video, Users, Palette, Info, ArrowLeft, Check, Bell } from 'lucid import { HeaderClient } from '@/components/header-client'; import { Footer } from '@/components/footer'; import { NotificationPreferences } from '@/components/notifications/NotificationPreferences'; +import { YouTubeConnect } from './youtube-connect'; import type { UserData } from '@/components/header'; import type { RecordingQuality } from '@/hooks/useRecording'; import type { CaptureQuality } from '@/hooks/useScreenCapture'; @@ -104,6 +105,9 @@ export function SettingsContent({ user }: { user: UserData | null }) {
+ {/* YouTube auto go-live */} + + {/* Account Section */}
diff --git a/apps/web/src/app/settings/youtube-connect.tsx b/apps/web/src/app/settings/youtube-connect.tsx new file mode 100644 index 0000000..33b1974 --- /dev/null +++ b/apps/web/src/app/settings/youtube-connect.tsx @@ -0,0 +1,96 @@ +'use client'; + +import { useState, useEffect, useCallback } from 'react'; +import { Radio, Check, Loader2 } from 'lucide-react'; + +interface YtStatus { + configured: boolean; + connected: boolean; + channelTitle: string | null; +} + +/** + * Connect a YouTube account so PairUX auto-transitions the broadcast to live + * when you start streaming (instead of it sitting on "Preparing stream"). + */ +export function YouTubeConnect() { + const [status, setStatus] = useState(null); + const [busy, setBusy] = useState(false); + + const load = useCallback(async () => { + try { + const res = await fetch('/api/youtube/status'); + if (res.ok) { + const json = (await res.json()) as { data: YtStatus }; + setStatus(json.data); + } + } catch { + // leave status null — section shows a spinner / nothing actionable + } + }, []); + + useEffect(() => { + void load(); + }, [load]); + + const disconnect = useCallback(async () => { + setBusy(true); + try { + await fetch('/api/youtube/disconnect', { method: 'POST' }); + await load(); + } finally { + setBusy(false); + } + }, [load]); + + return ( +
+
+
+
+ +
+
+

YouTube auto go-live

+

+ Take your YouTube broadcast live automatically when you start streaming — no more + “Preparing stream”. +

+
+
+
+
+ {status === null ? ( + + ) : !status.configured ? ( +

+ Not available yet — the server is missing Google OAuth configuration. +

+ ) : status.connected ? ( +
+ + + Connected{status.channelTitle ? ` — ${status.channelTitle}` : ''} + + +
+ ) : ( + + + Connect YouTube + + )} +
+
+ ); +} diff --git a/apps/web/src/lib/youtube.test.ts b/apps/web/src/lib/youtube.test.ts new file mode 100644 index 0000000..a8ddd7c --- /dev/null +++ b/apps/web/src/lib/youtube.test.ts @@ -0,0 +1,78 @@ +import { describe, it, expect, beforeEach, afterEach, vi } from 'vitest'; +import { + buildAuthUrl, + findTransitionableBroadcasts, + transitionToLive, + exchangeCodeForTokens, + youtubeOAuthConfigured, +} from './youtube'; + +describe('youtube lib', () => { + beforeEach(() => { + process.env.GOOGLE_CLIENT_ID = 'cid'; + process.env.GOOGLE_CLIENT_SECRET = 'secret'; + process.env.NEXT_PUBLIC_APP_URL = 'https://pairux.com'; + }); + afterEach(() => { + vi.restoreAllMocks(); + }); + + it('youtubeOAuthConfigured reflects env presence', () => { + expect(youtubeOAuthConfigured()).toBe(true); + delete process.env.GOOGLE_CLIENT_ID; + expect(youtubeOAuthConfigured()).toBe(false); + }); + + it('buildAuthUrl requests offline access + consent with scope, state and redirect', () => { + const url = new URL(buildAuthUrl('xyz')); + expect(url.searchParams.get('client_id')).toBe('cid'); + expect(url.searchParams.get('access_type')).toBe('offline'); + expect(url.searchParams.get('prompt')).toBe('consent'); + expect(url.searchParams.get('state')).toBe('xyz'); + expect(url.searchParams.get('scope')).toContain('youtube'); + expect(url.searchParams.get('redirect_uri')).toBe('https://pairux.com/api/youtube/callback'); + }); + + it('findTransitionableBroadcasts returns only ready/testing broadcasts', async () => { + global.fetch = vi.fn().mockResolvedValue({ + ok: true, + json: () => + Promise.resolve({ + items: [ + { id: 'a', status: { lifeCycleStatus: 'ready' }, snippet: { title: 'A' } }, + { id: 'b', status: { lifeCycleStatus: 'live' }, snippet: { title: 'B' } }, + { id: 'c', status: { lifeCycleStatus: 'testing' }, snippet: { title: 'C' } }, + { id: 'd', status: { lifeCycleStatus: 'complete' }, snippet: { title: 'D' } }, + ], + }), + }) as unknown as typeof fetch; + + const res = await findTransitionableBroadcasts('tok'); + expect(res.map((b) => b.id)).toEqual(['a', 'c']); + }); + + it('transitionToLive POSTs to the transition endpoint with broadcastStatus=live', async () => { + const fetchMock = vi.fn().mockResolvedValue({ ok: true, text: () => Promise.resolve('') }); + global.fetch = fetchMock as unknown as typeof fetch; + + await transitionToLive('tok', 'bid-1'); + + const [calledUrl, opts] = fetchMock.mock.calls[0] as [string, RequestInit]; + expect(calledUrl).toContain('/liveBroadcasts/transition'); + expect(calledUrl).toContain('broadcastStatus=live'); + expect(calledUrl).toContain('id=bid-1'); + expect(opts.method).toBe('POST'); + expect((opts.headers as Record).Authorization).toBe('Bearer tok'); + }); + + it('exchangeCodeForTokens returns the parsed token response', async () => { + global.fetch = vi.fn().mockResolvedValue({ + ok: true, + json: () => Promise.resolve({ access_token: 'at', refresh_token: 'rt', expires_in: 3600 }), + }) as unknown as typeof fetch; + + const t = await exchangeCodeForTokens('code'); + expect(t.refresh_token).toBe('rt'); + expect(t.access_token).toBe('at'); + }); +}); diff --git a/apps/web/src/lib/youtube.ts b/apps/web/src/lib/youtube.ts new file mode 100644 index 0000000..869fdc4 --- /dev/null +++ b/apps/web/src/lib/youtube.ts @@ -0,0 +1,132 @@ +/** + * YouTube Live API helpers — OAuth + auto-transition a stuck "Preparing" + * broadcast to live. + * + * The egress pushes a clean continuous RTMP stream to YouTube, but YouTube + * leaves the broadcast in `ready`/`testing` ("Preparing stream") until someone + * clicks "Go Live" (unless Auto-start is on). With the user's YouTube OAuth we + * can detect that state and call liveBroadcasts.transition → live automatically. + * + * Requires env: GOOGLE_CLIENT_ID, GOOGLE_CLIENT_SECRET. Redirect URI is derived + * from NEXT_PUBLIC_APP_URL (or APP_URL), defaulting to https://pairux.com. + */ + +const GOOGLE_AUTH_URL = 'https://accounts.google.com/o/oauth2/v2/auth'; +const GOOGLE_TOKEN_URL = 'https://oauth2.googleapis.com/token'; +const YT_API = 'https://www.googleapis.com/youtube/v3'; + +// Manage the user's live broadcasts (list + transition). +export const YOUTUBE_SCOPE = 'https://www.googleapis.com/auth/youtube'; + +export function youtubeOAuthConfigured(): boolean { + return Boolean(process.env.GOOGLE_CLIENT_ID && process.env.GOOGLE_CLIENT_SECRET); +} + +export function getRedirectUri(): string { + const base = process.env.NEXT_PUBLIC_APP_URL ?? process.env.APP_URL ?? 'https://pairux.com'; + return `${base.replace(/\/$/, '')}/api/youtube/callback`; +} + +/** Build the Google consent URL. `state` ties the callback back to the user. */ +export function buildAuthUrl(state: string): string { + const params = new URLSearchParams({ + client_id: process.env.GOOGLE_CLIENT_ID ?? '', + redirect_uri: getRedirectUri(), + response_type: 'code', + scope: YOUTUBE_SCOPE, + access_type: 'offline', // get a refresh token + prompt: 'consent', // force refresh token on re-consent + include_granted_scopes: 'true', + state, + }); + return `${GOOGLE_AUTH_URL}?${params.toString()}`; +} + +interface TokenResponse { + access_token: string; + refresh_token?: string; + expires_in: number; + scope?: string; +} + +/** Exchange an authorization code for tokens (includes a refresh_token). */ +export async function exchangeCodeForTokens(code: string): Promise { + const res = await fetch(GOOGLE_TOKEN_URL, { + method: 'POST', + headers: { 'Content-Type': 'application/x-www-form-urlencoded' }, + body: new URLSearchParams({ + code, + client_id: process.env.GOOGLE_CLIENT_ID ?? '', + client_secret: process.env.GOOGLE_CLIENT_SECRET ?? '', + redirect_uri: getRedirectUri(), + grant_type: 'authorization_code', + }), + }); + if (!res.ok) { + throw new Error(`Google token exchange failed: ${String(res.status)} ${await res.text()}`); + } + return (await res.json()) as TokenResponse; +} + +/** Trade a stored refresh token for a fresh access token. */ +export async function refreshAccessToken(refreshToken: string): Promise { + const res = await fetch(GOOGLE_TOKEN_URL, { + method: 'POST', + headers: { 'Content-Type': 'application/x-www-form-urlencoded' }, + body: new URLSearchParams({ + refresh_token: refreshToken, + client_id: process.env.GOOGLE_CLIENT_ID ?? '', + client_secret: process.env.GOOGLE_CLIENT_SECRET ?? '', + grant_type: 'refresh_token', + }), + }); + if (!res.ok) { + throw new Error(`Google token refresh failed: ${String(res.status)} ${await res.text()}`); + } + const json = (await res.json()) as TokenResponse; + return json.access_token; +} + +export interface LiveBroadcast { + id: string; + lifeCycleStatus: string; // created | ready | testing | live | complete | revoked + title: string; +} + +/** + * List the user's broadcasts that are waiting to go live (ready/testing). + * These are the ones stuck on "Preparing". + */ +export async function findTransitionableBroadcasts(accessToken: string): Promise { + const url = + `${YT_API}/liveBroadcasts?part=id,snippet,status` + + `&broadcastStatus=upcoming&maxResults=10&broadcastType=all`; + const res = await fetch(url, { headers: { Authorization: `Bearer ${accessToken}` } }); + if (!res.ok) { + throw new Error(`liveBroadcasts.list failed: ${String(res.status)} ${await res.text()}`); + } + const json = (await res.json()) as { + items?: { id: string; status?: { lifeCycleStatus?: string }; snippet?: { title?: string } }[]; + }; + return (json.items ?? []) + .map((it) => ({ + id: it.id, + lifeCycleStatus: it.status?.lifeCycleStatus ?? '', + title: it.snippet?.title ?? '', + })) + .filter((b) => b.lifeCycleStatus === 'ready' || b.lifeCycleStatus === 'testing'); +} + +/** Transition a broadcast to live. Idempotent-ish: YouTube 4xxs if not ready. */ +export async function transitionToLive(accessToken: string, broadcastId: string): Promise { + const url = + `${YT_API}/liveBroadcasts/transition` + + `?part=status&broadcastStatus=live&id=${encodeURIComponent(broadcastId)}`; + const res = await fetch(url, { + method: 'POST', + headers: { Authorization: `Bearer ${accessToken}` }, + }); + if (!res.ok) { + throw new Error(`liveBroadcasts.transition failed: ${String(res.status)} ${await res.text()}`); + } +} diff --git a/apps/web/src/lib/youtubeAutoLive.test.ts b/apps/web/src/lib/youtubeAutoLive.test.ts new file mode 100644 index 0000000..e2044af --- /dev/null +++ b/apps/web/src/lib/youtubeAutoLive.test.ts @@ -0,0 +1,55 @@ +import { describe, it, expect, vi, beforeEach } from 'vitest'; + +const mockMaybeSingle = vi.fn(); +vi.mock('@/lib/supabase/service', () => ({ + serviceClient: () => ({ + from: () => ({ select: () => ({ eq: () => ({ maybeSingle: mockMaybeSingle }) }) }), + }), +})); + +const mockRefresh = vi.fn(); +const mockFind = vi.fn(); +const mockTransition = vi.fn(); +vi.mock('@/lib/youtube', () => ({ + youtubeOAuthConfigured: () => true, + refreshAccessToken: (...a: unknown[]) => mockRefresh(...a), + findTransitionableBroadcasts: (...a: unknown[]) => mockFind(...a), + transitionToLive: (...a: unknown[]) => mockTransition(...a), +})); + +import { autoTransitionYouTube } from './youtubeAutoLive'; + +describe('autoTransitionYouTube', () => { + beforeEach(() => { + vi.clearAllMocks(); + }); + + it('transitions the first ready broadcast to live', async () => { + mockMaybeSingle.mockResolvedValue({ data: { refresh_token: 'rt' } }); + mockRefresh.mockResolvedValue('access-tok'); + mockFind.mockResolvedValue([{ id: 'b1', lifeCycleStatus: 'ready', title: 'X' }]); + mockTransition.mockResolvedValue(undefined); + + await autoTransitionYouTube('user-1'); + + expect(mockRefresh).toHaveBeenCalledWith('rt'); + expect(mockTransition).toHaveBeenCalledWith('access-tok', 'b1'); + }); + + it('does nothing when the user has not connected YouTube', async () => { + mockMaybeSingle.mockResolvedValue({ data: null }); + + await autoTransitionYouTube('user-1'); + + expect(mockRefresh).not.toHaveBeenCalled(); + expect(mockTransition).not.toHaveBeenCalled(); + }); + + it('never throws when the YouTube API fails', async () => { + mockMaybeSingle.mockResolvedValue({ data: { refresh_token: 'rt' } }); + mockRefresh.mockRejectedValue(new Error('google down')); + + await expect(autoTransitionYouTube('user-1')).resolves.toBeUndefined(); + expect(mockTransition).not.toHaveBeenCalled(); + }); +}); diff --git a/apps/web/src/lib/youtubeAutoLive.ts b/apps/web/src/lib/youtubeAutoLive.ts new file mode 100644 index 0000000..832476c --- /dev/null +++ b/apps/web/src/lib/youtubeAutoLive.ts @@ -0,0 +1,53 @@ +import { serviceClient } from '@/lib/supabase/service'; +import { + refreshAccessToken, + findTransitionableBroadcasts, + transitionToLive, + youtubeOAuthConfigured, +} from '@/lib/youtube'; + +/** How long/often to poll for a broadcast that's ready to go live. */ +const POLL_INTERVAL_MS = 6_000; +const MAX_ATTEMPTS = 15; // ~90s — enough for YouTube to ingest + reach "ready" + +const sleep = (ms: number) => new Promise((resolve) => setTimeout(resolve, ms)); + +/** Fetch a user's stored YouTube refresh token, or null if not connected. */ +async function getRefreshToken(userId: string): Promise { + const { data } = await serviceClient() + .from('youtube_credentials') + .select('refresh_token') + .eq('user_id', userId) + .maybeSingle(); + const token = (data as { refresh_token?: string } | null)?.refresh_token; + return token ?? null; +} + +/** + * Best-effort: after egress starts pushing to YouTube, poll the user's + * broadcasts and transition the first ready/testing one to live — so it never + * sits on "Preparing stream". Fire-and-forget; never throws. + */ +export async function autoTransitionYouTube(userId: string): Promise { + try { + if (!youtubeOAuthConfigured()) return; + + const refreshToken = await getRefreshToken(userId); + if (!refreshToken) return; // user hasn't connected YouTube + + const accessToken = await refreshAccessToken(refreshToken); + + for (let attempt = 0; attempt < MAX_ATTEMPTS; attempt++) { + const [first] = await findTransitionableBroadcasts(accessToken); + if (first) { + await transitionToLive(accessToken, first.id); + console.log(`[youtube] transitioned broadcast ${first.id} to live for user ${userId}`); + return; + } + await sleep(POLL_INTERVAL_MS); + } + console.warn(`[youtube] no ready broadcast to transition for user ${userId} after polling`); + } catch (err) { + console.warn('[youtube] auto-transition failed:', err); + } +} diff --git a/supabase/migrations/20260623000001_youtube_credentials.sql b/supabase/migrations/20260623000001_youtube_credentials.sql new file mode 100644 index 0000000..9f96759 --- /dev/null +++ b/supabase/migrations/20260623000001_youtube_credentials.sql @@ -0,0 +1,20 @@ +-- YouTube OAuth credentials (per user) +-- Stores the user's YouTube refresh token so the server can auto-transition a +-- "Preparing" broadcast to live via the YouTube Live API. +-- +-- The refresh_token is sensitive and SERVER-ONLY. RLS is enabled with NO +-- policies, so PostgREST/anon/authenticated clients cannot read it — only the +-- service-role key (which bypasses RLS) can, from server routes. Connected +-- status is surfaced to the client via a server route, never the token itself. + +CREATE TABLE IF NOT EXISTS public.youtube_credentials ( + user_id UUID PRIMARY KEY REFERENCES auth.users(id) ON DELETE CASCADE, + refresh_token TEXT NOT NULL, + channel_title TEXT, + scope TEXT, + created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(), + updated_at TIMESTAMPTZ NOT NULL DEFAULT NOW() +); + +-- Enable RLS with no policies => deny all client access; service role bypasses. +ALTER TABLE public.youtube_credentials ENABLE ROW LEVEL SECURITY;