diff --git a/apps/web/src/app/api/v1/agent/broadcasts/[sessionId]/route.ts b/apps/web/src/app/api/v1/agent/broadcasts/[sessionId]/route.ts new file mode 100644 index 00000000..41762846 --- /dev/null +++ b/apps/web/src/app/api/v1/agent/broadcasts/[sessionId]/route.ts @@ -0,0 +1,44 @@ +import { z } from 'zod'; +import { errorResponse, handleApiError, successResponse } from '@/lib/api'; +import { AccountAgents } from '@/lib/account-agents'; +import { BroadcastError, endBroadcast, heartbeatBroadcast } from '@/lib/agent-broadcast'; +import { serviceClient } from '@/lib/supabase/service'; + +export const runtime = 'nodejs'; +export const dynamic = 'force-dynamic'; + +interface RouteParams { + params: Promise<{ sessionId: string }>; +} + +const idSchema = z.guid('Invalid session ID'); + +/** POST /api/v1/agent/broadcasts/:id — heartbeat; { live } says whether it still is. */ +export async function POST(request: Request, { params }: RouteParams) { + try { + const agent = await new AccountAgents().fromRequest(request); + if (!agent) return errorResponse('Send Authorization: Bearer .', 401); + const sessionId = idSchema.parse((await params).sessionId); + return successResponse(await heartbeatBroadcast(serviceClient(), agent, sessionId)); + } catch (error) { + if (error instanceof BroadcastError) return errorResponse(error.message, error.status); + return handleApiError(error); + } +} + +/** DELETE /api/v1/agent/broadcasts/:id — the broadcast is over. */ +export async function DELETE(request: Request, { params }: RouteParams) { + try { + const agent = await new AccountAgents().fromRequest(request); + if (!agent) return errorResponse('Send Authorization: Bearer .', 401); + const sessionId = idSchema.parse((await params).sessionId); + const result = await endBroadcast(serviceClient(), agent, sessionId); + void import('@/lib/channel-restream').then(({ stopChannelRestream }) => + stopChannelRestream(sessionId) + ); + return successResponse(result); + } catch (error) { + if (error instanceof BroadcastError) return errorResponse(error.message, error.status); + return handleApiError(error); + } +} diff --git a/apps/web/src/app/api/v1/agent/broadcasts/route.ts b/apps/web/src/app/api/v1/agent/broadcasts/route.ts new file mode 100644 index 00000000..33503cb3 --- /dev/null +++ b/apps/web/src/app/api/v1/agent/broadcasts/route.ts @@ -0,0 +1,63 @@ +import { errorResponse, handleApiError, successResponse } from '@/lib/api'; +import { AccountAgents } from '@/lib/account-agents'; +import { BroadcastError, startBroadcast, startBroadcastSchema } from '@/lib/agent-broadcast'; +import { serviceClient } from '@/lib/supabase/service'; +import { getIceServers } from '@/lib/ice-servers'; +import { FixedWindowRateLimiter } from '@/lib/rate-limit'; + +export const runtime = 'nodejs'; +export const dynamic = 'force-dynamic'; + +const startsByAgent = new FixedWindowRateLimiter(10, 60 * 60_000); + +/** + * POST /api/v1/agent/broadcasts — an agent (Bearer pux_ag_...) goes live on + * one of its owner's channels: { channel: "moshcoding", title, description? }. + * Answers the room, its public page and a publish token for it. + */ +export async function POST(request: Request) { + try { + const agent = await new AccountAgents().fromRequest(request); + if (!agent) return errorResponse('Send Authorization: Bearer .', 401); + const limit = startsByAgent.check(agent.id); + if (!limit.success) { + return errorResponse( + `Too many broadcasts. Try again in ${String(limit.retryAfterSeconds)} seconds.`, + 429 + ); + } + const apiKey = process.env.LIVEKIT_API_KEY; + const apiSecret = process.env.LIVEKIT_API_SECRET; + const url = process.env.NEXT_PUBLIC_LIVEKIT_URL; + if (!apiKey || !apiSecret || !url) return errorResponse('LiveKit not configured', 503); + + const input = startBroadcastSchema.parse(await request.json().catch(() => ({}))); + const db = serviceClient(); + const { broadcast, channel, subject } = await startBroadcast(db, agent, input, { + apiKey, + apiSecret, + url, + }); + + // The same follow-ups a person's go-live gets: subscribers hear about it + // and the channel's restream destinations pick it up. + void import('@/lib/notify-live').then(({ notifyGoLive }) => + notifyGoLive({ + creator_id: agent.ownerUserId, + subject, + join_code: broadcast.joinCode, + channel_id: channel.id, + channel_handle: channel.handle, + channel_name: channel.name, + }) + ); + void import('@/lib/channel-restream').then(({ startChannelRestream }) => + startChannelRestream(broadcast.sessionId, channel.id) + ); + + return successResponse({ ...broadcast, iceServers: await getIceServers() }, 201); + } catch (error) { + if (error instanceof BroadcastError) return errorResponse(error.message, error.status); + return handleApiError(error); + } +} diff --git a/apps/web/src/app/docs/agents/page.tsx b/apps/web/src/app/docs/agents/page.tsx index 2447ff5d..ac160b77 100644 --- a/apps/web/src/app/docs/agents/page.tsx +++ b/apps/web/src/app/docs/agents/page.tsx @@ -58,6 +58,14 @@ const api: { route: string; what: string }[] = [ what: 'Mint a 24-hour LiveKit token scoped to this session and microphone audio. The session must use SFU mode.', }, { route: 'DELETE /api/v1/agents/:participantId', what: 'Leave the session.' }, + { + route: 'POST /api/v1/agent/broadcasts', + what: 'Account agents (Bearer pux_ag_ key): { channel, title, description? } goes live on a channel the agent owner may use and returns { sessionId, liveUrl, url, token } to publish audio into. Used to stream podcasts as they are recorded.', + }, + { + route: 'POST /api/v1/agent/broadcasts/:sessionId', + what: 'Heartbeat while publishing (a host unseen for 90 seconds drops off the listings). DELETE ends the broadcast.', + }, ]; export default function AgentDocsPage() { @@ -173,9 +181,9 @@ pairux voice join --json`}