init

Cargo.lock

@@ -2495,9 +2495,9 @@ dependencies = [
 
 [[package]]
 name = "jsonwebtoken"
-version = "11.0.0"
+version = "11.1.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "881733cbc631fc9e472e24447ce32a64bedf2da498d6d8570b08edc87de71f65"
+checksum = "e75fe14a82d81e5f5af639997db37d8b96045938a7ac6ab18cdbe1c7467e05e1"
 dependencies = [
  "aws-lc-rs",
  "base64 0.22.1",
@@ -3714,9 +3714,9 @@ dependencies = [
 
 [[package]]
 name = "rustrtc"
-version = "0.3.133"
+version = "0.3.138"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "6d8307edee45813b075a3a7dd237ab68f41d7220c53cf2752802431f68c055db"
+checksum = "282d086d81cf8908a2c3d6ecf068180b711a20f0412658da13f052608f00d936"
 dependencies = [
  "aes 0.9.3",
  "aes-gcm",
@@ -3787,9 +3787,9 @@ dependencies = [
 
 [[package]]
 name = "sea-orm"
-version = "2.0.2"
+version = "2.0.3"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "a334e83ced3ae3ee44db0f84d1fcf8d2087a1ad9bb9036f00f9f6067156ea197"
+checksum = "e01d46a6e22b8ce84aba64fe97011669859bc7f7120a47f5b58839dcaaa4545c"
 dependencies = [
  "async-stream",
  "async-trait",
@@ -3835,9 +3835,9 @@ dependencies = [
 
 [[package]]
 name = "sea-orm-cli"
-version = "2.0.2"
+version = "2.0.3"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "5a53505884d7c907bcf4f7b4ddb1b29425e62fef8b98aea9c99e17781cceb798"
+checksum = "2479056c3f9c96c8ec5b9091103050b5f3a0532c615d8ab01d3edb81af5c65a8"
 dependencies = [
  "chrono",
  "clap",
@@ -3852,9 +3852,9 @@ dependencies = [
 
 [[package]]
 name = "sea-orm-macros"
-version = "2.0.2"
+version = "2.0.3"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "4039a86f9acc4d3b52747508b347dddc6fd725bbc429902ebeb6d26225fc2528"
+checksum = "e24c82fc1e76c014dffe5ecdac8f654da73161d703a16ea13414581e07459c83"
 dependencies = [
  "heck 0.5.0",
  "itertools 0.14.0",
@@ -3868,9 +3868,9 @@ dependencies = [
 
 [[package]]
 name = "sea-orm-migration"
-version = "2.0.2"
+version = "2.0.3"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "bd09adbef87100d07131a60a8c5508b53d0cf2136521f654aae47af5e6a097fe"
+checksum = "9acbadc98b17aa94657122dae4b5a840c7cebe6359a5ba15d8eaa233768942a5"
 dependencies = [
  "async-trait",
  "clap",

Cargo.toml

@@ -15,7 +15,7 @@ tokio = { version = "1.53.1", features = ["full"] }
 axum = { version = "0.8", features = ["ws", "multipart"] }
 axum-extra = { version = "0.12.6", features = ["cookie"] }
 config = "0.15.25"
-sea-orm = { version = "2.0.2", features = ["sqlx-sqlite", "sqlx-postgres", "sqlx-mysql", "runtime-tokio", "with-chrono", "with-uuid", "with-json", "schema-sync"] }
+sea-orm = { version = "2.0.3", features = ["sqlx-sqlite", "sqlx-postgres", "sqlx-mysql", "runtime-tokio", "with-chrono", "with-uuid", "with-json", "schema-sync"] }
 migration = { path = "migration" }
 event_bus = { path = "event_bus" }
 parking_lot = "0.12.5"
@@ -31,7 +31,7 @@ utoipa-swagger-ui = { version = "9", features = ["axum"] }
 log = "0.4"
 bitflags = "2.13.2"
 argon2 = { version = "0.6.0", features = ["password-hash"] }
-jsonwebtoken = { version = "11.0.0", features = ["aws_lc_rs"] }
+jsonwebtoken = { version = "11.1.0", features = ["aws_lc_rs"] }
 tower = { version = "0.5", features = ["util"] }
 tower-http = { version = "0.7.1", features = ["catch-panic", "cors", "trace"] }
 chrono = "0.4.45"
@@ -42,4 +42,4 @@ futures-util = "0.3"
 form_urlencoded = "1.2.2"
 time = "0.3.55"
 sha2 = "0.11.0"
-rustrtc = "0.3.133"
+rustrtc = "0.3.138"

migration/Cargo.toml

@@ -13,7 +13,7 @@ async-std = { version = "1", features = ["attributes", "tokio1"] }
 uuid = { version = "1", features = ["v5"] }
 
 [dependencies.sea-orm-migration]
-version = "2.0.2"
+version = "2.0.3"
 features = [
     # Enable at least one `ASYNC_RUNTIME` and `DATABASE_DRIVER` feature if you want to run migration via CLI.
     # View the list of supported features at https://www.sea-ql.org/SeaORM/docs/install-and-config/database-and-async-runtime.

src/lib.rs

@@ -16,3 +16,5 @@ pub mod domain;
 
 pub mod services;
 pub mod utils;
+
+pub mod rtc;

src/routes/mod.rs

@@ -15,6 +15,7 @@ pub mod gateway;
 pub mod message;
 pub mod openapi;
 pub mod role;
+pub mod rtc;
 pub mod server;
 pub mod server_item_order;
 pub mod user;
@@ -44,7 +45,8 @@ pub fn router() -> OxRouter {
 
     let ws_routes = Router::new()
         .merge(gateway::routes::router())
-        .merge(voice::routes::router());
+        .merge(voice::routes::router())
+        .merge(rtc::routes::router());
 
     Router::new()
         .nest("/api", api_routes)

src/routes/rtc/handlers.rs

@@ -0,0 +1,22 @@
+use crate::core::AppState;
+use crate::http::context::CurrentUser;
+use crate::http::error::HTTPError;
+use crate::rtc::ws_entrypoint::ws_entrypoint_handler;
+use axum::extract::{Path, State, WebSocketUpgrade};
+use axum::response::IntoResponse;
+use uuid::Uuid;
+
+pub async fn ws_handler(
+    ws: WebSocketUpgrade,
+    Path(channel_id): Path<Uuid>,
+    State(state): State<AppState>,
+    CurrentUser(user): CurrentUser,
+) -> Result<impl IntoResponse, HTTPError> {
+    let channel = state
+        .repositories
+        .channel
+        .get_by_id(channel_id)
+        .await?
+        .ok_or(HTTPError::NotFound)?;
+    Ok(ws.on_upgrade(move |socket| ws_entrypoint_handler(socket, state, channel, user)))
+}

src/routes/rtc/mod.rs

@@ -0,0 +1,2 @@
+mod handlers;
+pub mod routes;

src/routes/rtc/routes.rs

@@ -0,0 +1,8 @@
+use super::handlers;
+use crate::core::AppState;
+use axum::Router;
+use axum::routing::get;
+
+pub fn router() -> Router<AppState> {
+    Router::new().route("/rtc/{channel_id}", get(handlers::ws_handler))
+}

src/routes/voice/README.md

@@ -0,0 +1,2 @@
+Recréer ce module de 0, afin de le comprendre à 100% (dans le dossier RTC)
+

src/rtc/client.rs

@@ -0,0 +1,35 @@
+use crate::models::{channel, user};
+use axum::extract::ws::Message;
+use rustrtc::peer_connection::PeerConnection;
+use std::sync::Arc;
+use tokio::sync::mpsc::UnboundedSender;
+
+/// Client connecté à un canal RTC.
+#[derive(Clone)]
+pub struct RtcClient {
+    pub user: user::Model,
+    pub channel: channel::Model,
+    pub peer_connection: Arc<PeerConnection>,
+    pub websocket_sender: UnboundedSender<Message>,
+}
+
+impl RtcClient {
+    pub fn new(
+        user: user::Model,
+        channel: channel::Model,
+        peer_connection: Arc<PeerConnection>,
+        websocket_sender: UnboundedSender<Message>,
+    ) -> Self {
+        Self {
+            user,
+            channel,
+            peer_connection,
+            websocket_sender,
+        }
+    }
+
+    /// Ferme proprement la connexion WebRTC.
+    pub fn close(&self) {
+        self.peer_connection.close();
+    }
+}

src/rtc/metrics.rs

@@ -0,0 +1,178 @@
+use std::sync::Arc;
+use std::sync::atomic::{AtomicU64, Ordering};
+use std::time::{Duration, Instant};
+
+use crate::metrics::{Metrics, MetricsSnapshot};
+
+/// Compteurs atomiques pour les métriques de la voix / WebRTC.
+#[derive(Debug, Default)]
+pub struct VoiceMetrics {
+    /// Nombre total de datagrammes / paquets reçus.
+    pub packets_received: AtomicU64,
+    /// Volume total d'octets reçus.
+    pub bytes_received: AtomicU64,
+    /// Nombre total de datagrammes / paquets retransmis.
+    pub packets_sent: AtomicU64,
+    /// Volume total d'octets retransmis.
+    pub bytes_sent: AtomicU64,
+    /// Paquets ignorés ou rejetés.
+    pub packets_dropped: AtomicU64,
+    /// Nombre d'erreurs d'émission.
+    pub send_errors: AtomicU64,
+    /// Nombre d'erreurs de réception.
+    pub recv_errors: AtomicU64,
+}
+
+impl VoiceMetrics {
+    /// Crée un jeu de métriques vide enveloppé dans un [`Arc`].
+    pub fn new() -> Arc<Self> {
+        Arc::new(Self::default())
+    }
+
+    /// Enregistre la réception d'un paquet de `bytes` octets.
+    #[inline]
+    pub fn inc_received(&self, bytes: u64) {
+        self.packets_received.fetch_add(1, Ordering::Relaxed);
+        self.bytes_received.fetch_add(bytes, Ordering::Relaxed);
+    }
+
+    /// Enregistre l'émission d'un paquet de `bytes` octets.
+    #[inline]
+    pub fn inc_sent(&self, bytes: u64) {
+        self.packets_sent.fetch_add(1, Ordering::Relaxed);
+        self.bytes_sent.fetch_add(bytes, Ordering::Relaxed);
+    }
+
+    /// Enregistre un paquet ignoré.
+    #[inline]
+    pub fn inc_dropped(&self) {
+        self.packets_dropped.fetch_add(1, Ordering::Relaxed);
+    }
+
+    /// Enregistre un échec d'émission non fatal.
+    #[inline]
+    pub fn inc_send_error(&self) {
+        self.send_errors.fetch_add(1, Ordering::Relaxed);
+    }
+
+    /// Enregistre un échec de réception.
+    #[inline]
+    pub fn inc_recv_error(&self) {
+        self.recv_errors.fetch_add(1, Ordering::Relaxed);
+    }
+
+    /// Prend un instantané cohérent de tous les compteurs.
+    pub fn snapshot(&self) -> VoiceMetricsSnapshot {
+        VoiceMetricsSnapshot {
+            taken_at: Instant::now(),
+            packets_received: self.packets_received.load(Ordering::Relaxed),
+            bytes_received: self.bytes_received.load(Ordering::Relaxed),
+            packets_sent: self.packets_sent.load(Ordering::Relaxed),
+            bytes_sent: self.bytes_sent.load(Ordering::Relaxed),
+            packets_dropped: self.packets_dropped.load(Ordering::Relaxed),
+            send_errors: self.send_errors.load(Ordering::Relaxed),
+            recv_errors: self.recv_errors.load(Ordering::Relaxed),
+        }
+    }
+}
+
+impl Metrics for VoiceMetrics {
+    type Snapshot = VoiceMetricsSnapshot;
+
+    fn snapshot(&self) -> VoiceMetricsSnapshot {
+        self.snapshot()
+    }
+}
+
+/// Lecture cohérente de l'ensemble des compteurs à un instant T.
+#[derive(Debug, Clone, Copy)]
+pub struct VoiceMetricsSnapshot {
+    pub taken_at: Instant,
+    pub packets_received: u64,
+    pub bytes_received: u64,
+    pub packets_sent: u64,
+    pub bytes_sent: u64,
+    pub packets_dropped: u64,
+    pub send_errors: u64,
+    pub recv_errors: u64,
+}
+
+impl VoiceMetricsSnapshot {
+    /// Calcule les taux moyens par seconde depuis un snapshot précédent.
+    pub fn rates_since(&self, previous: &Self) -> VoiceRates {
+        let secs = self
+            .taken_at
+            .duration_since(previous.taken_at)
+            .as_secs_f64()
+            .max(f64::EPSILON);
+
+        VoiceRates {
+            packets_received_per_sec: self
+                .packets_received
+                .saturating_sub(previous.packets_received)
+                as f64
+                / secs,
+            bytes_received_per_sec: self.bytes_received.saturating_sub(previous.bytes_received)
+                as f64
+                / secs,
+            packets_sent_per_sec: self.packets_sent.saturating_sub(previous.packets_sent) as f64
+                / secs,
+            bytes_sent_per_sec: self.bytes_sent.saturating_sub(previous.bytes_sent) as f64 / secs,
+            packets_dropped_per_sec: self
+                .packets_dropped
+                .saturating_sub(previous.packets_dropped)
+                as f64
+                / secs,
+        }
+    }
+}
+
+impl MetricsSnapshot for VoiceMetricsSnapshot {
+    fn taken_at(&self) -> Instant {
+        self.taken_at
+    }
+}
+
+/// Taux moyens par seconde calculés entre deux [`VoiceMetricsSnapshot`].
+#[derive(Debug, Clone, Copy)]
+pub struct VoiceRates {
+    pub packets_received_per_sec: f64,
+    pub bytes_received_per_sec: f64,
+    pub packets_sent_per_sec: f64,
+    pub bytes_sent_per_sec: f64,
+    pub packets_dropped_per_sec: f64,
+}
+
+pub fn spawn_reporter(metrics: Arc<VoiceMetrics>, interval: Duration) {
+    tokio::spawn(async move {
+        let mut ticker = tokio::time::interval(interval);
+        ticker.tick().await;
+
+        let mut prev_snapshot = metrics.snapshot();
+
+        loop {
+            ticker.tick().await;
+
+            let current = metrics.snapshot();
+            let rates = current.rates_since(&prev_snapshot);
+
+            tracing::info!(
+                pkts_rx = current.packets_received,
+                bytes_rx = current.bytes_received,
+                pkts_tx = current.packets_sent,
+                bytes_tx = current.bytes_sent,
+                pkts_dropped = current.packets_dropped,
+                send_errors = current.send_errors,
+                recv_errors = current.recv_errors,
+                pkts_rx_s = format!("{:.1}", rates.packets_received_per_sec),
+                bytes_rx_s = format!("{:.0}", rates.bytes_received_per_sec),
+                pkts_tx_s = format!("{:.1}", rates.packets_sent_per_sec),
+                bytes_tx_s = format!("{:.0}", rates.bytes_sent_per_sec),
+                pkts_dropped_s = format!("{:.1}", rates.packets_dropped_per_sec),
+                "Voice metrics"
+            );
+
+            prev_snapshot = current;
+        }
+    });
+}

src/rtc/mod.rs

@@ -0,0 +1,23 @@
+mod client;
+mod metrics;
+pub mod ws_entrypoint;
+
+use crate::config::NetworkConfig;
+use metrics::VoiceMetrics;
+use rustrtc::{RtcConfiguration, RtcConfigurationBuilder};
+use std::sync::Arc;
+
+pub struct RTCManager {
+    pub config: RtcConfiguration,
+}
+
+impl RTCManager {
+    pub fn new(network: &NetworkConfig, metrics: Arc<VoiceMetrics>) -> Self {
+        let builder = RtcConfigurationBuilder::new()
+            .ice_udp_mux(true)
+            .ice_udp_mux_port(network.udp_port)
+            .bind_ip(network.host.to_string());
+        let rtc_config = builder.build();
+        Self { config: rtc_config }
+    }
+}

src/rtc/ws_entrypoint.rs

@@ -0,0 +1,36 @@
+// This is the first point when WebRTC ask for a connection
+
+use super::client::RtcClient;
+use crate::core::AppState;
+use crate::models::{channel, user};
+use axum::extract::ws::{Message, WebSocket};
+use futures_util::{SinkExt, StreamExt};
+use tokio::sync::mpsc;
+
+pub async fn ws_entrypoint_handler(
+    socket: WebSocket,
+    state: AppState,
+    channel: channel::Model,
+    user: user::Model,
+) {
+    let (mut sender, mut receiver) = socket.split();
+    let (tx, mut rx) = mpsc::unbounded_channel::<Message>();
+    // todo : faire le peer connexion.
+    let mut rtc_client = RtcClient::new(user, channel, peer_connection, tx);
+
+    let send_task = tokio::spawn(async move {
+        while let Some(message) = rx.recv().await {
+            if sender.send(message).await.is_err() {
+                break;
+            }
+        }
+    });
+
+    let client_clone = rtc_client.clone();
+    let state_clone = state.clone();
+    let mut recv_task = tokio::spawn(async move {
+        while let Some(Ok(message)) = receiver.next().await {
+            client_clone.on_message(message, &state_clone).await;
+        }
+    });
+}

src/voice/README.md

@@ -0,0 +1,2 @@
+Recréer ce module de 0, afin de le comprendre à 100% (dans le dossier RTC)
+