diff --git a/CLAUDE.md b/CLAUDE.md index 417c454..6da9a2e 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -20,84 +20,114 @@ 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-WHEP bridge: it accepts an RTMP video publish stream and re-streams it to browsers via WebRTC (using the WHEP signaling protocol). +This is an RTMP-to-WHEP bridge: accepts an RTMP video publish stream and re-streams it to browsers via WebRTC (WHEP signaling protocol). **Signal flow:** ``` OBS/encoder → RTMP (port 8123) → H264Parser → async_broadcast channel ↓ -Browser ← WebRTC/UDP ← str0m Rtc ← WHEP HTTP (port 5000) +Browser ← WebRTC/UDP ← str0m Rtc ← WHEP HTTP (port 3000) ``` -### RTMP ingestion (port 8123) +**Runtime:** tokio (not smol). The HTTP server runs as a tokio task using hyper. -Each incoming TCP connection is handled in a detached smol task: +### RTMP ingestion (port 8123) — `crates/server/src/main.rs` -1. **Handshake** — reads C0+C1 (1537 bytes) using `rml_rtmp::Handshake`, sends S0+S1+S2, reads C2 (1536 bytes) to complete the handshake. -2. **Session setup** — creates an `rml_rtmp::ServerSession` and writes its initial response bytes to the stream. -3. **Event loop** — reads 4096-byte chunks and calls `rtmp_session.handle_input`. The library returns a mix of `OutboundResponse` packets (written back immediately) and `RaisedEvent` values: +Each incoming TCP connection is handled in a detached tokio task: + +1. **Handshake** — reads C0+C1 (1537 bytes) via `rml_rtmp::Handshake`, sends S0+S1+S2, reads C2 (1536 bytes). +2. **Session setup** — creates an `rml_rtmp::ServerSession`, writes its initial response bytes. +3. **Event loop** — reads 4096-byte chunks, calls `rtmp_session.handle_input`. Handles: - `ConnectionRequested` → accepted unconditionally. - - `PublishStreamRequested` → accepted only if `stream_key == "test"`, rejected otherwise. On accept, a `StreamSession` holding an `async_broadcast::Sender>` is inserted into the `AppState` `DashMap`. - - `VideoDataReceived` → checked for HEVC (bytes 1–4 == `hvc1`; connection dropped if true), then passed to `H264Parser::parse`. If a frame is returned it is broadcast on the channel. - - `PublishStreamFinished` → the entry is removed from the `DashMap`. + - `PublishStreamRequested` → accepted for any stream key; a `StreamSession` with an `async_broadcast::Sender>` is inserted into `AppState`. + - `VideoDataReceived` → HEVC check (bytes 1–4 == `hvc1`; drops connection if true), then passed to `H264Parser::parse`. Parsed frames are broadcast on the channel. + - `PublishStreamFinished` → entry removed from `AppState`. -### H.264 parsing (AVCC → Annex-B) +`AppState` is a `Arc>` wrapping a `DashMap`. -`H264Parser` processes the raw `VideoDataReceived` payload bytes: +### HTTP API (port 3000) — `crates/server/src/http.rs` -- **Byte 0**: upper nibble = frame type (1 = keyframe), lower nibble = codec ID (7 = H.264; anything else is dropped). -- **Byte 1**: AVC packet type — `0` = sequence header (AVCDecoderConfigurationRecord), `1` = NAL unit data. -- **Bytes 2–4**: composition time offset (ignored). -- **Bytes 5+**: payload. +Async hyper server running in a tokio task. Routes: -For packet type `0` the parser walks the `AVCDecoderConfigurationRecord` to cache the raw SPS and PPS byte arrays. +- `GET /api/catalog` — returns JSON `{ active_streams: [String] }` listing currently publishing stream keys. +- `GET /api/meow` — returns `"meow"` (health check / placeholder). +- WHEP signaling is not yet wired into this HTTP server (see webrtc.rs for the channel plumbing). -For packet type `1` the parser converts from AVCC (each NALU preceded by a 4-byte big-endian length) to Annex-B (each NALU preceded by the `00 00 00 01` start code). Before the first NALU of every keyframe it prepends the cached `SPS` and `PPS` in Annex-B form so that str0m's RTP packetizer can bundle them into a STAP-A alongside the IDR NALU. +SDP offer/answer exchange uses `tokio::sync::mpsc` channels between HttpServer and the Webrtc task. -### WHEP signaling (port 5000) +### WebRTC negotiation and media loop — `crates/server/src/webrtc.rs` -The browser opens a `RTCPeerConnection`, adds a `recvonly` video transceiver, calls `createOffer`, and POSTs the SDP to `POST /whep/test`. +The `Webrtc` task receives `(stream_key, sdp_body)` tuples from `offer_rx`: -`tiny_http` (running in an OS thread) receives the request, sends the tuple `("", sdp_body)` on `offer_tx`, then **blocks** on `accept_rx` waiting for the answer SDP. The 201 response with `Content-Type: application/sdp` is sent once the answer arrives. - -### WebRTC negotiation and media loop - -The smol `Webrtc` task receives the offer from `offer_rx`: - -1. Binds a UDP socket to `127.0.0.1:0` — this is the only ICE candidate advertised (host, UDP, loopback). Remote candidates coming from the browser are handled by str0m internally; the socket just needs to be reachable from the browser on the same machine. -2. Builds an `Rtc` with H.264 explicitly configured for payload types 102, 104, and 106 (profiles `0x42e01f`, `0x4d001f`, `0x64001f`). The default H.264 support is disabled first so only these three PTs are offered. -3. Adds a `SendOnly` video media track, then calls `changes.accept_offer(offer_sdp)` to produce the SDP answer. -4. Sends the answer back on `accept_tx` (unblocking the HTTP thread), then detaches a per-connection async loop. +1. Binds a UDP socket to `127.0.0.1:0` — only ICE candidate advertised (host, UDP, loopback). +2. Builds `str0m::Rtc` with H.264 explicitly configured for PTs 102, 104, 106 (profiles `0x42e01f`, `0x4d001f`, `0x64001f`). Default H.264 support is disabled first. +3. Adds a `SendOnly` video media track, calls `changes.accept_offer(offer_sdp)` to produce the SDP answer. +4. Sends the answer back on `accept_tx`, then spawns a per-connection tokio task. **Per-connection loop** (`Webrtc::detach_connection`): -- Calls `rtc.poll_output()` in a tight loop until it returns `Output::Timeout(deadline)`. Each iteration either sends a UDP datagram (`Output::Transmit`) or handles an event: - - `Event::MediaAdded` — iterates the writer's payload params and picks the PT with the highest `profile_level_id`, storing it in `video_pt`. - - `Event::Connected` — sets `connected = true`; media sending begins after this. -- Once connected, on each iteration it lazily subscribes to the `async_broadcast` channel for stream key `"test"` (if not already subscribed), then drains all available frames with `try_recv` and writes each one via `writer.write(pt, now, rtp_time, frame.data)`. The `rtp_time` is constructed from `frame.timestamp_ms` as `MediaTime::from_millis`. -- Then waits with `smol::future::or` for either the str0m deadline or a UDP datagram. Incoming datagrams are fed to `rtc.handle_input(Input::Receive(...))` for ICE/DTLS/RTCP processing; a timeout fires `rtc.handle_input(Input::Timeout(...))`. +- Drains `rtc.poll_output()` until `Output::Timeout`. Each iteration sends UDP datagrams (`Output::Transmit`) or handles events: + - `Event::MediaAdded` — picks the PT with the highest `profile_level_id`, stores in `video_pt`. + - `Event::Connected` — sets `connected = true`; media sending begins. +- When connected, lazily subscribes to the `async_broadcast` channel for the stream key, then drains up to 8 frames per iteration via `try_recv`, writing each with `writer.write(pt, now, rtp_time, frame.data)`. `rtp_time` is computed as `MediaTime::from_90khz(timestamp_ms * 90)`. +- Waits (capped at 20 ms) with `tokio::select!` for either the str0m deadline or a UDP datagram. Incoming datagrams are fed to `rtc.handle_input(Input::Receive(...))`. -**Module breakdown:** +### H.264 parsing — `crates/server/src/media.rs` -- `main.rs` — Entry point. Owns `AppState` (a `DashMap`). Spawns the HTTP and WebRTC tasks, then loops accepting RTMP TCP connections. Each RTMP connection runs the `rml_rtmp` handshake and session, parses H.264 frames via `H264Parser`, and broadcasts them on a per-stream `async_broadcast` channel stored in `AppState`. +`H264Parser` converts raw RTMP `VideoDataReceived` payloads (AVCC) to Annex-B: -- `src/http.rs` — Blocking `tiny_http` server on port 5000 running in its own OS thread. Handles CORS and WHEP `POST /whep/*` requests. Forwards the SDP offer to the WebRTC task via `smol::channel` and blocks waiting for the SDP answer before responding. +- **Byte 0**: upper nibble = frame type (1 = keyframe), lower nibble = codec ID (7 = H.264; anything else dropped). +- **Byte 1**: AVC packet type — `0` = sequence header, `1` = NAL unit data. +- **Bytes 5+**: payload. -- `src/webrtc.rs` — Async task (smol) that receives SDP offers, builds a `str0m` `Rtc` instance with H.264 codec config, negotiates the answer, and detaches a per-connection loop. That loop polls `str0m` for output (packets to send), listens on a per-connection UDP socket for incoming DTLS/ICE, and drains the `async_broadcast` video channel to write Annex-B frames into the `str0m` writer. +Packet type `0` walks `AVCDecoderConfigurationRecord` to cache SPS and PPS byte arrays. -- `src/media.rs` — `H264Parser` converts raw RTMP `VideoDataReceived` payloads (AVCC format) to Annex-B. Sequence-header packets (AVC packet type 0) update cached SPS/PPS; NAL unit packets (type 1) prepend SPS+PPS before each keyframe and convert length-prefixed NALUs to start-code NALUs. +Packet type `1` converts AVCC (4-byte big-endian length prefix per NALU) to Annex-B (`00 00 00 01` start code). Before the first NALU of every keyframe, prepends SPS+PPS in Annex-B form. -- `src/rtmp.rs` — An early stub for a manual RTMP handshake implementation, unused in the current flow (the actual RTMP handling uses `rml_rtmp` directly in `main.rs`). +### Unused stub — `crates/server/src/rtmp.rs` -**Key design choices:** +Early manual RTMP handshake implementation, not used in the current flow. -- Only the hardcoded stream key `"test"` is accepted; other keys are rejected. -- HEVC/H.265 is explicitly rejected at ingestion time. -- `async_broadcast` channels are overflow-enabled (old frames are silently dropped if no subscriber drains fast enough). -- The runtime is `smol` (not tokio). The HTTP server runs in a dedicated OS thread (`std::thread::spawn`) because `tiny_http` is synchronous. -- `str0m` handles RTP packetization, DTLS, and ICE internally; the code only provides Annex-B video bytes and a UDP socket. +## Database (SeaORM + SQLite) + +DB file: `stream.db`. 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(db, username)` — updates username +- `ActiveModel::update_password(db, hashed_password)` — updates password hash +- Passwords must be hashed before being passed to these methods + +**`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` + +**`stream_session`** (`crates/entity/src/stream_session.rs`) +- Fields: `id`, `stream_key_id`, `started_at`, `ended_at` (nullable) +- Relations: `belongs_to` → `stream_key` + +### SeaORM conventions + +- Query methods go on `Entity` (e.g. `Entity::find_by_x`). +- Mutation helpers that intercept save logic (e.g. setting timestamps, pre-save transforms) go on `ActiveModel`. +- For destructive schema changes in prod, use expand-contract: add new structure → backfill → switch app code → drop old structure in a later migration. **Test page:** `index.html` — open in a browser to view the stream via WHEP without any extra tooling. diff --git a/Cargo.lock b/Cargo.lock index 1cf9c6a..d7fc0c6 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -120,6 +120,18 @@ version = "1.0.102" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "7f202df86484c868dbad7eaa557ef785d5c66295e41b460ef922eca0723b842c" +[[package]] +name = "argon2" +version = "0.5.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "3c3610892ee6e0cbce8ae2700349fcf8f98adb0dbfbee85aec3c9179d29cc072" +dependencies = [ + "base64ct", + "blake2", + "cpufeatures 0.2.17", + "password-hash", +] + [[package]] name = "arrayvec" version = "0.7.6" @@ -254,6 +266,58 @@ dependencies = [ "fs_extra", ] +[[package]] +name = "axum" +version = "0.8.9" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "31b698c5f9a010f6573133b09e0de5408834d0c82f8d7475a89fc1867a71cd90" +dependencies = [ + "axum-core", + "bytes", + "form_urlencoded", + "futures-util", + "http", + "http-body", + "http-body-util", + "hyper", + "hyper-util", + "itoa", + "matchit", + "memchr", + "mime", + "percent-encoding", + "pin-project-lite", + "serde_core", + "serde_json", + "serde_path_to_error", + "serde_urlencoded", + "sync_wrapper", + "tokio", + "tower", + "tower-layer", + "tower-service", + "tracing", +] + +[[package]] +name = "axum-core" +version = "0.5.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "08c78f31d7b1291f7ee735c1c6780ccde7785daae9a9206026862dab7d8792d1" +dependencies = [ + "bytes", + "futures-core", + "http", + "http-body", + "http-body-util", + "mime", + "pin-project-lite", + "sync_wrapper", + "tower-layer", + "tower-service", + "tracing", +] + [[package]] name = "base16ct" version = "0.2.0" @@ -316,6 +380,15 @@ dependencies = [ "wyz", ] +[[package]] +name = "blake2" +version = "0.10.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "46502ad458c9a52b69d4d4d32775c788b7a1b85e8bc9d482d92250fc0e3f8efe" +dependencies = [ + "digest 0.10.7", +] + [[package]] name = "block-buffer" version = "0.9.0" @@ -452,8 +525,10 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "1aa79e62e7697b8e29b513a68abacf485adcd1fe8284a4316c5ae868e6633327" dependencies = [ "iana-time-zone", + "js-sys", "num-traits", "serde", + "wasm-bindgen", "windows-link", ] @@ -547,16 +622,6 @@ version = "0.9.6" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "c2459377285ad874054d797f3ccebf984978aa39129f6eafde5cdc8315b612f8" -[[package]] -name = "core-foundation" -version = "0.9.4" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "91e195e091a93c46f7102ec7818a2aa394e1e1771c3ab4825963fa03e45afb8f" -dependencies = [ - "core-foundation-sys", - "libc", -] - [[package]] name = "core-foundation-sys" version = "0.8.7" @@ -846,8 +911,12 @@ dependencies = [ name = "entity" version = "0.1.0" dependencies = [ + "argon2", + "chrono", + "rand 0.10.1", "sea-orm", "serde", + "uuid", ] [[package]] @@ -1098,25 +1167,6 @@ version = "0.3.3" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "0cc23270f6e1808e30a928bdc84dea0b9b4136a8bc82338574f23baf47bbd280" -[[package]] -name = "h2" -version = "0.4.15" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "6cb093c84e8bd9b188d4c4a8cb6579fc016968d14c99882163cd3ff402a4f155" -dependencies = [ - "atomic-waker", - "bytes", - "fnv", - "futures-core", - "futures-sink", - "http", - "indexmap", - "slab", - "tokio", - "tokio-util", - "tracing", -] - [[package]] name = "hashbrown" version = "0.12.3" @@ -1268,7 +1318,6 @@ dependencies = [ "bytes", "futures-channel", "futures-core", - "h2", "http", "http-body", "httparse", @@ -1277,7 +1326,6 @@ dependencies = [ "pin-project-lite", "smallvec", "tokio", - "want", ] [[package]] @@ -1286,24 +1334,13 @@ version = "0.1.20" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "96547c2556ec9d12fb1578c4eaf448b04993e7fb79cbaad930a656880a6bdfa0" dependencies = [ - "base64", "bytes", - "futures-channel", - "futures-util", "http", "http-body", "hyper", - "ipnet", - "libc", - "percent-encoding", "pin-project-lite", - "socket2", - "system-configuration", "tokio", - "tower-layer", "tower-service", - "tracing", - "windows-registry", ] [[package]] @@ -1477,12 +1514,6 @@ dependencies = [ "generic-array", ] -[[package]] -name = "ipnet" -version = "2.12.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "d98f6fed1fde3f8c21bc40a1abb88dd75e67924f9cffc3ef95607bad8017f8e2" - [[package]] name = "is" version = "0.9.1" @@ -1620,6 +1651,12 @@ dependencies = [ "regex-automata", ] +[[package]] +name = "matchit" +version = "0.8.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "47e1ffaa40ddd1f3ed91f717a33c8c0ee23fff369e3aa8772b9605cc1d22f4c3" + [[package]] name = "md-5" version = "0.10.6" @@ -1653,6 +1690,12 @@ dependencies = [ "tokio", ] +[[package]] +name = "mime" +version = "0.3.17" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "6877bb514081ee2a7ff5ef9de3281f14a4dd4bceac4c09388074a6b5df8a139a" + [[package]] name = "minimal-lexical" version = "0.2.1" @@ -1853,6 +1896,17 @@ dependencies = [ "windows-link", ] +[[package]] +name = "password-hash" +version = "0.5.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "346f04948ba92c43e8469c1ee6736c7563d71012b17d40745260fe106aac2166" +dependencies = [ + "base64ct", + "rand_core 0.6.4", + "subtle", +] + [[package]] name = "pem-rfc7468" version = "0.7.0" @@ -2623,6 +2677,17 @@ dependencies = [ "zmij", ] +[[package]] +name = "serde_path_to_error" +version = "0.1.20" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "10a9ff822e371bb5403e391ecd83e182e0e77ba7f6fe0160b795797109d1b457" +dependencies = [ + "itoa", + "serde", + "serde_core", +] + [[package]] name = "serde_urlencoded" version = "0.7.1" @@ -2639,18 +2704,21 @@ dependencies = [ name = "server" version = "0.1.0" dependencies = [ + "argon2", "async-broadcast", + "axum", "bytes", "dashmap", - "http-body-util", - "hyper", - "hyper-util", + "entity", + "migration", "rand 0.10.1", "rml_rtmp", + "sea-orm", "serde", "serde_json", "str0m", "tokio", + "uuid", ] [[package]] @@ -3092,6 +3160,12 @@ dependencies = [ "unicode-ident", ] +[[package]] +name = "sync_wrapper" +version = "1.0.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0bf256ce5efdfa370213c1dabab5935a12e49f2c58d15e9eac2870d3b4f27263" + [[package]] name = "synstructure" version = "0.13.2" @@ -3103,27 +3177,6 @@ dependencies = [ "syn 2.0.117", ] -[[package]] -name = "system-configuration" -version = "0.7.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "a13f3d0daba03132c0aa9767f98351b3488edc2c100cda2d2ec2b04f3d8d3c8b" -dependencies = [ - "bitflags", - "core-foundation", - "system-configuration-sys", -] - -[[package]] -name = "system-configuration-sys" -version = "0.6.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "8e1d1b10ced5ca923a1fcb8d03e96b8d3268065d724548c0211415ff6ac6bac4" -dependencies = [ - "core-foundation-sys", - "libc", -] - [[package]] name = "tap" version = "1.0.1" @@ -3274,19 +3327,6 @@ dependencies = [ "tokio", ] -[[package]] -name = "tokio-util" -version = "0.7.18" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "9ae9cec805b01e8fc3fd2fe289f89149a9b66dd16786abd8b19cfa7b48cb0098" -dependencies = [ - "bytes", - "futures-core", - "futures-sink", - "pin-project-lite", - "tokio", -] - [[package]] name = "toml_datetime" version = "1.1.1+spec-1.1.0" @@ -3317,6 +3357,22 @@ dependencies = [ "winnow", ] +[[package]] +name = "tower" +version = "0.5.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ebe5ef63511595f1344e2d5cfa636d973292adc0eec1f0ad45fae9f0851ab1d4" +dependencies = [ + "futures-core", + "futures-util", + "pin-project-lite", + "sync_wrapper", + "tokio", + "tower-layer", + "tower-service", + "tracing", +] + [[package]] name = "tower-layer" version = "0.3.3" @@ -3376,12 +3432,6 @@ dependencies = [ "tracing-core", ] -[[package]] -name = "try-lock" -version = "0.2.5" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "e421abadd41a4225275504ea4d6566923418b7f05506fbc9c0fe86ba7396114b" - [[package]] name = "typenum" version = "1.20.1" @@ -3463,6 +3513,7 @@ version = "1.23.3" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "144d6b123cef80b301b8f72a9e2ca4370ddec21950d0a103dd22c437006d2db7" dependencies = [ + "getrandom 0.4.2", "js-sys", "serde_core", "wasm-bindgen", @@ -3480,15 +3531,6 @@ version = "0.9.5" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "0b928f33d975fc6ad9f86c8f283853ad26bdd5b10b7f1542aa2fa15e2289105a" -[[package]] -name = "want" -version = "0.3.1" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "bfa7760aed19e106de2c7c0b581b509f2f25d3dacaf737cb82ac61bc6d760b0e" -dependencies = [ - "try-lock", -] - [[package]] name = "wasi" version = "0.11.1+wasi-snapshot-preview1" @@ -3690,17 +3732,6 @@ version = "0.2.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "f0805222e57f7521d6a62e36fa9163bc891acd422f971defe97d64e70d0a4fe5" -[[package]] -name = "windows-registry" -version = "0.6.1" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "02752bf7fbdcce7f2a27a742f798510f3e5ad88dbe84871e5168e2120c3d5720" -dependencies = [ - "windows-link", - "windows-result", - "windows-strings", -] - [[package]] name = "windows-result" version = "0.4.1" diff --git a/Cargo.toml b/Cargo.toml index 969ad2e..223e059 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -1,3 +1,9 @@ +[workspace.metadata.crane] +name = "rtmp-to-whip" + +[workspace.package] +version = "0.1.0" + [workspace] members = ["crates/server", "crates/entity", "crates/migration"] resolver = "2" diff --git a/crates/entity/Cargo.toml b/crates/entity/Cargo.toml index 4c81a5f..322c1a6 100644 --- a/crates/entity/Cargo.toml +++ b/crates/entity/Cargo.toml @@ -4,5 +4,9 @@ version = "0.1.0" edition = "2024" [dependencies] -sea-orm = { version = "1", features = ["sqlx-sqlite", "runtime-tokio-rustls", "macros"] } +sea-orm = { version = "1", features = [ "sqlx-sqlite", "runtime-tokio-rustls", "macros" ] } serde = { version = "1", features = ["derive"] } +chrono = { version = "0.4", features = ["serde"] } +rand = "0.10.1" +argon2 = "0.5.3" +uuid = { version = "1.23.3", features = ["v4"] } diff --git a/crates/entity/src/auth_session.rs b/crates/entity/src/auth_session.rs new file mode 100644 index 0000000..394ba48 --- /dev/null +++ b/crates/entity/src/auth_session.rs @@ -0,0 +1,54 @@ +use sea_orm::{ActiveValue::Set, entity::prelude::*}; +use serde::{Deserialize, Serialize}; + +#[derive(Clone, Debug, PartialEq, DeriveEntityModel, Serialize, Deserialize)] +#[sea_orm(table_name = "auth_session")] +pub struct Model { + #[sea_orm(primary_key)] + pub id: i32, + pub id_user: i32, + pub value: String, +} + +#[derive(Copy, Clone, Debug, EnumIter, DeriveRelation)] +pub enum Relation { + #[sea_orm( + belongs_to = "super::users::Entity", + from = "Column::IdUser", + to = "super::users::Column::Id" + )] + Users, +} + +impl Related for Entity { + fn to() -> RelationDef { + Relation::Users.def() + } +} + +impl ActiveModelBehavior for ActiveModel {} + +impl Entity { + pub async fn create(db: &DatabaseConnection, user_id: i32) -> Result { + let value = uuid::Uuid::new_v4().to_string(); + + ActiveModel { + id_user: Set(user_id), + value: Set(value), + ..Default::default() + } + .insert(db) + .await + } + pub async fn find_by_user_id( + db: &DatabaseConnection, + user_id: i32, + ) -> Result, DbErr> { + Entity::find() + .filter(Column::IdUser.eq(user_id)) + .one(db) + .await + } +} + +impl ActiveModel {} diff --git a/crates/entity/src/lib.rs b/crates/entity/src/lib.rs index 611a777..0e18543 100644 --- a/crates/entity/src/lib.rs +++ b/crates/entity/src/lib.rs @@ -1,3 +1,5 @@ +pub mod auth_session; pub mod prelude; pub mod stream_key; pub mod stream_session; +pub mod users; diff --git a/crates/entity/src/prelude.rs b/crates/entity/src/prelude.rs index 152f298..da25cbe 100644 --- a/crates/entity/src/prelude.rs +++ b/crates/entity/src/prelude.rs @@ -1,2 +1,4 @@ +pub use super::auth_session::Entity as AuthSession; pub use super::stream_key::Entity as StreamKey; pub use super::stream_session::Entity as StreamSession; +pub use super::users::Entity as Users; diff --git a/crates/entity/src/stream_key.rs b/crates/entity/src/stream_key.rs index 25631e6..46bace5 100644 --- a/crates/entity/src/stream_key.rs +++ b/crates/entity/src/stream_key.rs @@ -1,4 +1,4 @@ -use sea_orm::entity::prelude::*; +use sea_orm::{Set, entity::prelude::*}; use serde::{Deserialize, Serialize}; #[derive(Clone, Debug, PartialEq, DeriveEntityModel, Serialize, Deserialize)] @@ -8,8 +8,10 @@ pub struct Model { pub id: i32, #[sea_orm(unique)] pub key_value: String, + pub user_id: i32, pub label: String, pub is_active: bool, + pub is_unlisted: bool, pub created_at: DateTimeUtc, } @@ -17,6 +19,12 @@ pub struct Model { pub enum Relation { #[sea_orm(has_many = "super::stream_session::Entity")] StreamSession, + #[sea_orm( + belongs_to = "super::users::Entity", + from = "Column::UserId", + to = "super::users::Column::Id" + )] + User, } impl Related for Entity { @@ -25,4 +33,49 @@ impl Related for Entity { } } +impl Related for Entity { + fn to() -> RelationDef { + Relation::User.def() + } +} + impl ActiveModelBehavior for ActiveModel {} + +impl Entity { + pub async fn create( + db: &DatabaseConnection, + user_id: i32, + key_value: String, + label: String, + is_unlisted: bool, + ) -> Result { + ActiveModel { + user_id: Set(user_id), + key_value: Set(key_value), + label: Set(label), + is_active: Set(false), + is_unlisted: Set(is_unlisted), + created_at: Set(chrono::Utc::now()), + ..Default::default() + } + .insert(db) + .await + } + + pub async fn find_by_key( + db: &DatabaseConnection, + key_value: &str, + ) -> Result, DbErr> { + Entity::find() + .filter(Column::KeyValue.eq(key_value)) + .one(db) + .await + } + + pub async fn find_by_user(db: &DatabaseConnection, user_id: i32) -> Result, DbErr> { + Entity::find() + .filter(Column::UserId.eq(user_id)) + .all(db) + .await + } +} diff --git a/crates/entity/src/stream_session.rs b/crates/entity/src/stream_session.rs index 73784b0..a5f4ab2 100644 --- a/crates/entity/src/stream_session.rs +++ b/crates/entity/src/stream_session.rs @@ -1,4 +1,8 @@ -use sea_orm::entity::prelude::*; +use sea_orm::{ + ActiveValue::{NotSet, Set}, + entity::prelude::*, + sqlx::types::chrono::{self, Utc}, +}; use serde::{Deserialize, Serialize}; #[derive(Clone, Debug, PartialEq, DeriveEntityModel, Serialize, Deserialize)] @@ -28,3 +32,42 @@ impl Related for Entity { } impl ActiveModelBehavior for ActiveModel {} + +impl Model { + pub async fn create_stream_session( + db: &DatabaseConnection, + stream_key_id: i32, + started_at: chrono::DateTime, + ) -> Result { + ActiveModel { + id: NotSet, + stream_key_id: Set(stream_key_id), + started_at: Set(started_at), + ..Default::default() + } + .insert(db) + .await + } + pub async fn get_stream_session( + db: &DatabaseConnection, + stream_session_id: i32, + ) -> Result { + Entity::find_by_id(stream_session_id) + .one(db) + .await? + .ok_or(DbErr::RecordNotFound(format!( + "stream_session {stream_session_id}" + ))) + } +} + +impl ActiveModel { + pub async fn finish_stream_session( + mut self, + db: &DatabaseConnection, + ended_at: chrono::DateTime, + ) -> Result { + self.ended_at = Set(Some(ended_at)); + self.update(db).await + } +} diff --git a/crates/entity/src/users.rs b/crates/entity/src/users.rs new file mode 100644 index 0000000..ab4b6c9 --- /dev/null +++ b/crates/entity/src/users.rs @@ -0,0 +1,92 @@ +use sea_orm::{Set, entity::prelude::*}; +use serde::{Deserialize, Serialize}; + +#[derive(Clone, Debug, PartialEq, DeriveEntityModel, Serialize, Deserialize)] +#[sea_orm(table_name = "users")] +pub struct Model { + #[sea_orm(primary_key)] + pub id: i32, + pub username: String, + pub hashed_password: String, + pub stream_key_limit: i32, +} + +#[derive(Copy, Clone, Debug, EnumIter, DeriveRelation)] +pub enum Relation { + #[sea_orm(has_many = "super::stream_key::Entity")] + StreamKey, + #[sea_orm(has_many = "super::auth_session::Entity")] + AuthSession, +} + +impl Related for Entity { + fn to() -> RelationDef { + Relation::StreamKey.def() + } +} + +impl Related for Entity { + fn to() -> RelationDef { + Relation::AuthSession.def() + } +} + +impl ActiveModelBehavior for ActiveModel {} + +impl Entity { + pub async fn create( + db: &DatabaseConnection, + username: String, + password: String, + ) -> Result { + ActiveModel { + username: Set(username), + hashed_password: Set(password), + ..Default::default() + } + .insert(db) + .await + } + pub async fn find_by_auth_session( + db: &DatabaseConnection, + auth_session: String, + ) -> Result, DbErr> { + let sessions = crate::auth_session::Entity::find() + .filter(crate::auth_session::Column::Value.eq(auth_session.to_string())) + .one(db) + .await?; + if let Some(x) = sessions { + Entity::find_by_id(x.id_user).one(db).await + } else { + return Ok(None); + } + } +} + +impl ActiveModel { + pub async fn update_username( + mut self, + db: &DatabaseConnection, + username: String, + ) -> Result { + self.username = Set(username); + self.update(db).await + } + + pub async fn update_password( + mut self, + db: &DatabaseConnection, + hashed_password: String, + ) -> Result { + self.hashed_password = Set(hashed_password); + self.update(db).await + } + pub async fn change_stream_key_limit( + mut self, + db: &DatabaseConnection, + new_limit: i32, + ) -> Result { + self.stream_key_limit = Set(new_limit); + self.update(db).await + } +} diff --git a/crates/migration/src/lib.rs b/crates/migration/src/lib.rs index 3d0886a..454239f 100644 --- a/crates/migration/src/lib.rs +++ b/crates/migration/src/lib.rs @@ -1,12 +1,20 @@ pub use sea_orm_migration::prelude::*; -mod m20260616_000001_create_tables; +mod m20260616_000001_create_users; +mod m20260616_000002_create_stream_key; +mod m20260616_000003_create_stream_session; +mod m20260616_000004_create_auth_session; pub struct Migrator; #[async_trait::async_trait] impl MigratorTrait for Migrator { fn migrations() -> Vec> { - vec![Box::new(m20260616_000001_create_tables::Migration)] + vec![ + Box::new(m20260616_000001_create_users::Migration), + Box::new(m20260616_000002_create_stream_key::Migration), + Box::new(m20260616_000003_create_stream_session::Migration), + Box::new(m20260616_000004_create_auth_session::Migration), + ] } } diff --git a/crates/migration/src/m20260616_000001_create_users.rs b/crates/migration/src/m20260616_000001_create_users.rs new file mode 100644 index 0000000..4c41932 --- /dev/null +++ b/crates/migration/src/m20260616_000001_create_users.rs @@ -0,0 +1,48 @@ +use sea_orm_migration::prelude::*; + +#[derive(DeriveMigrationName)] +pub struct Migration; + +#[async_trait::async_trait] +impl MigrationTrait for Migration { + async fn up(&self, manager: &SchemaManager) -> Result<(), DbErr> { + manager + .create_table( + Table::create() + .table(Users::Table) + .if_not_exists() + .col( + ColumnDef::new(Users::Id) + .integer() + .not_null() + .auto_increment() + .primary_key(), + ) + .col( + ColumnDef::new(Users::Username) + .string() + .not_null() + .unique_key(), + ) + .col(ColumnDef::new(Users::HashedPassword).string()) + .col(ColumnDef::new(Users::StreamKeyLimit).integer().default(3)) + .to_owned(), + ) + .await + } + + async fn down(&self, manager: &SchemaManager) -> Result<(), DbErr> { + manager + .drop_table(Table::drop().table(Users::Table).to_owned()) + .await + } +} + +#[derive(Iden)] +pub enum Users { + Table, + Id, + Username, + HashedPassword, + StreamKeyLimit, +} diff --git a/crates/migration/src/m20260616_000001_create_tables.rs b/crates/migration/src/m20260616_000002_create_stream_key.rs similarity index 54% rename from crates/migration/src/m20260616_000001_create_tables.rs rename to crates/migration/src/m20260616_000002_create_stream_key.rs index 507928b..b101f8f 100644 --- a/crates/migration/src/m20260616_000001_create_tables.rs +++ b/crates/migration/src/m20260616_000002_create_stream_key.rs @@ -1,5 +1,7 @@ use sea_orm_migration::prelude::*; +use super::m20260616_000001_create_users::Users; + #[derive(DeriveMigrationName)] pub struct Migration; @@ -18,6 +20,7 @@ impl MigrationTrait for Migration { .auto_increment() .primary_key(), ) + .col(ColumnDef::new(StreamKey::UserId).integer().not_null()) .col( ColumnDef::new(StreamKey::KeyValue) .string() @@ -31,73 +34,38 @@ impl MigrationTrait for Migration { .not_null() .default(true), ) - .col(ColumnDef::new(StreamKey::CreatedAt).date_time().not_null()) - .to_owned(), - ) - .await?; - - manager - .create_table( - Table::create() - .table(StreamSession::Table) - .if_not_exists() .col( - ColumnDef::new(StreamSession::Id) - .integer() + ColumnDef::new(StreamKey::IsUnlisted) + .boolean() .not_null() - .auto_increment() - .primary_key(), + .default(true), ) - .col(ColumnDef::new(StreamSession::StreamKeyId).integer().not_null()) - .col(ColumnDef::new(StreamSession::StartedAt).date_time().not_null()) - .col(ColumnDef::new(StreamSession::EndedAt).date_time().null()) + .col(ColumnDef::new(StreamKey::CreatedAt).date_time().not_null()) .foreign_key( ForeignKey::create() - .from(StreamSession::Table, StreamSession::StreamKeyId) - .to(StreamKey::Table, StreamKey::Id), + .from(StreamKey::Table, StreamKey::UserId) + .to(Users::Table, Users::Id), ) .to_owned(), ) - .await?; - - // Seed the default stream key so OBS can connect out of the box - manager - .get_connection() - .execute_unprepared( - "INSERT OR IGNORE INTO stream_key (key_value, label, is_active, created_at) \ - VALUES ('test', 'Default', 1, datetime('now'))", - ) - .await?; - - Ok(()) + .await } async fn down(&self, manager: &SchemaManager) -> Result<(), DbErr> { - manager - .drop_table(Table::drop().table(StreamSession::Table).to_owned()) - .await?; manager .drop_table(Table::drop().table(StreamKey::Table).to_owned()) - .await?; - Ok(()) + .await } } #[derive(Iden)] -enum StreamKey { +pub enum StreamKey { Table, Id, + UserId, KeyValue, Label, IsActive, + IsUnlisted, CreatedAt, } - -#[derive(Iden)] -enum StreamSession { - Table, - Id, - StreamKeyId, - StartedAt, - EndedAt, -} diff --git a/crates/migration/src/m20260616_000003_create_stream_session.rs b/crates/migration/src/m20260616_000003_create_stream_session.rs new file mode 100644 index 0000000..318d12a --- /dev/null +++ b/crates/migration/src/m20260616_000003_create_stream_session.rs @@ -0,0 +1,58 @@ +use sea_orm_migration::prelude::*; + +use super::m20260616_000002_create_stream_key::StreamKey; + +#[derive(DeriveMigrationName)] +pub struct Migration; + +#[async_trait::async_trait] +impl MigrationTrait for Migration { + async fn up(&self, manager: &SchemaManager) -> Result<(), DbErr> { + manager + .create_table( + Table::create() + .table(StreamSession::Table) + .if_not_exists() + .col( + ColumnDef::new(StreamSession::Id) + .integer() + .not_null() + .auto_increment() + .primary_key(), + ) + .col( + ColumnDef::new(StreamSession::StreamKeyId) + .integer() + .not_null(), + ) + .col( + ColumnDef::new(StreamSession::StartedAt) + .date_time() + .not_null(), + ) + .col(ColumnDef::new(StreamSession::EndedAt).date_time().null()) + .foreign_key( + ForeignKey::create() + .from(StreamSession::Table, StreamSession::StreamKeyId) + .to(StreamKey::Table, StreamKey::Id), + ) + .to_owned(), + ) + .await + } + + async fn down(&self, manager: &SchemaManager) -> Result<(), DbErr> { + manager + .drop_table(Table::drop().table(StreamSession::Table).to_owned()) + .await + } +} + +#[derive(Iden)] +pub enum StreamSession { + Table, + Id, + StreamKeyId, + StartedAt, + EndedAt, +} diff --git a/crates/migration/src/m20260616_000004_create_auth_session.rs b/crates/migration/src/m20260616_000004_create_auth_session.rs new file mode 100644 index 0000000..0ea66db --- /dev/null +++ b/crates/migration/src/m20260616_000004_create_auth_session.rs @@ -0,0 +1,48 @@ +use sea_orm_migration::prelude::*; + +use crate::m20260616_000001_create_users::Users; + +#[derive(DeriveMigrationName)] +pub struct Migration; + +#[async_trait::async_trait] +impl MigrationTrait for Migration { + async fn up(&self, manager: &SchemaManager) -> Result<(), DbErr> { + manager + .create_table( + Table::create() + .table(AuthSession::Table) + .if_not_exists() + .col( + ColumnDef::new(AuthSession::Id) + .integer() + .not_null() + .auto_increment() + .primary_key(), + ) + .col(ColumnDef::new(AuthSession::IdUser).integer().not_null()) + .col(ColumnDef::new(AuthSession::Value).string()) + .foreign_key( + ForeignKey::create() + .from(AuthSession::Table, AuthSession::IdUser) + .to(Users::Table, Users::Id), + ) + .to_owned(), + ) + .await + } + + async fn down(&self, manager: &SchemaManager) -> Result<(), DbErr> { + manager + .drop_table(Table::drop().table(AuthSession::Table).to_owned()) + .await + } +} + +#[derive(Iden)] +pub enum AuthSession { + Table, + Id, + IdUser, + Value, +} diff --git a/crates/server/Cargo.toml b/crates/server/Cargo.toml index 7d0b185..7dff8cb 100644 --- a/crates/server/Cargo.toml +++ b/crates/server/Cargo.toml @@ -9,14 +9,17 @@ path = "src/main.rs" [dependencies] async-broadcast = "0.7.2" -bytes = "1.11.1" +bytes = "1" dashmap = "6.2.1" rand = "0.10.1" rml_rtmp = "0.8.0" str0m = "0.20.0" tokio = { version = "1", features = ["full"] } -hyper = { version = "1", features = ["full"] } -http-body-util = "0.1" -hyper-util = { version = "0.1", features = ["full"] } +axum = "0.8" serde = { version = "1.0.228", features = ["serde_derive"] } serde_json = "1.0.150" +sea-orm = { version = "1", features = [ "sqlx-sqlite", "runtime-tokio-rustls", "macros" ] } +entity = {path = "../entity"} +migration = {path = "../migration"} +argon2 = "0.5.3" +uuid = { version = "1.23.3", features = ["v4"] } diff --git a/crates/server/src/hash.rs b/crates/server/src/hash.rs new file mode 100644 index 0000000..9efe8ea --- /dev/null +++ b/crates/server/src/hash.rs @@ -0,0 +1,15 @@ +use argon2::password_hash::{SaltString, rand_core::OsRng}; +use argon2::{Argon2, PasswordHash, PasswordHasher, PasswordVerifier}; + +pub fn hash_password(password: &str) -> Result { + let salt = SaltString::generate(&mut OsRng); + let hash = Argon2::default().hash_password(password.as_bytes(), &salt)?; + Ok(hash.to_string()) +} + +pub fn verify_password(password: &str, stored_hash: &str) -> bool { + let parsed = PasswordHash::new(stored_hash).unwrap(); + Argon2::default() + .verify_password(password.as_bytes(), &parsed) + .is_ok() +} diff --git a/crates/server/src/http.rs b/crates/server/src/http.rs index 88be626..89e367d 100644 --- a/crates/server/src/http.rs +++ b/crates/server/src/http.rs @@ -1,30 +1,30 @@ -use std::{ - hash::{DefaultHasher, Hash, Hasher}, - net::SocketAddr, - sync::Arc, -}; +use std::{net::SocketAddr, sync::Arc}; use axum::{ Json, Router, - body::Bytes, - extract::{Form, Query, State}, - http::StatusCode, + body::{Body, Bytes}, + extract::{Form, FromRequestParts, Path, Query, State}, + http::{HeaderMap, StatusCode, header::SET_COOKIE, request::Parts}, response::{IntoResponse, Response}, routing::{get, post}, }; +use entity::{auth_session, stream_key, users}; +use sea_orm::DatabaseConnection; use serde::{Deserialize, Serialize}; use tokio::{ net::TcpListener, sync::{Mutex, mpsc::Sender}, }; +use uuid::Uuid; -use crate::AppState; +use crate::{AppState, hash::hash_password}; pub struct HttpServer { - pub offer_tx: Sender<(i32, String, String)>, - pub accept_rx: async_broadcast::Receiver<(i32, String)>, + pub offer_tx: Sender<(i32, i32, String)>, + pub accept_rx: async_broadcast::Receiver<(i32, Option)>, pub appstate: Arc>, pub request_count: Mutex, + pub db: DatabaseConnection, } impl HttpServer { @@ -35,7 +35,12 @@ impl HttpServer { let app = Router::new() .route("/api/catalog", get(catalog_handler)) .route("/api/user", post(create_user_handler)) - .route("/api/stream", get(stream_handler)) + .route( + "/api/stream-key", + post(create_stream_key_handler).get(get_all_stream_keys), + ) + .route("/api/login", post(login_handler)) + .route("/api/stream/{slug}", post(stream_handler)) .route("/api/meow", get(meow_handler)) .with_state(state); @@ -47,17 +52,103 @@ impl HttpServer { } } -fn calculate_hash(t: &T) -> u64 { - let mut s = DefaultHasher::new(); - t.hash(&mut s); - s.finish() -} - async fn catalog_handler(State(state): State>) -> impl IntoResponse { let catalog = catalog_from_state(&state.appstate).await; Json(catalog) } +#[derive(Deserialize)] +struct CreateStreamKeyBody { + label: String, +} + +struct AuthUser(entity::users::Model); + +impl FromRequestParts> for AuthUser { + type Rejection = StatusCode; + + async fn from_request_parts( + parts: &mut Parts, + state: &Arc, + ) -> Result { + let token = parts + .headers + .get("session") + // parse token... + .ok_or(StatusCode::UNAUTHORIZED)?; + + let user = + users::Entity::find_by_auth_session(&state.db, token.to_str().unwrap().to_string()) + .await + .map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)? + .ok_or(StatusCode::UNAUTHORIZED)?; + + Ok(AuthUser(user)) + } +} + +async fn create_stream_key_handler( + State(state): State>, + auth: AuthUser, + Json(payload): Json, +) -> impl IntoResponse { + let uuid = uuid::Uuid::new_v4(); + let value = format!("stream-key-{uuid}"); + let key = stream_key::Entity::create(&state.db, auth.0.id, value, payload.label, false).await; + + if let Ok(_key) = key { + StatusCode::CREATED + } else { + StatusCode::INTERNAL_SERVER_ERROR + } +} + +struct StreamKeys { + keys: Vec, +} + +struct StreamKey { + id: i32, + label: String, + value: String, +} + +async fn get_all_stream_keys( + State(state): State>, + auth: AuthUser, +) -> impl IntoResponse { + let user_keys = stream_key::Entity::find_by_user(&state.db, auth.0.id) + .await + .unwrap(); + + match stream_key::Entity::find_by_user(&state.db, auth.0.id).await { + Ok(keys) => Json(keys).into_response(), + Err(_) => StatusCode::INTERNAL_SERVER_ERROR.into_response(), + } +} + +#[derive(Deserialize)] +struct LoginForm { + username: String, + password: String, +} + +#[derive(Serialize)] +struct LoginResponse { + session_token: String, +} + +async fn login_handler( + State(state): State>, + Json(payload): Json, +) -> StatusCode { + // TODO: hash payload.password with a real password hasher (argon2/bcrypt), look up user by username + // TODO: verify hash matches stored hashed_password + // TODO: call auth_session::Entity::create, return session token + // TODO: return 401 on bad credentials + todo!("login not yet implemented") +} + #[derive(Deserialize)] struct CreateUserForm { username: String, @@ -65,44 +156,91 @@ struct CreateUserForm { ref_token: String, } -async fn create_user_handler(Form(form): Form) -> StatusCode { - // TODO: ref_token validation, db integration - let _ = ( - form.username, - calculate_hash(&form.password), - form.ref_token, - ); - todo!("user creation not yet implemented") -} +async fn create_user_handler( + State(state): State>, + Json(payload): Json, +) -> (HeaderMap, StatusCode) { + if payload.ref_token != "TEST" { + return (HeaderMap::new(), StatusCode::UNAUTHORIZED); + } -#[derive(Deserialize)] -struct StreamQuery { - stream_label: String, + let meow = users::Entity::create( + &state.db, + payload.username, + hash_password(&payload.password).unwrap(), + ) + .await; + + // Create session + let session: auth_session::Model = if let Ok(meow) = meow { + auth_session::Entity::create(&state.db, meow.id) + .await + .unwrap() // This should be ok (hopefully) + } else { + return (HeaderMap::new(), StatusCode::CONFLICT); + }; + let token = session.value; + let mut headers = HeaderMap::new(); + headers.insert( + SET_COOKIE, + format!("session={token}; HttpOnly; SameSite=Strict; Path=/") + .parse() + .unwrap(), + ); + (headers, StatusCode::OK) } async fn stream_handler( State(state): State>, - Query(query): Query, - body: Bytes, + Path(slug): Path, + body: String, ) -> impl IntoResponse { - let body_str = String::from_utf8(body.to_vec()).unwrap_or_default(); - let mut request_id = state.request_count.lock().await; *request_id += 1; let request_id_clone = *request_id; drop(request_id); + let stream_key_id = { + let app = state.appstate.lock().await; + app.stream_sessions + .iter() + .find(|e| e.value().stream_key_label == slug) + .map(|e| *e.key()) + }; + + let stream_key_id = if let Some(id) = stream_key_id { + id + } else { + return Response::builder() + .status(StatusCode::NOT_FOUND) + .header("content-type", "application/text") + .body("".to_string()) + .unwrap(); + }; + let mut accept_rx = state.accept_rx.new_receiver(); let _ = state .offer_tx - .send((request_id_clone, query.stream_label, body_str)) - .await; + .send((request_id_clone, stream_key_id, body)) + .await + .unwrap(); let mut reply_body = String::new(); while let Ok(answer) = accept_rx.recv().await { - if answer.0 == request_id_clone { - reply_body = answer.1; - break; + if let Some(reply) = answer.1 { + if answer.0 == request_id_clone { + Response::builder() + .status(StatusCode::CREATED) + .header("content-type", "application/sdp") + .body(reply) + .unwrap(); + } + } else { + Response::builder() + .status(StatusCode::NOT_FOUND) + // .header("content-type", "application/sdp") + .body("") + .unwrap(); } } @@ -119,9 +257,10 @@ async fn meow_handler() -> &'static str { #[derive(Serialize)] struct StreamCatalog { - active_streams: Vec, + active_streams: Vec, } +// TODO: Make this return streak labels and their id instead of the whole stream key omfg async fn catalog_from_state(state: &Arc>) -> StreamCatalog { let app = state.lock().await; let active_streams = app diff --git a/crates/server/src/main.rs b/crates/server/src/main.rs index 27e1fae..a9c78d1 100644 --- a/crates/server/src/main.rs +++ b/crates/server/src/main.rs @@ -2,10 +2,12 @@ use std::{error::Error, sync::Arc}; use async_broadcast::broadcast; use dashmap::DashMap; +use migration::{Migrator, MigratorTrait}; use rml_rtmp::{ handshake::{Handshake, HandshakeProcessResult, PeerType}, sessions::{ServerSession, ServerSessionConfig, ServerSessionEvent, ServerSessionResult}, }; +use sea_orm::{Database, sqlx::types::chrono}; use tokio::{ io::{AsyncReadExt, AsyncWriteExt}, net::TcpListener, @@ -17,33 +19,49 @@ use crate::{ media::{H264Parser, VideoFrame}, }; +mod hash; mod http; mod media; mod rtmp; mod webrtc; pub struct AppState { - pub stream_sessions: Arc>, + pub stream_sessions: Arc>, } pub struct StreamSession { - pub stream_key: String, + pub stream_key_id: i32, + pub stream_key_label: String, pub frame_channel: async_broadcast::Sender>, } #[tokio::main] async fn main() -> Result<(), Box> { let listener = TcpListener::bind("0.0.0.0:8123").await?; + let db = Database::connect("sqlite://./db.sqlite?mode=rwc") + .await + .unwrap(); + Migrator::up(&db, None).await.unwrap(); let appstate = Arc::new(Mutex::new(AppState { stream_sessions: Arc::new(DashMap::new()), })); - let (offer_tx, offer_rx) = tokio::sync::mpsc::channel::<(String, String)>(4); - let (answer_tx, answer_rx) = tokio::sync::mpsc::channel::<(String, String)>(4); + let (offer_tx, offer_rx) = tokio::sync::mpsc::channel::<(i32, i32, String)>(4); + // Request_Id, + // String_Label, + // Offer_body + // -- + // We are using broadcast instead of a regular channel because the way answers are processed are + // mulithreaded and we could easily process the answer of a completely different offer request + let (answer_tx, answer_rx) = broadcast::<(i32, Option)>(4); + // Request_Id, + // Answer_body - let mut http = HttpServer { + let http = HttpServer { offer_tx, accept_rx: answer_rx, appstate: appstate.clone(), + request_count: Mutex::new(0), + db: db.clone(), }; http.start()?; @@ -52,6 +70,7 @@ async fn main() -> Result<(), Box> { offer_rx, accept_tx: answer_tx, sessions_ref: app.stream_sessions.clone(), + db: db.clone(), }; webrtc.start()?; drop(app); @@ -59,6 +78,7 @@ async fn main() -> Result<(), Box> { loop { let (stream, _) = listener.accept().await?; let appstate = appstate.clone(); + let db = db.clone(); tokio::spawn(async move { let mut stream = stream; @@ -108,22 +128,45 @@ async fn main() -> Result<(), Box> { } ServerSessionResult::RaisedEvent(x) => match x { ServerSessionEvent::PublishStreamFinished { stream_key, .. } => { - appstate.lock().await.stream_sessions.remove(&stream_key); + let key = entity::stream_key::Entity::find_by_key(&db, &stream_key) + .await + .unwrap() + .unwrap(); + appstate.lock().await.stream_sessions.remove(&key.id); } ServerSessionEvent::PublishStreamRequested { request_id, stream_key, .. } => { + let key = + entity::stream_key::Entity::find_by_key(&db, &stream_key).await; + + let key = if let Ok(Some(key)) = key { + key + } else { + // Early return, gives error to client + let reply = rtmp_session + .reject_request(request_id, "", "Stream key invalid") + .unwrap(); + for x in reply { + if let ServerSessionResult::OutboundResponse(y) = x { + stream.write_all(&y.bytes).await.unwrap(); + } + } + break; + }; + let session = StreamSession { - stream_key: stream_key.clone(), + stream_key_id: key.id, + stream_key_label: key.label, frame_channel: video_channel.clone(), }; appstate .lock() .await .stream_sessions - .insert(stream_key.clone(), session); + .insert(key.id, session); let reply = rtmp_session.accept_request(request_id).unwrap(); for x in reply { diff --git a/crates/server/src/webrtc.rs b/crates/server/src/webrtc.rs index 0b921cf..c65ef10 100644 --- a/crates/server/src/webrtc.rs +++ b/crates/server/src/webrtc.rs @@ -1,13 +1,15 @@ use dashmap::DashMap; -use tokio::{ - net::UdpSocket, - sync::mpsc::{Receiver, Sender}, -}; +use entity::stream_key; +use sea_orm::DatabaseConnection; use std::{ error::Error, sync::Arc, time::{Duration, Instant}, }; +use tokio::{ + net::UdpSocket, + sync::mpsc::{Receiver, Sender}, +}; use str0m::{ Candidate, Event, Input, Output, Rtc, @@ -19,16 +21,28 @@ use str0m::{ use crate::{StreamSession, media::VideoFrame}; pub struct Webrtc { - pub offer_rx: Receiver<(String, String)>, - pub accept_tx: Sender<(String, String)>, - pub sessions_ref: Arc>, + pub offer_rx: Receiver<(i32, i32, String)>, + pub accept_tx: async_broadcast::Sender<(i32, Option)>, + pub sessions_ref: Arc>, + pub db: DatabaseConnection, } +const PER_CLIENT_CONNECTION_BUF: usize = 65535; + impl Webrtc { pub fn start(mut self) -> Result<(), Box> { tokio::spawn(async move { while let Some(offer) = self.offer_rx.recv().await { - let (stream_key, sdp_body) = offer; + let (request_id, stream_id, sdp_body) = offer; + + let stream_key = + stream_key::Entity::find_by_key(&self.db, &stream_id.to_string()).await; + + if let Ok(None) = stream_key { + self.accept_tx.broadcast((request_id, None)).await.unwrap(); + continue; + } + let socket = UdpSocket::bind("127.0.0.1:0").await.unwrap(); let local_addr = socket.local_addr().unwrap(); @@ -50,7 +64,7 @@ impl Webrtc { let mid = changes.add_media( MediaKind::Video, str0m::media::Direction::SendOnly, - Some(stream_key.clone()), + Some(stream_id.to_string()), Some("video0".to_string()), None, ); @@ -64,13 +78,13 @@ impl Webrtc { let answer_sdp = offer_answer.to_sdp_string(); self.accept_tx - .send((stream_key.clone(), answer_sdp)) + .broadcast((request_id, Some(answer_sdp))) .await .unwrap(); let sessions_ref = self.sessions_ref.clone(); tokio::spawn(async move { - Webrtc::detach_connection(socket, rtc, sessions_ref, mid).await; + Webrtc::detach_connection(socket, rtc, sessions_ref, stream_id, mid).await; }); } }); @@ -80,7 +94,8 @@ impl Webrtc { async fn detach_connection( socket: UdpSocket, mut rtc: Rtc, - sessions_ref: Arc>, + sessions_ref: Arc>, + stream_id: i32, _hint_mid: Mid, ) { let mut video_mid: Option = None; @@ -88,7 +103,7 @@ impl Webrtc { let mut connected = false; let mut video_stream: Option>> = None; - let mut recv_buf = vec![0u8; 65535]; + let mut recv_buf = vec![0u8; PER_CLIENT_CONNECTION_BUF]; let local_addr = socket.local_addr().unwrap(); loop { @@ -104,11 +119,9 @@ impl Webrtc { Event::MediaAdded(ma) => { if ma.kind == MediaKind::Video { if let Some(writer) = rtc.writer(ma.mid) { - let best = writer - .payload_params() - .max_by_key(|p| { - p.spec().format.profile_level_id.unwrap_or(0) - }); + let best = writer.payload_params().max_by_key(|p| { + p.spec().format.profile_level_id.unwrap_or(0) + }); if let Some(params) = best { println!("Selected PT {:?}", params.pt()); video_pt = Some(params.pt()); @@ -132,7 +145,7 @@ impl Webrtc { if connected { if video_stream.is_none() { - if let Some(session) = sessions_ref.get("test") { + if let Some(session) = sessions_ref.get(&stream_id) { video_stream = Some(session.frame_channel.new_receiver()); } } @@ -141,7 +154,8 @@ impl Webrtc { match stream.try_recv() { Ok(frame) => { let now = Instant::now(); - let rtp_time = MediaTime::from_90khz(frame.timestamp_ms as u64 * 90); + let rtp_time = + MediaTime::from_90khz(frame.timestamp_ms as u64 * 90); if let (Some(pt), Some(writer)) = (video_pt, video_mid.and_then(|m| rtc.writer(m))) { @@ -160,7 +174,9 @@ impl Webrtc { } } - let wait_until = deadline.min(Instant::now() + Duration::from_millis(20)).max(Instant::now()); + let wait_until = deadline + .min(Instant::now() + Duration::from_millis(20)) + .max(Instant::now()); let sleep = tokio::time::sleep_until(wait_until.into()); tokio::select! { diff --git a/flake.nix b/flake.nix index c9b3514..d41df60 100644 --- a/flake.nix +++ b/flake.nix @@ -94,6 +94,8 @@ # Inherit inputs from checks. checks = self.checks.${system}; + ADMIN_REF_CODE = "meowmeowpurrrmeow"; + # Additional dev-shell environment variables can be set directly # MY_CUSTOM_DEVELOPMENT_VAR = "something else"; @@ -101,6 +103,7 @@ packages = [ pkgs.clang pkgs.mold + pkgs.sea-orm-cli ]; }; }