fix: session cookie not saving

.gitignore

@@ -10,3 +10,5 @@ devenv.local.yaml
 
 # pre-commit
 .pre-commit-config.yaml
+/stream.db
+/db.sqlite

.new

@@ -0,0 +1,3 @@
+
+[Default Applications]
+x-scheme-handler/claude-cli=claude-code-url-handler.desktop

CLAUDE.md

@@ -0,0 +1,133 @@
+# 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-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 3000)
+```
+
+**Runtime:** tokio (not smol). The HTTP server runs as a tokio task using hyper.
+
+### RTMP ingestion (port 8123) — `crates/server/src/main.rs`
+
+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 for any stream key; a `StreamSession` with an `async_broadcast::Sender<Arc<VideoFrame>>` 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`.
+
+`AppState` is a `Arc<Mutex<AppState>>` wrapping a `DashMap<String, StreamSession>`.
+
+### HTTP API (port 3000) — `crates/server/src/http.rs`
+
+Async hyper server running in a tokio task. Routes:
+
+- `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).
+
+SDP offer/answer exchange uses `tokio::sync::mpsc` channels between HttpServer and the Webrtc task.
+
+### WebRTC negotiation and media loop — `crates/server/src/webrtc.rs`
+
+The `Webrtc` task receives `(stream_key, sdp_body)` tuples from `offer_rx`:
+
+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`):
+
+- 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(...))`.
+
+### H.264 parsing — `crates/server/src/media.rs`
+
+`H264Parser` converts raw RTMP `VideoDataReceived` payloads (AVCC) to Annex-B:
+
+- **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.
+
+Packet type `0` walks `AVCDecoderConfigurationRecord` to cache SPS and PPS byte arrays.
+
+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.
+
+### Unused stub — `crates/server/src/rtmp.rs`
+
+Early manual RTMP handshake implementation, not used in the current flow.
+
+## 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.

Cargo.toml

@@ -1,9 +1,17 @@
-[package]
+[profile.dev]
-name = "rtmp-to-whip-simple-server"
+debug = true
-version = "0.1.0"
-edition = "2024"
 
-[dependencies]
+[profile.flamegraph]
-macro_rules_attribute = "0.2.2"
+inherits = "release"
-smol = "2.0.2"
+debug = true
-smol-macros = "0.1.1"
+force-frame-pointers = true
+
+[workspace.metadata.crane]
+name = "rtmp-to-whip"
+
+[workspace.package]
+version = "0.1.0"
+
+[workspace]
+members = ["crates/server", "crates/entity", "crates/migration"]
+resolver = "2"

Dockerfile

@@ -0,0 +1,46 @@
+FROM rust:1-bookworm AS builder
+
+RUN apt-get update && apt-get install -y \
+        libopus-dev \
+        libsqlite3-dev \
+        clang \
+    && rm -rf /var/lib/apt/lists/*
+
+WORKDIR /app
+
+COPY Cargo.toml Cargo.lock ./
+COPY crates/server/Cargo.toml crates/server/Cargo.toml
+COPY crates/entity/Cargo.toml crates/entity/Cargo.toml
+COPY crates/migration/Cargo.toml crates/migration/Cargo.toml
+
+RUN mkdir -p crates/server/src crates/entity/src crates/migration/src \
+    && echo "fn main() {}" > crates/server/src/main.rs \
+    && echo "" > crates/entity/src/lib.rs \
+    && echo "" > crates/migration/src/lib.rs
+
+ENV RUSTFLAGS=""
+
+RUN cargo build --release --bin rtmp-to-whip
+
+COPY crates crates
+RUN touch crates/server/src/main.rs crates/entity/src/lib.rs crates/migration/src/lib.rs \
+    && cargo build --release --bin rtmp-to-whip \
+    && strip target/release/rtmp-to-whip
+
+RUN mkdir /deps \
+    && ldd target/release/rtmp-to-whip \
+        | awk 'NF==4{print $3} NF==2{print $1}' \
+        | grep -v vdso \
+        | xargs -I{} cp --parents {} /deps
+
+FROM scratch
+
+COPY --from=builder /deps /
+COPY --from=builder /etc/ssl/certs /etc/ssl/certs
+COPY --from=builder /app/target/release/rtmp-to-whip /rtmp-to-whip
+
+EXPOSE 1935
+EXPOSE 3000
+EXPOSE 6969/udp
+
+CMD ["/rtmp-to-whip"]

Dockerfile.aarch64

@@ -0,0 +1,58 @@
+FROM --platform=linux/amd64 rust:1-bookworm AS builder
+
+RUN dpkg --add-architecture arm64 \
+    && apt-get update && apt-get install -y \
+        gcc-aarch64-linux-gnu \
+        libopus-dev:arm64 \
+        libsqlite3-dev:arm64 \
+        libc6:arm64 \
+        ca-certificates \
+    && rm -rf /var/lib/apt/lists/*
+
+RUN rustup target add aarch64-unknown-linux-gnu
+
+ENV CARGO_TARGET_AARCH64_UNKNOWN_LINUX_GNU_LINKER=aarch64-linux-gnu-gcc \
+    PKG_CONFIG_ALLOW_CROSS=1 \
+    PKG_CONFIG_PATH=/usr/lib/aarch64-linux-gnu/pkgconfig \
+    RUSTFLAGS=""
+
+WORKDIR /app
+
+COPY Cargo.toml Cargo.lock ./
+COPY crates/server/Cargo.toml crates/server/Cargo.toml
+COPY crates/entity/Cargo.toml crates/entity/Cargo.toml
+COPY crates/migration/Cargo.toml crates/migration/Cargo.toml
+
+RUN mkdir -p crates/server/src crates/entity/src crates/migration/src \
+    && echo "fn main() {}" > crates/server/src/main.rs \
+    && echo "" > crates/entity/src/lib.rs \
+    && echo "" > crates/migration/src/lib.rs
+
+RUN cargo build --release --target aarch64-unknown-linux-gnu --bin rtmp-to-whip
+
+COPY crates crates
+RUN touch crates/server/src/main.rs crates/entity/src/lib.rs crates/migration/src/lib.rs \
+    && cargo build --release --target aarch64-unknown-linux-gnu --bin rtmp-to-whip \
+    && aarch64-linux-gnu-strip target/aarch64-unknown-linux-gnu/release/rtmp-to-whip
+
+RUN mkdir /deps \
+    && find /usr/lib/aarch64-linux-gnu -name 'libopus.so*' -exec cp --parents {} /deps \; \
+    && find /usr/lib/aarch64-linux-gnu -name 'libsqlite3.so*' -exec cp --parents {} /deps \; \
+    && find /lib/aarch64-linux-gnu -name 'libc*' -exec cp --parents {} /deps \; \
+    && find /lib/aarch64-linux-gnu -name 'libm*' -exec cp --parents {} /deps \; \
+    && find /lib/aarch64-linux-gnu -name 'libpthread*' -exec cp --parents {} /deps \; \
+    && find /lib/aarch64-linux-gnu -name 'libdl*' -exec cp --parents {} /deps \; \
+    && find /lib/aarch64-linux-gnu -name 'libgcc_s*' -exec cp --parents {} /deps \; \
+    && cp --parents /lib/ld-linux-aarch64.so.1 /deps
+
+FROM --platform=linux/arm64/v8 scratch
+
+COPY --from=builder /deps /
+COPY --from=builder /etc/ssl/certs /etc/ssl/certs
+COPY --from=builder /app/target/aarch64-unknown-linux-gnu/release/rtmp-to-whip /rtmp-to-whip
+
+EXPOSE 1935
+EXPOSE 3000
+EXPOSE 6969/udp
+
+CMD ["/rtmp-to-whip"]

Justfile

@@ -0,0 +1,17 @@
+build-dev:
+    docker build ./ --tag reg.h.doloro.co.uk/doloro/rtmp-to-whip:dev
+
+push-dev:
+    docker push reg.h.doloro.co.uk/doloro/rtmp-to-whip:dev
+
+build-latest:
+    docker build ./ --tag reg.h.doloro.co.uk/doloro/rtmp-to-whip:latest
+
+push-latest: build-latest
+    docker push reg.h.doloro.co.uk/doloro/rtmp-to-whip:latest
+
+build-aarch64:
+    docker build ./ --file Dockerfile.aarch64 --tag reg.h.doloro.co.uk/doloro/rtmp-to-whip:latest-aarch64
+
+push-aarch64: build-aarch64
+    docker push reg.h.doloro.co.uk/doloro/rtmp-to-whip:latest-aarch64

compose.yaml

@@ -0,0 +1,17 @@
+services:
+  server:
+    build: .
+    ports:
+      - "1935:1935"       # RTMP
+      - "3000:3000"       # HTTP API
+      - "6969:6969/udp"   # WebRTC UDP proxy
+    environment:
+      - PUBLIC_DOMAIN=   # set to your domain, e.g. rtmp.example.com; falls back to STUN if unset
+      - RTC_PORT=6969
+      - SIGNUP_CODE=     # required to create accounts; leave empty to disable signup
+    volumes:
+      - db:/app/db.sqlite
+    restart: unless-stopped
+
+volumes:
+  db:

crates/entity/Cargo.toml

@@ -0,0 +1,12 @@
+[package]
+name = "entity"
+version = "0.1.0"
+edition = "2024"
+
+[dependencies]
+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"] }

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<super::users::Entity> for Entity {
+    fn to() -> RelationDef {
+        Relation::Users.def()
+    }
+}
+
+impl ActiveModelBehavior for ActiveModel {}
+
+impl Entity {
+    pub async fn create(db: &DatabaseConnection, user_id: i32) -> Result<Model, DbErr> {
+        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<Option<Model>, DbErr> {
+        Entity::find()
+            .filter(Column::IdUser.eq(user_id))
+            .one(db)
+            .await
+    }
+}
+
+impl ActiveModel {}

crates/entity/src/lib.rs

@@ -0,0 +1,5 @@
+pub mod auth_session;
+pub mod prelude;
+pub mod stream_key;
+pub mod stream_session;
+pub mod users;

crates/entity/src/prelude.rs

@@ -0,0 +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;

crates/entity/src/stream_key.rs

@@ -0,0 +1,81 @@
+use sea_orm::{Set, entity::prelude::*};
+use serde::{Deserialize, Serialize};
+
+#[derive(Clone, Debug, PartialEq, DeriveEntityModel, Serialize, Deserialize)]
+#[sea_orm(table_name = "stream_key")]
+pub struct Model {
+    #[sea_orm(primary_key)]
+    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,
+}
+
+#[derive(Copy, Clone, Debug, EnumIter, DeriveRelation)]
+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<super::stream_session::Entity> for Entity {
+    fn to() -> RelationDef {
+        Relation::StreamSession.def()
+    }
+}
+
+impl Related<super::users::Entity> 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<Model, DbErr> {
+        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<Option<Model>, DbErr> {
+        Entity::find()
+            .filter(Column::KeyValue.eq(key_value))
+            .one(db)
+            .await
+    }
+
+    pub async fn find_by_user(db: &DatabaseConnection, user_id: i32) -> Result<Vec<Model>, DbErr> {
+        Entity::find()
+            .filter(Column::UserId.eq(user_id))
+            .all(db)
+            .await
+    }
+}

crates/entity/src/stream_session.rs

@@ -0,0 +1,100 @@
+use sea_orm::{
+    ActiveValue::{NotSet, Set},
+    entity::prelude::*,
+    sea_query::Expr,
+    sqlx::types::chrono::{self, Utc},
+};
+use serde::{Deserialize, Serialize};
+
+#[derive(Clone, Debug, PartialEq, DeriveEntityModel, Serialize, Deserialize)]
+#[sea_orm(table_name = "stream_session")]
+pub struct Model {
+    #[sea_orm(primary_key)]
+    pub id: i32,
+    pub stream_key_id: i32,
+    pub started_at: DateTimeUtc,
+    pub ended_at: Option<DateTimeUtc>,
+}
+
+#[derive(Copy, Clone, Debug, EnumIter, DeriveRelation)]
+pub enum Relation {
+    #[sea_orm(
+        belongs_to = "super::stream_key::Entity",
+        from = "Column::StreamKeyId",
+        to = "super::stream_key::Column::Id"
+    )]
+    StreamKey,
+}
+
+impl Related<super::stream_key::Entity> for Entity {
+    fn to() -> RelationDef {
+        Relation::StreamKey.def()
+    }
+}
+
+impl ActiveModelBehavior for ActiveModel {}
+
+impl Model {
+    pub async fn create_stream_session(
+        db: &DatabaseConnection,
+        stream_key_id: i32,
+        started_at: chrono::DateTime<Utc>,
+    ) -> Result<Model, DbErr> {
+        ActiveModel {
+            id: NotSet,
+            stream_key_id: Set(stream_key_id),
+            started_at: Set(started_at),
+            ended_at: NotSet,
+            ..Default::default()
+        }
+        .insert(db)
+        .await
+    }
+    pub async fn get_stream_session(
+        db: &DatabaseConnection,
+        stream_session_id: i32,
+    ) -> Result<Model, DbErr> {
+        Entity::find_by_id(stream_session_id)
+            .one(db)
+            .await?
+            .ok_or(DbErr::RecordNotFound(format!(
+                "stream_session {stream_session_id}"
+            )))
+    }
+    pub async fn get_all_active_sessions(db: &DatabaseConnection) -> Result<Vec<Model>, DbErr> {
+        Ok(Entity::find()
+            .filter(Column::EndedAt.is_null())
+            .all(db)
+            .await?)
+    }
+    pub async fn get_active_by_stream_key_id(
+        db: &DatabaseConnection,
+        stream_key_id: i32,
+    ) -> Result<Option<Model>, DbErr> {
+        Ok(Entity::find()
+            .filter(Column::StreamKeyId.eq(stream_key_id))
+            .filter(Column::EndedAt.is_null())
+            .one(db)
+            .await?)
+    }
+
+    pub async fn clean_unended_streams(db: &DatabaseConnection) -> Result<u64, DbErr> {
+        let result = Entity::update_many()
+            .filter(Column::EndedAt.is_null())
+            .col_expr(Column::EndedAt, Expr::col(Column::StartedAt).into())
+            .exec(db)
+            .await?;
+        Ok(result.rows_affected)
+    }
+}
+
+impl ActiveModel {
+    pub async fn finish_stream_session(
+        mut self,
+        db: &DatabaseConnection,
+        ended_at: chrono::DateTime<Utc>,
+    ) -> Result<Model, DbErr> {
+        self.ended_at = Set(Some(ended_at));
+        self.update(db).await
+    }
+}

crates/entity/src/users.rs

@@ -0,0 +1,102 @@
+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<super::stream_key::Entity> for Entity {
+    fn to() -> RelationDef {
+        Relation::StreamKey.def()
+    }
+}
+
+impl Related<super::auth_session::Entity> 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<Model, DbErr> {
+        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<Option<Model>, 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);
+        }
+    }
+    pub async fn find_by_username(
+        db: &DatabaseConnection,
+        username: String,
+    ) -> Result<Option<Model>, DbErr> {
+        let user = Entity::find()
+            .filter(Column::Username.eq(username))
+            .one(db)
+            .await;
+        user
+    }
+}
+
+impl ActiveModel {
+    pub async fn update_username(
+        mut self,
+        db: &DatabaseConnection,
+        username: String,
+    ) -> Result<Model, DbErr> {
+        self.username = Set(username);
+        self.update(db).await
+    }
+
+    pub async fn update_password(
+        mut self,
+        db: &DatabaseConnection,
+        hashed_password: String,
+    ) -> Result<Model, DbErr> {
+        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<Model, DbErr> {
+        self.stream_key_limit = Set(new_limit);
+        self.update(db).await
+    }
+}

crates/migration/Cargo.toml

@@ -0,0 +1,16 @@
+[package]
+name = "migration"
+version = "0.1.0"
+edition = "2024"
+
+[lib]
+name = "migration"
+path = "src/lib.rs"
+
+[[bin]]
+name = "migration"
+path = "src/main.rs"
+
+[dependencies]
+sea-orm-migration = { version = "1", features = ["runtime-tokio-rustls", "sqlx-sqlite"] }
+tokio = { version = "1", features = ["rt-multi-thread", "macros"] }

crates/migration/src/lib.rs

@@ -0,0 +1,20 @@
+pub use sea_orm_migration::prelude::*;
+
+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<Box<dyn MigrationTrait>> {
+        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),
+        ]
+    }
+}

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,
+}

crates/migration/src/m20260616_000002_create_stream_key.rs

@@ -0,0 +1,71 @@
+use sea_orm_migration::prelude::*;
+
+use super::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(StreamKey::Table)
+                    .if_not_exists()
+                    .col(
+                        ColumnDef::new(StreamKey::Id)
+                            .integer()
+                            .not_null()
+                            .auto_increment()
+                            .primary_key(),
+                    )
+                    .col(ColumnDef::new(StreamKey::UserId).integer().not_null())
+                    .col(
+                        ColumnDef::new(StreamKey::KeyValue)
+                            .string()
+                            .not_null()
+                            .unique_key(),
+                    )
+                    .col(ColumnDef::new(StreamKey::Label).string().not_null())
+                    .col(
+                        ColumnDef::new(StreamKey::IsActive)
+                            .boolean()
+                            .not_null()
+                            .default(true),
+                    )
+                    .col(
+                        ColumnDef::new(StreamKey::IsUnlisted)
+                            .boolean()
+                            .not_null()
+                            .default(true),
+                    )
+                    .col(ColumnDef::new(StreamKey::CreatedAt).date_time().not_null())
+                    .foreign_key(
+                        ForeignKey::create()
+                            .from(StreamKey::Table, StreamKey::UserId)
+                            .to(Users::Table, Users::Id),
+                    )
+                    .to_owned(),
+            )
+            .await
+    }
+
+    async fn down(&self, manager: &SchemaManager) -> Result<(), DbErr> {
+        manager
+            .drop_table(Table::drop().table(StreamKey::Table).to_owned())
+            .await
+    }
+}
+
+#[derive(Iden)]
+pub enum StreamKey {
+    Table,
+    Id,
+    UserId,
+    KeyValue,
+    Label,
+    IsActive,
+    IsUnlisted,
+    CreatedAt,
+}

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,
+}

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,
+}

crates/migration/src/main.rs

@@ -0,0 +1,6 @@
+use sea_orm_migration::prelude::*;
+
+#[tokio::main]
+async fn main() {
+    cli::run_cli(migration::Migrator).await;
+}

crates/server/Cargo.toml

@@ -0,0 +1,36 @@
+[package]
+name = "server"
+version = "0.1.1"
+edition = "2024"
+
+[target.x86_64-unknown-linux-gnu]
+linker = "clang"
+rustflags = ["-Clink-arg=-fuse-ld=/usr/local/bin/mold", "-Clink-arg=-Wl,--no-rosegment"]
+
+[[bin]]
+name = "rtmp-to-whip"
+path = "src/main.rs"
+
+[dependencies]
+async-broadcast = "0.7.2"
+bytes = "1"
+dashmap = "6.2.1"
+rand = "0.10.1"
+rml_rtmp = "0.8.0"
+str0m = "0.20.0"
+tokio = { version = "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"] }
+tower-http = { version = "0.6", features = ["cors"] }
+futures = "0.3.32"
+tracing = "0.1"
+tracing-subscriber = { version = "0.3", features = ["env-filter"] }
+symphonia = { version = "0.5", features = ["aac"] }
+opus = "0.3.1"
+rubato = "3.0.0"

crates/server/src/audio.rs

@@ -0,0 +1,176 @@
+use std::{error::Error, fmt::Display};
+
+use bytes::Bytes;
+use rubato::{Fft, Resampler, audioadapter_buffers::direct::InterleavedSlice};
+use symphonia::core::{
+    audio::SampleBuffer,
+    codecs::{CODEC_TYPE_AAC, CodecParameters, Decoder, DecoderOptions},
+    formats::Packet,
+};
+
+// Opus at 48kHz: valid frame sizes are 120/240/480/960/1920/2880 samples per channel.
+// We use 960 (20ms), the standard VoIP size.
+const OPUS_FRAME_SAMPLES: usize = 960;
+const OPUS_FRAME_INTERLEAVED: usize = OPUS_FRAME_SAMPLES * 2; // stereo
+
+// Assumes 44100 Hz stereo input from AAC-LC.
+pub struct AudioProcesser {
+    resampler: Fft<f32>,
+    resample_out: Vec<f32>,
+    // Accumulates resampled stereo f32 PCM until we have a full Opus frame.
+    pcm_buf: Vec<f32>,
+    encoder: opus::Encoder,
+    // Monotonic 48kHz sample counter; drives RTP timestamps independent of RTMP timestamps.
+    samples_emitted: u64,
+}
+
+impl AudioProcesser {
+    pub fn new() -> AudioProcesser {
+        let resampler =
+            Fft::<f32>::new(44100, 48000, 1024, 2, 2, rubato::FixedSync::Input).unwrap();
+        let resample_out = vec![0.0f32; resampler.output_frames_max() * 2];
+        AudioProcesser {
+            resampler,
+            resample_out,
+            pcm_buf: Vec::with_capacity(OPUS_FRAME_INTERLEAVED * 2),
+            encoder: opus::Encoder::new(48000, opus::Channels::Stereo, opus::Application::LowDelay)
+                .unwrap(),
+            samples_emitted: 0,
+        }
+    }
+
+    // Takes a decoded PCM AudioFrame, resamples it, buffers it, and returns however
+    // many complete 20ms Opus frames could be produced.
+    pub fn encode(&mut self, frame: AudioFrame) -> Vec<OpusAudioFrame> {
+        let mut samples: Vec<f32> = frame
+            .data
+            .chunks_exact(4)
+            .map(|b| f32::from_le_bytes([b[0], b[1], b[2], b[3]]))
+            .collect();
+
+        if frame.sample_rate == 48000 {
+            self.pcm_buf.extend_from_slice(&samples);
+        } else {
+            // Resample from 44100 Hz to 48000 Hz.
+            let input_frames = samples.len() / 2;
+            let input_buf = InterleavedSlice::new_mut(&mut samples, 2, input_frames).unwrap();
+            let max_out = self.resample_out.len() / 2;
+            let mut output_buf =
+                InterleavedSlice::new_mut(&mut self.resample_out, 2, max_out).unwrap();
+            let (_, output_frames) = self
+                .resampler
+                .process_into_buffer(&input_buf, &mut output_buf, None)
+                .unwrap();
+            self.pcm_buf
+                .extend_from_slice(&self.resample_out[..output_frames * 2]);
+        }
+
+        let mut result = Vec::new();
+        while self.pcm_buf.len() >= OPUS_FRAME_INTERLEAVED {
+            let chunk: Vec<f32> = self.pcm_buf.drain(..OPUS_FRAME_INTERLEAVED).collect();
+            let encoded = self.encoder.encode_vec_float(&chunk, 4096).unwrap();
+            result.push(OpusAudioFrame {
+                data: Bytes::copy_from_slice(&encoded),
+                // timestamp_ms here represents the 48kHz sample clock ÷ 48,
+                // so webrtc.rs can multiply by 48 to get the RTP clock tick.
+                timestamp_ms: (self.samples_emitted / 48) as u32,
+            });
+            self.samples_emitted += OPUS_FRAME_SAMPLES as u64;
+        }
+        result
+    }
+}
+
+pub struct AudioFrame {
+    pub data: Bytes, // interleaved f32 PCM, little-endian
+    pub timestamp_ms: u32,
+    pub sample_rate: u32,
+}
+
+pub struct OpusAudioFrame {
+    pub data: Bytes,
+    pub timestamp_ms: u32,
+}
+
+pub struct AACParser {
+    decoder: Option<Box<dyn Decoder>>,
+}
+
+#[derive(Debug)]
+enum AudioParseError {
+    InvalidCodec,
+    NoConfigPacket,
+}
+
+impl Display for AudioParseError {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        match self {
+            Self::InvalidCodec => write!(f, "Not the right codec provided"),
+            Self::NoConfigPacket => write!(f, "No config packet cached"),
+        }
+    }
+}
+
+impl Error for AudioParseError {}
+
+// Byte 0: upper nibble = sound format (10 = AAC)
+// Byte 1: 0 = AudioSpecificConfig, 1 = raw AAC frame
+impl AACParser {
+    pub fn new() -> Self {
+        Self { decoder: None }
+    }
+
+    pub fn parse(
+        &mut self,
+        bytes: &[u8],
+        timestamp_ms: u32,
+    ) -> Result<Option<AudioFrame>, Box<dyn Error>> {
+        if bytes.len() < 2 {
+            return Ok(None);
+        }
+
+        if (bytes[0] >> 4) != 10 {
+            return Err(Box::new(AudioParseError::InvalidCodec));
+        }
+
+        if bytes[1] == 0 {
+            let mut params = CodecParameters::new();
+            params
+                .for_codec(CODEC_TYPE_AAC)
+                .with_extra_data(bytes[2..].to_vec().into_boxed_slice());
+            self.decoder =
+                Some(symphonia::default::get_codecs().make(&params, &DecoderOptions::default())?);
+            return Ok(None);
+        }
+
+        let decoder = self
+            .decoder
+            .as_mut()
+            .ok_or(AudioParseError::NoConfigPacket)?;
+
+        let packet = Packet::new_from_boxed_slice(
+            0,
+            0,
+            1024, // AAC-LC frame is always 1024 samples
+            bytes[2..].to_vec().into_boxed_slice(),
+        );
+
+        let audio_buf = decoder.decode(&packet)?;
+
+        let spec = *audio_buf.spec();
+        let mut sample_buf = SampleBuffer::<f32>::new(audio_buf.capacity() as u64, spec);
+        sample_buf.copy_interleaved_ref(audio_buf);
+
+        let pcm_bytes: Vec<u8> = sample_buf
+            .samples()
+            .iter()
+            .flat_map(|s| s.to_le_bytes())
+            .collect();
+
+        Ok(Some(AudioFrame {
+            data: Bytes::from(pcm_bytes),
+            timestamp_ms,
+            sample_rate: spec.rate,
+        }))
+    }
+}

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<String, argon2::password_hash::Error> {
+    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()
+}

crates/server/src/http.rs

@@ -0,0 +1,411 @@
+use std::{net::SocketAddr, sync::Arc};
+
+use axum::{
+    Json, Router,
+    body::{Body, Bytes},
+    extract::{Form, FromRequestParts, Path, Query, State},
+    http::{
+        HeaderMap, HeaderName, Method, StatusCode,
+        header::{ACCEPT, AUTHORIZATION, CONTENT_TYPE, SET_COOKIE},
+        request::Parts,
+    },
+    response::{IntoResponse, Response},
+    routing::{get, post},
+};
+use entity::{auth_session, stream_key, stream_session, users};
+use sea_orm::{DatabaseConnection, EntityTrait, QueryFilter};
+use serde::{Deserialize, Serialize};
+use tokio::{
+    net::TcpListener,
+    sync::{Mutex, mpsc::Sender},
+};
+use tower_http::cors::CorsLayer;
+use uuid::Uuid;
+
+use tracing::{debug, info, warn};
+
+use crate::{
+    AppState,
+    hash::{hash_password, verify_password},
+    webrtc_ingest::handle_whip_injest,
+};
+
+const MAX_USERNAME_LEN: usize = 32;
+const MAX_LABEL_LEN: usize = 64;
+
+pub struct HttpServerConfig {
+    pub signup_code: String,
+}
+
+pub struct HttpServer {
+    pub offer_tx: Sender<(i32, i32, String)>,
+    pub accept_rx: async_broadcast::InactiveReceiver<(i32, Option<String>)>,
+    pub appstate: Arc<Mutex<AppState>>,
+    pub request_count: Mutex<i32>,
+    pub db: DatabaseConnection,
+    pub config: HttpServerConfig,
+}
+
+impl HttpServer {
+    pub fn start(self) -> Result<(), Box<dyn std::error::Error>> {
+        let state = Arc::new(self);
+
+        let origins = [
+            "http://localhost:5173".parse().unwrap(),
+            "https://stream.h.doloro.co.uk".parse().unwrap(),
+        ];
+
+        let cors = CorsLayer::new()
+            .allow_methods([Method::GET, Method::POST, Method::OPTIONS])
+            .allow_headers([
+                AUTHORIZATION,
+                ACCEPT,
+                CONTENT_TYPE,
+                HeaderName::from_static("session"),
+            ])
+            .allow_credentials(true)
+            .allow_origin(origins);
+
+        tokio::spawn(async move {
+            let app = Router::new()
+                .route("/api/catalog", get(catalog_handler))
+                .route("/api/user", post(create_user_handler))
+                .route(
+                    "/api/stream-key",
+                    post(create_stream_key_handler).get(get_all_stream_keys),
+                )
+                // .route("/api/admin/server_stats", get(todo!()))
+                .route("/api/whip", post(handle_whip_injest))
+                .route("/api/login", post(login_handler))
+                .route("/api/stream/{slug}", post(stream_handler))
+                .route("/api/meow", get(meow_handler))
+                .layer(cors)
+                .with_state(state);
+
+            let addr = SocketAddr::from(([0, 0, 0, 0], 3000));
+            let listener = TcpListener::bind(addr).await.unwrap();
+            axum::serve(listener, app).await.unwrap();
+        });
+        Ok(())
+    }
+}
+
+#[derive(Serialize)]
+struct StreamCatalog {
+    active_streams: Vec<StreamListing>,
+}
+
+#[derive(Serialize)]
+struct StreamListing {
+    label: String,
+    id: i32,
+    user: String,
+}
+
+async fn catalog_handler(State(state): State<Arc<HttpServer>>) -> impl IntoResponse {
+    let streams = stream_session::Model::get_all_active_sessions(&state.db)
+        .await
+        .unwrap();
+    debug!("{:#?}", streams);
+    let catalog: Vec<StreamListing> = futures::future::join_all(streams.iter().map(|listing| {
+        let db = state.db.clone();
+        let stream_key_id = listing.stream_key_id;
+        async move {
+            let key_info = stream_key::Entity::find_by_id(stream_key_id)
+                .one(&db)
+                .await
+                .unwrap()
+                .unwrap();
+            let user = users::Entity::find_by_id(key_info.user_id)
+                .one(&db)
+                .await
+                .unwrap()
+                .unwrap();
+            StreamListing {
+                id: stream_key_id,
+                label: key_info.label,
+                user: user.username,
+            }
+        }
+    }))
+    .await;
+    Json(catalog)
+}
+
+#[derive(Deserialize)]
+struct CreateStreamKeyBody {
+    label: String,
+}
+
+fn extract_session_token(headers: &HeaderMap) -> Option<String> {
+    // Check bare `session` header first (curl / API clients)
+    if let Some(v) = headers.get("session") {
+        return Some(v.to_str().ok()?.to_string());
+    }
+    // Fall back to Cookie header (browsers)
+    let cookie_header = headers.get("cookie")?.to_str().ok()?;
+    cookie_header
+        .split(';')
+        .find_map(|pair| {
+            let pair = pair.trim();
+            pair.strip_prefix("session=")
+        })
+        .map(|v| v.to_string())
+}
+
+struct AuthUser(entity::users::Model);
+
+impl FromRequestParts<Arc<HttpServer>> for AuthUser {
+    type Rejection = StatusCode;
+
+    async fn from_request_parts(
+        parts: &mut Parts,
+        state: &Arc<HttpServer>,
+    ) -> Result<Self, StatusCode> {
+        let token = extract_session_token(&parts.headers).ok_or(StatusCode::UNAUTHORIZED)?;
+
+        let user = users::Entity::find_by_auth_session(&state.db, token)
+            .await
+            .map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)?
+            .ok_or(StatusCode::UNAUTHORIZED)?;
+
+        Ok(AuthUser(user))
+    }
+}
+
+async fn create_stream_key_handler(
+    State(state): State<Arc<HttpServer>>,
+    auth: AuthUser,
+    Json(payload): Json<CreateStreamKeyBody>,
+) -> impl IntoResponse {
+    let uuid = uuid::Uuid::new_v4();
+    let value = format!("stream-key-{uuid}");
+    let key_amount = stream_key::Entity::find_by_user(&state.db, auth.0.id)
+        .await
+        .unwrap()
+        .len();
+    if payload.label.is_empty() || payload.label.len() > MAX_LABEL_LEN {
+        warn!(
+            user_id = auth.0.id,
+            label_len = payload.label.len(),
+            max = MAX_LABEL_LEN,
+            "stream key creation rejected: label length invalid"
+        );
+        return StatusCode::UNPROCESSABLE_ENTITY;
+    }
+    if key_amount >= auth.0.stream_key_limit.try_into().unwrap() {
+        warn!(
+            user_id = auth.0.id,
+            limit = auth.0.stream_key_limit,
+            "stream key limit reached"
+        );
+        return StatusCode::NOT_ACCEPTABLE;
+    }
+    let key = stream_key::Entity::create(&state.db, auth.0.id, value, payload.label, false).await;
+
+    if let Ok(ref k) = key {
+        info!(user_id = auth.0.id, stream_key_id = k.id, label = %k.label, "stream key created");
+        StatusCode::CREATED
+    } else {
+        StatusCode::INTERNAL_SERVER_ERROR
+    }
+}
+
+struct StreamKeys {
+    keys: Vec<StreamKey>,
+}
+
+struct StreamKey {
+    id: i32,
+    label: String,
+    value: String,
+}
+
+async fn get_all_stream_keys(
+    State(state): State<Arc<HttpServer>>,
+    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<Arc<HttpServer>>,
+    Json(payload): Json<LoginForm>,
+) -> impl IntoResponse {
+    if let Ok(x) = users::Entity::find_by_username(&state.db, payload.username.clone()).await {
+        if let Some(x) = x {
+            let pass = verify_password(&payload.password, &x.hashed_password);
+            if !pass {
+                warn!(username = %payload.username, "login failed: wrong password");
+                let mut meow = Response::new("".to_string());
+                *meow.status_mut() = StatusCode::UNAUTHORIZED;
+                return meow;
+            };
+            info!(user_id = x.id, username = %x.username, "login successful");
+            let auth = auth_session::Entity::create(&state.db, x.id).await.unwrap();
+            let token = auth.value;
+            let mut meow = Response::new("".to_string());
+            meow.headers_mut().insert(
+                SET_COOKIE,
+                format!("session={token}; SameSite=Strict; Path=/; Max-Age=2592000")
+                    .parse()
+                    .unwrap(),
+            );
+            *meow.status_mut() = StatusCode::OK;
+            return meow;
+        } else {
+            warn!(username = %payload.username, "login failed: user not found");
+            let mut meow = Response::new("".to_string());
+            *meow.status_mut() = StatusCode::UNAUTHORIZED;
+            return meow;
+        }
+    } else {
+        warn!(username = %payload.username, "login failed: DB error");
+        let mut meow = Response::new("".to_string());
+        *meow.status_mut() = StatusCode::UNAUTHORIZED;
+        return meow;
+    }
+}
+
+#[derive(Deserialize)]
+struct CreateUserForm {
+    username: String,
+    password: String,
+    ref_token: String,
+}
+
+async fn create_user_handler(
+    State(state): State<Arc<HttpServer>>,
+    Json(payload): Json<CreateUserForm>,
+) -> (HeaderMap, StatusCode) {
+    if state.config.signup_code.is_empty() || payload.ref_token != state.config.signup_code {
+        warn!(username = %payload.username, "signup rejected: invalid signup code");
+        return (HeaderMap::new(), StatusCode::UNAUTHORIZED);
+    }
+    if payload.username.is_empty() || payload.username.len() > MAX_USERNAME_LEN {
+        warn!(username = %payload.username, max = MAX_USERNAME_LEN, "signup rejected: username length invalid");
+        return (HeaderMap::new(), StatusCode::UNPROCESSABLE_ENTITY);
+    }
+
+    let meow = users::Entity::create(
+        &state.db,
+        payload.username.clone(),
+        hash_password(&payload.password).unwrap(),
+    )
+    .await;
+
+    // Create session
+    let session: auth_session::Model = if let Ok(ref user) = meow {
+        info!(user_id = user.id, username = %user.username, "user created");
+        auth_session::Entity::create(&state.db, user.id)
+            .await
+            .unwrap()
+    } else {
+        warn!(username = %payload.username, "user creation failed (likely username conflict)");
+        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<Arc<HttpServer>>,
+    Path(slug): Path<String>,
+    body: String,
+) -> impl IntoResponse {
+    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_id.to_string() == slug)
+            .map(|e| *e.key())
+    };
+
+    let stream_key_id = if let Some(id) = stream_key_id {
+        id
+    } else {
+        warn!(slug = %slug, "WHEP request for unknown or inactive stream");
+        return Response::builder()
+            .status(StatusCode::NOT_FOUND)
+            .header("content-type", "application/text")
+            .body("".to_string())
+            .unwrap();
+    };
+
+    info!(request_id = request_id_clone, slug = %slug, stream_key_id, "WHEP offer received");
+    let mut accept_rx = state.accept_rx.activate_cloned();
+    let _ = state
+        .offer_tx
+        .send((request_id_clone, stream_key_id, body))
+        .await
+        .unwrap();
+    debug!(
+        request_id = request_id_clone,
+        "offer sent, waiting for answer"
+    );
+
+    let mut reply_body = String::new();
+    while let Ok(answer) = accept_rx.recv().await {
+        debug!(request_id = request_id_clone, "received answer candidate");
+        if let Some(reply) = answer.1 {
+            if answer.0 == request_id_clone {
+                return Response::builder()
+                    .status(StatusCode::CREATED)
+                    .header("content-type", "application/sdp")
+                    .body(reply)
+                    .unwrap();
+            } else {
+                continue;
+            }
+        } else if answer.0 == request_id_clone {
+            return Response::builder()
+                .status(StatusCode::NOT_FOUND)
+                .body(String::new())
+                .unwrap();
+        }
+    }
+    info!(
+        request_id = request_id_clone,
+        "answer channel closed without a match"
+    );
+
+    Response::builder()
+        .status(StatusCode::CREATED)
+        .header("content-type", "application/sdp")
+        .body(reply_body)
+        .unwrap()
+}
+
+async fn meow_handler() -> &'static str {
+    "meow"
+}

crates/server/src/main.rs

@@ -0,0 +1,134 @@
+use std::{env, error::Error, sync::Arc};
+use tracing::{info, level_filters::LevelFilter, warn};
+
+use async_broadcast::broadcast;
+use dashmap::DashMap;
+use entity::stream_session;
+use migration::{Migrator, MigratorTrait};
+use rml_rtmp::{
+    handshake::{Handshake, HandshakeProcessResult, PeerType},
+    sessions::{ServerSession, ServerSessionConfig, ServerSessionEvent, ServerSessionResult},
+};
+use sea_orm::{
+    Database, IntoActiveModel,
+    sqlx::types::chrono::{self, Local},
+};
+use tokio::{
+    io::{AsyncReadExt, AsyncWriteExt},
+    net::TcpListener,
+    sync::Mutex,
+    time::Instant,
+};
+use tracing_subscriber::EnvFilter;
+
+use crate::{
+    audio::OpusAudioFrame,
+    http::{HttpServer, HttpServerConfig},
+    media::{H264Parser, VideoFrame},
+    webrtc_proxy::WebRtcProxyConfig,
+};
+
+mod audio;
+mod hash;
+mod http;
+mod media;
+mod rtmp;
+mod webrtc;
+mod webrtc_ingest;
+mod webrtc_proxy;
+
+// #[derive(Debug)]
+pub struct AppState {
+    pub stream_sessions: Arc<DashMap<i32, StreamSession>>,
+}
+
+pub struct StreamSession {
+    pub stream_key_id: i32,
+    pub stream_key_label: String,
+    pub frame_channel: async_broadcast::Sender<Arc<VideoFrame>>,
+    pub audio_channel: async_broadcast::Sender<Arc<OpusAudioFrame>>,
+}
+
+#[tokio::main]
+async fn main() -> Result<(), Box<dyn Error>> {
+    let env_filter = EnvFilter::builder()
+        .with_default_directive(LevelFilter::DEBUG.into())
+        .from_env()
+        // .parse("")
+        .unwrap();
+
+    tracing_subscriber::fmt().with_env_filter(env_filter).init();
+
+    let listener = TcpListener::bind("0.0.0.0:1935").await?;
+    info!("RTMP listening on 0.0.0.0:1935");
+    info!("HTTP API listening on 0.0.0.0:3000");
+
+    let db = Database::connect("sqlite://./db/db.sqlite?mode=rwc")
+        .await
+        .unwrap();
+    info!("database connected");
+    Migrator::up(&db, None).await.unwrap();
+    info!("migrations complete");
+
+    stream_session::Model::clean_unended_streams(&db)
+        .await
+        .unwrap();
+
+    let appstate = Arc::new(Mutex::new(AppState {
+        stream_sessions: Arc::new(DashMap::new()),
+    }));
+    let (offer_tx, offer_rx) = tokio::sync::mpsc::channel::<(i32, i32, String)>(64);
+    // Request_Id,
+    // String_Label,
+    // Offer_body
+
+    let (answer_tx, answer_rx) = broadcast::<(i32, Option<String>)>(64);
+    // Request_Id,
+    // Answer_body
+
+    let http = HttpServer {
+        offer_tx,
+        accept_rx: answer_rx.deactivate(),
+        appstate: appstate.clone(),
+        request_count: Mutex::new(0),
+        db: db.clone(),
+        config: HttpServerConfig {
+            signup_code: env::var("SIGNUP_CODE").unwrap_or_else(|_| {
+                warn!("SIGNUP_CODE not set; signup will be disabled");
+                String::new()
+            }),
+        },
+    };
+    http.start()?;
+
+    let proxyconfig = WebRtcProxyConfig {
+        proxy_port: env::var("RTC_PORT")
+            .unwrap_or("6969".into())
+            .parse()
+            .expect("RTC_PORT needs to be a number (i32)"),
+    };
+    let proxy = webrtc_proxy::WebrtcProxy::new(proxyconfig).await.unwrap();
+    proxy.start().unwrap();
+
+    let app = appstate.lock().await;
+    let webrtc = webrtc::Webrtc {
+        offer_rx,
+        accept_tx: answer_tx,
+        sessions_ref: app.stream_sessions.clone(),
+        db: db.clone(),
+        proxy: proxy.into(),
+    };
+    webrtc.start()?;
+
+    let rtmp = rtmp::Rtmp {
+        db: db.clone(),
+        stream_sessions: app.stream_sessions.clone(),
+        listener: listener,
+    };
+    rtmp.start()?;
+
+    drop(app);
+
+    tokio::signal::ctrl_c().await?;
+    Ok(())
+}

crates/server/src/media.rs

@@ -0,0 +1,147 @@
+// Claude slop... im not skilled amount to do this bullshit
+
+use bytes::Bytes;
+
+pub struct VideoFrame {
+    pub data: Bytes,
+    pub is_keyframe: bool,
+    pub timestamp_ms: u32,
+}
+
+pub struct AudioFrame {
+    pub data: Bytes,
+    pub timestamp_ms: u32,
+}
+
+pub struct H264Parser {
+    sps: Option<Vec<u8>>,
+    pps: Option<Vec<u8>>,
+}
+
+impl H264Parser {
+    pub fn new() -> Self {
+        Self {
+            sps: None,
+            pps: None,
+        }
+    }
+
+    /// Parse an RTMP VideoDataReceived payload. Returns None for sequence
+    /// header packets (which carry SPS/PPS but no displayable frame).
+    pub fn parse(&mut self, bytes: &[u8], timestamp_ms: u32) -> Option<VideoFrame> {
+        if bytes.len() < 5 {
+            return None;
+        }
+
+        let frame_type = (bytes[0] >> 4) & 0x0F;
+        let codec_id = bytes[0] & 0x0F;
+
+        if codec_id != 7 {
+            return None; // not H.264
+        }
+
+        let avc_packet_type = bytes[1];
+        // bytes[2..5] are the composition time offset — not needed for sending
+        let payload = &bytes[5..];
+
+        match avc_packet_type {
+            0 => {
+                self.parse_sequence_header(payload);
+                None
+            }
+            1 => {
+                let is_keyframe = frame_type == 1;
+                let data = self.avcc_to_annexb(payload, is_keyframe)?;
+                Some(VideoFrame {
+                    data: Bytes::from(data),
+                    is_keyframe,
+                    timestamp_ms,
+                })
+            }
+            _ => None,
+        }
+    }
+
+    fn parse_sequence_header(&mut self, payload: &[u8]) {
+        // AVCDecoderConfigurationRecord layout:
+        // [0] configurationVersion
+        // [1] AVCProfileIndication
+        // [2] profile_compatibility
+        // [3] AVCLevelIndication
+        // [4] 0xFF (lower 2 bits = lengthSizeMinusOne, always 3 meaning 4-byte lengths)
+        // [5] 0xE0 | numSPS
+        // [6..] SPS entries: 2-byte length + bytes
+        // then: numPPS, PPS entries: 2-byte length + bytes
+        if payload.len() < 7 {
+            return;
+        }
+
+        let mut i = 5;
+
+        let num_sps = (payload[i] & 0x1F) as usize;
+        i += 1;
+
+        for _ in 0..num_sps {
+            if i + 2 > payload.len() {
+                return;
+            }
+            let len = u16::from_be_bytes([payload[i], payload[i + 1]]) as usize;
+            i += 2;
+            if i + len > payload.len() {
+                return;
+            }
+            self.sps = Some(payload[i..i + len].to_vec());
+            i += len;
+        }
+
+        if i >= payload.len() {
+            return;
+        }
+
+        let num_pps = payload[i] as usize;
+        i += 1;
+
+        for _ in 0..num_pps {
+            if i + 2 > payload.len() {
+                return;
+            }
+            let len = u16::from_be_bytes([payload[i], payload[i + 1]]) as usize;
+            i += 2;
+            if i + len > payload.len() {
+                return;
+            }
+            self.pps = Some(payload[i..i + len].to_vec());
+            i += len;
+        }
+    }
+
+    fn avcc_to_annexb(&self, payload: &[u8], is_keyframe: bool) -> Option<Vec<u8>> {
+        let mut out = Vec::new();
+
+        // Prepend SPS+PPS before every keyframe so str0m's packetizer
+        // can bundle them into a STAP-A alongside the IDR NALU.
+        if is_keyframe {
+            if let (Some(sps), Some(pps)) = (&self.sps, &self.pps) {
+                out.extend_from_slice(&[0, 0, 0, 1]);
+                out.extend_from_slice(sps);
+                out.extend_from_slice(&[0, 0, 0, 1]);
+                out.extend_from_slice(pps);
+            }
+        }
+
+        // Convert each length-prefixed NALU to an Annex B start-code NALU.
+        let mut i = 0;
+        while i + 4 <= payload.len() {
+            let nalu_len = u32::from_be_bytes(payload[i..i + 4].try_into().unwrap()) as usize;
+            i += 4;
+            if i + nalu_len > payload.len() {
+                break;
+            }
+            out.extend_from_slice(&[0, 0, 0, 1]);
+            out.extend_from_slice(&payload[i..i + nalu_len]);
+            i += nalu_len;
+        }
+
+        if out.is_empty() { None } else { Some(out) }
+    }
+}

crates/server/src/rtmp.rs

@@ -0,0 +1,247 @@
+use std::{error::Error, sync::Arc};
+
+use async_broadcast::broadcast;
+use dashmap::DashMap;
+use entity::stream_session;
+use rml_rtmp::{
+    handshake::{Handshake, HandshakeProcessResult, PeerType},
+    sessions::{ServerSession, ServerSessionConfig, ServerSessionEvent, ServerSessionResult},
+};
+use sea_orm::{DatabaseConnection, IntoActiveModel, sqlx::types::chrono::Local};
+use tokio::{
+    io::{AsyncReadExt, AsyncWriteExt},
+    net::{TcpListener, TcpStream},
+};
+use tracing::{debug, info, warn};
+
+use crate::{
+    StreamSession,
+    audio::{AACParser, AudioProcesser, OpusAudioFrame},
+    media::{H264Parser, VideoFrame},
+};
+
+pub struct Rtmp {
+    pub listener: TcpListener,
+    pub stream_sessions: Arc<DashMap<i32, StreamSession>>,
+    pub db: DatabaseConnection,
+}
+
+async fn write_outbound(socket: &mut TcpStream, results: Vec<ServerSessionResult>) {
+    for r in results {
+        if let ServerSessionResult::OutboundResponse(p) = r {
+            socket.write_all(&p.bytes).await.unwrap();
+        }
+    }
+}
+
+impl Rtmp {
+    async fn handshake(
+        mut socket: TcpStream,
+    ) -> Result<(ServerSession, TcpStream), Box<dyn Error>> {
+        let mut server = Handshake::new(PeerType::Server);
+
+        let mut c0_c1 = [0u8; 1537];
+        socket.read_exact(&mut c0_c1).await.unwrap();
+        let s0_s1_s2 = match server.process_bytes(&c0_c1) {
+            Ok(HandshakeProcessResult::InProgress { response_bytes }) => response_bytes,
+            _ => panic!("handshake failed"),
+        };
+        socket.write_all(&s0_s1_s2).await.unwrap();
+
+        let mut c2 = [0u8; 1536];
+        socket.read_exact(&mut c2).await.unwrap();
+        match server.process_bytes(&c2) {
+            Ok(HandshakeProcessResult::Completed { .. }) => {}
+            Ok(HandshakeProcessResult::InProgress { response_bytes }) => {
+                socket.write_all(&response_bytes).await.unwrap()
+            }
+            x => panic!("Unexpected process_bytes response: {:?}", x),
+        }
+
+        let (rtmp_session, init_bytes) = ServerSession::new(ServerSessionConfig::new()).unwrap();
+        write_outbound(&mut socket, init_bytes).await;
+
+        Ok((rtmp_session, socket))
+    }
+
+    pub fn start(self) -> Result<(), Box<dyn Error>> {
+        let Self {
+            listener,
+            stream_sessions,
+            db,
+        } = self;
+        tokio::spawn(async move {
+            loop {
+                let (socket, peer_addr) = listener.accept().await.unwrap();
+                info!(%peer_addr, "RTMP connection accepted");
+                let (mut session, mut socket) = Rtmp::handshake(socket).await.unwrap();
+                info!(%peer_addr, "RTMP handshake complete");
+                let (mut video_tx, mut video_rx) = broadcast::<Arc<VideoFrame>>(32);
+                let (mut audio_tx, mut audio_rx) = broadcast::<Arc<OpusAudioFrame>>(32);
+                // video_rx.cycle
+
+                video_tx.set_overflow(true);
+                audio_tx.set_overflow(true);
+                let mut parser = H264Parser::new();
+                let mut aac_parser = AACParser::new();
+                let mut audio_proc = AudioProcesser::new();
+                let db = db.clone();
+                let stream_sessions = stream_sessions.clone();
+
+                tokio::spawn(async move {
+                    loop {
+                        let mut buf = [0u8; 4096];
+                        let n = socket.read(&mut buf).await.unwrap();
+                        if n == 0 {
+                            debug!("RTMP connection closed by peer");
+                            return;
+                        }
+                        let events = session.handle_input(&buf[..n]).unwrap();
+                        // Blankly using it, so it doesnt drop
+                        video_rx.is_closed();
+                        audio_rx.is_closed();
+
+                        for event in events {
+                            match event {
+                                ServerSessionResult::OutboundResponse(p) => {
+                                    socket.write_all(&p.bytes).await.unwrap();
+                                }
+                                ServerSessionResult::RaisedEvent(e) => match e {
+                                    ServerSessionEvent::ConnectionRequested {
+                                        request_id, ..
+                                    } => {
+                                        debug!("RTMP ConnectionRequested, accepting");
+                                        let reply = session.accept_request(request_id).unwrap();
+                                        write_outbound(&mut socket, reply).await;
+                                    }
+                                    ServerSessionEvent::PublishStreamRequested {
+                                        request_id,
+                                        stream_key,
+                                        ..
+                                    } => {
+                                        info!(stream_key = %stream_key, "publish stream requested");
+                                        let key = entity::stream_key::Entity::find_by_key(
+                                            &db,
+                                            &stream_key,
+                                        )
+                                        .await;
+
+                                        let key = if let Ok(Some(key)) = key {
+                                            key
+                                        } else {
+                                            warn!(stream_key = %stream_key, "stream key not found, rejecting");
+                                            let reply = session
+                                                .reject_request(
+                                                    request_id,
+                                                    "",
+                                                    "Stream key invalid",
+                                                )
+                                                .unwrap();
+                                            write_outbound(&mut socket, reply).await;
+                                            break;
+                                        };
+
+                                        let already_live =
+                                            stream_session::Model::get_active_by_stream_key_id(
+                                                &db, key.id,
+                                            )
+                                            .await
+                                            .unwrap();
+                                        if already_live.is_some() {
+                                            warn!(stream_key_id = key.id, label = %key.label, "stream key already live, rejecting duplicate publish");
+                                            let reply = session
+                                                .reject_request(
+                                                    request_id,
+                                                    "",
+                                                    "You're already streaming...",
+                                                )
+                                                .unwrap();
+                                            write_outbound(&mut socket, reply).await;
+                                            break;
+                                        }
+
+                                        info!(stream_key_id = key.id, label = %key.label, "stream started");
+                                        stream_sessions.insert(
+                                            key.id,
+                                            StreamSession {
+                                                stream_key_id: key.id,
+                                                stream_key_label: key.label,
+                                                frame_channel: video_tx.clone(),
+                                                audio_channel: audio_tx.clone(),
+                                            },
+                                        );
+
+                                        let reply = session.accept_request(request_id).unwrap();
+                                        stream_session::Model::create_stream_session(
+                                            &db,
+                                            key.id,
+                                            Local::now().into(),
+                                        )
+                                        .await
+                                        .unwrap();
+                                        write_outbound(&mut socket, reply).await;
+                                    }
+                                    ServerSessionEvent::PublishStreamFinished {
+                                        stream_key,
+                                        ..
+                                    } => {
+                                        info!(stream_key = %stream_key, "publish stream finished");
+                                        let key = entity::stream_key::Entity::find_by_key(
+                                            &db,
+                                            &stream_key,
+                                        )
+                                        .await
+                                        .unwrap()
+                                        .unwrap();
+                                        stream_sessions.remove(&key.id);
+                                        stream_session::Model::get_active_by_stream_key_id(
+                                            &db, key.id,
+                                        )
+                                        .await
+                                        .unwrap()
+                                        .unwrap()
+                                        .into_active_model()
+                                        .finish_stream_session(&db, Local::now().into())
+                                        .await
+                                        .unwrap();
+                                        stream_sessions.get(&key.id).unwrap().frame_channel.close();
+                                    }
+                                    ServerSessionEvent::VideoDataReceived {
+                                        data,
+                                        timestamp,
+                                        ..
+                                    } => {
+                                        if data.len() >= 5 && &data[1..5] == b"hvc1" {
+                                            warn!("HEVC/H.265 not supported, closing connection");
+                                            return;
+                                        }
+                                        if let Some(frame) = parser.parse(&data, timestamp.value) {
+                                            video_tx.broadcast(Arc::new(frame)).await.ok();
+                                        }
+                                    }
+                                    ServerSessionEvent::AudioDataReceived {
+                                        data,
+                                        timestamp,
+                                        ..
+                                    } => {
+                                        // Consume the non-Send error before any await point.
+                                        let opus_frames: Vec<_> = match aac_parser.parse(&data, timestamp.value) {
+                                            Err(e) => { warn!("AAC parse error: {}", e); vec![] }
+                                            Ok(frame) => frame.map(|f| audio_proc.encode(f)).unwrap_or_default(),
+                                        };
+                                        for frame in opus_frames {
+                                            audio_tx.broadcast(Arc::new(frame)).await.ok();
+                                        }
+                                    }
+                                    _ => {}
+                                },
+                                _ => {}
+                            }
+                        }
+                    }
+                });
+            }
+        });
+        Ok(())
+    }
+}

crates/server/src/webrtc.rs

@@ -0,0 +1,354 @@
+use bytes::Bytes;
+use dashmap::DashMap;
+use entity::stream_key;
+use sea_orm::{DatabaseConnection, EntityTrait};
+use std::{
+    error::Error,
+    net::SocketAddr,
+    sync::Arc,
+    time::{Duration, Instant},
+};
+use tokio::{
+    net::UdpSocket,
+    sync::mpsc::{Receiver, Sender},
+};
+use tracing::{debug, error, info, warn};
+
+use str0m::{
+    Candidate, Event, Input, Output, Rtc,
+    change::SdpOffer,
+    media::{Frequency, MediaKind, MediaTime, Mid},
+    net::{Protocol, Receive},
+};
+
+use crate::{StreamSession, audio::OpusAudioFrame, media::VideoFrame, webrtc_proxy::WebrtcProxy};
+
+pub struct Webrtc {
+    pub offer_rx: Receiver<(i32, i32, String)>,
+    pub accept_tx: async_broadcast::Sender<(i32, Option<String>)>,
+    pub sessions_ref: Arc<DashMap<i32, StreamSession>>,
+    pub proxy: Arc<WebrtcProxy>,
+    pub db: DatabaseConnection,
+}
+
+const PER_CLIENT_CONNECTION_BUF: usize = 65535;
+
+impl Webrtc {
+    pub fn start(mut self) -> Result<(), Box<dyn Error>> {
+        tokio::spawn(async move {
+            while let Some(offer) = self.offer_rx.recv().await {
+                let (request_id, stream_id, sdp_body) = offer;
+                info!(request_id, stream_id, "processing offer");
+
+                let stream_key = stream_key::Entity::find_by_id(stream_id)
+                    .one(&self.db)
+                    .await;
+
+                if let Ok(None) = stream_key {
+                    warn!(request_id, stream_id, "stream key not found in DB, rejecting offer");
+                    self.accept_tx.broadcast((request_id, None)).await.unwrap();
+                    continue;
+                }
+                if let Err(ref e) = stream_key {
+                    warn!(request_id, stream_id, "DB error looking up stream key: {:?}", e);
+                    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();
+
+                let local_addr = self.proxy.public_addr();
+
+                let mut builder = Rtc::builder();
+                {
+                    let cc = builder.codec_config();
+                    cc.enable_h264(false);
+                    cc.add_h264(102.into(), None, true, 0x42e01f);
+                    cc.add_h264(104.into(), None, true, 0x4d001f);
+                    cc.add_h264(106.into(), None, true, 0x64001f);
+                }
+                let mut rtc = builder.build(Instant::now());
+
+                let candidate = Candidate::host(local_addr, Protocol::Udp).unwrap();
+                rtc.add_local_candidate(candidate);
+
+                let offer_sdp = SdpOffer::from_sdp_string(&sdp_body).unwrap();
+                let mut changes = rtc.sdp_api();
+                let mid = changes.add_media(
+                    MediaKind::Video,
+                    str0m::media::Direction::SendOnly,
+                    Some(stream_id.to_string()),
+                    Some("0".to_string()),
+                    None,
+                );
+                changes.add_media(
+                    MediaKind::Audio,
+                    str0m::media::Direction::SendOnly,
+                    Some(stream_id.to_string()),
+                    Some("0".to_string()),
+                    None,
+                );
+                changes.add_channel("meow".into());
+                let offer_answer = match changes.accept_offer(offer_sdp) {
+                    Ok(a) => a,
+                    Err(e) => {
+                        error!("accept_offer failed: {:?}", e);
+                        continue;
+                    }
+                };
+                let answer_sdp = offer_answer.to_sdp_string();
+                let ufrag = answer_sdp
+                    .lines()
+                    .find(|l| l.starts_with("a=ice-ufrag:"))
+                    .and_then(|l| l.strip_prefix("a=ice-ufrag:"))
+                    .map(|s| s.trim().to_string());
+                info!("Serving webrtc ufrag: {:?}", ufrag);
+                let (socket, rx) = self.proxy.add_client(ufrag.unwrap());
+
+                debug!(request_id, "sending answer back");
+                self.accept_tx
+                    .broadcast((request_id, Some(answer_sdp)))
+                    .await
+                    .unwrap();
+
+                let sessions_ref = self.sessions_ref.clone();
+                let public_addr = self.proxy.public_addr();
+                tokio::spawn(async move {
+                    Webrtc::detach_connection(socket, rx, rtc, sessions_ref, stream_id, mid, public_addr).await;
+                });
+            }
+        });
+        Ok(())
+    }
+
+    async fn detach_connection(
+        socket: Arc<UdpSocket>,
+        mut rx: Receiver<(Bytes, SocketAddr)>,
+        mut rtc: Rtc,
+        sessions_ref: Arc<DashMap<i32, StreamSession>>,
+        stream_id: i32,
+        _hint_mid: Mid,
+        local_addr: SocketAddr,
+    ) {
+        let mut video_mid: Option<Mid> = None;
+        let mut video_pt = None;
+        let mut audio_mid: Option<Mid> = None;
+        let mut audio_pt = None;
+        let mut connected = false;
+        let mut video_stream: Option<async_broadcast::Receiver<Arc<VideoFrame>>> = None;
+        let mut audio_stream: Option<async_broadcast::Receiver<Arc<OpusAudioFrame>>> = None;
+        let mut saw_keyframe = false;
+
+        loop {
+            let deadline = loop {
+                match rtc.poll_output() {
+                    Ok(Output::Timeout(t)) => break t,
+                    Ok(Output::Transmit(t)) => {
+                        if let Err(e) = socket.send_to(&t.contents, t.destination).await {
+                            warn!("UDP send error: {:?}", e);
+                            return;
+                        }
+                    }
+                    Ok(Output::Event(e)) => match e {
+                        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)
+                                    });
+                                    if let Some(params) = best {
+                                        info!("selected PT {:?}", params.pt());
+                                        video_pt = Some(params.pt());
+                                        video_mid = Some(ma.mid);
+                                    }
+                                }
+                            }
+                            if ma.kind == MediaKind::Audio {
+                                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)
+                                    });
+                                    if let Some(params) = best {
+                                        info!("selected PT {:?}", params.pt());
+                                        audio_pt = Some(params.pt());
+                                        audio_mid = Some(ma.mid);
+                                    }
+                                }
+                            }
+                        }
+                        Event::IceConnectionStateChange(state) => {
+                            use str0m::IceConnectionState;
+                            info!("ICE state: {:?}", state);
+                            if matches!(state, IceConnectionState::Disconnected) {
+                                info!("ICE disconnected, closing connection");
+                                return;
+                            }
+                        }
+                        Event::Connected => {
+                            info!("DTLS+ICE connected, ready for media");
+                            connected = true;
+                        }
+                        _ => {}
+                    },
+                    Err(e) => {
+                        error!("poll_output error (connection closing): {:?}", e);
+                        return;
+                    }
+                }
+            };
+
+            if connected {
+                if video_stream.is_none() {
+                    if let Some(session) = sessions_ref.get(&stream_id) {
+                        video_stream = Some(session.frame_channel.new_receiver());
+                        debug!(stream_id, "subscribed to video channel");
+                    } else {
+                        warn!(
+                            stream_id,
+                            "stream session not found in map, cannot subscribe"
+                        );
+                    }
+                }
+                if let Some(video) = &video_stream {
+                    if video.is_closed() {
+                        if let Some(session) = sessions_ref.get(&stream_id) {
+                            video_stream = Some(session.frame_channel.new_receiver());
+                            debug!(stream_id, "subscribed to video channel");
+                        }
+                    }
+                }
+                if audio_stream.is_none() {
+                    if let Some(session) = sessions_ref.get(&stream_id) {
+                        audio_stream = Some(session.audio_channel.new_receiver());
+                        debug!(stream_id, "subscribed to audio channel");
+                    }
+                }
+                if let Some(audio) = &audio_stream {
+                    if audio.is_closed() {
+                        if let Some(session) = sessions_ref.get(&stream_id) {
+                            audio_stream = Some(session.audio_channel.new_receiver());
+                            debug!(stream_id, "subscribed to audio channel");
+                        }
+                    }
+                }
+                if let Some(ref mut stream) = video_stream {
+                    let mut wrote_any = false;
+                    for _ in 0..8 {
+                        match stream.try_recv() {
+                            Ok(frame) => {
+                                if !saw_keyframe {
+                                    if !frame.is_keyframe {
+                                        continue;
+                                    }
+                                    saw_keyframe = true;
+                                    debug!(stream_id, "first keyframe, starting RTP send");
+                                }
+                                let now = Instant::now();
+                                let rtp_time =
+                                    MediaTime::from_90khz(frame.timestamp_ms as u64 * 90);
+                                match (video_pt, video_mid.and_then(|m| rtc.writer(m))) {
+                                    (Some(pt), Some(writer)) => {
+                                        match writer.write(pt, now, rtp_time, frame.data.to_vec()) {
+                                            Ok(_) => wrote_any = true,
+                                            Err(e) => warn!("video RTP write error: {:?}", e),
+                                        }
+                                    }
+                                    _ => warn!(
+                                        stream_id,
+                                        "video_pt or writer not ready, dropping frame (video_pt={:?} video_mid={:?})",
+                                        video_pt,
+                                        video_mid
+                                    ),
+                                }
+                            }
+                            Err(async_broadcast::TryRecvError::Empty) => break,
+                            Err(async_broadcast::TryRecvError::Closed) => {
+                                warn!("video channel closed, stream ended");
+                                break;
+                            }
+                            Err(async_broadcast::TryRecvError::Overflowed(_)) => {
+                                // saw_keyframe = false;
+                                continue;
+                            }
+                        }
+                    }
+                    if let Some(ref mut audio) = audio_stream {
+                        for _ in 0..8 {
+                            match audio.try_recv() {
+                                Ok(frame) => {
+                                    let now = Instant::now();
+                                    let rtp_time = MediaTime::new(
+                                        frame.timestamp_ms as u64 * 48,
+                                        Frequency::FORTY_EIGHT_KHZ,
+                                    );
+                                    if let (Some(pt), Some(writer)) =
+                                        (audio_pt, audio_mid.and_then(|m| rtc.writer(m)))
+                                    {
+                                        match writer.write(pt, now, rtp_time, frame.data.to_vec()) {
+                                            Ok(_) => wrote_any = true,
+                                            Err(e) => warn!("RTP write error: {:?}", e),
+                                        }
+                                    }
+                                }
+                                Err(async_broadcast::TryRecvError::Empty) => break,
+                                Err(async_broadcast::TryRecvError::Closed) => {
+                                    info!("audio channel closed, stream ended");
+                                    break;
+                                }
+                                Err(async_broadcast::TryRecvError::Overflowed(_)) => continue,
+                            }
+                        }
+                    }
+                    // If we queued RTP data, drain poll_output immediately so packets are
+                    // transmitted in this iteration rather than 20ms later. But still drive
+                    // str0m's timeout if the deadline has passed — ICE consent refresh depends on it.
+                    if wrote_any {
+                        let now = Instant::now();
+                        if now >= deadline {
+                            if let Err(e) = rtc.handle_input(Input::Timeout(now)) {
+                                error!("handle_input(Timeout) error: {:?}", e);
+                                return;
+                            }
+                        }
+                        continue;
+                    }
+                }
+            }
+
+            // debug!("waiting");
+
+            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! {
+                _ = sleep => {
+                    if let Err(e) = rtc.handle_input(Input::Timeout(Instant::now())) {
+                        error!("handle_input(Timeout) error: {:?}", e);
+                        return;
+                    }
+                }
+                result = rx.recv() => {
+                    if let Some((data, from)) = result {
+                        if let Ok(contents) = (&data[..]).try_into() {
+                            if let Err(e) = rtc.handle_input(Input::Receive(
+                                Instant::now(),
+                                Receive {
+                                    proto: Protocol::Udp,
+                                    source: from,
+                                    destination: local_addr,
+                                    contents,
+                                },
+                            )) {
+                                error!("handle_input(Receive) error: {:?}", e);
+                                return;
+                            }
+                        }
+                    }
+                }
+            }
+        }
+    }
+}

crates/server/src/webrtc_ingest.rs

@@ -0,0 +1,17 @@
+use std::sync::Arc;
+
+use axum::{extract::State, response::IntoResponse};
+use str0m::change::SdpOffer;
+use tracing::info;
+
+use crate::http::HttpServer;
+
+pub async fn handle_whip_injest(
+    State(state): State<Arc<HttpServer>>,
+    offer: String,
+) -> impl IntoResponse {
+    let sdp_offer = SdpOffer::from_sdp_string(&offer).unwrap();
+    for x in &sdp_offer.media_lines {
+        info!("{}", x);
+    }
+}

crates/server/src/webrtc_proxy.rs

@@ -0,0 +1,218 @@
+use std::{env, error::Error, net::SocketAddr, sync::Arc};
+
+use bytes::Bytes;
+use dashmap::DashMap;
+use str0m::net::DatagramRecv;
+use tokio::{
+    net::UdpSocket,
+    sync::mpsc::{self, Receiver},
+};
+use tracing::{debug, error, info, warn};
+
+pub struct WebRtcProxyConfig {
+    pub proxy_port: i32,
+}
+
+pub struct WebrtcProxy {
+    clients_ufrag: Arc<DashMap<String, tokio::sync::mpsc::Sender<(Bytes, SocketAddr)>>>,
+    clients_addr: Arc<DashMap<SocketAddr, tokio::sync::mpsc::Sender<(Bytes, SocketAddr)>>>,
+    socket: Arc<UdpSocket>,
+    public_addr: SocketAddr,
+}
+
+const STUN_MAGIC: u32 = 0x2112A442;
+
+impl WebrtcProxy {
+    pub async fn new(config: WebRtcProxyConfig) -> Result<Self, Box<dyn Error>> {
+        let sock = UdpSocket::bind(format!("0.0.0.0:{}", config.proxy_port)).await?;
+        let port = sock.local_addr()?.port();
+
+        let public_ip = match env::var("PUBLIC_DOMAIN") {
+            Ok(domain) => {
+                let ip = resolve_domain(&domain).await?;
+                info!(%domain, %ip, "resolved PUBLIC_DOMAIN for WebRTC candidates");
+                ip
+            }
+            Err(_) => {
+                let ip = stun_public_ip().await?;
+                info!(%ip, "discovered public IP via STUN for WebRTC candidates");
+                ip
+            }
+        };
+        let public_addr = SocketAddr::new(public_ip, port);
+        info!(%public_addr, "WebRTC UDP proxy listening");
+
+        Ok(WebrtcProxy {
+            socket: Arc::new(sock),
+            clients_ufrag: Arc::new(DashMap::new()),
+            clients_addr: Arc::new(DashMap::new()),
+            public_addr,
+        })
+    }
+    pub fn start(&self) -> Result<(), Box<dyn Error>> {
+        // let self_arc = Arc::new(self);
+        let by_ufrag = self.clients_ufrag.clone();
+        let by_addr = self.clients_addr.clone();
+        let socket = self.socket.clone();
+
+        tokio::spawn(async move {
+            let mut buf = vec![0u8; 65535];
+            loop {
+                let (b, from) = match socket.recv_from(&mut buf).await {
+                    Ok(data) => data,
+                    Err(err) => {
+                        warn!("proxy couldnt eat data from socket, (({}))", err);
+                        continue;
+                    }
+                };
+                let data = Bytes::copy_from_slice(&buf[..b]);
+
+                // By addr
+                if let Some(tx) = by_addr.get(&from) {
+                    match tx.try_send((data, from)) {
+                        Ok(_) => continue,
+                        Err(e) => {
+                            match e {
+                                mpsc::error::TrySendError::Full(_) => continue,
+                                mpsc::error::TrySendError::Closed(_) => {
+                                    // the rv is ded
+                                    drop(tx);
+                                    by_addr.remove(&from);
+                                    continue;
+                                }
+                            };
+                        }
+                    };
+                };
+
+                let Some(ufrag) = self::WebrtcProxy::ufrag(&data) else {
+                    debug!("huh, packet isnt stun or added as client.");
+                    continue;
+                };
+
+                let Some((_, tx)) = by_ufrag.remove(&ufrag) else {
+                    warn!("STUN packet ({}), isnt registored", ufrag);
+                    continue;
+                };
+
+                by_addr.insert(from, tx.clone());
+                debug!("got ufrag {}", ufrag);
+                debug!("sending data");
+                if let Err(e) = tx.try_send((data, from)) {
+                    match e {
+                        mpsc::error::TrySendError::Full(_) => {
+                            error!("Channel full")
+                        }
+                        mpsc::error::TrySendError::Closed(_) => {
+                            error!("Channel is closed");
+                        }
+                    }
+                };
+            }
+        });
+        Ok(())
+    }
+    pub fn add_client(&self, ufrag: String) -> (Arc<UdpSocket>, Receiver<(Bytes, SocketAddr)>) {
+        debug!("Added client {}", ufrag);
+        let (tx, rx) = mpsc::channel(256);
+        self.clients_ufrag.insert(ufrag, tx);
+        (self.socket.clone(), rx)
+    }
+    pub fn local_addr(&self) -> SocketAddr {
+        self.socket.local_addr().unwrap()
+    }
+
+    pub fn public_addr(&self) -> SocketAddr {
+        self.public_addr
+    }
+    pub fn ufrag(b: &Bytes) -> Option<String> {
+        if b.len() <= 20 {
+            return None;
+        }
+        let magic = u32::from_be_bytes(b[4..8].try_into().ok()?);
+        if magic != STUN_MAGIC {
+            return None;
+        }
+        // attribies start at 20
+        let mut pos = 20usize;
+        while (pos + 4) <= b.len() {
+            let attr_type: u16 = u16::from_be_bytes(b[pos..pos + 2].try_into().ok()?);
+            let attr_len: u16 = u16::from_be_bytes(b[pos + 2..pos + 4].try_into().ok()?);
+            pos = pos + 4;
+            if attr_type == 0x0006 {
+                let value = std::str::from_utf8(b[pos..pos + (attr_len as usize)].try_into().ok()?);
+                let local = value.unwrap().split(":").next();
+                return Some(local.unwrap().to_string());
+            }
+            pos += (attr_len as usize + 3) & !3;
+        }
+        None
+    }
+}
+
+async fn resolve_domain(domain: &str) -> Result<std::net::IpAddr, Box<dyn Error>> {
+    let addr = tokio::net::lookup_host(format!("{}:0", domain))
+        .await?
+        .find(|a| a.is_ipv4())
+        .ok_or_else(|| format!("no IPv4 address found for {}", domain))?;
+    Ok(addr.ip())
+}
+
+// Send a STUN Binding Request to a public STUN server and extract our public IP
+// from the XOR-MAPPED-ADDRESS attribute in the response.
+async fn stun_public_ip() -> Result<std::net::IpAddr, Box<dyn Error>> {
+    let sock = UdpSocket::bind("0.0.0.0:0").await?;
+    sock.connect("stun.l.google.com:19302").await?;
+
+    // Build a minimal STUN Binding Request (RFC 5389).
+    // Header: type(2) | length(2) | magic(4) | transaction-id(12)
+    let mut req = [0u8; 20];
+    req[0..2].copy_from_slice(&0x0001u16.to_be_bytes()); // Binding Request
+    req[2..4].copy_from_slice(&0u16.to_be_bytes()); // no attributes
+    req[4..8].copy_from_slice(&STUN_MAGIC.to_be_bytes());
+    req[8..20].copy_from_slice(b"rtmp2whip_tx"); // transaction ID (12 bytes)
+
+    sock.send(&req).await?;
+
+    let mut buf = [0u8; 512];
+    let n = tokio::time::timeout(std::time::Duration::from_secs(5), sock.recv(&mut buf)).await??;
+
+    let data = &buf[..n];
+    parse_xor_mapped_address(data).ok_or("no XOR-MAPPED-ADDRESS in STUN response".into())
+}
+
+// Parse XOR-MAPPED-ADDRESS (0x0020) from a STUN response.
+// The IP is XOR'd with the magic cookie (IPv4) or magic+transaction-id (IPv6).
+fn parse_xor_mapped_address(data: &[u8]) -> Option<std::net::IpAddr> {
+    if data.len() < 20 {
+        return None;
+    }
+    let magic = u32::from_be_bytes(data[4..8].try_into().ok()?);
+    if magic != STUN_MAGIC {
+        return None;
+    }
+
+    let mut pos = 20usize;
+    while pos + 4 <= data.len() {
+        let attr_type = u16::from_be_bytes(data[pos..pos + 2].try_into().ok()?);
+        let attr_len = u16::from_be_bytes(data[pos + 2..pos + 4].try_into().ok()?) as usize;
+        pos += 4;
+        if pos + attr_len > data.len() {
+            break;
+        }
+        if attr_type == 0x0020 && attr_len >= 8 {
+            // byte 0: reserved, byte 1: family (0x01=IPv4, 0x02=IPv6)
+            let family = data[pos + 1];
+            let x_port = u16::from_be_bytes(data[pos + 2..pos + 4].try_into().ok()?);
+            let _ = x_port ^ (STUN_MAGIC >> 16) as u16; // port (unused here)
+
+            if family == 0x01 {
+                let x_addr = u32::from_be_bytes(data[pos + 4..pos + 8].try_into().ok()?);
+                let addr = x_addr ^ STUN_MAGIC;
+                return Some(std::net::IpAddr::V4(std::net::Ipv4Addr::from(addr)));
+            }
+        }
+        pos += (attr_len + 3) & !3;
+    }
+    None
+}

flake.lock

@@ -35,11 +35,11 @@
     },
     "nixpkgs": {
       "locked": {
-        "lastModified": 1780747962,
+        "lastModified": 1781454065,
-        "narHash": "sha256-IX7G1dlKrOqPOImfbo7ADDfV5yU1+j+MRChI3TL4tAA=",
+        "narHash": "sha256-d2xfDjnfRuf/xYGdu9VVRHiav/2w5hDL/5cw2TuVAXw=",
         "owner": "NixOS",
         "repo": "nixpkgs",
-        "rev": "cbb5cf358f50aa6acc9efd6113b7bcfbc352cd73",
+        "rev": "9eac87a12312b8f60dd52e1c6e1a265f6fc7f5fc",
         "type": "github"
       },
       "original": {

flake.nix

@@ -21,6 +21,7 @@
       system:
       let
         pkgs = nixpkgs.legacyPackages.${system};
+        inherit (pkgs) lib;
 
         craneLib = crane.mkLib pkgs;
 
@@ -30,48 +31,63 @@
           src = craneLib.cleanCargoSource ./.;
           strictDeps = true;
 
-          buildInputs = [
+          buildInputs =
-            # Add additional build inputs here
+            [ ]
-          ]
+            ++ pkgs.lib.optionals pkgs.stdenv.isDarwin [
-          ++ pkgs.lib.optionals pkgs.stdenv.isDarwin [
+              pkgs.libiconv
-            # Additional darwin specific inputs can be set here
+            ];
-            pkgs.libiconv
-          ];
         };
 
-        my-crate = craneLib.buildPackage (
+        fileSetForCrate =
+          crate:
+          lib.fileset.toSource {
+            root = ./.;
+            fileset = lib.fileset.unions [
+              ./Cargo.toml
+              ./Cargo.lock
+              (craneLib.fileset.commonCargoSources ./crates/entity)
+              (craneLib.fileset.commonCargoSources ./crates/migration)
+              (craneLib.fileset.commonCargoSources ./crates/server)
+              (craneLib.fileset.commonCargoSources crate)
+            ];
+          };
+
+        rtmp-to-whip-simple-server = craneLib.buildPackage (
           commonArgs
           // {
+            pname = "rtmp-to-whip";
+            version = "0.1.0";
             cargoArtifacts = craneLib.buildDepsOnly commonArgs;
-
+            cargoExtraArgs = "-p server";
-            # Additional environment variables or build phases/hooks can be set
+            src = fileSetForCrate ./crates/server;
-            # here *without* rebuilding all dependency crates
-            # MY_CUSTOM_VAR = "some value";
           }
         );
+
       in
       {
         checks = {
-          inherit my-crate;
+          inherit rtmp-to-whip-simple-server;
         };
 
-        packages.default = my-crate;
+        packages.default = rtmp-to-whip-simple-server;
 
         apps.default = flake-utils.lib.mkApp {
-          drv = my-crate;
+          drv = rtmp-to-whip-simple-server;
         };
 
         devShells.default = craneLib.devShell {
-          # Inherit inputs from checks.
           checks = self.checks.${system};
 
-          # Additional dev-shell environment variables can be set directly
+          ADMIN_REF_CODE = "meowmeowpurrrmeow";
-          # MY_CUSTOM_DEVELOPMENT_VAR = "something else";
 
-          # Extra inputs can be added here; cargo and rustc are provided by default.
           packages = [
             pkgs.clang
             pkgs.mold
+            pkgs.sea-orm-cli
+            pkgs.cmake
+            pkgs.opus
+            pkgs.pkgconf
+            pkgs.just
           ];
         };
       }

index.html

@@ -0,0 +1,119 @@
+<!DOCTYPE html>
+<html>
+<head>
+  <meta charset="utf-8">
+  <title>WHEP Viewer</title>
+  <style>
+    body { margin: 0; background: #111; color: #eee; font-family: monospace; }
+    #video { display: block; width: 100%; max-width: 800px; }
+    #stats {
+      max-width: 800px;
+      padding: 8px 12px;
+      background: rgba(0,0,0,0.6);
+      font-size: 13px;
+      line-height: 1.8;
+    }
+    .label { color: #aaa; }
+    .value { color: #7ef; font-weight: bold; }
+    .warn  { color: #fa0; }
+    .bad   { color: #f44; }
+  </style>
+</head>
+<body>
+  <video id="video" autoplay muted playsinline></video>
+  <div id="stats">Waiting for stream…</div>
+<script>
+  const fmt = ms => ms < 0 ? '—' : ms.toFixed(1) + ' ms';
+
+  const colorClass = ms => ms < 0 ? '' : ms < 80 ? 'value' : ms < 200 ? 'warn' : 'bad';
+
+  let prevStats = null;
+
+  async function pollStats(pc) {
+    const reports = await pc.getStats();
+
+    let inbound = null;
+    let networkRttMs = -1;
+    reports.forEach(r => {
+      if (r.type === 'inbound-rtp' && r.kind === 'video') inbound = r;
+      // The nominated ICE candidate pair carries the active STUN ping RTT.
+      if (r.type === 'candidate-pair' && r.nominated && r.currentRoundTripTime != null) {
+        networkRttMs = r.currentRoundTripTime * 1000;
+      }
+    });
+
+    if (!inbound) return;
+
+    // Network one-way: ICE STUN ping RTT / 2.
+    const networkOneWayMs = networkRttMs >= 0 ? networkRttMs / 2 : -1;
+
+    // Jitter buffer latency: average time a packet waits before being emitted to decoder.
+    const jitterMs = inbound.jitterBufferEmittedCount > 0
+      ? (inbound.jitterBufferDelay / inbound.jitterBufferEmittedCount) * 1000
+      : -1;
+
+    // Decode latency: average time spent decoding each frame.
+    const decodeMs = inbound.framesDecoded > 0
+      ? (inbound.totalDecodeTime / inbound.framesDecoded) * 1000
+      : -1;
+
+    // Server → client: network transit + jitter buffer + decode.
+    const serverToClientMs =
+      (networkOneWayMs >= 0 && jitterMs >= 0 && decodeMs >= 0)
+        ? networkOneWayMs + jitterMs + decodeMs
+        : -1;
+
+    // Frames per second (received from network, before decode).
+    const fps = inbound.framesPerSecond ?? -1;
+
+    // Packets lost ratio.
+    const totalPkts = (inbound.packetsReceived || 0) + (inbound.packetsLost || 0);
+    const lossRatio = totalPkts > 0
+      ? ((inbound.packetsLost || 0) / totalPkts * 100).toFixed(1) + '%'
+      : '—';
+
+    const el = document.getElementById('stats');
+    const row = (label, val, cls) =>
+      `<span class="label">${label}:</span> <span class="${cls}">${val}</span>`;
+
+    el.innerHTML = [
+      row('Server → Client',  fmt(serverToClientMs), colorClass(serverToClientMs)),
+      row('  Network (1-way)', fmt(networkOneWayMs),  colorClass(networkOneWayMs)),
+      row('  Jitter buffer',   fmt(jitterMs),         colorClass(jitterMs)),
+      row('  Decode',          fmt(decodeMs),          colorClass(decodeMs)),
+      row('FPS',               fps >= 0 ? fps.toFixed(1) : '—', 'value'),
+      row('Packet loss',       lossRatio, 'value'),
+      row('Jitter',            fmt(inbound.jitter * 1000), colorClass(inbound.jitter * 1000)),
+    ].join('<br>');
+
+    prevStats = inbound;
+  }
+
+  const start = async () => {
+    window.pc = new RTCPeerConnection();
+    const pc = window.pc;
+    pc.addTransceiver('video', { direction: 'recvonly' });
+    pc.ontrack = (e) => {
+      const video = document.getElementById('video');
+      video.srcObject = new MediaStream([e.track]);
+      video.play().catch(err => console.error('play() failed:', err));
+      // Start polling once we have a track.
+      setInterval(() => pollStats(pc), 500);
+    };
+    pc.oniceconnectionstatechange = () => console.log('ice state:', pc.iceConnectionState);
+
+    const offer = await pc.createOffer();
+    await pc.setLocalDescription(offer);
+    const res = await fetch('http://localhost:5000/whep/test', {
+      method: 'POST',
+      headers: { 'Content-Type': 'application/sdp' },
+      body: offer.sdp,
+    });
+    const answer = await res.text();
+    await pc.setRemoteDescription({ type: 'answer', sdp: answer });
+  };
+
+  start();
+</script>
+</body>
+</html>

src/main.rs

@@ -1,23 +0,0 @@
-use std::error::Error;
-
-use macro_rules_attribute::apply;
-use smol::{io::AsyncReadExt, net::TcpListener, stream::StreamExt};
-use smol_macros::main;
-
-#[apply(main!)]
-async fn main() -> Result<(), Box<dyn Error>> {
-    let listener = TcpListener::bind("0.0.0.0:8123").await?;
-    let mut incoming = listener.incoming();
-
-    while let Some(connection) = incoming.next().await {
-        let mut stream = connection?;
-        smol::spawn(async move {
-            let mut buf = vec![0; 1024];
-            stream.read(&mut buf).await.unwrap();
-            print!("{:#?}", buf);
-        })
-        .await;
-    }
-
-    Ok(())
-}