From d4b77bd71714971269691995fc59beb2afdee868 Mon Sep 17 00:00:00 2001 From: Doloro1978 Date: Sun, 19 Jul 2026 16:59:00 +0100 Subject: [PATCH] feat: High quality AGENTS.md file --- AGENTS.md | 642 ++++++++++++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 642 insertions(+) create mode 100644 AGENTS.md diff --git a/AGENTS.md b/AGENTS.md new file mode 100644 index 0000000..0c65a1a --- /dev/null +++ b/AGENTS.md @@ -0,0 +1,642 @@ +# AGENTS.md + +This file provides comprehensive guidance for AI agents working with this repository. +It is the canonical reference; CLAUDE.md may be a subset of this. + +--- + +## Project Summary + +**RTMP-to-WHIP/WHEP bridge.** Accepts RTMP video+audio publish streams and re-streams to +browsers via WebRTC (WHEP). Also has a stub for WHIP ingest. + +**Binary name:** `rtmp-to-whip` +**Workspace:** Rust 2024 edition, 3 crates (`server`, `entity`, `migration`) +**No automated tests exist** (except one ignored FLV replay test). + +--- + +## Commands + +```bash +# Build (all crates) +cargo build + +# Run (server binary) +cargo run + +# Build (Nix) +nix build + +# Dev shell (provides clang + mold linker + sea-orm-cli + opus + just) +nix develop +``` + +--- + +## Workspace Layout + +``` +crates/ + server/ — main binary (RTMP, HTTP, WebRTC, codecs, audio) + entity/ — SeaORM entity definitions (users, stream_key, stream_session, auth_session) + migration/ — SeaORM migrations (4 files) +target/ — build output (gitignored) +db/ — SQLite database at runtime (db/db.sqlite, gitignored) +index.html — WHEP test page (browser-based viewer) +``` + +--- + +## Ports & Networking + +| Port | Protocol | Purpose | +|-------|----------|----------------------------------| +| 1935 | TCP | RTMP ingest (OBS/encoder) | +| 3000 | TCP | HTTP API (axum) | +| 6969 | UDP | WebRTC media (configurable via RTC_PORT env) | + +--- + +## Environment Variables + +| Variable | Default | Purpose | +|-----------------|---------|------------------------------------------------| +| `RTC_PORT` | `6969` | UDP port for WebRTC media traffic | +| `PUBLIC_DOMAIN` | — | Domain for ICE candidates (DNS-resolved); falls back to STUN discovery if unset | +| `SIGNUP_CODE` | — | Required token for `/api/user` signup; empty = disabled | +| `RUST_LOG` | — | Tracing filter (e.g. `info,warn`) | + +--- + +## Entry Point — `crates/server/src/main.rs` + +Initializes all components, spawns 4 workers in a `JoinSet`: + +```rust +workers.spawn(http.run()); +workers.spawn(proxy.run()); +workers.spawn(webrtc.run()); +workers.spawn(rtmp.run()); +``` + +Shuts down on Ctrl-C or any worker exit (`workers.abort_all()`). + +**`AppState`** — `Arc>` wrapping `DashMap`. +Keyed by `stream_key.id`. + +**`StreamSession`** fields: +- `stream_key_id: i32` +- `stream_key_label: String` +- `frame_channel: async_broadcast::Sender>` +- `audio_channel: async_broadcast::Sender>` +- `codec: Option` — `H264 | H265 | AV1` +- `started_at: DateTime` +- `viewers: AtomicU32` + +**Shared channels:** +- `offer_tx/offer_rx: mpsc::channel<(i32, i32, String)>(64)` — HTTP → WebRTC (request_id, stream_key_id, sdp_offer) +- `answer_tx/answer_rx: broadcast::<(i32, Option)>(64)` — WebRTC → HTTP (request_id, answer_or_none) + +**DB:** `sqlite://./db/db.sqlite?mode=rwc`. Migrations run at startup via `Migrator::up`. +`stream_session::Model::clean_unended_streams()` repairs crash-residual sessions. + +--- + +## Architecture — Signal Flow + +``` +OBS/encoder → RTMP (port 1935) → H264Parser / H265Parser / Av1Parser → async_broadcast frame channel + ↓ +Browser ← WebRTC/UDP ← str0m Rtc ← WHEP HTTP POST /api/stream/{slug} + ↑ + WebrtcProxy (UDP port from RTC_PORT env, default 6969) + +WHIP ingest: Browser/encoder → WHIP POST /api/whip → (stub: parses SDP, logs media) +``` + +--- + +## RTMP Ingestion — `crates/server/src/rtmp.rs` + +`pub struct Rtmp { listener, stream_sessions, db }` + +`async fn run(self)` — accepts TCP connections on port 1935, spawns one task per connection: + +1. **Handshake** — `rml_rtmp::Handshake` (C0+C1 → S0+S1+S2 → C2), 30s timeout. +2. **Session setup** — `rml_rtmp::ServerSession`. +3. **Event loop** — 4096-byte reads, 15s idle timeout: + - `ConnectionRequested` → accepted unconditionally. + - `PublishStreamRequested` → DB lookup of stream key; rejects if not found or already live. + Inserts `StreamSession` into `AppState` + DB row. + - `VideoDataReceived` → dispatched to `CodecParser::parse` (H.264, H.265, or AV1). + Frame broadcast on `frame_channel`. + - `AudioDataReceived` → `AACParser` decodes AAC → PCM, `AudioProcesser` resamples + encodes Opus. + `OpusAudioFrame` broadcast on `audio_channel`. + - `PublishStreamFinished` → removes `StreamSession` from `AppState`, sets `ended_at` in DB. + +**Codec detection** (`parse_video_codec`): +- Enhanced RTMP (byte 0 bit 7 = 1): bytes 1–4 = FourCC (`hvc1`→H.265, `avc1`→H.264, `av01`→AV1). +- Legacy RTMP (byte 0 bit 7 = 0): nibble `& 0x0F == 7` → H.264. +- `StreamSession.codec` is stamped on first `VideoDataReceived` and never changed. + +--- + +## HTTP API — `crates/server/src/http.rs` + +`pub struct HttpServer` fields: `offer_tx`, `accept_rx`, `appstate`, `request_count: AtomicI32`, `db`, `config`. + +Serves on `0.0.0.0:3000` with CORS for `http://localhost:5173` and `https://stream.h.doloro.co.uk`. + +### Routes + +| Method | Path | Handler | Auth? | Notes | +|--------|-------------------------|--------------------------------|-------|-------------------------------------------------| +| GET | `/api/catalog` | `catalog_handler` | No | Returns `{ active_streams: [...] }` from DB + AppState | +| POST | `/api/user` | `create_user_handler` | No | Signup; requires `SIGNUP_CODE` header | +| POST | `/api/login` | `login_handler` | No | Verifies password, creates auth_session, sets cookie | +| GET | `/api/stream-key` | `get_all_stream_keys` | Yes | Returns user's stream keys | +| POST | `/api/stream-key` | `create_stream_key_handler` | Yes | Creates key with UUID value (`stream-key-`) | +| PATCH | `/api/stream-key` | `edit_stream_key` | Yes | Edits label (validated by `KEY_RE` regex) | +| POST | `/api/whip` | `handle_whip_injest` | No | **Stub** — parses SDP, logs media lines | +| POST | `/api/stream/{slug}` | `stream_handler` | No | WHEP offer; forwards to WebRTC, waits for answer | +| GET | `/api/meow` | `meow_handler` | No | Returns `"meow"` | + +### Auth + +- Session token sent as `session` cookie or header. +- `AuthUser` extractor: looks up `auth_session` by token value, then joins to `users`. +- Protected routes require `AuthUser`. +- Cookie flags: `HttpOnly`, `Path=/`, `SameSite=None` (dev) or `Lax` (prod). +- `DevFlag` extractor: reads `?dev=1` query param. + +### Validation + +- Username: max 32 chars, non-empty. +- Stream key label: 1–64 chars, regex `^[A-Za-z0-9 _-]{1,67}$`, must contain at least one letter. +- User has `stream_key_limit` (default 3) enforced at creation time. + +--- + +## WebRTC Proxy — `crates/server/src/webrtc_proxy.rs` + +`pub struct WebrtcProxy` — all fields `Arc`-wrapped, `Clone`-able. + +- `clients_ufrag: DashMap>` — pending ICE ufrag → per-client channel +- `clients_addr: DashMap>` — established addr → per-client channel +- `socket: Arc` — shared UDP socket bound to `0.0.0.0:{RTC_PORT}` +- `public_addr: SocketAddr` — from `PUBLIC_DOMAIN` DNS or STUN discovery + +**`async fn run(self)`** — UDP receive loop: +1. Receives datagrams on shared socket. +2. If source addr in `clients_addr`, forwards to that client. +3. Otherwise parses STUN binding request to extract ufrag (username attribute, split on `:`), + looks up `clients_ufrag`, promotes to `clients_addr`, forwards. + +**`fn add_client(ufrag)`** — called by Webrtc when setting up peer connection. +Returns `(Arc, Receiver<(Bytes, SocketAddr)>)`. + +**Public IP discovery:** +- If `PUBLIC_DOMAIN` env set: DNS-resolve for IPv4. +- Else if `cfg!(debug_assertions)`: returns `127.0.0.1` (for testing). +- Else: sends STUN Binding Request to `stun.l.google.com:19302`, parses XOR-MAPPED-ADDRESS. + +--- + +## WebRTC Negotiation — `crates/server/src/webrtc.rs` + +`pub struct Webrtc { offer_rx, accept_tx, sessions_ref, db, proxy }` + +`async fn run(mut self)` — receives `(request_id, stream_id, sdp_body)` from `offer_rx`: + +1. Looks up `stream_key` in DB; rejects if not found. +2. Gets codec from `StreamSession.codec` (determines which str0m codec config to enable). +3. Configures str0m `Rtc::builder`: + - Opus always enabled. + - H.265: `cc.enable_h265(true)`. + - AV1: `cc.enable_av1(true)`. + - H.264 (default): adds 3 profiles — Constrained Baseline (0x42e01f), Main (0x4d001f), High (0x64001f). + Firefox only offers Constrained Baseline — without it, no codec matches. +4. Accepts SDP offer → produces SDP answer. +5. Validates answer has non-empty video PT list (str0m returns empty list if no codec matched). +6. Extracts ICE ufrag from answer, registers with proxy. +7. Broadcasts answer on `accept_tx`. +8. Spawns `detach_connection` per peer. + +**`detach_connection`** per-peer loop: +- Drains `rtc.poll_output()` — transmits via proxy UDP, handles events. +- `Event::MediaAdded` — selects best video PT by profile_level_id (H.264), + h265_profile_tier_level (H.265), or level_idx (AV1). +- `Event::Connected` — starts subscribing to frame/audio channels from AppState. +- `tokio::select!` on: str0m deadline, incoming UDP from proxy, video frame, audio frame. +- **Keyframe gating:** drops non-keyframe frames until first keyframe is seen. +- **Overflow handling:** on ring buffer overflow, resets `saw_keyframe` to wait for next keyframe. +- Video RTP timestamp: `MediaTime::from_90khz(ts * 90)` where `ts` is VideoFrame timestamp in ms. +- Audio RTP timestamp: `MediaTime::new(ts * 48, Frequency::FORTY_EIGHT_KHZ)`. +- Drains up to 7 pending frames per tick to catch up. + +--- + +## Codecs — `crates/server/src/codec/` + +### Trait — `CodecParser` + +```rust +pub trait CodecParser: Send { + fn parse(&mut self, data: &[u8], timestamp_ms: u32) -> Option; +} +``` + +### `VideoFrame` + +```rust +pub struct VideoFrame { + pub data: Bytes, // Annex-B elementary stream + pub is_keyframe: bool, + pub timestamp_ms: u32, // Presentation timestamp (DTS + CTS for B-frame streams) +} +``` + +### H.264 — `h264.rs` + +Handles both legacy and enhanced RTMP. + +**Legacy RTMP:** byte 0 = `(frame_type << 4) | codec_id`, byte 1 = packet type, bytes 2–4 = CTS. +**Enhanced RTMP:** byte 0 = `0x80 | (frame_type << 4) | packet_type`, bytes 1–4 = FourCC. + +- Packet type `0`: parses `AVCDecoderConfigurationRecord`, caches SPS+PPS. +- Packet type `1`: AVCC → Annex-B conversion, prepends SPS+PPS before keyframes. + Uses `PTS = DTS + CTS` to avoid B-frame stuttering. +- Packet type `3`: same as type 1 but no CTS field. + +### H.265 — `h265.rs` + +Enhanced RTMP only (FourCC `hvc1`). + +- Packet type `0`: bytes[5..] = `HEVCDecoderConfigurationRecord`. +- Packet type `1`: bytes 5–7 = CTS, bytes 8+ = HVCC NALUs. Uses `PTS = DTS + CTS`. +- Packet type `3`: bytes 5+ = HVCC NALUs, no CTS. + +`HEVCDecoderConfigurationRecord` parsing: skips first 22 bytes (profile/level/tier), +parses arrays at byte 22, caches VPS (NAL type 32), SPS (33), PPS (34). + +### AV1 — `av1.rs` + +Enhanced RTMP only (FourCC `av01`). + +- Packet type `0`: bytes[5..] = `AV1CodecConfigurationRecord`, extracts `configOBUs` from `payload[4..]`. +- Packet type `1`/`3`: bytes[5..] = OBU stream. + +**Keyframe detection:** +1. RTMP FrameType=1. +2. Scan OBUs for `OBU_SEQUENCE_HEADER` (obu_type=1). +3. First coded frame after config (bootstrap fallback — OBS may never set FrameType=1 for AV1). + +Config OBUs prepended to keyframe payloads so str0m's Av1Packetizer sees a Sequence Header OBU. + +### FLV Replay Test — `flv_replay_test.rs` + +Ignored test harness for codec validation: +```bash +FLV_IN=test.flv ES_OUT=output.es cargo test -p server flv_replay -- --ignored --nocapture +``` +Parses FLV video tags, runs through codec parser, dumps Annex-B elementary stream with +temporal delimiter OBUs for ffprobe validation. Checks for duplicate PTS values. + +--- + +## Audio — `crates/server/src/audio.rs` + +### AACParser + +Parses RTMP audio payloads: +- Byte 0: `>> 4 == 10` → AAC codec. +- Byte 1: `0` = AudioSpecificConfig → initializes Symphonia AAC decoder with `extra_data`. +- Byte 1: `1` = raw AAC frame → decodes via Symphonia → interleaved f32 PCM. + +### AudioFrame + +```rust +pub struct AudioFrame { + pub data: Bytes, // interleaved f32 PCM, little-endian + pub timestamp_ms: u32, + pub sample_rate: u32, // typically 44100 Hz from AAC-LC +} +``` + +### AudioProcesser + +AAC → Opus transcoding pipeline: +- **Resampler:** `rubato::Fft` from 44100 Hz → 48000 Hz (if needed). +- **Encoder:** `opus::Encoder` — 48kHz stereo, `LowDelay` mode. +- **Frame size:** 960 samples per channel (20ms), 1920 interleaved. +- **Timestamp:** monotonic 48kHz counter: `timestamp_ms = samples_emitted / 48` (independent of RTMP timestamps). +- **Buffer:** accumulates resampled PCM until full Opus frame, then encodes and emits `OpusAudioFrame`. + +### OpusAudioFrame + +```rust +pub struct OpusAudioFrame { + pub data: Bytes, + pub timestamp_ms: u32, +} +``` + +--- + +## Password Hashing — `crates/server/src/hash.rs` + +- `hash_password(password)` — Argon2id with random `OsRng` salt, returns PHC-format string. +- `verify_password(password, hash)` — parses PHC string, verifies with Argon2. + **Panics** if `hash` is not valid PHC format (uses `.unwrap()` on `PasswordHash::new`). + +--- + +## Error Handling — `crates/server/src/http_error.rs` + +`#[derive(Error, Debug)] pub enum HttpError { ... }` + +Maps to HTTP status codes: +- `DbErr`/`Hash`/`Internal` → 500 +- `NotFound` → 404 +- `Unauthorized` → 401 +- `Forbidden` → 403 +- `Conflict` → 409 +- `BadRequest` → 400 +- `Unprocessable` → 422 +- `NotAcceptable` → 406 + +4xx errors return message body; 5xx returns empty body (no internal details leaked). + +--- + +## WebRTC Ingest (WHIP) — `crates/server/src/webrtc_ingest.rs` + +**STUB — not implemented.** Currently: +- Accepts `POST /api/whip` with SDP body. +- Parses SDP via `str0m::change::SdpOffer`. +- Logs media lines. + +--- + +## Database (SeaORM + SQLite) + +**DB file:** `./db/db.sqlite` +**Migrations:** `crates/migration/` (4 migrations) + +### Tables + +#### `users` +| Column | Type | Constraints | +|---------------------|----------|--------------------------| +| `id` | INTEGER | PK, autoincrement | +| `username` | TEXT | NOT NULL, UNIQUE | +| `hashed_password` | TEXT | NOT NULL | +| `stream_key_limit` | INTEGER | NOT NULL, default 3 | + +Relations: `has_many` → `stream_key`, `auth_session` + +Custom methods: +- `Entity::create(db, username, password_hash)` — inserts user +- `Entity::find_by_username(db, username)` — lookup +- `Entity::find_by_auth_session(db, token)` — join auth_session → user +- `ActiveModel::update_username`, `update_password`, `change_stream_key_limit` + +#### `stream_key` +| Column | Type | Constraints | +|--------------|----------|--------------------------| +| `id` | INTEGER | PK, autoincrement | +| `key_value` | TEXT | NOT NULL, UNIQUE | +| `user_id` | INTEGER | NOT NULL, FK → users | +| `label` | TEXT | NOT NULL | +| `is_active` | BOOLEAN | NOT NULL, default true | +| `is_unlisted`| BOOLEAN | NOT NULL, default true | +| `created_at` | DATETIME | NOT NULL | + +Relations: `belongs_to` → `users`, `has_many` → `stream_session` + +Custom methods: +- `Entity::create(db, user_id, key_value, label, is_unlisted)` — generates `stream-key-` as key_value +- `Entity::find_by_key(db, key_value)` — lookup by raw key string +- `Entity::find_by_user(db, user_id)` — list user's keys +- `ActiveModel::change_label_value` + +#### `stream_session` +| Column | Type | Constraints | +|----------------|----------|--------------------------| +| `id` | INTEGER | PK, autoincrement | +| `stream_key_id`| INTEGER | NOT NULL, FK → stream_key| +| `started_at` | DATETIME | NOT NULL | +| `ended_at` | DATETIME | nullable | + +Relations: `belongs_to` → `stream_key` + +Custom methods: +- `Model::create_stream_session(db, stream_key_id, started_at)` — creates open session +- `Model::get_stream_session(db, id)` — get by ID +- `Model::get_all_active_sessions(db)` — all sessions where `ended_at IS NULL` +- `Model::get_active_by_stream_key_id(db, key_id)` — single active session +- `Model::clean_unended_streams(db)` — sets `ended_at = started_at` for crash recovery +- `ActiveModel::finish_stream_session(db, ended_at)` — marks session ended + +#### `auth_session` +| Column | Type | Constraints | +|-----------|----------|--------------------------| +| `id` | INTEGER | PK, autoincrement | +| `id_user` | INTEGER | NOT NULL, FK → users | +| `value` | TEXT | UUID v4 string | + +Relations: `belongs_to` → `users` + +Custom methods: +- `Entity::create(db, user_id)` — generates UUID token, creates session +- `Entity::find_by_user_id(db, user_id)` — lookup by user + +### Conventions +- Query methods on `Entity` (e.g. `Entity::find_by_x`). +- Mutation helpers on `ActiveModel`. +- For destructive schema changes: expand-contract pattern (add → backfill → switch code → drop old). + +--- + +## Dependencies + +### Server (`crates/server/Cargo.toml`) + +| Crate | Version | Purpose | +|---------------|---------|-----------------------------| +| `tokio` | 1 | Async runtime (full) | +| `str0m` | 0.20.0 | WebRTC | +| `rml_rtmp` | 0.8.0 | RTMP server | +| `axum` | 0.8 | HTTP framework | +| `axum-extra` | 0.12.6 | Cookie extraction | +| `sea-orm` | 1 | ORM (SQLite, tokio-rustls) | +| `symphonia` | 0.5 | Audio decode (AAC) | +| `opus` | 0.3.1 | Audio encode (Opus) | +| `rubato` | 3.0.0 | Audio resampling | +| `argon2` | 0.5.3 | Password hashing | +| `uuid` | 1.23.3 | UUID v4 | +| `dashmap` | 6.2.1 | Concurrent map | +| `async-broadcast` | 0.7.2 | Ring buffer channels | +| `tower-http` | 0.6 | CORS | +| `regex` | 1 | Stream key label validation | +| `thiserror` | 2.0.18 | Error types | +| `tracing` | 0.1 | Structured logging | +| `chrono` | 0.4.45 | Date/time | +| `bytes` | 1 | Byte buffers | +| `futures` | 0.3.32 | async block_on | +| `serde` | 1.0.228 | Serialization | + +### Entity (`crates/entity/Cargo.toml`) + +| Crate | Version | Purpose | +|---------------|---------|----------------| +| `sea-orm` | 1 | ORM macros | +| `serde` | 1 | Serialization | +| `chrono` | 0.4 | DateTime | +| `rand` | 0.10.1 | Random salts | +| `argon2` | 0.5.3 | Password hash | +| `uuid` | 1.23.3 | UUID v4 | + +### Migration (`crates/migration/Cargo.toml`) + +| Crate | Version | Purpose | +|--------------------|---------|----------------| +| `sea-orm-migration`| 1 | Migration CLI | +| `tokio` | 1 | Async runtime | + +--- + +## Build & Release + +### Profiles (root `Cargo.toml`) + +- **dev:** debug = true +- **release:** lto = true, codegen-units = 1, panic = "abort" +- **flamegraph:** inherits release + debug + force-frame-pointers + +### Cross-compilation + +- x86_64: clang linker + mold (via rustflags). +- aarch64: cross-toolchain (`aarch64-linux-gnu-gcc`), no mold. + +### Docker + +- x86_64: multi-stage, strips binary, copies only needed .so deps. +- aarch64: QEMU cross-build from x86_64 builder, copies arm64 .so deps. +- Both use `scratch` final stage. +- Exposes: 1935/tcp, 3000/tcp, 6969/udp. + +### Nix + +- Uses `crane` for cargo builds. +- Dev shell: clang, mold, sea-orm-cli, cmake, opus, pkgconf, just. +- ADMIN_REF_CODE set in dev shell: `meowmeowpurrrmeow`. + +--- + +## WHEP Test Page — `index.html` + +- Connects to `http://localhost:5000/whep/test` (note: hardcoded to port 5000, not 3000). +- Creates RTCPeerConnection with video recvonly transceiver. +- Sends WHEP offer POST, receives SDP answer. +- Streams video to `