Bump: 0.5.0 : unlisted, password and locked stream support

Dockerfile

@@ -43,6 +43,6 @@ EXPOSE 1935
 EXPOSE 3000
 EXPOSE 6969/udp
 
-ENV RUST_LOG="info,warn"
+ENV RUST_LOG="info"
 
 CMD ["/rtmp-to-whip"]

Dockerfile.aarch64

@@ -55,4 +55,6 @@ EXPOSE 1935
 EXPOSE 3000
 EXPOSE 6969/udp
 
+ENV RUST_LOG="info"
+
 CMD ["/rtmp-to-whip"]

TODO.md

@@ -0,0 +1,36 @@
+# TODO
+
+Audit findings from ponytail-audit (ranked biggest cut first). Net: ~-300 lines, -5 deps possible.
+
+## Deletions
+
+- [ ] Delete `Dockerfile.aarch64` — merge into `Dockerfile` with `ARG TARGETARCH` (buildx sets it); both files are 90% identical except the dep-copy block. [Dockerfile.aarch64]
+- [ ] Delete entity dead methods: `update_username`, `update_password`, `change_stream_key_limit`, `find_by_user_id`, `get_stream_session`, `get_all_active_sessions`. Zero callers; scaffolding for the unbuilt admin panel. [crates/entity/src/users.rs:77, crates/entity/src/auth_session.rs:43, crates/entity/src/stream_session.rs:53]
+- [ ] Delete `/api/health`, `/api/uptime`, `/api/version` + `HealthResponse`/`UptimeResponse` — all subsets of `/api/stats`; also kills `serde_json` (only used by `version_handler`). [crates/server/src/http.rs:497]
+- [ ] Delete `SessionCookie` extractor — its value is only debug-logged, never used; kill the extractor + 2 logs. [crates/server/src/http.rs:213]
+- [ ] Delete `WebrtcProxy::local_addr()` (no callers), dead `let _ = x_port ^ …` stmt, and single-field `WebRtcProxyConfig` struct → pass `u16` (also fixes i32 port type). [crates/server/src/webrtc_proxy.rs:167, crates/server/src/webrtc_proxy.rs:139]
+- [ ] Delete `webrtc.rs` `mid` binding from `add_media` + `_hint_mid` param — passed straight into an underscore. [crates/server/src/webrtc.rs:142]
+- [ ] Delete `stream_key` `is_active`/`is_unlisted` columns — never read anywhere; `create()` hardcodes `is_active=false` and `is_unlisted` is always passed `false`. Drop param + column (migration, expand-contract). [crates/entity/src/stream_key.rs:45]
+- [ ] Delete `rtmp.rs` redundant `stream_id` var (warn can use `current_stream_key_id`), `warn!("")` empty-log on parse error, and `parse_video_codec`'s `Result<_, Box<dyn Error>>` → `Option`. [crates/server/src/rtmp.rs:206, crates/server/src/rtmp.rs:369]
+- [ ] Delete `catalog_handler`'s `get_all_active_sessions` query — result only feeds a `debug!`; kills the entity method too. [crates/server/src/http.rs:185]
+- [ ] Delete commented-out routes/code (`admin/server_stats`, duplicate whip route, `.max_age`, `fs::File`) + `#[axum::debug_handler]`. [crates/server/src/http.rs:109]
+- [ ] Delete `StreamSession.active_clients` — written 0, never read; drops `AtomicU32` import. [crates/server/src/main.rs:65]
+- [ ] Delete `webrtc_ingest` `_connected` flag — set true, never read. [crates/server/src/webrtc_ingest.rs:363]
+- [ ] Delete `users.rs` `let user = …; user` pointless binding in `find_by_username`. [crates/entity/src/users.rs:70]
+- [ ] Delete `meow_handler` + route — joke endpoint, zero consumers. [crates/server/src/http.rs:492]
+- [ ] Delete `index.html` `prevStats` — assigned, never read. [index.html:57]
+- [ ] Delete deps — server: `futures`, `rand`; entity: `rand`, `argon2` (hashing lives in server crate now); `serde_json` (with health/uptime/version cut). [crates/server/Cargo.toml, crates/entity/Cargo.toml]
+
+## Shrinks
+
+- [ ] Extract shared "remove session + finish_stream_session" fn — closure duplicated 3× (rtmp.rs cleanup, webrtc_ingest detach cleanup, whip DELETE handler). Also lets `handle_whip_injest` drop its manual `err()` closure for `Result<HttpError>` like its siblings. [crates/server/src/rtmp.rs:141, crates/server/src/webrtc_ingest.rs:338, crates/server/src/webrtc_ingest.rs:34]
+- [ ] Extract `fixup_answer_sdp()` — `whip_sdp_probe.rs` re-encodes the SDP string-fixups verbatim; probe drops ~30 lines. [crates/server/tests/whip_sdp_probe.rs:66]
+- [ ] Derive `thiserror` on `AudioParseError` instead of hand-rolled `Display`+`Error` impls (~15 lines). [crates/server/src/audio.rs:66]
+- [ ] Collapse `webrtc.rs` four near-identical channel re-subscribe blocks (`is_none`/`is_closed` × video/audio) → one helper. [crates/server/src/webrtc.rs:336]
+
+## Out of scope (correctness — route to normal review)
+
+- `Local::now()` vs `Utc::now()` inconsistency in session creation
+- `KEY_RE` `{1,67}` vs `MAX_LABEL_LEN = 64` mismatch
+- `index.html` hardcoded port 5000
+- `/api/stream/{slug}` unauthenticated

