290 lines
14 KiB
Markdown
290 lines
14 KiB
Markdown
# CLAUDE.md
|
||
|
||
This file provides guidance to Claude Code (claude.ai/code) when working with code in this repository.
|
||
|
||
## Commands
|
||
|
||
```bash
|
||
# Build
|
||
cargo build
|
||
|
||
# Run
|
||
cargo run
|
||
|
||
# Build (Nix)
|
||
nix build
|
||
|
||
# Dev shell (provides clang + mold linker)
|
||
nix develop
|
||
```
|
||
|
||
No tests exist yet.
|
||
|
||
## Workspace layout
|
||
|
||
```
|
||
crates/
|
||
server/ — main binary (RTMP ingestion, HTTP API, WebRTC)
|
||
entity/ — SeaORM entity definitions
|
||
migration/ — SeaORM migrations
|
||
```
|
||
|
||
## Architecture
|
||
|
||
This is an RTMP-to-WHIP/WHEP bridge: accepts RTMP video+audio publish streams and re-streams to browsers via WebRTC.
|
||
|
||
**Signal flow:**
|
||
|
||
```
|
||
OBS/encoder → RTMP (port 1935) → H264Parser / AACParser → async_broadcast channels
|
||
↓
|
||
Browser ← WebRTC/UDP ← str0m Rtc ← WHIP/WHEP HTTP (port 3000)
|
||
↑
|
||
WebrtcProxy (UDP port from RTC_PORT env, default 6969)
|
||
```
|
||
|
||
**Runtime:** tokio. All top-level workers run in a `JoinSet`; if any exits unexpectedly the process aborts all others and exits.
|
||
|
||
---
|
||
|
||
### Entry point — `crates/server/src/main.rs`
|
||
|
||
Initialises everything, then hands each worker to a `tokio::task::JoinSet`:
|
||
|
||
```rust
|
||
workers.spawn(http.run());
|
||
workers.spawn(proxy.run());
|
||
workers.spawn(webrtc.run());
|
||
workers.spawn(rtmp.run());
|
||
```
|
||
|
||
`tokio::select!` waits for either Ctrl-C or a worker exiting; on either branch `workers.abort_all()` is called.
|
||
|
||
**`AppState`** — `Arc<Mutex<AppState>>` wrapping a `DashMap<i32, StreamSession>` keyed by `stream_key.id`.
|
||
|
||
**`StreamSession`** fields:
|
||
- `stream_key_id: i32`
|
||
- `stream_key_label: String`
|
||
- `frame_channel: async_broadcast::Sender<Arc<VideoFrame>>`
|
||
- `audio_channel: async_broadcast::Sender<Arc<OpusAudioFrame>>`
|
||
- `codec: Option<StreamCodec>` — `H264 | H265 | AV1` (AV1 unsupported)
|
||
|
||
Shared channels between main components:
|
||
- `offer_tx/offer_rx: mpsc::channel<(request_id: i32, stream_key_id: i32, sdp_body: String)>(64)` — HTTP → Webrtc
|
||
- `answer_tx/answer_rx: async_broadcast<(request_id: i32, Option<String>)>(64)` — Webrtc → HTTP
|
||
|
||
DB: `sqlite://./db/db.sqlite?mode=rwc`. Migrations run at startup via `Migrator::up`. `stream_session::Model::clean_unended_streams` is called at startup to repair sessions left open by a previous crash.
|
||
|
||
---
|
||
|
||
### RTMP ingestion — `crates/server/src/rtmp.rs`
|
||
|
||
`pub struct Rtmp { listener, stream_sessions, db }`
|
||
|
||
`async fn run(self)` — accepts TCP connections on port 1935, one `tokio::spawn` per connection:
|
||
|
||
1. **Handshake** — `rml_rtmp::Handshake` (C0+C1 → S0+S1+S2 → C2).
|
||
2. **Session setup** — `rml_rtmp::ServerSession`.
|
||
3. **Event loop** — 4096-byte reads, `session.handle_input`:
|
||
- `ConnectionRequested` → accepted unconditionally.
|
||
- `PublishStreamRequested` → looks up stream key in DB; rejects if not found or if a `stream_session` record is already active. On accept: inserts a `StreamSession` into `AppState` and creates a `stream_session` row in DB.
|
||
- `VideoDataReceived` → dispatched to `H264Parser::parse` (H.265/AV1 drops the connection). Parsed `VideoFrame`s are broadcast on `frame_channel`.
|
||
- `AudioDataReceived` → dispatched to `AACParser::parse_aac`, then `AudioProcesser` transcodes AAC→Opus. `OpusAudioFrame`s broadcast on `audio_channel`.
|
||
- `PublishStreamFinished` → removes `StreamSession` from `AppState`, sets `ended_at` on the `stream_session` DB row.
|
||
|
||
---
|
||
|
||
### HTTP API — `crates/server/src/http.rs`
|
||
|
||
`pub struct HttpServer` fields: `offer_tx`, `accept_rx` (broadcast), `appstate`, `request_count: AtomicI32`, `db`, `config: Arc<HttpServerConfig>`.
|
||
|
||
`async fn run(self)` — wraps self in `Arc`, builds axum router with CORS (allows `localhost:5173` and `stream.h.doloro.co.uk`), serves on `0.0.0.0:3000`.
|
||
|
||
**Routes:**
|
||
|
||
| Method | Path | Handler |
|
||
|--------|------|---------|
|
||
| `GET` | `/api/catalog` | `catalog_handler` — returns `{ active_streams: [{ label, id, user }] }` from `AppState` joined with DB user lookup |
|
||
| `POST` | `/api/user` | `create_user_handler` — creates user; requires `SIGNUP_CODE` header matching env var |
|
||
| `GET/POST` | `/api/stream-key` | `get_all_stream_keys` / `create_stream_key_handler` |
|
||
| `POST` | `/api/whip` | `handle_whip_injest` (in `webrtc_ingest.rs`) — WHIP ingest endpoint |
|
||
| `POST` | `/api/login` | `login_handler` — verifies password hash, creates `auth_session`, returns session cookie |
|
||
| `POST` | `/api/stream/{slug}` | `stream_handler` — WHEP offer: sends SDP offer over `offer_tx`, waits on `accept_rx` for matching `request_id`, returns SDP answer |
|
||
|
||
Auth: session token sent as `session` header or cookie; `auth_session` entity looked up from DB. `FromRequestParts` extractor `AuthSession` handles this for protected routes.
|
||
|
||
`request_count: AtomicI32` tracks in-flight request IDs (monotonic, `fetch_add(1, Relaxed)`).
|
||
|
||
---
|
||
|
||
### WebRTC proxy — `crates/server/src/webrtc_proxy.rs`
|
||
|
||
`#[derive(Clone)] pub struct WebrtcProxy` — all fields are `Arc`-wrapped:
|
||
- `clients_ufrag: Arc<DashMap<String, mpsc::Sender<(Bytes, SocketAddr)>>>` — pending ICE ufrag → per-client channel
|
||
- `clients_addr: Arc<DashMap<SocketAddr, mpsc::Sender<(Bytes, SocketAddr)>>>` — established addr → per-client channel
|
||
- `socket: Arc<UdpSocket>` — shared UDP socket bound to `0.0.0.0:{RTC_PORT}` (default 6969)
|
||
- `public_addr: SocketAddr` — resolved via `PUBLIC_DOMAIN` env var DNS lookup or STUN discovery
|
||
|
||
**`async fn run(self)`** — UDP receive loop:
|
||
1. Receives datagrams on the shared socket.
|
||
2. If source addr is already in `clients_addr`, forwards to that client's channel.
|
||
3. Otherwise parses STUN binding request to extract ufrag (`username` attribute, part before `:`), looks up `clients_ufrag`, promotes to `clients_addr`, forwards.
|
||
|
||
**`fn add_client(ufrag, …) -> (Arc<UdpSocket>, Receiver<…>)`** — called by `Webrtc` when setting up a new peer connection. Registers the ufrag and returns the shared socket + a per-client receive channel.
|
||
|
||
**`fn public_addr()`** — returns the public address advertised in ICE candidates.
|
||
|
||
---
|
||
|
||
### WebRTC negotiation — `crates/server/src/webrtc.rs`
|
||
|
||
`pub struct Webrtc { offer_rx, accept_tx, sessions_ref, db, proxy: Arc<WebrtcProxy> }`
|
||
|
||
**`async fn run(mut self)`** — receives `(request_id, stream_id, sdp_body)` from `offer_rx`:
|
||
|
||
1. Looks up `stream_key` in DB; sends `None` answer and continues on error/not-found.
|
||
2. Gets `public_addr` from proxy for ICE candidate.
|
||
3. Builds `str0m::Rtc` with H.264 PTs 102 (`0x42e01f`), 104 (`0x4d001f`), 106 (`0x64001f`); default H.264 disabled.
|
||
4. Calls `add_client(ufrag)` on proxy to register ICE ufrag and get the UDP socket + channel.
|
||
5. Accepts SDP offer → produces SDP answer → broadcasts answer on `accept_tx`.
|
||
6. Spawns `detach_connection` task.
|
||
|
||
**`detach_connection`** per-peer loop:
|
||
- Drains `rtc.poll_output()`: sends transmits via the shared proxy UDP socket, handles `Event::MediaAdded` (selects best PT by `profile_level_id`) and `Event::Connected`.
|
||
- When connected, subscribes to `frame_channel` from `AppState` for the stream, drains up to 8 frames per tick via `try_recv`, writes with `writer.write(pt, now, MediaTime::from_90khz(ts * 90))`.
|
||
- `tokio::select!` (capped 20 ms) on str0m deadline or incoming UDP datagram from proxy channel.
|
||
|
||
---
|
||
|
||
### WHIP ingest — `crates/server/src/webrtc_ingest.rs`
|
||
|
||
`async fn handle_whip_injest` — axum handler for `POST /api/whip`. Currently a stub; extracts `State<Arc<HttpServer>>` and the request body. (Implementation in progress.)
|
||
|
||
---
|
||
|
||
### Adding a new codec
|
||
|
||
To add support for a new ingest codec:
|
||
|
||
1. **`rtmp.rs` — `parse_video_codec`**: add a FourCC arm (enhanced RTMP) or legacy codec ID. `StreamCodec` enum lives in `main.rs`.
|
||
|
||
2. **New parser struct** (e.g. `crates/server/src/codec/mycodec.rs`):
|
||
- Field for each parameter set (`Vec<u8>`)
|
||
- `parse_sequence_header(&mut self, payload: &[u8])` — parses the decoder config record, caches parameter sets
|
||
- `to_annexb(&self, payload: &[u8], is_keyframe: bool) -> Option<Vec<u8>>` — converts length-prefixed NALUs to Annex-B, prepends parameter sets before keyframes
|
||
|
||
3. **`rtmp.rs` — `VideoDataReceived` handler**: branch on `StreamCodec`, slice the payload correctly for each packet type, call `parse_sequence_header` on type `0` and `to_annexb` on types `1`/`3`, broadcast the resulting `VideoFrame`.
|
||
|
||
4. **`webrtc.rs` — codec config**: configure the correct PT via `codec_config()` (e.g. `enable_h265`, `add_h264`). Fix PT selection in `Event::MediaAdded` if the new codec uses a different profile field than `profile_level_id`.
|
||
|
||
5. **`StreamSession`** (`main.rs`): `codec: StreamCodec` field — set it when inserting into `stream_sessions` in `rtmp.rs` so the WebRTC layer can know what codec the session is using.
|
||
|
||
---
|
||
|
||
### H.264 parsing — `crates/server/src/media.rs`
|
||
|
||
`H264Parser` converts RTMP `VideoDataReceived` AVCC payloads → Annex-B `VideoFrame`s:
|
||
|
||
- Byte 0: frame type (upper nibble, 1=keyframe) + codec ID (lower nibble, 7=H.264).
|
||
- Byte 1: AVC packet type — `0`=sequence header, `1`=NAL data.
|
||
- Packet type `0`: parses `AVCDecoderConfigurationRecord`, caches SPS+PPS.
|
||
- Packet type `1`: converts AVCC length-prefixed NALUs to `00 00 00 01` Annex-B. Prepends SPS+PPS before the first NALU of each keyframe.
|
||
|
||
`pub struct VideoFrame { pub data: Bytes, pub is_keyframe: bool, pub timestamp_ms: u32 }`
|
||
|
||
Also defines `pub struct AudioFrame { pub data: Bytes, pub timestamp_ms: u32 }` (distinct from `OpusAudioFrame`).
|
||
|
||
---
|
||
|
||
### Audio — `crates/server/src/audio.rs`
|
||
|
||
**`AACParser`** — parses raw RTMP `AudioDataReceived` payloads:
|
||
- Byte 0: codec (upper nibble, 10=AAC).
|
||
- Byte 1: AAC packet type — `0`=AudioSpecificConfig (codec init), `1`=raw AAC frame.
|
||
- Packet type `0`: initialises a **Symphonia** AAC decoder with the config bytes as `extra_data`.
|
||
- Packet type `1`: decodes via Symphonia, converts to interleaved f32 PCM, returns `AudioFrame { data, timestamp_ms, sample_rate }`.
|
||
|
||
**`AudioProcesser`** — AAC→Opus transcoder:
|
||
- `encoder: opus::Encoder` — 48kHz stereo, `LowDelay` application mode.
|
||
- `resampler: rubato::FftFixedIn<f32>` — resamples 44100 Hz → 48000 Hz when needed.
|
||
- `pcm_buf: Vec<f32>` — accumulates samples until a full 960-sample (20 ms) Opus frame is ready.
|
||
- `samples_emitted: u64` — monotonic 48kHz counter; `timestamp_ms = samples_emitted / 48` (independent of RTMP timestamps).
|
||
- `encode(frame) -> Vec<OpusAudioFrame>`: resamples if not already 48kHz, drains `pcm_buf` in 960-sample chunks, emits one `OpusAudioFrame` per chunk.
|
||
|
||
`pub struct OpusAudioFrame { pub data: Bytes, pub timestamp_ms: u32 }`
|
||
|
||
---
|
||
|
||
### Password hashing — `crates/server/src/hash.rs`
|
||
|
||
- `hash_password(password: &str) -> Result<String>` — Argon2id hash via `Argon2::default()` with a random `OsRng` salt; returns PHC-format string.
|
||
- `verify_password(password: &str, hash: &str) -> bool` — parses PHC string and verifies with Argon2. Panics if `hash` is not valid PHC format.
|
||
|
||
---
|
||
|
||
### H.265 parsing — `crates/server/src/codec/h265.rs`
|
||
|
||
`H265Parser` converts enhanced RTMP HEVC payloads → Annex-B `VideoFrame`s.
|
||
|
||
**Enhanced RTMP detection** (`parse_video_codec` in `rtmp.rs`):
|
||
- Byte 0 bit 7 (`0x80`) set = ExVideoHeader (enhanced RTMP format)
|
||
- Bits 4–6 of byte 0 = frame type (1=keyframe)
|
||
- Bits 0–3 of byte 0 = packet type: `0`=SequenceStart, `1`=CodedFrames, `3`=CodedFramesX
|
||
- Bytes 1–4 = FourCC: `hvc1`=H.265, `avc1`=H.264, `av01`=AV1
|
||
|
||
**Payload offsets by packet type:**
|
||
- Type `0` (SequenceStart): payload at `data[5..]` — `HEVCDecoderConfigurationRecord`
|
||
- Type `1` (CodedFrames): payload at `data[8..]` — 3 bytes composition time skipped
|
||
- Type `3` (CodedFramesX): payload at `data[5..]` — no composition time
|
||
|
||
**`HEVCDecoderConfigurationRecord`** parsing:
|
||
- Skip first 22 bytes (profile/level/tier info, not needed for forwarding)
|
||
- Byte 22 = `numOfArrays`; each array: 1 byte `nal_unit_type` (lower 6 bits) + 2 byte NALU count + length-prefixed NALUs
|
||
- NAL types: VPS=32, SPS=33, PPS=34 — cached as `Vec<u8>` on the parser struct
|
||
|
||
**HVCC → Annex-B conversion:**
|
||
- Same as AVCC: replace 4-byte big-endian length prefix with `00 00 00 01` start code
|
||
- Prepend VPS+SPS+PPS (each with start code) before every keyframe
|
||
|
||
---
|
||
|
||
## Database (SeaORM + SQLite)
|
||
|
||
DB file: `./db/db.sqlite`. Migrations: `crates/migration/`. Entities: `crates/entity/`.
|
||
|
||
Run migrations: `Migrator::up(&db, None).await` — idempotent, tracked in `seaql_migrations`.
|
||
|
||
### Entities
|
||
|
||
**`users`** (`crates/entity/src/users.rs`)
|
||
- Fields: `id`, `username`, `hashed_password`
|
||
- Relations: `has_many` → `stream_key`
|
||
- `Entity::create(db, username, hashed_password)` — inserts a new user
|
||
- `ActiveModel::update_username` / `update_password` — mutation helpers
|
||
- Passwords must be hashed via `hash::hash_password` (Argon2id) before storing
|
||
|
||
**`stream_key`** (`crates/entity/src/stream_key.rs`)
|
||
- Fields: `id`, `key_value` (unique), `user_id`, `label`, `is_active`, `is_unlisted`, `created_at`
|
||
- Relations: `belongs_to` → `users`, `has_many` → `stream_session`
|
||
- `Entity::find_by_key(db, key_value)` — lookup by raw stream key string
|
||
|
||
**`stream_session`** (`crates/entity/src/stream_session.rs`)
|
||
- Fields: `id`, `stream_key_id`, `started_at`, `ended_at` (nullable)
|
||
- Relations: `belongs_to` → `stream_key`
|
||
- `Model::get_active_by_stream_key_id(db, id)` — finds open session (no `ended_at`)
|
||
- `Model::clean_unended_streams(db)` — sets `ended_at = now` on all sessions missing it (crash recovery)
|
||
|
||
**`auth_session`** (`crates/entity/src/auth_session.rs`)
|
||
- Fields: `id`, `user_id`, `token`, `created_at`
|
||
- Used for cookie-based auth; token matched against `session` header/cookie
|
||
|
||
### SeaORM conventions
|
||
|
||
- Query methods go on `Entity` (e.g. `Entity::find_by_x`).
|
||
- Mutation helpers that intercept save logic go on `ActiveModel`.
|
||
- For destructive schema changes in prod, use expand-contract: add → backfill → switch code → drop old in a later migration.
|
||
|
||
---
|
||
|
||
**Test page:** `index.html` — open in a browser to play the stream via WHEP without extra tooling.
|