cargo clippy fix

Cargo.lock

@@ -2856,9 +2856,9 @@ dependencies = [
 
 [[package]]
 name = "serde_json"
-version = "1.0.150"
+version = "1.0.151"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "e8014e44b4736ed0538adeecded0fce2a272f22dc9578a7eb6b2d9993c74cfb9"
+checksum = "c841b55ecdae098c80dcae9cf767f6f8a0c2cdb3416bbef72181df4d0fe73f14"
 dependencies = [
  "itoa",
  "memchr",
@@ -2892,7 +2892,7 @@ dependencies = [
 
 [[package]]
 name = "server"
-version = "0.5.4"
+version = "0.6.0"
 dependencies = [
  "argon2",
  "async-broadcast",
@@ -2908,6 +2908,7 @@ dependencies = [
  "rubato",
  "sea-orm",
  "serde",
+ "serde_json",
  "str0m",
  "symphonia",
  "sysinfo",

crates/entity/src/stream_session.rs

@@ -51,20 +51,20 @@ impl Model {
         .await
     }
     pub async fn get_all_active_sessions(db: &DatabaseConnection) -> Result<Vec<Model>, DbErr> {
-        Ok(Entity::find()
+        Entity::find()
             .filter(Column::EndedAt.is_null())
             .all(db)
-            .await?)
+            .await
     }
     pub async fn get_active_by_stream_key_id(
         db: &DatabaseConnection,
         stream_key_id: i32,
     ) -> Result<Option<Model>, DbErr> {
-        Ok(Entity::find()
+        Entity::find()
             .filter(Column::StreamKeyId.eq(stream_key_id))
             .filter(Column::EndedAt.is_null())
             .one(db)
-            .await?)
+            .await
     }
 
     pub async fn clean_unended_streams(db: &DatabaseConnection) -> Result<u64, DbErr> {

crates/entity/src/users.rs

@@ -58,18 +58,18 @@ impl Entity {
         if let Some(x) = sessions {
             Entity::find_by_id(x.id_user).one(db).await
         } else {
-            return Ok(None);
+            Ok(None)
         }
     }
     pub async fn find_by_username(
         db: &DatabaseConnection,
         username: String,
     ) -> Result<Option<Model>, DbErr> {
-        let user = Entity::find()
+        
+        Entity::find()
             .filter(Column::Username.eq(username))
             .one(db)
-            .await;
-        user
+            .await
     }
 }
 

crates/server/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "server"
-version = "0.5.4"
+version = "0.6.0"
 edition = "2024"
 
 [target.x86_64-unknown-linux-gnu]
@@ -36,3 +36,4 @@ time = "0.3"
 thiserror = "2.0.18"
 sysinfo = "0.36"
 axum-extra = { version = "0.12.6", features = ["cookie"] }
+serde_json = "1.0.151"

crates/server/src/audio.rs

@@ -44,7 +44,7 @@ impl AudioProcesser {
     pub fn encode(&mut self, frame: AudioFrame) -> Vec<OpusAudioFrame> {
         let mut samples: Vec<f32> = frame
             .data
-            .chunks_exact(4)
+            .as_chunks::<4>().0.iter()
             .map(|b| f32::from_le_bytes([b[0], b[1], b[2], b[3]]))
             .collect();
 

crates/server/src/http.rs

@@ -125,9 +125,11 @@ impl HttpServer {
 #[derive(Serialize)]
 struct StreamListing {
     label: String,
+    custom_url_label: String,
     id: i32,
     user: String,
-    started_at: DateTime<Utc>, //UNIX TIMESTAMP
+    started_at: DateTime<Utc>,
+    is_password_protected: bool,
 }
 
 #[derive(Deserialize)]
@@ -160,16 +162,14 @@ async fn edit_stream_key(
         }
     }
     // Empty values are allowed (they clear the field).
-    if let Some(custom_id) = payload.custom_id.as_deref() {
-        if !custom_id.is_empty() && !valid_charset(custom_id) {
+    if let Some(custom_id) = payload.custom_id.as_deref()
+        && !custom_id.is_empty() && !valid_charset(custom_id) {
             return Err(HttpError::BadRequest("invalid custom id".into()));
         }
-    }
-    if let Some(pwd) = payload.password.as_deref() {
-        if !pwd.is_empty() && !valid_charset(pwd) {
+    if let Some(pwd) = payload.password.as_deref()
+        && !pwd.is_empty() && !valid_charset(pwd) {
             return Err(HttpError::BadRequest("invalid password".into()));
         }
-    }
 
     let stream_key = stream_key::Entity::find_by_id(payload.id)
         .one(&state.db)
@@ -252,8 +252,10 @@ async fn catalog_handler(
         .map(|x| StreamListing {
             id: x.stream_key_id,
             label: x.stream_key_label.clone(),
+            custom_url_label: x.custom_id.clone().unwrap_or_default(),
             user: x.stream_key_user.clone(),
             started_at: x.started_at,
+            is_password_protected: x.password.is_some(),
         })
         .collect::<Vec<_>>();
     Ok(Json(catalog))

crates/server/src/main.rs

@@ -27,6 +27,7 @@ mod hash;
 mod http;
 mod http_error;
 mod rtmp;
+mod stream_session_2;
 mod webrtc;
 mod webrtc_ingest;
 mod webrtc_proxy;

crates/server/src/stream_session_2.rs

@@ -0,0 +1,18 @@
+use serde::Serialize;
+
+use crate::StreamSession;
+
+#[derive(Serialize)]
+pub struct StreamUpdateData {
+    viewers: u32,
+}
+
+impl StreamSession {
+    pub fn stream_update_data(&self) -> StreamUpdateData {
+        StreamUpdateData {
+            viewers: self
+                .active_clients
+                .load(std::sync::atomic::Ordering::Relaxed),
+        }
+    }
+}

crates/server/src/webrtc.rs

@@ -2,13 +2,18 @@ use bytes::Bytes;
 use dashmap::DashMap;
 use entity::stream_key;
 use sea_orm::{DatabaseConnection, EntityTrait};
-use std::{net::SocketAddr, sync::Arc, time::Instant};
+use std::{
+    net::SocketAddr,
+    sync::Arc,
+    time::{Duration, Instant},
+};
 use tokio::{net::UdpSocket, sync::mpsc::Receiver};
 use tracing::{debug, error, info, trace, warn};
 
 use str0m::{
     Candidate, Event, Input, Output, Rtc,
     change::SdpOffer,
+    channel::ChannelId,
     media::{Frequency, MediaKind, MediaTime, Mid, Pt},
     net::{Protocol, Receive},
 };
@@ -236,10 +241,34 @@ impl Webrtc {
         let mut video_pt = None;
         let mut audio_mid: Option<Mid> = None;
         let mut audio_pt = None;
+        let mut channel_id = 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;
+        // Update client with stream info through webrtc data channel ()
+        let mut tick = tokio::time::interval(Duration::from_millis(2000));
+
+        if let Some(ses) = sessions_ref.get(&stream_id) {
+            ses.active_clients
+                .fetch_add(1, std::sync::atomic::Ordering::Relaxed);
+        }
+        struct ActiveClientGuard {
+            sessions: Arc<DashMap<i32, StreamSession>>,
+            id: i32,
+        }
+        impl Drop for ActiveClientGuard {
+            fn drop(&mut self) {
+                if let Some(ses) = self.sessions.get(&self.id) {
+                    ses.active_clients
+                        .fetch_sub(1, std::sync::atomic::Ordering::Relaxed);
+                }
+            }
+        }
+        let _active_client_guard = ActiveClientGuard {
+            sessions: sessions_ref.clone(),
+            id: stream_id,
+        };
 
         loop {
             let deadline = loop {
@@ -252,6 +281,9 @@ impl Webrtc {
                         }
                     }
                     Ok(Output::Event(e)) => match e {
+                        Event::ChannelOpen(channelId, _name) => {
+                            channel_id = Some(channelId);
+                        }
                         Event::MediaAdded(ma) => {
                             info!(stream_id, kind = ?ma.kind, mid = ?ma.mid, "MediaAdded");
                             if ma.kind == MediaKind::Video {
@@ -431,6 +463,14 @@ impl Webrtc {
                         Err(async_broadcast::RecvError::Overflowed(_)) => {}
                     }
                 }
+                _interval = tick.tick() => {
+               if let Some(id) = channel_id
+                   && let Some(session) = sessions_ref.get(&stream_id)
+                   && let Ok(json) = serde_json::to_vec(&session.stream_update_data())
+                   && let Some(mut ch) = rtc.channel(id) {
+                       let _ = ch.write(false, &json);
+                   }
+                }
             }
         }
     }
@@ -497,4 +537,12 @@ impl Webrtc {
             warn!("RTP write error: {:?}", e);
         }
     }
+
+    fn write_channel_data(rtc: &mut Rtc, channel_id: ChannelId, data: &[u8]) {
+        if let Some(mut channel) = rtc.channel(channel_id)
+            && let Err(e) = channel.write(false, data)
+        {
+            warn!("Channel write error: {:?}", e)
+        }
+    }
 }

flake.lock

@@ -2,11 +2,11 @@
   "nodes": {
     "crane": {
       "locked": {
-        "lastModified": 1785284101,
-        "narHash": "sha256-ghcXEpYEM4a7pbEkoqbn8c0ptJJqgGzuFiG3T6W5g4I=",
+        "lastModified": 1788465171,
+        "narHash": "sha256-Y1/TTVXjYXGF068IThQH9fPSZ0SIE74PABlUxnWTUH0=",
         "owner": "ipetkov",
         "repo": "crane",
-        "rev": "756d6d07c3818ea95d1e2cdac63fa7d02fe3e61b",
+        "rev": "eb35abda9f232cc6610b1d1e3200d15c49b7ac54",
         "type": "github"
       },
       "original": {
@@ -35,11 +35,11 @@
     },
     "nixpkgs": {
       "locked": {
-        "lastModified": 1785301185,
-        "narHash": "sha256-eoS3KQTO0aPWXZvIaRbRAzSSHW3l5wdMFXtT1ISfoKA=",
+        "lastModified": 1789012029,
+        "narHash": "sha256-1CBkBf+Nhggykzlx0jvXrj5rk20btl+tK8Fa8Ml/BL4=",
         "owner": "NixOS",
         "repo": "nixpkgs",
-        "rev": "9bc02893134c733dd85de46ee4fb2fac696b5529",
+        "rev": "d5dfd8e6716dde34398bc14bc87c10dece9c8c68",
         "type": "github"
       },
       "original": {
@@ -64,11 +64,11 @@
         ]
       },
       "locked": {
-        "lastModified": 1786160316,
-        "narHash": "sha256-oLoc3ZLg1LX/S5Jb3v6MrF415AzDyC4vgeWy9UcYTQk=",
+        "lastModified": 1789024335,
+        "narHash": "sha256-kCy/MVLRIr95DJ4vspVzWj+kO/x+JuwSHmYZCvShg8w=",
         "owner": "oxalica",
         "repo": "rust-overlay",
-        "rev": "4e1c940c96560ceab7c547f89642231371a66646",
+        "rev": "577bb1e1fc5af0713169176c5c76622c21fa3ec0",
         "type": "github"
       },
       "original": {