crates/entity/src/stream_key.rs

@@ -12,6 +12,8 @@ pub struct Model {
     pub label: String,
     pub is_active: bool,
     pub is_unlisted: bool,
+    pub password: Option<String>,
+    pub custom_id: Option<String>,
     pub created_at: DateTimeUtc,
 }
 
@@ -78,6 +80,16 @@ impl Entity {
             .all(db)
             .await
     }
+
+    pub async fn find_by_custom_id(
+        db: &DatabaseConnection,
+        custom_id: String,
+    ) -> Result<Option<Model>, DbErr> {
+        Entity::find()
+            .filter(Column::CustomId.eq(custom_id))
+            .one(db)
+            .await
+    }
 }
 
 impl ActiveModel {
@@ -89,4 +101,17 @@ impl ActiveModel {
         self.label = Set(value);
         self.update(db).await
     }
+
+    pub async fn password(
+        mut self,
+        db: &DatabaseConnection,
+        value: String,
+    ) -> Result<Model, DbErr> {
+        self.password = if value.is_empty() {
+            Set(None)
+        } else {
+            Set(Some(value))
+        };
+        self.update(db).await
+    }
 }

crates/migration/src/lib.rs

@@ -4,6 +4,8 @@ mod m20260616_000001_create_users;
 mod m20260616_000002_create_stream_key;
 mod m20260616_000003_create_stream_session;
 mod m20260616_000004_create_auth_session;
+mod m20260815_000005_add_password_to_stream_key;
+mod m20260815_000006_set_stream_keys_unlisted;
 
 pub struct Migrator;
 
@@ -15,6 +17,8 @@ impl MigratorTrait for Migrator {
             Box::new(m20260616_000002_create_stream_key::Migration),
             Box::new(m20260616_000003_create_stream_session::Migration),
             Box::new(m20260616_000004_create_auth_session::Migration),
+            Box::new(m20260815_000005_add_password_to_stream_key::Migration),
+            Box::new(m20260815_000006_set_stream_keys_unlisted::Migration),
         ]
     }
 }

crates/migration/src/m20260616_000002_create_stream_key.rs

@@ -38,7 +38,7 @@ impl MigrationTrait for Migration {
                         ColumnDef::new(StreamKey::IsUnlisted)
                             .boolean()
                             .not_null()
-                            .default(true),
+                            .default(false),
                     )
                     .col(ColumnDef::new(StreamKey::CreatedAt).date_time().not_null())
                     .foreign_key(
@@ -67,5 +67,7 @@ pub enum StreamKey {
     Label,
     IsActive,
     IsUnlisted,
+    Password,
+    CustomId,
     CreatedAt,
 }

crates/migration/src/m20260815_000005_add_password_to_stream_key.rs

@@ -0,0 +1,47 @@
+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
+            .alter_table(
+                Table::alter()
+                    .table(StreamKey::Table)
+                    .add_column(ColumnDef::new(StreamKey::Password).string().null())
+                    .to_owned(),
+            )
+            .await?;
+        manager
+            .alter_table(
+                Table::alter()
+                    .table(StreamKey::Table)
+                    .add_column(ColumnDef::new(StreamKey::CustomId).string().null())
+                    .to_owned(),
+            )
+            .await
+    }
+
+    async fn down(&self, manager: &SchemaManager) -> Result<(), DbErr> {
+        manager
+            .alter_table(
+                Table::alter()
+                    .table(StreamKey::Table)
+                    .drop_column(StreamKey::Password)
+                    .to_owned(),
+            )
+            .await?;
+        manager
+            .alter_table(
+                Table::alter()
+                    .table(StreamKey::Table)
+                    .drop_column(StreamKey::CustomId)
+                    .to_owned(),
+            )
+            .await
+    }
+}

