livestream

What this skill does

# Livestream (LiveKit Egress + MUX)

Livestreaming from a [LiveKit](https://livekit.io) room to external platforms via RTMP egress. Creates a MUX live stream, starts an RTMP egress from the LiveKit room to the MUX ingest URL, and auto-archives to a MUX VOD asset when the stream ends. Also supports custom RTMP destinations like YouTube Live and Twitch.

## Prerequisites

- Next.js app with `src/` directory and App Router
- `video-room` skill applied (provides `livekit-server-sdk`, LiveKit server configuration)
- `stream-mux` skill applied (provides `@mux/mux-node` client, MUX API credentials)
- `recording` skill applied (provides the `recordings` schema and Egress patterns)
- `env-config` skill applied (provides Zod env validation)

## Installation

No additional packages required. Uses `livekit-server-sdk` from the `video-room` skill and `@mux/mux-node` from the `stream-mux` skill.

## Environment Variables

Uses existing environment variables from dependencies:

```env
# LiveKit (from video-room)
LIVEKIT_API_KEY=your_api_key
LIVEKIT_API_SECRET=your_api_secret
NEXT_PUBLIC_LIVEKIT_URL=wss://your-project.livekit.cloud

# MUX (from stream-mux)
MUX_TOKEN_ID=your_mux_token_id
MUX_TOKEN_SECRET=your_mux_token_secret
```

Add to `src/env.ts` server schema (if not already present):

```typescript
MUX_TOKEN_ID: z.string().min(1),
MUX_TOKEN_SECRET: z.string().min(1),
```

## What Gets Created

```
src/
├── lib/
│   ├── video/
│   │   ├── livestream.ts              # startLivestream, stopLivestream, getLivestreamStatus
│   │   └── stream-destinations.ts     # StreamDestination type, RTMP URL helpers
│   └── db/
│       └── schema/
│           └── livestreams.ts         # Drizzle schema: livestreams table
└── app/
    └── api/
        └── video/
            └── livestream/
                ├── route.ts           # POST go live, DELETE stop stream
                └── status/
                    └── route.ts       # GET current stream status + viewer count
```

## Setup Steps

### Step 1: Create `src/lib/video/stream-destinations.ts`

```typescript
export type StreamDestinationType = "mux" | "youtube" | "twitch" | "custom";

export type StreamDestination = {
  type: StreamDestinationType;
  name: string;
  rtmpUrl: string;
  streamKey: string;
};

/**
 * Constructs the full RTMP URL for a MUX live stream.
 * MUX RTMP ingest endpoint: rtmp://global-live.mux.com:5222/app/{stream_key}
 */
export function buildMuxRtmpUrl(streamKey: string): string {
  return `rtmp://global-live.mux.com:5222/app/${streamKey}`;
}

/**
 * Constructs the full RTMP URL for YouTube Live.
 * YouTube RTMP ingest: rtmp://a.rtmp.youtube.com/live2/{stream_key}
 */
export function buildYouTubeRtmpUrl(streamKey: string): string {
  return `rtmp://a.rtmp.youtube.com/live2/${streamKey}`;
}

/**
 * Constructs the full RTMP URL for Twitch.
 * Twitch RTMP ingest: rtmp://live.twitch.tv/app/{stream_key}
 * Use the nearest ingest server for lower latency.
 */
export function buildTwitchRtmpUrl(
  streamKey: string,
  ingestServer?: string
): string {
  const server = ingestServer ?? "live.twitch.tv";
  return `rtmp://${server}/app/${streamKey}`;
}

/**
 * Builds a StreamDestination from a destination type and stream key.
 */
export function buildDestination(
  type: StreamDestinationType,
  streamKey: string,
  options?: { name?: string; customRtmpUrl?: string; twitchIngestServer?: string }
): StreamDestination {
  const name = options?.name ?? type;

  switch (type) {
    case "mux":
      return {
        type,
        name,
        rtmpUrl: buildMuxRtmpUrl(streamKey),
        streamKey,
      };
    case "youtube":
      return {
        type,
        name,
        rtmpUrl: buildYouTubeRtmpUrl(streamKey),
        streamKey,
      };
    case "twitch":
      return {
        type,
        name,
        rtmpUrl: buildTwitchRtmpUrl(streamKey, options?.twitchIngestServer),
        streamKey,
      };
    case "custom":
      if (!options?.customRtmpUrl) {
        throw new Error("customRtmpUrl is required for custom destinations");
      }
      return {
        type,
        name,
        rtmpUrl: `${options.customRtmpUrl}/${streamKey}`,
        streamKey,
      };
  }
}

