feat(channels): channel-level RTMP restreaming (configure once, auto on go-live)

Per-channel external restream destinations (YouTube/Twitch/etc.) + a master
'External restreaming' toggle in the dashboard. Keys stored service-only under
RLS (repo's youtube_credentials pattern) — the client never sends/receives a
key; the server builds rtmp_url/key at go-live. When a channel with
restream_enabled + enabled destinations goes public, the visibility route
auto-starts a LiveKit egress fanning out to them; stopped on host-leave.

- migration: channel_restream_destinations + channels.restream_enabled +
  sessions.restream_egress_id + owner RPCs; list_my_channels returns the flag.
- lib/channel-restream.ts start/stop; wired into visibility + host_leave.
- /api/channels/[id]/restream CRUD + RestreamManager dashboard panel.

Desktop global toggle to follow as its own release (per-channel switch already
makes this work end-to-end).

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

Author: ralyodio

apps/web/src/app/api/channels/[channelId]/restream/[destId]/route.ts

@@ -0,0 +1,29 @@
+import { createClient, getAuthenticatedUser } from '@/lib/supabase/server';
+import { successResponse, errorResponse, handleApiError } from '@/lib/api';
+
+interface RouteParams {
+  params: Promise<{ channelId: string; destId: string }>;
+}
+
+// DELETE — remove a restream destination.
+export async function DELETE(_request: Request, { params }: RouteParams) {
+  try {
+    const { channelId, destId } = await params;
+    const supabase = await createClient();
+    const { user, error: authError } = await getAuthenticatedUser(supabase);
+    if (authError || !user) return errorResponse('Authentication required', 401);
+
+    // eslint-disable-next-line @typescript-eslint/no-unsafe-assignment, @typescript-eslint/no-unsafe-call, @typescript-eslint/no-explicit-any
+    const { error } = await (supabase.rpc as any)('delete_channel_restream_destination', {
+      p_channel_id: channelId,
+      p_id: destId,
+    });
+    if (error) {
+      // eslint-disable-next-line @typescript-eslint/no-unsafe-argument, @typescript-eslint/no-unsafe-member-access
+      return errorResponse(error.message, 400);
+    }
+    return successResponse({ ok: true });
+  } catch (error) {
+    return handleApiError(error);
+  }
+}

apps/web/src/app/api/channels/[channelId]/restream/route.ts

@@ -0,0 +1,85 @@
+/* eslint-disable @typescript-eslint/no-unsafe-assignment, @typescript-eslint/no-unsafe-member-access, @typescript-eslint/no-unsafe-call, @typescript-eslint/no-unsafe-argument, @typescript-eslint/no-explicit-any */
+import { createClient, getAuthenticatedUser } from '@/lib/supabase/server';
+import { successResponse, errorResponse, handleApiError } from '@/lib/api';
+
+interface RouteParams {
+  params: Promise<{ channelId: string }>;
+}
+
+interface UpsertBody {
+  id?: string;
+  platform?: string;
+  label?: string;
+  rtmpUrl?: string;
+  streamKey?: string;
+  enabled?: boolean;
+}
+
+// GET — list a channel's restream destinations (never returns stream keys).
+export async function GET(_request: Request, { params }: RouteParams) {
+  try {
+    const { channelId } = await params;
+    const supabase = await createClient();
+    const { user, error: authError } = await getAuthenticatedUser(supabase);
+    if (authError || !user) return errorResponse('Authentication required', 401);
+
+    const { data, error } = await (supabase.rpc as any)('list_channel_restream_destinations', {
+      p_channel_id: channelId,
+    });
+    if (error) {
+      return errorResponse(error.message, 400);
+    }
+    return successResponse({ destinations: data });
+  } catch (error) {
+    return handleApiError(error);
+  }
+}
+
+// POST — add (no id) or update a destination.
+export async function POST(request: Request, { params }: RouteParams) {
+  try {
+    const { channelId } = await params;
+    const body = (await request.json().catch(() => ({}))) as UpsertBody;
+    const supabase = await createClient();
+    const { user, error: authError } = await getAuthenticatedUser(supabase);
+    if (authError || !user) return errorResponse('Authentication required', 401);
+
+    const { data, error } = await (supabase.rpc as any)('upsert_channel_restream_destination', {
+      p_channel_id: channelId,
+      p_id: body.id ?? null,
+      p_platform: body.platform ?? 'custom',
+      p_label: body.label ?? null,
+      p_rtmp_url: body.rtmpUrl ?? '',
+      p_stream_key: body.streamKey ?? null,
+      p_enabled: body.enabled ?? true,
+    });
+    if (error) {
+      return errorResponse(error.message, 400);
+    }
+    return successResponse({ id: data });
+  } catch (error) {
+    return handleApiError(error);
+  }
+}
+
+// PATCH — toggle the channel's restream master switch. Body: { enabled }.
+export async function PATCH(request: Request, { params }: RouteParams) {
+  try {
+    const { channelId } = await params;
+    const body = (await request.json().catch(() => ({}))) as { enabled?: boolean };
+    const supabase = await createClient();
+    const { user, error: authError } = await getAuthenticatedUser(supabase);
+    if (authError || !user) return errorResponse('Authentication required', 401);
+
+    const { error } = await (supabase.rpc as any)('set_channel_restream_enabled', {
+      p_channel_id: channelId,
+      p_enabled: Boolean(body.enabled),
+    });
+    if (error) {
+      return errorResponse(error.message, 400);
+    }
+    return successResponse({ enabled: Boolean(body.enabled) });
+  } catch (error) {
+    return handleApiError(error);
+  }
+}

apps/web/src/app/api/sessions/[sessionId]/route.ts

@@ -57,11 +57,14 @@ export async function DELETE(_request: Request, { params }: RouteParams) {
       return errorResponse(error.message, 400);
     }
 
-    // Host left → stop any server-side recording for this session (the egress
-    // also self-stops when the room empties, so this is best-effort).
+    // Host left → stop any server-side recording + external restream for this
+    // session (the egress also self-stops when the room empties; best-effort).
     void import('@/lib/livekit-recording').then(({ stopSessionRecording }) =>
       stopSessionRecording(sessionId)
     );
+    void import('@/lib/channel-restream').then(({ stopChannelRestream }) =>
+      stopChannelRestream(sessionId)
+    );
 
     return successResponse(data);
   } catch (error) {

apps/web/src/app/api/sessions/[sessionId]/visibility/route.ts

@@ -52,6 +52,26 @@ export async function PATCH(request: Request, { params }: RouteParams) {
       if (live?.creator_id) {
         void import('@/lib/notify-live').then(({ notifyGoLive }) => notifyGoLive(live));
       }
+
+      // Auto-restream to external RTMP if this live's channel has it enabled.
+      // Resolve the channel from the flip, else query (flip is null if the
+      // went-live moment already fired via the heartbeat path).
+      let channelId = live?.channel_id ?? null;
+      if (!channelId) {
+        // eslint-disable-next-line @typescript-eslint/no-unsafe-assignment, @typescript-eslint/no-explicit-any
+        const { data: s } = (await supabase
+          .from('sessions')
+          .select('channel_id')
+          .eq('id', sessionId)
+          .single()) as { data: { channel_id: string | null } | null };
+        channelId = s?.channel_id ?? null;
+      }
+      if (channelId) {
+        const cid = channelId;
+        void import('@/lib/channel-restream').then(({ startChannelRestream }) =>
+          startChannelRestream(sessionId, cid)
+        );
+      }
     }
 
     return successResponse(data);

apps/web/src/app/dashboard/ChannelsManager.tsx

@@ -13,6 +13,7 @@ import {
   Sparkles,
 } from 'lucide-react';
 import type { MyChannel } from '@pairux/shared-types';
+import { RestreamManager } from './RestreamManager';
 
 // Where OBS / any RTMP client points. The stream key selects the channel.
 const RTMP_INGEST_URL = 'rtmp://rtmp.pairux.com/live';
@@ -430,6 +431,10 @@ export function ChannelsManager() {
                     </button>
                   </div>
                 </div>
+
+                <div className="mt-4">
+                  <RestreamManager channelId={ch.id} initialEnabled={ch.restream_enabled} />
+                </div>
               </div>
             </div>
           );