simple-instant-stream/CLAUDE.md

CLAUDE.md

This file provides guidance to Claude Code (claude.ai/code) when working with code in this repository.

Commands

# 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:

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 VideoFrames are broadcast on frame_channel.
   • AudioDataReceived → dispatched to AACParser::parse_aac, then AudioProcesser transcodes AAC→Opus. OpusAudioFrames 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 VideoFrames:

• 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 VideoFrames.

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.