crates/migration/src/m20260815_000006_set_stream_keys_unlisted.rs

@@ -0,0 +1,26 @@
+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> {
+        // By accident the is_unlisted column defaulted to true; existing stream
+        // keys were meant to be listed. Reset all rows to false here.
+        manager
+            .exec_stmt(
+                Query::update()
+                    .table(StreamKey::Table)
+                    .values([(StreamKey::IsUnlisted, false.into())])
+                    .to_owned(),
+            )
+            .await
+    }
+
+    async fn down(&self, _manager: &SchemaManager) -> Result<(), DbErr> {
+        Ok(())
+    }
+}

crates/server/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "server"
-version = "0.4.0"
+version = "0.5.0"
 edition = "2024"
 
 [target.x86_64-unknown-linux-gnu]

crates/server/src/http.rs

@@ -22,7 +22,7 @@ use axum::{
     Json, Router,
     extract::{FromRequestParts, Path, State},
     http::{
-        HeaderName, Method, StatusCode,
+        HeaderMap, HeaderName, Method, StatusCode,
         header::{ACCEPT, AUTHORIZATION, CONTENT_TYPE},
         request::Parts,
     },
@@ -31,7 +31,7 @@ use axum::{
 };
 use chrono::{DateTime, Utc};
 use entity::{auth_session, stream_key, stream_session, users};
-use sea_orm::{DatabaseConnection, EntityTrait, IntoActiveModel};
+use sea_orm::{ActiveModelTrait, DatabaseConnection, EntityTrait, IntoActiveModel, Set};
 use serde::{Deserialize, Serialize};
 use sysinfo::System;
 use tokio::{
@@ -63,7 +63,7 @@ pub struct ServerInfo {
 
 pub struct HttpServer {
     pub offer_tx: Sender<(i32, i32, String)>,
-    pub accept_rx: async_broadcast::InactiveReceiver<(i32, Option<String>)>,
+    pub accept_rx: async_broadcast::InactiveReceiver<(i32, Result<String, String>)>,
     pub appstate: Arc<Mutex<AppState>>,
     pub request_count: AtomicI32,
     pub db: DatabaseConnection,
@@ -149,7 +149,14 @@ struct StreamListing {
 #[derive(Deserialize)]
 struct EditStreamKeyRequest {
     id: i32,
-    new: String,
+    #[serde(default)]
+    new: Option<String>,
+    #[serde(default)]
+    password: Option<String>,
+    #[serde(default)]
+    unlisted: Option<bool>,
+    #[serde(default)]
+    custom_id: Option<String>,
 }
 
 async fn edit_stream_key(
@@ -158,14 +165,16 @@ async fn edit_stream_key(
     Json(payload): Json<EditStreamKeyRequest>,
 ) -> Result<impl IntoResponse, HttpError> {
     // Trim surrounding whitespace so labels aren't stored with leading/trailing spaces.
-    let new_label = payload.new.trim();
+    let new_label = payload.new.as_deref().map(str::trim);
     // Length (1..=67) and allowed charset ([A-Za-z0-9 _-], space included).
-    if !KEY_RE.is_match(new_label) {
+    if let Some(label) = new_label {
-        return Err(HttpError::BadRequest("invalid label".into()));
+        if !KEY_RE.is_match(label) {
-    }
+            return Err(HttpError::BadRequest("invalid label".into()));
-    // Replaces the JS lookahead: label must contain at least one letter.
+        }
-    if !new_label.chars().any(|c| c.is_ascii_alphabetic()) {
+        // Replaces the JS lookahead: label must contain at least one letter.
-        return Err(HttpError::BadRequest("label must contain a letter".into()));
+        if !label.chars().any(|c| c.is_ascii_alphabetic()) {
+            return Err(HttpError::BadRequest("label must contain a letter".into()));
+        }
     }
 
     let stream_key = stream_key::Entity::find_by_id(payload.id)
@@ -176,10 +185,60 @@ async fn edit_stream_key(
     if stream_key.user_id != auth.0.id {
         return Err(HttpError::Forbidden);
     }
-    stream_key
+
-        .into_active_model()
+    let lock = state.appstate.lock().await;
-        .change_label_value(&state.db, new_label.to_string())
+    let mut ses = { lock.stream_sessions.get_mut(&payload.id) };
-        .await?;
+    // drop(lock);
+
+    let mut am = stream_key.into_active_model();
+    if let Some(custom_id) = payload.custom_id {
+        am.custom_id = if custom_id.is_empty() {
+            if let Some(ref mut ses) = ses {
+                ses.custom_id = None;
+            }
+            Set(None)
+        } else {
+            // Check for conflict.
+            if stream_key::Entity::find_by_custom_id(&state.db, custom_id.clone())
+                .await?
+                .is_some()
+            {
+                return Err(HttpError::Conflict);
+            }
+            if let Some(ref mut ses) = ses {
+                ses.custom_id = Some(custom_id.clone());
+            }
+            Set(Some(custom_id))
+        };
+    }
+    if let Some(label) = new_label {
+        if let Some(ref mut ses) = ses {
+            ses.stream_key_label = label.to_string();
+        }
+        am.label = Set(label.to_string());
+    }
+    if let Some(pwd) = payload.password {
+        am.password = if pwd.is_empty() {
+            if let Some(ref mut ses) = ses {
+                ses.password = None;
+            }
+            Set(None)
+        } else {
+            if let Some(ref mut ses) = ses {
+                ses.password = Some(pwd.clone());
+            }
+            Set(Some(pwd))
+        };
+    }
+    if let Some(unlisted) = payload.unlisted {
+        if let Some(ref mut ses) = ses {
+            ses.is_unlisted = unlisted;
+        }
+        am.is_unlisted = Set(unlisted);
+    }
+    // This hopefully will not fail, if it does, our values for the stream key will be mismatched,
+    // that would be bad
+    am.update(&state.db).await?;
 
     Ok(StatusCode::OK)
 }
@@ -195,6 +254,7 @@ async fn catalog_handler(
         .await
         .stream_sessions
         .iter()
+        .filter(|x| !x.is_unlisted)
         .map(|x| StreamListing {
             id: x.stream_key_id,
             label: x.stream_key_label.clone(),
@@ -412,16 +472,18 @@ async fn create_user_handler(
 async fn stream_handler(
     State(state): State<Arc<HttpServer>>,
     Path(slug): Path<String>,
+    headers: HeaderMap,
     body: String,
 ) -> Result<impl IntoResponse, HttpError> {
     let request_id_clone = state.request_count.fetch_add(1, Ordering::Relaxed);
+    let app = state.appstate.lock().await;
 
     let stream_key_id = {
-        let app = state.appstate.lock().await;
+        app.stream_sessions.iter().find(|e| {
-        app.stream_sessions
+            e.value().stream_key_id.to_string() == slug
-            .iter()
+                || e.value().custom_id.as_deref() == Some(slug.as_str())
-            .find(|e| e.value().stream_key_id.to_string() == slug)
+        })
-            .map(|e| *e.key())
+        // .map(|e| *e.key())
     };
 
     let stream_key_id = stream_key_id.ok_or_else(|| {
@@ -429,12 +491,29 @@ async fn stream_handler(
         HttpError::NotFound
     })?;
 
-    info!(request_id = request_id_clone, slug = %slug, stream_key_id, "WHEP offer received");
+    // Dont return stream by its id in the db if its unlisted
+    if stream_key_id.is_unlisted && stream_key_id.custom_id.clone().ok_or("") != Ok(slug.clone()) {
+        return Err(HttpError::NotFound);
+    }
+
+    let auth_header = headers.get("auth");
+
+    if let Some(password) = &stream_key_id.password {
+        if let Some(auth) = auth_header {
+            if auth.to_str().unwrap() != password {
+                return Err(HttpError::Unauthorized);
+            }
+        } else {
+            return Err(HttpError::Unauthorized);
+        };
+    };
+
+    info!(request_id = request_id_clone, slug = %slug, stream_key_id.stream_key_id, "WHEP offer received");
     let accept_rx = state.accept_rx.activate_cloned();
     // The webrtc worker owning the receiver died if this fails.
     state
         .offer_tx
-        .send((request_id_clone, stream_key_id, body))
+        .send((request_id_clone, stream_key_id.stream_key_id, body))
         .await
         .map_err(|_| HttpError::Internal)?;
     debug!(
@@ -456,18 +535,15 @@ async fn stream_handler(
     })
     .await
     {
-        Ok(Some(Some(reply))) => {
+        Ok(Some(Ok(reply))) => {
             return Ok(Response::builder()
                 .status(StatusCode::CREATED)
                 .header("content-type", "application/sdp")
                 .body(reply)
                 .unwrap());
         }
-        Ok(Some(None)) => {
+        Ok(Some(Err(codec))) => {
-            return Ok(Response::builder()
+            return Err(HttpError::WhepCodecError(codec));
-                .status(StatusCode::UNSUPPORTED_MEDIA_TYPE)
-                .body(String::new())
-                .unwrap());
         }
         Ok(None) => {
             info!(

crates/server/src/http_error.rs

@@ -25,6 +25,8 @@ pub enum HttpError {
     Unprocessable(String),
     #[error("not acceptable: {0}")]
     NotAcceptable(String),
+    #[error("unsupported codec: {0}")]
+    WhepCodecError(String),
     #[error("internal error")]
     Internal,
 }
@@ -40,6 +42,7 @@ impl HttpError {
             Self::BadRequest(_) => StatusCode::BAD_REQUEST,
             Self::Unprocessable(_) => StatusCode::UNPROCESSABLE_ENTITY,
             Self::NotAcceptable(_) => StatusCode::NOT_ACCEPTABLE,
+        Self::WhepCodecError(_) => StatusCode::UNSUPPORTED_MEDIA_TYPE,
         }
     }
 }
@@ -47,12 +50,14 @@ impl HttpError {
 impl IntoResponse for HttpError {
     fn into_response(self) -> Response<Body> {
         let status = self.status();
-        // Only surface a message body for client (4xx) errors; keep an empty
+        let body = match self {
-        // body for 5xx so internal details aren't leaked.
+            // The WHEP client (frontend) reads this body as the rejected
-        let body = if status.is_client_error() {
+            // codec, so send it bare rather than the full error string.
-            self.to_string()
+            Self::WhepCodecError(codec) => codec,
-        } else {
+            // Only surface a message body for client (4xx) errors; keep an
-            String::new()
+            // empty body for 5xx so internal details aren't leaked.
+            e if status.is_client_error() => e.to_string(),
+            _ => String::new(),
         };
         (status, body).into_response()
     }

crates/server/src/main.rs

@@ -50,6 +50,9 @@ pub struct StreamSession {
     pub stream_key_id: i32,
     pub stream_key_label: String,
     pub stream_key_user: String,
+    pub custom_id: Option<String>,
+    pub is_unlisted: bool,
+    pub password: Option<String>,
     pub frame_channel: async_broadcast::Sender<Arc<VideoFrame>>,
     pub audio_channel: async_broadcast::Sender<Arc<OpusAudioFrame>>,
     pub codec: Option<StreamCodec>,
@@ -98,11 +101,12 @@ async fn main() -> Result<(), Box<dyn Error>> {
         webrtc_proxy: proxy.clone(),
     }));
     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);
+    let (answer_tx, answer_rx) = broadcast::<(i32, Result<String, String>)>(64);
     // Request_Id,
     // Answer_body
 

crates/server/src/rtmp.rs

@@ -14,7 +14,7 @@ use tokio::{
     net::{TcpListener, TcpStream},
     time::timeout,
 };
-use tracing::{debug, info, warn};
+use tracing::{debug, info, trace, warn};
 
 use crate::{
     StreamCodec, StreamSession,
@@ -312,6 +312,9 @@ impl Rtmp {
                                             stream_key_id: key.id,
                                             stream_key_label: key.label,
                                             stream_key_user: user.username,
+                                            custom_id: key.custom_id,
+                                            is_unlisted: key.is_unlisted,
+                                            password: key.password,
                                             frame_channel: video_tx.clone(),
                                             audio_channel: audio_tx.clone(),
                                             codec: None,
@@ -406,7 +409,7 @@ impl Rtmp {
                                                 let pkt_type = data[0] & 0x0F;
                                                 match p.parse(&data, timestamp.value) {
                                                     Some(frame) => {
-                                                        debug!(
+                                                        trace!(
                                                             pkt_type,
                                                             is_keyframe = frame.is_keyframe,
                                                             ts = frame.timestamp_ms,
@@ -419,7 +422,7 @@ impl Rtmp {
                                                             .ok();
                                                     }
                                                     None => {
-                                                        debug!(
+                                                        trace!(
                                                             pkt_type,
                                                             "AV1 packet produced no frame (seq header or unknown type)"
                                                         );

crates/server/src/webrtc.rs

@@ -4,7 +4,7 @@ use entity::stream_key;
 use sea_orm::{DatabaseConnection, EntityTrait};
 use std::{net::SocketAddr, sync::Arc, time::Instant};
 use tokio::{net::UdpSocket, sync::mpsc::Receiver};
-use tracing::{debug, error, info, warn};
+use tracing::{debug, error, info, trace, warn};
 
 use str0m::{
     Candidate, Event, Input, Output, Rtc,
@@ -18,11 +18,11 @@ use crate::{
 };
 
 pub struct Webrtc {
-pub offer_rx: Receiver<(i32, i32, String)>,
+    pub offer_rx: Receiver<(i32, i32, String)>,
-pub accept_tx: async_broadcast::Sender<(i32, Option<String>)>,
+    pub accept_tx: async_broadcast::Sender<(i32, Result<String, String>)>,
-pub sessions_ref: Arc<DashMap<i32, StreamSession>>,
+    pub sessions_ref: Arc<DashMap<i32, StreamSession>>,
-pub proxy: Arc<WebrtcProxy>,
+    pub proxy: Arc<WebrtcProxy>,
-pub db: DatabaseConnection,
+    pub db: DatabaseConnection,
 }
 
 impl Webrtc {
@@ -40,7 +40,10 @@ impl Webrtc {
                     request_id,
                     stream_id, "stream key not found in DB, rejecting offer"
                 );
-                self.accept_tx.broadcast((request_id, None)).await.unwrap();
+                self.accept_tx
+                    .broadcast((request_id, Err(String::new())))
+                    .await
+                    .unwrap();
                 continue;
             }
             if let Err(ref e) = stream_key {
@@ -48,7 +51,10 @@ impl Webrtc {
                     request_id,
                     stream_id, "DB error looking up stream key: {:?}", e
                 );
-                self.accept_tx.broadcast((request_id, None)).await.unwrap();
+                self.accept_tx
+                    .broadcast((request_id, Err(String::new())))
+                    .await
+                    .unwrap();
                 continue;
             }
 
@@ -134,7 +140,10 @@ impl Webrtc {
                 Ok(sdp) => sdp,
                 Err(e) => {
                     warn!(request_id, stream_id, "malformed SDP offer: {:?}", e);
-                    self.accept_tx.broadcast((request_id, None)).await.unwrap();
+                    self.accept_tx
+                        .broadcast((request_id, Err(String::new())))
+                        .await
+                        .unwrap();
                     continue;
                 }
             };
@@ -162,7 +171,7 @@ impl Webrtc {
                 }
             };
             let answer_sdp = offer_answer.to_sdp_string();
-            info!(request_id, "SDP answer:\n{}", answer_sdp);
+            trace!(request_id, "SDP answer:\n{}", answer_sdp);
 
             // Detect the case where str0m couldn't match any video codec.
             // str0m serialises the m-line with an empty PT list, which is invalid SDP
@@ -179,7 +188,10 @@ impl Webrtc {
                     "no video codec negotiated — browser likely doesn't support {:?}; rejecting offer",
                     stream_codec
                 );
-                self.accept_tx.broadcast((request_id, None)).await.unwrap();
+                self.accept_tx
+                    .broadcast((request_id, Err(format!("{:?}", stream_codec))))
+                    .await
+                    .unwrap();
                 continue;
             }
 
@@ -193,7 +205,7 @@ impl Webrtc {
 
             debug!(request_id, "sending answer back");
             self.accept_tx
-                .broadcast((request_id, Some(answer_sdp)))
+                .broadcast((request_id, Ok(answer_sdp)))
                 .await
                 .unwrap();
 
@@ -310,8 +322,8 @@ impl Webrtc {
                         }
                         _ => {}
                     },
-                    Err(e) => {
+                    Err(_e) => {
-                        error!("poll_output error (connection closing): {:?}", e);
+                        // error!("poll_output error (connection closing): {:?}", e);
                         return;
                     }
                 }
@@ -487,6 +499,5 @@ impl Webrtc {
         {
             warn!("RTP write error: {:?}", e);
         }
+    }
 }
-}
-    

crates/server/src/webrtc_ingest.rs

@@ -1,4 +1,8 @@
-use std::{net::SocketAddr, sync::Arc, time::Instant};
+use std::{
+    net::SocketAddr,
+    sync::Arc,
+    time::{Duration, Instant},
+};
 
 use async_broadcast::broadcast;
 use axum::{
@@ -18,13 +22,16 @@ use str0m::{
     net::{Protocol, Receive},
 };
 use tokio::{net::UdpSocket, sync::mpsc::Receiver};
-use tracing::{debug, error, info, warn};
+use tracing::{debug, error, info, trace, warn};
 
 use crate::{
     StreamSession, audio::OpusAudioFrame, codec::VideoFrame, http::HttpServer,
     http_error::HttpError,
 };
 
+/// Kill a WHIP publish if it delivers no media (video or audio) this long.
+const NO_MEDIA_TIMEOUT: Duration = Duration::from_secs(30);
+
 pub async fn handle_whip_injest_delete(
     State(state): State<Arc<HttpServer>>,
     ConnectInfo(remote): ConnectInfo<SocketAddr>,
@@ -295,6 +302,9 @@ pub async fn handle_whip_injest(
             stream_key_id: key.id,
             stream_key_label: key.label,
             stream_key_user: user.username,
+            custom_id: key.custom_id,
+            is_unlisted: key.is_unlisted,
+            password: key.password,
             frame_channel: video_tx,
             audio_channel: audio_tx,
             codec: negotiated_codec,
@@ -343,12 +353,12 @@ pub async fn handle_whip_injest(
         .status(StatusCode::CREATED)
         .header(header::CONTENT_TYPE, "application/sdp")
         .header(header::LOCATION, &location)
-        // Provide a STUN server via Link header so OBS can gather
+        // // Provide a STUN server via Link header so OBS can gather
-        // ICE candidates even without explicit STUN configuration.
+        // // ICE candidates even without explicit STUN configuration.
-        .header(
+        // .header(
-            header::LINK,
+        //     header::LINK,
-            "<stun:stun.l.google.com:19302>; rel=\"ice-server\"",
+        //     "<stun:stun.l.google.com:19302>; rel=\"ice-server\"",
-        )
+        // )
         .body(axum::body::Body::from(answer_sdp))
         .unwrap()
 }
@@ -369,6 +379,7 @@ async fn detach_inject_rtc(
         let sessions_ref = sessions_ref.clone();
         let db = db.clone();
         async move {
+            debug!("Cleaning up {:?}", &stream_key_id);
             sessions_ref.remove(&stream_key_id);
             if let Ok(Some(s)) =
                 stream_session::Model::get_active_by_stream_key_id(&db, stream_key_id).await
@@ -385,6 +396,7 @@ async fn detach_inject_rtc(
     let mut video_tx: Option<async_broadcast::Sender<Arc<VideoFrame>>> = None;
     let mut audio_tx: Option<async_broadcast::Sender<Arc<OpusAudioFrame>>> = None;
     let mut _connected = false;
+    let mut disconnect_timer: Option<Instant> = None;
 
     loop {
         // Blankly using them so they don't drop (like RTMP).
@@ -395,7 +407,8 @@ async fn detach_inject_rtc(
             match rtc.poll_output() {
                 Ok(Output::Timeout(t)) => break t,
                 Ok(Output::Transmit(t)) => {
-                    debug!(
+                    // The keep alive loop, by default is every 1 second.
+                    trace!(
                         "Whip TX: {} bytes → {}:{}",
                         t.contents.len(),
                         t.destination.ip(),
@@ -406,6 +419,22 @@ async fn detach_inject_rtc(
                         cleanup().await;
                         return;
                     }
+                    // 30 Sec time out if disconnected, then we clean up and disconnect.
+                    if let Some(instant_since_disconnet) = disconnect_timer
+                        && Instant::now()
+                            .duration_since(instant_since_disconnet)
+                            .as_secs()
+                            > NO_MEDIA_TIMEOUT.as_secs()
+                    {
+                        info!(
+                            "WHIP connection on stream_key_id: {:?}, has been disconnected for {:?} seconds. Cleaning up and destroying the connection,",
+                            stream_key_id,
+                            NO_MEDIA_TIMEOUT.as_secs()
+                        );
+                        cleanup().await;
+                        rtc.disconnect();
+                        return;
+                    }
                 }
                 Ok(Output::Event(e)) => match e {
                     Event::MediaAdded(ma) => {
@@ -445,10 +474,19 @@ async fn detach_inject_rtc(
                     }
                     Event::IceConnectionStateChange(state) => {
                         info!(stream_key_id, ?state, "Whip ICE state change");
-                        if matches!(state, str0m::IceConnectionState::Disconnected) {
+                        match state {
-                            info!("Whip ICE disconnected, closing connection");
+                            str0m::IceConnectionState::Disconnected => {
-                            cleanup().await;
+                                info!(
-                            return;
+                                    "Whip ICE disconnected... (State changed to Disconnected for {:?}) ((This is usually due to network jitter))",
+                                    &stream_key_id
+                                );
+                                disconnect_timer = Some(Instant::now());
+                            }
+                            str0m::IceConnectionState::Connected
+                            | str0m::IceConnectionState::Completed => {
+                                disconnect_timer = None;
+                            }
+                            _ => {}
                         }
                     }
                     Event::Connected => {
@@ -498,13 +536,21 @@ async fn detach_inject_rtc(
                         }
                     }
                 }
-                result = rx.recv() => {
+                    // No UDP input at all for 30s: closing connection.
-                let Some((data, from)) = result else {
+                    // (ICE keepalives still arrive when only the encoder is stalled,
-                    info!(stream_key_id, "Whip proxy channel closed, cleaning up");
+                    // so this fires on a dead peer, not a paused one.)
-                    cleanup().await;
+                    result = tokio::time::timeout(NO_MEDIA_TIMEOUT, rx.recv()) => {
-                    return;
+                    let Ok(Some((data, from))) = result else {
-                };
+                        if result.is_err() {
-                debug!(
+                            warn!(stream_key_id, "Whip: no input for {NO_MEDIA_TIMEOUT:?}, closing connection");
+                            rtc.disconnect();
+                        } else {
+                            info!(stream_key_id, "Whip proxy channel closed, cleaning up");
+                        }
+                        cleanup().await;
+                        return;
+                    };
+                trace!(
                     stream_key_id,
                     len = data.len(),
                     %from,

crates/server/src/webrtc_proxy.rs

@@ -11,7 +11,7 @@ use tokio::{
     net::UdpSocket,
     sync::mpsc::{self, Receiver},
 };
-use tracing::{debug, error, info, warn};
+use tracing::{debug, error, info, trace, warn};
 
 pub struct WebRtcProxyConfig {
     pub proxy_port: i32,
@@ -97,7 +97,7 @@ impl WebrtcProxy {
 
                 // By addr
                 if let Some(tx) = by_addr.get(&from) {
-                    debug!(
+                    trace!(
                         "proxy: routing {} bytes by addr {}:{} → channel",
                         b,
                         from.ip(),
@@ -120,7 +120,7 @@ impl WebrtcProxy {
                 };
 
                 let Some((part1, part2)) = self::WebrtcProxy::ufrag_pair(&data) else {
-                    debug!("huh, packet isnt stun or added as client.");
+                    trace!("huh, packet isnt stun or added as client.");
                     continue;
                 };
                 // Try both parts of the STUN username — the first packet

flake.lock

@@ -53,7 +53,28 @@
       "inputs": {
         "crane": "crane",
         "flake-utils": "flake-utils",
-        "nixpkgs": "nixpkgs"
+        "nixpkgs": "nixpkgs",
+        "rust-overlay": "rust-overlay"
+      }
+    },
+    "rust-overlay": {
+      "inputs": {
+        "nixpkgs": [
+          "nixpkgs"
+        ]
+      },
+      "locked": {
+        "lastModified": 1786160316,
+        "narHash": "sha256-oLoc3ZLg1LX/S5Jb3v6MrF415AzDyC4vgeWy9UcYTQk=",
+        "owner": "oxalica",
+        "repo": "rust-overlay",
+        "rev": "4e1c940c96560ceab7c547f89642231371a66646",
+        "type": "github"
+      },
+      "original": {
+        "owner": "oxalica",
+        "repo": "rust-overlay",
+        "type": "github"
       }
     },
     "systems": {

flake.nix

@@ -7,6 +7,12 @@
     crane.url = "github:ipetkov/crane";
 
     flake-utils.url = "github:numtide/flake-utils";
+
+    rust-overlay = {
+      url = "github:oxalica/rust-overlay";
+      inputs.nixpkgs.follows = "nixpkgs";
+    };
+
   };
 
   outputs =
@@ -16,14 +22,27 @@
       crane,
       flake-utils,
       ...
-    }:
+    }@inputs:
     flake-utils.lib.eachDefaultSystem (
       system:
       let
-        pkgs = nixpkgs.legacyPackages.${system};
+        pkgs = import inputs.nixpkgs {
+          inherit system;
+          overlays = [ (import inputs.rust-overlay) ];
+        };
+
         inherit (pkgs) lib;
 
-        craneLib = crane.mkLib pkgs;
+        craneLib = (inputs.crane.mkLib pkgs).overrideToolchain (
+          p:
+          p.rust-bin.nightly.latest.default.override {
+            extensions = [
+              "rustc-codegen-cranelift-preview"
+              "rust-analyzer"
+              "rust-src"
+            ];
+          }
+        );
 
         # Common arguments can be set here to avoid repeating them later
         # Note: changes here will rebuild all dependency crates
@@ -56,7 +75,7 @@
           commonArgs
           // {
             pname = "rtmp-to-whip";
-            version = "0.1.0";
+            version = "0.4.0";
             cargoArtifacts = craneLib.buildDepsOnly commonArgs;
             cargoExtraArgs = "-p server";
             src = fileSetForCrate ./crates/server;