/**
 * Constructs the full RTMP URL with stream key appended, ready for LiveKit egress.
 * LiveKit expects the full URL including the stream key.
 */
export function getFullRtmpUrl(destination: StreamDestination): string {
  // For MUX, the rtmpUrl already includes the stream key in the path
  return destination.rtmpUrl;
}

/**
 * Gets a MUX HLS playback URL from a playback ID.
 */
export function getMuxPlaybackUrl(playbackId: string): string {
  return `https://stream.mux.com/${playbackId}.m3u8`;
}

/**
 * Gets a MUX thumbnail URL from a playback ID.
 */
export function getMuxThumbnailUrl(
  playbackId: string,
  options?: { width?: number; height?: number; time?: number }
): string {
  const params = new URLSearchParams();
  if (options?.width) params.set("width", String(options.width));
  if (options?.height) params.set("height", String(options.height));
  if (options?.time) params.set("time", String(options.time));

  const queryString = params.toString();
  return `https://image.mux.com/${playbackId}/thumbnail.jpg${queryString ? `?${queryString}` : ""}`;
}
```

### Step 2: Create `src/lib/db/schema/livestreams.ts`

```typescript
import {
  pgTable,
  text,
  timestamp,
  uuid,
} from "drizzle-orm/pg-core";

export const livestreams = pgTable("livestreams", {
  id: uuid("id").primaryKey().defaultRandom(),
  roomName: text("room_name").notNull(),
  egressId: text("egress_id"),
  muxLiveStreamId: text("mux_live_stream_id"),
  muxPlaybackId: text("mux_playback_id"),
  muxStreamKey: text("mux_stream_key"),
  rtmpUrl: text("rtmp_url"),
  status: text("status", {
    enum: ["idle", "starting", "active", "stopping", "completed", "failed"],
  })
    .notNull()
    .default("idle"),
  muxAssetId: text("mux_asset_id"),
  startedAt: timestamp("started_at", { withTimezone: true }),
  endedAt: timestamp("ended_at", { withTimezone: true }),
  userId: text("user_id").notNull(),
  createdAt: timestamp("created_at", { withTimezone: true })
    .notNull()
    .defaultNow(),
  updatedAt: timestamp("updated_at", { withTimezone: true })
    .notNull()
    .defaultNow(),
});

export type Livestream = typeof livestreams.$inferSelect;
export type NewLivestream = typeof livestreams.$inferInsert;
```

### Step 3: Add export to `src/lib/db/schema/index.ts`

```typescript
export * from "./livestreams";
```

### Step 4: Create `src/lib/video/livestream.ts`

```typescript
import { EgressClient, EncodingOptions, StreamOutput, StreamProtocol } from "livekit-server-sdk";
import Mux from "@mux/mux-node";
import { db } from "@/lib/db";
import { livestreams, type Livestream } from "@/lib/db/schema";
import { eq, and, desc } from "drizzle-orm";
import {
  buildMuxRtmpUrl,
  getMuxPlaybackUrl,
  getFullRtmpUrl,
} from "./stream-destinations";
import type { StreamDestination } from "./stream-destinations";

type MuxLiveStream = {
  id: string;
  stream_key: string;
  playback_ids?: Array<{ id: string; policy: string }>;
  status: string;
};

type MuxAsset = {
  id: string;
  playback_ids?: Array<{ id: string; policy: string }>;
  status: string;
  duration?: number;
};

function getEgressClient(): EgressClient {
  const livekitUrl = process.env.NEXT_PUBLIC_LIVEKIT_URL;
  const apiKey = process.env.LIVEKIT_API_KEY;
  const apiSecret = process.env.LIVEKIT_API_SECRET;

  if (!livekitUrl || !apiKey || !apiSecret) {
    throw new Error(
      "Missing NEXT_PUBLIC_LIVEKIT_URL, LIVEKIT_API_KEY, or LIVEKIT_API_SECRET"
    );
  }

  // Convert wss:// to https:// for REST API
  const httpUrl = livekitUrl
    .replace("wss://", "https://")
    .replace("ws://", "http://");

  return new EgressClient(httpUrl, apiKey, apiSecret);
}

function getMuxClient(): Mux {
  const tokenId = process.env.MUX_TOKEN_ID;
  const tokenSecret = process.env.MUX_TOKEN_SECRET;

  if (!tokenId || !tokenS