feat(api): agent broadcasts - an account agent goes live on a channel

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 <pux_ag_ agent key>.', 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 <pux_ag_ agent key>.', 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);
+  }
+}

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 <pux_ag_ agent key>.', 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);
+  }
+}

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`}</code>
               <MessageSquare className="h-6 w-6" aria-hidden="true" /> HTTP API
             </h2>
             <p className="mt-4 text-gray-600">
-              The CLI is a thin wrapper over five routes. Responses are <code>{'{ data }'}</code> or{' '}
-              <code>{'{ error }'}</code>. The participant id returned by join is the credential for
-              the other agent routes, so keep it private.
+              The CLI is a thin wrapper over the first five routes. Responses are{' '}
+              <code>{'{ data }'}</code> or <code>{'{ error }'}</code>. The participant id returned
+              by join is the credential for the other agent routes, so keep it private.
             </p>
             <div className="mt-6 overflow-hidden rounded-xl border border-gray-200">
               {api.map((a) => (

apps/web/src/lib/agent-broadcast.test.ts

@@ -0,0 +1,174 @@
+import { describe, expect, it, vi } from 'vitest';
+import {
+  BroadcastError,
+  endBroadcast,
+  heartbeatBroadcast,
+  publisherGrant,
+  startBroadcast,
+  startBroadcastSchema,
+} from './agent-broadcast';
+import type { Db } from '@/lib/orgs';
+
+const tokenCtor = vi.fn();
+const tokenGrant = vi.fn();
+vi.mock('livekit-server-sdk', () => ({
+  TrackSource: { MICROPHONE: 2, CAMERA: 1 },
+  AccessToken: vi.fn().mockImplementation((...args: unknown[]) => {
+    tokenCtor(...args);
+    return { addGrant: tokenGrant, toJwt: () => Promise.resolve('jwt') };
+  }),
+}));
+vi.mock('@/lib/orgs', () => ({ resolveUserPlan: () => Promise.resolve('free') }));
+
+const OWNER = '11111111-1111-4111-8111-111111111111';
+const AGENT = { id: '22222222-2222-4222-8222-222222222222', name: 'riotcoder', ownerUserId: OWNER };
+const CHANNEL = {
+  id: '33333333-3333-4333-8333-333333333333',
+  handle: 'moshcoding',
+  name: 'Moshcoding',
+  owner_id: OWNER,
+  org_id: null,
+  team_id: null,
+};
+const LIVEKIT = {
+  apiKey: 'APIkey',
+  apiSecret: 'a-secret-long-enough-for-hs256-signing!!',
+  url: 'wss://sfu.test',
+};
+
+/** Just enough of the supabase builder for the calls the lib makes. */
+function fakeDb(opts: { channel?: unknown; allowed?: boolean; session?: unknown }) {
+  const calls: { table: string; op: string; payload?: unknown; filters: unknown[] }[] = [];
+  const builder = (table: string) => {
+    const call = { table, op: 'select', payload: undefined as unknown, filters: [] as unknown[] };
+    calls.push(call);
+    const b: Record<string, unknown> = {};
+    const chain =
+      (name: string) =>
+      (...args: unknown[]) => {
+        call.filters.push([name, ...args]);
+        return b;
+      };
+    for (const name of ['select', 'eq', 'ilike', 'not']) b[name] = chain(name);
+    b.insert = (payload: unknown) => {
+      call.op = 'insert';
+      call.payload = payload;
+      return b;
+    };
+    b.update = (payload: unknown) => {
+      call.op = 'update';
+      call.payload = payload;
+      return b;
+    };
+    const answer = () => {
+      if (table === 'channels') return { data: opts.channel ?? null };
+      if (call.op === 'insert')
+        return {
+          data: { id: '44444444-4444-4444-8444-444444444444', join_code: 'ABC234' },
+          error: null,
+        };
+      if (call.op === 'update') return { data: [{ id: 'x' }] };
+      return { data: opts.session ?? null };
+    };
+    b.maybeSingle = () => Promise.resolve(answer());
+    b.single = () => Promise.resolve(answer());
+    b.then = (resolve: (value: unknown) => void) => resolve(answer());
+    return b;
+  };
+  const db = { from: builder, rpc: () => Promise.resolve({ data: opts.allowed ?? false }) };
+  return { db: db as unknown as Db, calls };
+}
+
+describe('startBroadcastSchema', () => {
+  it('takes a handle with or without @', () => {
+    expect(startBroadcastSchema.parse({ channel: '@MoshCoding', title: 'Ep 1' }).channel).toBe(
+      'moshcoding'
+    );
+    expect(() => startBroadcastSchema.parse({ channel: 'bad handle', title: 'x' })).toThrow();
+    expect(() => startBroadcastSchema.parse({ channel: 'ok', title: '' })).toThrow();
+  });
+});
+
+describe('publisherGrant', () => {
+  it('publishes into one room and subscribes to nothing', () => {
+    const grant = publisherGrant('abc');
+    expect(grant.room).toBe('session-abc');
+    expect(grant.canSubscribe).toBe(false);
+    expect(grant.canPublishData).toBe(false);
+  });
+});
+
+describe('startBroadcast', () => {
+  it('opens a public live room on the channel with the owner as host', async () => {
+    const { db, calls } = fakeDb({ channel: CHANNEL, allowed: true });
+    const { broadcast } = await startBroadcast(
+      db,
+      AGENT,
+      { channel: 'moshcoding', title: 'TUI Weekly' },
+      LIVEKIT
+    );
+    const insert = calls.find((c) => c.op === 'insert');
+    expect(insert?.payload).toMatchObject({
+      channel_id: CHANNEL.id,
+      current_host_id: OWNER,
+      host_user_id: OWNER,
+      is_public: true,
+      mode: 'sfu',
+      subject: 'TUI Weekly',
+      settings: { broadcast: { agentId: AGENT.id } },
+    });
+    expect(broadcast.joinCode).toBe('ABC234');
+    expect(broadcast.liveUrl).toMatch(/\/l\/ABC234$/);
+    expect(broadcast.channelUrl).toMatch(/\/@moshcoding$/);
+    expect(broadcast.token).toBe('jwt');
+    // The host identity is what lets the webhook end the live when the publisher leaves.
+    expect(tokenCtor.mock.calls.at(-1)?.[2]).toMatchObject({ identity: OWNER });
+    expect(tokenGrant.mock.calls.at(-1)?.[0]).toMatchObject({
+      room: `session-${broadcast.sessionId}`,
+      canSubscribe: false,
+    });
+  });
+
+  it('refuses a channel the owner may not use, and an unknown one', async () => {
+    await expect(
+      startBroadcast(
+        fakeDb({ channel: CHANNEL, allowed: false }).db,
+        AGENT,
+        { channel: 'moshcoding', title: 'x' },
+        LIVEKIT
+      )
+    ).rejects.toMatchObject({ status: 403 });
+    await expect(
+      startBroadcast(fakeDb({}).db, AGENT, { channel: 'nope', title: 'x' }, LIVEKIT)
+    ).rejects.toBeInstanceOf(BroadcastError);
+  });
+});
+
+describe('heartbeat + end', () => {
+  const mine = { id: 's', status: 'active', settings: { broadcast: { agentId: AGENT.id } } };
+
+  it('only the agent that started it may ping or end it', async () => {
+    const other = { ...mine, settings: { broadcast: { agentId: 'someone-else' } } };
+    await expect(
+      heartbeatBroadcast(fakeDb({ session: other }).db, AGENT, 's')
+    ).rejects.toMatchObject({ status: 404 });
+    await expect(endBroadcast(fakeDb({ session: null }).db, AGENT, 's')).rejects.toMatchObject({
+      status: 404,
+    });
+  });
+
+  it('pings a live one and ends it', async () => {
+    expect(await heartbeatBroadcast(fakeDb({ session: mine }).db, AGENT, 's')).toEqual({
+      live: true,
+    });
+    expect(
+      await heartbeatBroadcast(fakeDb({ session: { ...mine, status: 'ended' } }).db, AGENT, 's')
+    ).toEqual({ live: false });
+    const { db, calls } = fakeDb({ session: mine });
+    await endBroadcast(db, AGENT, 's');
+    expect(calls.find((c) => c.op === 'update')?.payload).toMatchObject({
+      status: 'ended',
+      current_host_id: null,
+    });
+  });
+});