diff --git a/.junie/plans/replace-udp-with-rustrtc-2.md b/.junie/plans/replace-udp-with-rustrtc-2.md new file mode 100644 index 0000000..2411acf --- /dev/null +++ b/.junie/plans/replace-udp-with-rustrtc-2.md @@ -0,0 +1,140 @@ +--- +sessionId: session-260912-132834-fuse +--- + +# Requirements + +### Overview & Goals +Replace the current placeholder raw-UDP voice relay (`src/udp/*`) with a real WebRTC media stack built on the `rustrtc` crate (already declared in `Cargo.toml`). The server becomes a centralized **SFU** (Selective Forwarding Unit): each connected client opens **one `PeerConnection`** with the server, the server decrypts/receives that client's audio (and later video) track and forwards it to every other member of the same voice `Channel`. + +### Scope +**In Scope** +- Remove `src/udp/server.rs`, `src/udp/router.rs` raw-socket logic (metrics module is kept/adapted). +- Add a `voice` module hosting: `PeerConnection` lifecycle management via `rustrtc`, SDP offer/answer exchange, ICE candidate exchange, and SFU-style track forwarding scoped by `channel_id`. +- Reuse the existing WebSocket gateway (`GatewayManager` / `RealtimeRouter` / `on_message` in `src/routes/gateway/mod.rs`) as the signaling transport — add a `Voice` namespace to the existing JSON event envelope (`GatewayEvent { namespace, action, content }`) instead of introducing a new endpoint. +- Reuse `channel::ChannelType::Voice`, `channel_user`, and the `JOIN_VOICE` / `SPEAK` permission bits already defined in `src/permissions.rs` to authorize who may join/publish in a voice channel. +- Update `config.toml` / `NetworkConfig` so the single `udp_port` continues to be the one UDP port used, bound through `rustrtc`'s `ice_udp_mux` (single-port ICE) instead of a raw `UdpSocket`. +- Update `AppMetrics`/`core::mod::App::run` wiring so the new voice service starts/stops alongside the HTTP server, replacing `UdpServer`. + +**Out of Scope** +- TURN/STUN server configuration (kept minimal for LAN/local network scenarios for now). +- Video/screen-share specific handling (structure will allow it later, but only audio forwarding is implemented now). +- Client-side (frontend) implementation details beyond the signaling contract needed by the server. + +### User Stories +- As a user with `JOIN_VOICE` permission on a voice channel, I want to connect and have my microphone audio heard by other members of that same channel. +- As a user already in a voice channel, I want to hear every other member's audio forwarded by the server without opening a direct connection to each peer. +- As a server operator, I want the previous ad-hoc UDP relay removed and replaced by a maintained WebRTC stack (rustrtc) so voice traffic is encrypted (SRTP) and NAT-traversal (ICE) works out of the box. + +### Functional Requirements +- A client sends an SDP offer through the gateway (`namespace: "Voice"`), scoped to a `channel_id`; the server validates channel membership/permissions, creates a `rustrtc::PeerConnection`, and returns an SDP answer through the same gateway connection. +- ICE candidates generated by either side are exchanged as additional `Voice` gateway events. +- Once connected, the server subscribes the client's inbound audio track to every other `PeerConnection` currently joined to the same voice channel (SFU fan-out), and adds new joiners' tracks to previously-connected peers as they arrive. +- When a user disconnects (gateway socket closes) or leaves the voice channel, their `PeerConnection` is closed and their track is removed from all other peers' forwarding sets. +- Existing `UdpMetrics`-style counters (packets/bytes in/out, errors) are preserved in spirit, adapted for voice metrics and reported by the periodic `reporter`. + + +# Technical Design + +### Current Implementation +- `src/udp/server.rs`: binds a raw `tokio::net::UdpSocket` on `NetworkConfig.udp_port`, loops on `recv_from`, and blindly forwards datagrams to registered peers under a hard-coded `"default"` channel without protocol isolation or encryption. +- `src/udp/router.rs`: `RoutingTable` maps `ChannelId -> Vec`; dead placeholder API. +- `src/udp/metrics.rs`: atomic counters (`packets_received`, `bytes_sent`, etc.) + `spawn_reporter` logging every interval, plugged into `AppMetrics.udp` (`src/metrics/mod.rs`). +- `src/core/mod.rs`: `App::build`/`App::run` constructs `UdpServer::new(&config.network, udp_metrics)` and `tokio::spawn(udp_server.run())`, joined into shutdown-broadcast pattern. +- `src/routes/gateway/mod.rs`: `GatewayManager` tracks client WebSocket connections; `GatewayClient::on_message` handles incoming messages and acts as the extension point for voice signaling. +- `src/services/realtime_registry.rs`: maintains `channel_id -> HashSet` membership computed from `computed_permission` (READ_CHANNEL). +- `src/permissions.rs`: `JOIN_VOICE`, `SPEAK`, `STREAM`, `MOVE_OTHERS`, `DISCONNECT_OTHERS`, `MANAGE_VOICE_CHANNEL` bits already defined. +- `src/models/channel.rs`: `ChannelType::Voice` variant already exists. + +### Key Decisions +1. **Topology: centralized SFU** — one `rustrtc::PeerConnection` per connected client per voice channel; the server forwards each publisher's RTP track to every other subscriber in that channel. No direct peer-to-peer mesh. +2. **Signaling transport: existing WebSocket gateway** — SDP offer/answer and ICE candidates are carried as `Voice` namespace events inside the current `GatewayEvent` envelope, handled in `GatewayClient::on_message` and dispatched to `VoiceService`. +3. **Single UDP port reuse** — `rustrtc`'s `RtcConfiguration` is set up with `ice_udp_mux = true` and bound to `NetworkConfig.udp_port`, preserving existing firewall/network port configurations. +4. **No TURN/STUN for now** — `RtcConfiguration.ice_servers` left default/empty for local and direct network setups. +5. **Voice presence tracked separately from `RealtimeRegistry`** — an in-memory registry (`VoiceRoom` per `channel_id`) tracks active WebRTC connections independently from general WebSocket channel presence. + +### Proposed Changes +- Replace `src/udp` module with `src/voice`: + - `voice/mod.rs`: public API, exports `VoiceService`. + - `voice/service.rs`: `VoiceService` owns `rustrtc::RtcConfiguration`, a map `channel_id -> VoiceRoom`, and handler methods for offer, ICE candidates, and channel leaves. + - `voice/room.rs`: `VoiceRoom` holds `HashMap>` for a channel and orchestrates track fan-out across peers. + - `voice/metrics.rs`: tracks voice counters (packets/bytes/errors), exposed through `AppMetrics.voice` and the existing reporter cadence. +- Extend `GatewayEvent` handling: add `"Voice"` namespace with actions `offer`, `answer`, `ice-candidate`, `leave`. +- Permission check on `offer`: verify `ChannelPermission::JOIN_VOICE` and `SPEAK` before initializing a `PeerConnection`. +- Update `src/core/mod.rs`: replace `UdpServer` initialization with `VoiceService` held inside `AppState`. +- Update `src/config.rs`: keep `NetworkConfig.udp_port` feeding `rustrtc`'s `ice_udp_mux_port`. + +### Data Models / Contracts +```rust +// Gateway JSON contract additions (namespace = "Voice") +{ "namespace": "Voice", "action": "offer", "content": { "channel_id": Uuid, "sdp": String } } +{ "namespace": "Voice", "action": "answer", "content": { "channel_id": Uuid, "sdp": String } } +{ "namespace": "Voice", "action": "ice-candidate", "content": { "channel_id": Uuid, "candidate": String } } +{ "namespace": "Voice", "action": "leave", "content": { "channel_id": Uuid } } +``` + +```rust +pub struct VoiceService { + config: rustrtc::RtcConfiguration, + rooms: RwLock>, +} + +impl VoiceService { + pub async fn handle_offer(&self, user_id: Uuid, channel_id: Uuid, sdp: String) -> anyhow::Result; + pub async fn handle_ice_candidate(&self, user_id: Uuid, channel_id: Uuid, candidate: String) -> anyhow::Result<()>; + pub fn leave(&self, user_id: Uuid, channel_id: Uuid); +} +``` + +### Architecture Diagram +```mermaid +graph TD + ClientA[Client A] -- WebSocket: Voice offer/answer/ICE --> Gateway[GatewayManager / on_message] + ClientB[Client B] -- WebSocket: Voice offer/answer/ICE --> Gateway + Gateway -- dispatch --> VoiceService + VoiceService -- creates/owns --> RoomA[VoiceRoom per channel_id] + RoomA -- PeerConnection A --> PCA[rustrtc PeerConnection A] + RoomA -- PeerConnection B --> PCB[rustrtc PeerConnection B] + ClientA == RTP/SRTP media on UDP mux port ==> PCA + ClientB == RTP/SRTP media on UDP mux port ==> PCB + PCA -- forward audio track --> PCB + PCB -- forward audio track --> PCA +``` + +### Risks +- `rustrtc` API details: verify track forwarding primitives against `rustrtc 0.3.133` during implementation and encapsulate track re-subscription in `voice/room.rs`. +- Permission enforcement: ensure permission checks take place before creating peer connections to prevent unauthorized audio relay. +- Signaling message volume: ensure gateway handling remains non-blocking for high frequency ICE candidates. + + +# Delivery Steps + +### Step 1: Scaffold the voice module and RtcConfiguration from NetworkConfig +A new `src/voice` module exists with a `VoiceService` capable of building a `rustrtc::RtcConfiguration` from the app's network config, replacing the old `src/udp` module's role in `core/mod.rs`. +- Create `src/voice/mod.rs`, `src/voice/service.rs`, `src/voice/room.rs`, and `src/voice/metrics.rs`. +- Port `UdpMetrics`-style counters into `voice/metrics.rs` (packets/bytes in/out, errors), keeping the `Metrics`/`MetricsSnapshot` trait implementations used by `crate::metrics::reporter`. +- Build `RtcConfiguration`/`RtcConfigurationBuilder` in `VoiceService::new(&NetworkConfig, metrics)` using `ice_udp_mux = true` and `ice_udp_mux_port = network.udp_port`, with no external ICE servers configured. +- Remove `src/udp/server.rs` and `src/udp/router.rs`; delete unused `RoutingTable`. +- Update `src/core/mod.rs` (`App::build`/`App::run`) to construct `VoiceService` instead of `UdpServer`, store it on `AppState`, and drop unneeded `udp_shutdown_tx`/`udp_handle` tasks. +- Update `src/metrics/mod.rs` (`AppMetrics`) to reference the new voice metrics type. + +### Step 2: Implement per-channel PeerConnection lifecycle in VoiceRoom +`VoiceService`/`VoiceRoom` can accept an SDP offer for a given channel, create a `rustrtc::PeerConnection`, and return an SDP answer, tracking connections per `channel_id`. +- Implement `VoiceRoom` (`channel_id -> HashMap>`). +- Implement `VoiceService::handle_offer(user_id, channel_id, sdp)`: looks up or creates the room, creates a `PeerConnection` via `rustrtc`, invokes `set_remote_description`/`create_answer`/`set_local_description`, and returns the answer SDP. +- Implement `VoiceService::handle_ice_candidate` and `VoiceService::leave`, closing and removing the `PeerConnection` from its room. +- Enforce `ChannelPermission::JOIN_VOICE` (and `SPEAK` for publishing) before creating a `PeerConnection`, reusing the `computed_permission` lookup pattern from `src/services/realtime_registry.rs`. + +### Step 3: Wire SFU track forwarding between peers in the same channel +Audio published by one connected client in a voice channel is forwarded by the server to every other client connected to the same channel. +- On `PeerConnection::on_track` for a given user's connection, register the inbound track on the owning `VoiceRoom`. +- For every other `PeerConnection` already in that `VoiceRoom`, add and forward the new track (SFU fan-out) using `rustrtc`'s track/`MediaCapabilities` APIs. +- When a new peer joins an existing room, subscribe it to all tracks already being forwarded by other room members. +- On `leave` or disconnect, remove the peer's published track from all other peers' forwarding sets and close its `PeerConnection`. + +### Step 4: Expose Voice signaling through the existing WebSocket gateway +Clients can perform the full offer/answer/ICE handshake over the existing gateway WebSocket connection using a new `Voice` namespace, with no new HTTP/WS endpoint introduced. +- Extend `GatewayClient::on_message` in `src/routes/gateway/mod.rs` to parse `GatewayEvent{namespace: "Voice", action, content}` messages (`offer`, `answer`, `ice-candidate`, `leave`). +- Dispatch parsed messages to `VoiceService` (accessed via `AppState`) and send the resulting answer/ICE-candidate events back through the client's existing `mpsc::UnboundedSender`. +- On gateway disconnect (`GatewayClient::on_disconnect`), call `VoiceService::leave` for any channel the user was actively connected to in voice. +- Update `config.toml`'s `DEFAULT_CONFIG_TOML` comment for `udp_port` to reflect its new role as the rustrtc ICE/media mux port. \ No newline at end of file diff --git a/.junie/plans/replace-udp-with-rustrtc-3.md b/.junie/plans/replace-udp-with-rustrtc-3.md new file mode 100644 index 0000000..ded52cd --- /dev/null +++ b/.junie/plans/replace-udp-with-rustrtc-3.md @@ -0,0 +1,134 @@ +--- +sessionId: session-260912-132945-1get +--- + +# Requirements + +### Overview & Goals +Separate WebRTC voice signaling from the main real-time chat gateway by introducing a **dedicated Voice WebSocket endpoint** (`/ws/voice`). The voice signaling protocol (SDP offer/answer exchange, ICE candidates, room joins/leaves) is isolated from chat events, mirroring the architecture of production communication platforms (e.g. Discord's separate Gateway and Voice Gateway). + +### Scope +**In Scope** +- Implement a dedicated WebSocket route `GET /ws/voice` (nested alongside `/ws/gateway` under `ws_routes`). +- Authenticate incoming voice WebSocket connections using the existing `CurrentUser` extractor (supporting Authorization headers, cookies, and `?token=` query parameters). +- Define and implement a lightweight, dedicated JSON signaling protocol for voice sessions (`offer`, `answer`, `ice-candidate`, `leave`, `error`). +- Bind the voice connection lifecycle directly to the WebSocket lifecycle: connecting/joining initiates the session; closing the socket cleanly tears down the `PeerConnection` and removes the participant from the `VoiceRoom`. +- Remove voice signaling handlers (`namespace: "Voice"`) from the general chat gateway (`src/routes/gateway/mod.rs`), restoring clean separation of concerns. +- Maintain full compatibility with `VoiceService` and `rustrtc` SFU track fan-out. + +**Out of Scope** +- Changing the underlying `rustrtc` SFU forwarding engine or UDP multiplexing port (`udp_port`). +- Implementing TURN/STUN relay configurations. +- Video/screen-sharing UI controls (handled in future iterations). + +### User Stories +- As a client joining a voice channel, I want to open a dedicated WebSocket connection for voice so that media signaling traffic (large SDP payloads, ICE candidates) does not delay or congest my chat message stream. +- As a client leaving a voice channel, I want closing my voice WebSocket connection to immediately and cleanly release all server WebRTC resources without affecting my general gateway connection. +- As a backend developer, I want voice signaling isolated in its own module/handler so the main gateway router remains focused strictly on application domain events. + +### Functional Requirements +- **Dedicated Route**: `GET /ws/voice` upgrades to a WebSocket connection authenticated via `CurrentUser`. +- **Signaling Exchange**: + - Client sends `offer` with `{ "channel_id": Uuid, "sdp": String }`. The server validates permissions (`JOIN_VOICE`), creates/updates the `rustrtc::PeerConnection`, and returns an `answer` event with the server SDP. + - Client and server exchange `ice-candidate` events with `{ "channel_id": Uuid, "candidate": String }`. + - Client sends `leave` or closes the socket; server removes peer from `VoiceRoom` and unsubscribes tracks. +- **Error Handling**: Server returns `{ "action": "error", "message": String }` on permission denial, invalid payload, or WebRTC negotiation failure. +- **Gateway Decoupling**: `/ws/gateway` no longer processes `"Voice"` namespace events. + +### Non-Functional Requirements +- **Performance**: High throughput and low latency signaling with asynchronous message handling via Tokio tasks. +- **Reliability**: Deterministic cleanup on socket disconnection preventing orphaned WebRTC sessions or memory leaks. + +# Technical Design + +### Current Implementation +- `src/routes/mod.rs`: Mounts `ws_routes` containing `/ws/gateway`. +- `src/routes/gateway/mod.rs`: `GatewayClient::on_message` inspects JSON events; if `namespace == "Voice"`, it dispatches `offer`, `ice-candidate`, and `leave` to `AppState.voice`. +- `src/voice/service.rs`: Manages `RtcConfiguration`, `VoiceRoom` instances, SDP answers, ICE candidates, and permissions. +- `src/http/middleware.rs` & `src/http/context.rs`: `auth_middleware` extracts JWTs from Bearer headers, cookies, or `?token=` query parameters and populates `CurrentUser`. + +### Key Decisions +1. **Dedicated WebSocket Route (`/ws/voice`)**: Voice signaling is completely extracted into its own endpoint and handler module (`src/routes/voice/` or `src/voice/ws.rs`), eliminating protocol coupling in `src/routes/gateway/`. +2. **Connection-bound Voice Lifecycle**: The dedicated WebSocket lifetime directly reflects voice room occupancy. When the WebSocket connection drops, `VoiceService::leave` is triggered automatically. +3. **Simplified Signaling Protocol**: Elimination of the generic `GatewayEvent` envelope wrapper (`{ namespace, action, content }`) in favor of direct, typed voice signaling messages (`{ action, ... }`). +4. **Standardized Authentication**: Reuse of `CurrentUser` extractor ensures uniform JWT validation across REST, Gateway, and Voice WebSocket endpoints. + +### Proposed Changes +- **New Route & Handler**: + - Create `src/routes/voice/` with `mod.rs`, `handlers.rs`, and `routes.rs` (or integrate under `src/voice/` and expose via `src/routes/mod.rs`). + - Add `GET /voice` route to `ws_routes` in `src/routes/mod.rs` (resulting in `/ws/voice`). + - Implement `voice_ws_handler` accepting `WebSocketUpgrade`, `State(AppState)`, and `CurrentUser(user)`. +- **Voice WebSocket Protocol**: + - Define incoming message types: `Offer { channel_id, sdp }`, `IceCandidate { channel_id, candidate }`, `Leave { channel_id }`. + - Define outgoing message types: `Answer { channel_id, sdp }`, `IceCandidate { channel_id, candidate }`, `Error { message }`. +- **Gateway Cleanup**: + - Remove `namespace == "Voice"` handling from `GatewayClient::on_message`. + - Remove voice-specific imports from `src/routes/gateway/mod.rs`. +- **VoiceService Integration**: + - Update `VoiceService::handle_offer` to take the dedicated voice WebSocket sender channel for emitting server ICE candidates. + +### Data Models / Contracts +```rust +// Voice WebSocket Client -> Server messages +#[derive(Debug, Deserialize)] +#[serde(tag = "action", rename_all = "kebab-case")] +pub enum VoiceClientMessage { + Offer { + channel_id: Uuid, + sdp: String, + }, + IceCandidate { + channel_id: Uuid, + candidate: String, + }, + Leave { + channel_id: Uuid, + }, +} + +// Voice WebSocket Server -> Client messages +#[derive(Debug, Serialize)] +#[serde(tag = "action", rename_all = "kebab-case")] +pub enum VoiceServerMessage { + Answer { + channel_id: Uuid, + sdp: String, + }, + IceCandidate { + channel_id: Uuid, + candidate: String, + }, + Error { + message: String, + }, +} +``` + +### Components +- `VoiceWsHandler` (`src/routes/voice/handlers.rs`): Handles WebSocket upgrades, authentication, event loop, message serialization/deserialization, and lifecycle teardown. +- `VoiceRoutes` (`src/routes/voice/routes.rs`): Defines Axum router for `/voice`. +- `VoiceService` (`src/voice/service.rs`): Coordinates `rustrtc::PeerConnection` instances, SFU track forwarding, and room management. +- `GatewayManager` (`src/routes/gateway/mod.rs`): Retains responsibility solely for domain event broadcasting (chat, server, channel updates). + +### File Structure +- `src/routes/mod.rs`: Register `voice::routes::router()` under `/ws`. +- `src/routes/voice/mod.rs`: Public module definition. +- `src/routes/voice/routes.rs`: Voice WebSocket route definition. +- `src/routes/voice/handlers.rs`: Voice WebSocket connection handler and message loop. +- `src/routes/voice/messages.rs`: Strongly typed signaling message definitions. +- `src/routes/gateway/mod.rs`: Cleaned up to remove voice signaling logic. + +### Architecture Diagram +```mermaid +graph TD + Client[Client App] + Client -- Chat events / presence --> GW["/ws/gateway (GatewayManager)"] + Client -- Voice signaling (SDP/ICE) --> VWS["/ws/voice (VoiceWsHandler)"] + VWS -- Manage connections & rooms --> VS[VoiceService] + VS -- SFU media fan-out --> RR[rustrtc / VoiceRoom] + Client == SRTP media over single UDP port ==> RR +``` + +### Risks +- **Socket Disconnect Race Conditions**: Ensure that client disconnection promptly removes the user from `VoiceRoom` without deadlocking `RwLock` guards. +- **Message Deserialization Errors**: Malformed payloads must result in an `Error` response over the WebSocket rather than terminating the connection loop prematurely. \ No newline at end of file diff --git a/.junie/plans/replace-udp-with-rustrtc.md b/.junie/plans/replace-udp-with-rustrtc.md new file mode 100644 index 0000000..916b11e --- /dev/null +++ b/.junie/plans/replace-udp-with-rustrtc.md @@ -0,0 +1,138 @@ +--- +sessionId: session-260912-101455-1pxv +--- + +# Requirements + +### Overview & Goals +Replace the current placeholder raw-UDP voice relay (`src/udp/*`) with a real WebRTC media stack built on the `rustrtc` crate (already declared in `Cargo.toml`). The server becomes a centralized **SFU** (Selective Forwarding Unit): each connected client opens **one `PeerConnection`** with the server, the server decrypts/receives that client's audio (and later video) track and re-encodes/forwards it to every other member of the same voice `Channel`. + +### Scope +**In Scope** +- Remove `src/udp/server.rs`, `src/udp/router.rs` raw-socket logic (metrics module is kept/adapted). +- Add a `voice` module hosting: `PeerConnection` lifecycle management via `rustrtc`, SDP offer/answer exchange, ICE candidate exchange, and SFU-style track forwarding scoped by `channel_id`. +- Reuse the existing WebSocket gateway (`GatewayManager` / `RealtimeRouter` / `on_message` in `src/routes/gateway/mod.rs`) as the signaling transport — add a `Voice` namespace to the existing JSON event envelope (`GatewayEvent { namespace, action, content }`) instead of introducing a new endpoint, per explicit request to avoid breaking the current structure. +- Reuse `channel::ChannelType::Voice`, `channel_user`, and the `JOIN_VOICE` / `SPEAK` permission bits already defined in `src/permissions.rs` to authorize who may join/publish in a voice channel. +- Update `config.toml` / `NetworkConfig` so the single `udp_port` continues to be the one UDP port used, now bound through `rustrtc`'s `ice_udp_mux` (single-port ICE) instead of a raw `UdpSocket`. +- Update `AppMetrics`/`core::mod::App::run` wiring so the new voice service starts/stops alongside the HTTP server, replacing `UdpServer`. + +**Out of Scope** +- TURN/STUN server configuration (kept minimal per user's answer — LAN/local network scenario for now). +- Video/screen-share specific handling (structure will allow it later, but only audio forwarding is implemented now). +- Client-side (frontend) implementation details beyond the signaling contract needed by the server. + +### User Stories +- As a user with `JOIN_VOICE` permission on a voice channel, I want to connect and have my microphone audio heard by other members of that same channel. +- As a user already in a voice channel, I want to hear every other member's audio mixed/forwarded by the server without opening a direct connection to each peer. +- As a server operator, I want the previous ad-hoc UDP relay removed and replaced by a maintained WebRTC stack (rustrtc) so voice traffic is encrypted (SRTP) and NAT-traversal (ICE) works out of the box. + +### Functional Requirements +- A client sends an SDP offer through the gateway (`namespace: "Voice"`), scoped to a `channel_id`; the server validates channel membership/permissions, creates a `rustrtc::PeerConnection`, and returns an SDP answer through the same gateway connection. +- ICE candidates generated by either side are exchanged as additional `Voice` gateway events. +- Once connected, the server subscribes the client's inbound audio track to every other `PeerConnection` currently joined to the same voice channel (SFU fan-out), and adds new joiners' tracks to previously-connected peers as they arrive. +- When a user disconnects (gateway socket closes) or leaves the voice channel, their `PeerConnection` is closed and their track is removed from all other peers' forwarding sets. +- Existing `UdpMetrics`-style counters (packets/bytes in/out, errors) are preserved in spirit, adapted to whatever counters `rustrtc` exposes (or wrapped manually) for the periodic `reporter`. + +# Technical Design + +### Current Implementation +- `src/udp/server.rs`: binds a raw `tokio::net::UdpSocket` on `NetworkConfig.udp_port`, loops on `recv_from`, and blindly forwards every datagram to all peers registered in a `RoutingTable` under a **hard-coded** `"default"` channel — there is no real protocol, no per-voice-channel isolation, no encryption. +- `src/udp/router.rs`: `RoutingTable` maps `ChannelId -> Vec`; `join`/`leave`/`routing_table_mut()` exist but are **never called** anywhere in the codebase — dead placeholder API. +- `src/udp/metrics.rs`: atomic counters (`packets_received`, `bytes_sent`, …) + `spawn_reporter` logging every interval. Plugged into `AppMetrics.udp` (`src/metrics/mod.rs`). +- `src/core/mod.rs`: `App::build`/`App::run` construct `UdpServer::new(&config.network, udp_metrics)` and `tokio::spawn(udp_server.run())`, joined into the same `tokio::select!`/shutdown-broadcast pattern as `HttpServer`. +- `src/routes/gateway/mod.rs`: `GatewayManager` keeps `ConnectionKey{user_id, connection_id} -> GatewayClient{sender: mpsc::UnboundedSender}`; events are pushed as JSON `GatewayEvent{namespace, action, content}`. `GatewayClient::on_message` currently just logs incoming text — this is the extension point for voice signaling. +- `src/services/realtime_registry.rs`: already maintains `channel_id -> HashSet` membership computed from `computed_permission` (READ_CHANNEL) — reusable to know who is allowed in a channel, but voice-specific "currently connected to voice" state does not exist yet and must be tracked separately (WebSocket presence in a channel != actively broadcasting audio). +- `src/permissions.rs`: `JOIN_VOICE`, `SPEAK`, `STREAM`, `MOVE_OTHERS`, `DISCONNECT_OTHERS`, `MANAGE_VOICE_CHANNEL` bits already defined but unused by any voice logic today. +- `src/models/channel.rs`: `ChannelType::Voice` variant already exists. + +### Key Decisions +1. **Topology: centralized SFU** (confirmed by user) — one `rustrtc::PeerConnection` per connected client per voice channel; the server forwards each publisher's decoded RTP track to every other subscriber in that channel. No client-to-client PeerConnections. +2. **Signaling transport: existing WebSocket gateway** (confirmed by user, to avoid restructuring) — SDP offer/answer and ICE candidates are carried as new `Voice` namespace events inside the current `GatewayEvent` envelope, handled in `GatewayClient::on_message` (currently a stub) and dispatched to a new `VoiceService`. +3. **Single UDP port reuse** — `rustrtc`'s `RtcConfiguration` will be set up with `ice_udp_mux = true` and bound to the existing `NetworkConfig.udp_port`, so the media/ICE traffic keeps using the same single port previously owned by the raw `UdpServer`, minimizing config/infra changes (firewall rules, `config.toml` stay compatible). +4. **No TURN/STUN for now** (confirmed by user) — `RtcConfiguration.ice_servers` left empty/default; can be added later via `config.toml` without further architecture changes. +5. **Voice presence tracked separately from `RealtimeRegistry`** — a new lightweight in-memory registry (`VoiceRoom` per `channel_id`) tracks which `PeerConnection`s are actively publishing/subscribing in a voice channel, since being subscribed to gateway events (`RealtimeRegistry`) is not the same as being connected to voice media. + +### Proposed Changes +- Replace the `src/udp` module with a new `src/voice` module: + - `voice/mod.rs`: public API, exports `VoiceService`. + - `voice/service.rs`: `VoiceService` owns a `rustrtc::RtcConfiguration` (built from `NetworkConfig`), a map `channel_id -> VoiceRoom`, and methods `handle_offer(user_id, channel_id, sdp) -> answer_sdp`, `handle_ice_candidate(...)`, `leave(user_id, channel_id)`. + - `voice/room.rs`: `VoiceRoom` holds `HashMap>` for one channel; implements track fan-out — on receiving a remote track from peer A, it calls into every other peer's `PeerConnection` to add/forward that track (rustrtc `PeerConnection` API, per its `Usage` example: create connection, `set_remote_description`, `create_answer`, `set_local_description`, subscribe to `on_track`). + - `voice/metrics.rs`: keep counters analogous to today's `UdpMetrics` (packets/bytes/errors), fed by hooks around track forwarding, still exposed through `AppMetrics.udp` (renamed `AppMetrics.voice` where feasible) and the existing `reporter::spawn_reporter` cadence. +- Extend `GatewayEvent` handling: add `"Voice"` namespace with actions like `offer`, `answer`, `ice-candidate`, `leave`; `GatewayClient::on_message` parses these and calls `VoiceService` through `AppState`, then pushes the answer/ICE-candidate response back on the same `mpsc::UnboundedSender` used for all other gateway events (no protocol reinvention). +- Permission check on `offer`: verify the user has `ChannelPermission::JOIN_VOICE` (and `SPEAK` to publish) on the target channel, reusing the same `computed_permission` lookups already used by `RealtimeRegistry`/`http/permissions.rs`. +- Update `src/core/mod.rs`: drop `UdpServer::new` / `udp_server.run()` / `udp_shutdown_tx`; construct `VoiceService` instead and store it in `AppState` (alongside `gateway`, `services`) so gateway handlers can call it; no UDP socket bind/run task is spawned by `App::run` anymore — `rustrtc` manages its own I/O internally once configured. +- Update `src/config.rs`: keep `NetworkConfig.udp_port` (renamed conceptually to "voice/media port" in comments) feeding `rustrtc`'s `ice_udp_mux_port`; update `DEFAULT_CONFIG_TOML` comment accordingly. +- Remove `src/udp/` directory entirely once `voice/` fully replaces its responsibilities; update `Cargo.toml`/module declarations (`lib.rs`) accordingly. + +### Data Models / Contracts +```rust +// Gateway JSON contract additions (namespace = "Voice") +{ "namespace": "Voice", "action": "offer", "content": { "channel_id": Uuid, "sdp": String } } +{ "namespace": "Voice", "action": "answer", "content": { "channel_id": Uuid, "sdp": String } } +{ "namespace": "Voice", "action": "ice-candidate", "content": { "channel_id": Uuid, "candidate": String } } +{ "namespace": "Voice", "action": "leave", "content": { "channel_id": Uuid } } +``` +```rust +pub struct VoiceService { + config: rustrtc::RtcConfiguration, + rooms: RwLock>, // channel_id -> room +} + +impl VoiceService { + pub async fn handle_offer(&self, user_id: Uuid, channel_id: Uuid, sdp: String) -> anyhow::Result; + pub async fn handle_ice_candidate(&self, user_id: Uuid, channel_id: Uuid, candidate: String) -> anyhow::Result<()>; + pub fn leave(&self, user_id: Uuid, channel_id: Uuid); +} +``` + +### Architecture Diagram +```mermaid +graph TD + ClientA[Client A - browser] -- WebSocket gateway: Voice offer/answer/ICE --> Gateway[GatewayManager / on_message] + ClientB[Client B - browser] -- WebSocket gateway: Voice offer/answer/ICE --> Gateway + Gateway -- dispatch --> VoiceService + VoiceService -- creates/owns --> RoomA[VoiceRoom per channel_id] + RoomA -- PeerConnection A --> PCA[rustrtc PeerConnection A] + RoomA -- PeerConnection B --> PCB[rustrtc PeerConnection B] + ClientA == RTP/SRTP media, single UDP port ==> PCA + ClientB == RTP/SRTP media, single UDP port ==> PCB + PCA -- forward audio track --> PCB + PCB -- forward audio track --> PCA +``` + +### Risks +- `rustrtc` is a young/fast-moving crate (frequent point releases per the benchmark notes found) — API surface for track-forwarding/SFU usage should be validated against the pinned `0.3.133` docs during implementation; if a needed primitive (e.g., explicit track re-publishing helper) is missing, a thin adapter layer will be needed inside `voice/room.rs`. +- Moving from "no real protocol" to full SDP/ICE negotiation is a larger surface than the previous placeholder; permission checks (`JOIN_VOICE`/`SPEAK`) must be enforced before any `PeerConnection` is created to avoid unauthorized channel joins. +- Since signaling now flows through the same WebSocket used for all other realtime events, malformed/large SDP payloads must not block the `GatewayClient` message loop — the voice service calls will be dispatched without blocking other event types. + +# Delivery Steps + +### * Step 1: Scaffold the voice module and RtcConfiguration from NetworkConfig +A new `src/voice` module exists with a `VoiceService` capable of building a `rustrtc::RtcConfiguration` from the app's network config, replacing the old `src/udp` module's role in `core/mod.rs`. +- Create `src/voice/mod.rs`, `src/voice/service.rs`, `src/voice/room.rs`, `src/voice/metrics.rs`. +- Port `UdpMetrics`-style counters into `voice/metrics.rs` (packets/bytes in/out, errors), keeping the `Metrics`/`MetricsSnapshot` trait implementations used by `crate::metrics::reporter`. +- Build `RtcConfiguration`/`RtcConfigurationBuilder` in `VoiceService::new(&NetworkConfig, metrics)` using `ice_udp_mux = true` and `ice_udp_mux_port = network.udp_port`, no ICE servers configured. +- Remove `src/udp/server.rs` and `src/udp/router.rs`; delete unused `RoutingTable` (dead code confirmed unused elsewhere). +- Update `src/core/mod.rs` (`App::build`/`App::run`) to construct `VoiceService` instead of `UdpServer`, store it on `AppState`, and drop the now-unneeded `udp_shutdown_tx`/`udp_handle` spawn/join wiring. +- Update `src/metrics/mod.rs` (`AppMetrics`) to reference the new voice metrics type. + +### Step 2: Implement per-channel PeerConnection lifecycle in VoiceRoom +`VoiceService`/`VoiceRoom` can accept an SDP offer for a given channel, create a `rustrtc::PeerConnection`, and return an SDP answer, tracking connections per `channel_id`. +- Implement `VoiceRoom` (`channel_id -> HashMap>`). +- Implement `VoiceService::handle_offer(user_id, channel_id, sdp)`: looks up/creates the room, creates a `PeerConnection` via `rustrtc`, calls `set_remote_description`/`create_answer`/`set_local_description`, returns the answer SDP. +- Implement `VoiceService::handle_ice_candidate` and `VoiceService::leave`, closing and removing the `PeerConnection` from its room. +- Enforce `ChannelPermission::JOIN_VOICE` (and `SPEAK` for publishing) before creating a `PeerConnection`, reusing the `computed_permission` lookup pattern from `src/services/realtime_registry.rs`. + +### Step 3: Wire SFU track forwarding between peers in the same channel +Audio published by one connected client in a voice channel is forwarded by the server to every other client connected to the same channel. +- On `PeerConnection::on_track` for a given user's connection, register the inbound track on the owning `VoiceRoom`. +- For every other `PeerConnection` already in that `VoiceRoom`, add/forward the new track (SFU fan-out) using `rustrtc`'s track/`MediaCapabilities` APIs. +- When a new peer joins an existing room, subscribe it to all tracks already being forwarded by other room members. +- On `leave`/disconnect, remove the peer's published track from all other peers' forwarding sets and close its `PeerConnection`. + +### Step 4: Expose Voice signaling through the existing WebSocket gateway +Clients can perform the full offer/answer/ICE handshake over the existing gateway WebSocket connection using a new `Voice` namespace, with no new HTTP/WS endpoint introduced. +- Extend `GatewayClient::on_message` in `src/routes/gateway/mod.rs` to parse `GatewayEvent{namespace: "Voice", action, content}` messages (`offer`, `answer`, `ice-candidate`, `leave`). +- Dispatch parsed messages to `VoiceService` (accessed via `AppState`) and send the resulting answer/ICE-candidate events back through the client's existing `mpsc::UnboundedSender`. +- On gateway disconnect (`GatewayClient::on_disconnect`), call `VoiceService::leave` for any channel the user was actively connected to in voice. +- Update `config.toml`'s `DEFAULT_CONFIG_TOML` comment for `udp_port` to reflect its new role as the rustrtc ICE/media mux port. \ No newline at end of file diff --git a/Cargo.lock b/Cargo.lock index 0a5509a..5861631 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -8,6 +8,52 @@ version = "2.0.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "320119579fcad9c21884f5c4861d16174d0e06250625266f50fe6898340abefa" +[[package]] +name = "aead" +version = "0.5.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d122413f284cf2d62fb1b7db97e02edb8cda96d769b16e443a4f6195e35662b0" +dependencies = [ + "crypto-common 0.1.6", + "generic-array", +] + +[[package]] +name = "aes" +version = "0.8.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b169f7a6d4742236a0a00c541b845991d0ac43e546831af1249753ab4c3aa3a0" +dependencies = [ + "cfg-if", + "cipher 0.4.4", + "cpufeatures 0.2.17", +] + +[[package]] +name = "aes" +version = "0.9.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "35f0f96ce78e38c3dc6d8948aa8163d06385be74000f3c7a95bf1eef35d3ea32" +dependencies = [ + "cipher 0.5.2", + "cpubits", + "cpufeatures 0.3.0", +] + +[[package]] +name = "aes-gcm" +version = "0.10.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "831010a0f742e1209b3bcea8fab6a8e149051ba6099432c8cb2cc117dec3ead1" +dependencies = [ + "aead", + "aes 0.8.4", + "cipher 0.4.4", + "ctr 0.9.2", + "ghash", + "subtle", +] + [[package]] name = "ahash" version = "0.7.8" @@ -114,7 +160,7 @@ version = "1.1.5" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "40c48f72fd53cd289104fc64099abca73db4166ad86ea0b4341abe65af83dadc" dependencies = [ - "windows-sys", + "windows-sys 0.61.2", ] [[package]] @@ -125,7 +171,7 @@ checksum = "291e6a250ff86cd4a820112fb8898808a366d8f9f58ce16d1f538353ad55747d" dependencies = [ "anstyle", "once_cell_polyfill", - "windows-sys", + "windows-sys 0.61.2", ] [[package]] @@ -242,7 +288,7 @@ dependencies = [ "arrow-schema", "arrow-select", "atoi", - "base64", + "base64 0.22.1", "chrono", "half", "lexical-core", @@ -326,6 +372,45 @@ dependencies = [ "regex-syntax", ] +[[package]] +name = "asn1-rs" +version = "0.7.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b7f43a50ac4fdca5df8e885c21b835997f0a1cdee65494a6847694a98652d9d8" +dependencies = [ + "asn1-rs-derive", + "asn1-rs-impl", + "displaydoc", + "nom", + "num-traits", + "rusticata-macros", + "thiserror", + "time", +] + +[[package]] +name = "asn1-rs-derive" +version = "0.6.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "3109e49b1e4909e9db6515a30c633684d68cdeaa252f215214cb4fa1a5bfee2c" +dependencies = [ + "proc-macro2", + "quote", + "syn 2.0.119", + "synstructure", +] + +[[package]] +name = "asn1-rs-impl" +version = "0.2.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7b18050c2cd6fe86c3a76584ef5e0baf286d038cda203eb6223df2cc413565f7" +dependencies = [ + "proc-macro2", + "quote", + "syn 2.0.119", +] + [[package]] name = "async-attributes" version = "1.1.2" @@ -404,7 +489,7 @@ dependencies = [ "polling", "rustix", "slab", - "windows-sys", + "windows-sys 0.61.2", ] [[package]] @@ -499,6 +584,18 @@ version = "1.1.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "1505bd5d3d116872e7271a6d4e16d81d0c8570876c8de68093a09ac269d8aac0" +[[package]] +name = "attohttpc" +version = "0.16.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "fdb8867f378f33f78a811a8eb9bf108ad99430d7aad43315dd9319c827ef6247" +dependencies = [ + "http 0.2.12", + "log", + "url", + "wildmatch", +] + [[package]] name = "autocfg" version = "1.5.1" @@ -512,7 +609,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "ce2b2dcc879c3bae0d371e77c99f2238400ef24ec001394befa67b6e543add9e" dependencies = [ "aws-lc-sys", - "untrusted", + "untrusted 0.7.1", "zeroize", ] @@ -536,14 +633,14 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "31b698c5f9a010f6573133b09e0de5408834d0c82f8d7475a89fc1867a71cd90" dependencies = [ "axum-core", - "base64", + "base64 0.22.1", "bytes", "form_urlencoded", "futures-util", - "http", - "http-body", + "http 1.5.0", + "http-body 1.1.0", "http-body-util", - "hyper", + "hyper 1.11.0", "hyper-util", "itoa", "matchit", @@ -574,8 +671,8 @@ checksum = "08c78f31d7b1291f7ee735c1c6780ccde7785daae9a9206026862dab7d8792d1" dependencies = [ "bytes", "futures-core", - "http", - "http-body", + "http 1.5.0", + "http-body 1.1.0", "http-body-util", "mime", "pin-project-lite", @@ -597,8 +694,8 @@ dependencies = [ "cookie", "futures-core", "futures-util", - "http", - "http-body", + "http 1.5.0", + "http-body 1.1.0", "http-body-util", "mime", "pin-project-lite", @@ -607,12 +704,24 @@ dependencies = [ "tracing", ] +[[package]] +name = "base16ct" +version = "0.2.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "4c7f02d4ea65f2c1853089ffd8d2787bdbc63de2f0d29dedbcf8ccdfa0ccd4cf" + [[package]] name = "base64" version = "0.22.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "72b3254f16251a8381aa12e40e3c4d2f0199f8c6508fbecb9d91f575e0fbb8c6" +[[package]] +name = "base64" +version = "0.23.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ac07cdecf99051d9a5238b80f35af32cdeba5b336e55d957b318b50137e18da5" + [[package]] name = "base64ct" version = "1.8.3" @@ -634,10 +743,19 @@ dependencies = [ ] [[package]] -name = "bitflags" -version = "2.13.1" +name = "bit-vec" +version = "0.9.1" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "b588b76d00fde79687d7646a9b5bdf3cc0f655e0bbd080335a95d7e96f3587da" +checksum = "b71798fca2c1fe1086445a7258a4bc81e6e49dcd24c8d0dd9a1e57395b603f51" +dependencies = [ + "serde", +] + +[[package]] +name = "bitflags" +version = "2.13.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "3ded4057c258ba199e2d26386d3af3780957ecaee6c4ef4041c6b4b8b97c0b06" dependencies = [ "serde_core", ] @@ -757,6 +875,9 @@ name = "bytes" version = "1.12.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "fc652a48c352aef3ea3aed32080501cf3ef6ed5da78602a020c991775b0aff04" +dependencies = [ + "serde", +] [[package]] name = "cast" @@ -840,6 +961,27 @@ dependencies = [ "half", ] +[[package]] +name = "cipher" +version = "0.4.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "773f3b9af64447d2ce9850330c473515014aa235e6a783b02db81ff39e4a3dad" +dependencies = [ + "crypto-common 0.1.6", + "inout 0.1.4", +] + +[[package]] +name = "cipher" +version = "0.5.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e8cf2a2c93cd704877c0858356ed03480ff301ee950b43f1cbe4573b088bfa6c" +dependencies = [ + "block-buffer 0.12.1", + "crypto-common 0.2.2", + "inout 0.2.2", +] + [[package]] name = "clap" version = "4.6.6" @@ -930,6 +1072,12 @@ dependencies = [ "yaml-rust2", ] +[[package]] +name = "const-oid" +version = "0.9.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c2459377285ad874054d797f3ccebf984978aa39129f6eafde5cdc8315b612f8" + [[package]] name = "const-oid" version = "0.10.2" @@ -982,6 +1130,12 @@ version = "0.8.7" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "773648b94d0e5d620f64f280777445740e61fe701025087ec8b57f45c791888b" +[[package]] +name = "cpubits" +version = "0.1.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "15b85f9c39137c3a891689859392b1bd49812121d0d61c9caf00d46ed5ce06ae" + [[package]] name = "cpufeatures" version = "0.2.17" @@ -1015,6 +1169,15 @@ version = "2.5.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "217698eaf96b4a3f0bc4f3662aaa55bdf913cd54d7204591faa790070c6d0853" +[[package]] +name = "crc32c" +version = "0.6.8" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "3a47af21622d091a8f0fb295b88bc886ac74efcc613efc19f5d0b21de5c89e47" +dependencies = [ + "rustc_version", +] + [[package]] name = "crc32fast" version = "1.5.0" @@ -1100,6 +1263,18 @@ version = "0.2.4" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "460fbee9c2c2f33933d720630a6a0bac33ba7053db5344fac858d4b8952d77d5" +[[package]] +name = "crypto-bigint" +version = "0.5.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0dc92fb57ca44df6db8059111ab3af99a63d5d0f8375d9972e319a379c6bab76" +dependencies = [ + "generic-array", + "rand_core 0.6.4", + "subtle", + "zeroize", +] + [[package]] name = "crypto-common" version = "0.1.6" @@ -1107,6 +1282,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "1bfb12502f3fc46cca1bb51ac28df9d618d813cdc3d2f25b9fe775a34af26bb3" dependencies = [ "generic-array", + "rand_core 0.6.4", "typenum", ] @@ -1119,6 +1295,24 @@ dependencies = [ "hybrid-array", ] +[[package]] +name = "ctr" +version = "0.9.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0369ee1ad671834580515889b80f2ea915f23b8be8d0daa4bbaf2ac5c7590835" +dependencies = [ + "cipher 0.4.4", +] + +[[package]] +name = "ctr" +version = "0.10.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "baaca1c4b237092596f64d571e9db6ce4109c4ef9742e27590f1709594461f21" +dependencies = [ + "cipher 0.5.2", +] + [[package]] name = "ctutils" version = "0.4.2" @@ -1158,6 +1352,7 @@ dependencies = [ "ident_case", "proc-macro2", "quote", + "strsim", "syn 2.0.119", ] @@ -1202,6 +1397,31 @@ version = "2.11.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "4583a4551df46e2792f82ceeac45e850d2e2d5debba0b91f102385cda5b11f06" +[[package]] +name = "der" +version = "0.7.10" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e7c1832837b905bbfb5101e07cc24c8deddf52f93225eee6ead5f4d63d53ddcb" +dependencies = [ + "const-oid 0.9.6", + "pem-rfc7468", + "zeroize", +] + +[[package]] +name = "der-parser" +version = "10.0.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "07da5016415d5a3c4dd39b11ed26f915f52fc4e0dc197d87908bc916e51bc1a6" +dependencies = [ + "asn1-rs", + "displaydoc", + "nom", + "num-bigint", + "num-traits", + "rusticata-macros", +] + [[package]] name = "deranged" version = "0.5.8" @@ -1233,6 +1453,37 @@ dependencies = [ "syn 2.0.119", ] +[[package]] +name = "derive_builder" +version = "0.20.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "507dfb09ea8b7fa618fcf76e953f4f5e192547945816d5358edffe39f6f94947" +dependencies = [ + "derive_builder_macro", +] + +[[package]] +name = "derive_builder_core" +version = "0.20.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "2d5bcf7b024d6835cfb3d473887cd966994907effbe9227e8c8219824d06c4e8" +dependencies = [ + "darling 0.20.11", + "proc-macro2", + "quote", + "syn 2.0.119", +] + +[[package]] +name = "derive_builder_macro" +version = "0.20.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ab63b0e2bf4d5928aff72e83a7dace85d7bba5fe12dcc3c5a572d78caffd3f3c" +dependencies = [ + "derive_builder_core", + "syn 2.0.119", +] + [[package]] name = "derive_more" version = "2.1.1" @@ -1262,7 +1513,9 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "9ed9a281f7bc9b7576e61468ba615a66a5c8cfdff42420a70aa82701a3b1e292" dependencies = [ "block-buffer 0.10.4", + "const-oid 0.9.6", "crypto-common 0.1.6", + "subtle", ] [[package]] @@ -1272,7 +1525,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "f1dd6dbb5841937940781866fa1281a1ff7bd3bf827091440879f9994983d5c2" dependencies = [ "block-buffer 0.12.1", - "const-oid", + "const-oid 0.10.2", "crypto-common 0.2.2", "ctutils", ] @@ -1309,6 +1562,20 @@ version = "1.0.5" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "92773504d58c093f6de2459af4af33faa518c13451eb8f2b5698ed3d36e7c813" +[[package]] +name = "ecdsa" +version = "0.16.9" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ee27f32b5c5292967d2d4a9d7f1e0b0aed2c15daded5a60300e4abb9d8020bca" +dependencies = [ + "der", + "digest 0.10.7", + "elliptic-curve", + "rfc6979", + "signature", + "spki", +] + [[package]] name = "either" version = "1.17.0" @@ -1318,6 +1585,27 @@ dependencies = [ "serde", ] +[[package]] +name = "elliptic-curve" +version = "0.13.8" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b5e6043086bf7973472e0c7dff2142ea0b680d30e18d9cc40f267efbf222bd47" +dependencies = [ + "base16ct", + "crypto-bigint", + "digest 0.10.7", + "ff", + "generic-array", + "group", + "hkdf 0.12.4", + "pem-rfc7468", + "pkcs8", + "rand_core 0.6.4", + "sec1", + "subtle", + "zeroize", +] + [[package]] name = "encoding_rs" version = "0.8.35" @@ -1351,7 +1639,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "39cab71617ae0d63f51a36d69f866391735b51691dbda63cf6f96d042b63efeb" dependencies = [ "libc", - "windows-sys", + "windows-sys 0.61.2", ] [[package]] @@ -1361,7 +1649,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "de48cc4d1c1d97a20fd819def54b890cadde72ed3ad0c614822a0a433361be96" dependencies = [ "cfg-if", - "windows-sys", + "windows-sys 0.61.2", ] [[package]] @@ -1407,6 +1695,16 @@ version = "2.5.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "da7c62ceae207dd37ea5b845da6a0696c799f85e97da1ab5b7910be3c1c80223" +[[package]] +name = "ff" +version = "0.13.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c0b50bfb653653f9ca9095b427bed08ab8d75a137839d9ad64eb11810d5b6393" +dependencies = [ + "rand_core 0.6.4", + "subtle", +] + [[package]] name = "find-msvc-tools" version = "0.1.11" @@ -1467,6 +1765,21 @@ version = "2.0.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "e6d5a32815ae3f33302d95fdcb2ce17862f8c65363dcfd29360480ba1001fc9c" +[[package]] +name = "futures" +version = "0.3.34" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9a31d2a3fbaaeb2af2368bbdd904aa8e812d3c04a1ee10d3171f52d556e5d0a3" +dependencies = [ + "futures-channel", + "futures-core", + "futures-executor", + "futures-io", + "futures-sink", + "futures-task", + "futures-util", +] + [[package]] name = "futures-channel" version = "0.3.34" @@ -1553,6 +1866,7 @@ version = "0.3.34" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "0d50a92467f8ba5dd6e3ee5d4bd04d73ab2e4e1c44474a0674821dfce14b79bc" dependencies = [ + "futures-channel", "futures-core", "futures-io", "futures-macro", @@ -1571,6 +1885,7 @@ checksum = "4bb6743198531e02858aeaea5398fcc883e71851fcbcb5a2f773e2fb6cb1edf2" dependencies = [ "typenum", "version_check", + "zeroize", ] [[package]] @@ -1608,6 +1923,27 @@ dependencies = [ "rand_core 0.10.1", ] +[[package]] +name = "getset" +version = "0.1.7" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "6cf442baaabe4213ce7d1239afc26c039180b6456da2cededa316ae2c8a77a77" +dependencies = [ + "proc-macro2", + "quote", + "syn 2.0.119", +] + +[[package]] +name = "ghash" +version = "0.5.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f0d8a4362ccb29cb0b265253fb0a2728f592895ee6854fd9bc13f2ffda266ff1" +dependencies = [ + "opaque-debug", + "polyval", +] + [[package]] name = "glob" version = "0.3.4" @@ -1626,6 +1962,36 @@ dependencies = [ "wasm-bindgen", ] +[[package]] +name = "group" +version = "0.13.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f0f9ef7462f7c099f518d754361858f86d8a07af53ba9af0fe635bbccb151a63" +dependencies = [ + "ff", + "rand_core 0.6.4", + "subtle", +] + +[[package]] +name = "h2" +version = "0.3.27" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0beca50380b1fc32983fc1cb4587bfa4bb9e78fc259aad4a0032d2080309222d" +dependencies = [ + "bytes", + "fnv", + "futures-core", + "futures-sink", + "futures-util", + "http 0.2.12", + "indexmap", + "slab", + "tokio", + "tokio-util", + "tracing", +] + [[package]] name = "half" version = "2.7.1" @@ -1703,13 +2069,31 @@ version = "0.4.3" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "7f24254aa9a54b5c858eaee2f5bccdb46aaf0e486a595ed5fd8f86ba55232a70" +[[package]] +name = "hkdf" +version = "0.12.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7b5f8eb2ad728638ea2c7d47a21db23b7b58a72ed6a38256b8a1849f15fbbdf7" +dependencies = [ + "hmac 0.12.1", +] + [[package]] name = "hkdf" version = "0.13.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "4aaa26c720c68b866f2c96ef5c1264b3e6f473fe5d4ce61cd44bbe913e553018" dependencies = [ - "hmac", + "hmac 0.13.0", +] + +[[package]] +name = "hmac" +version = "0.12.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "6c49c37c09c17a53d937dfbb742eb3a961d65a994e6bcdcf37e7399d0cc8ab5e" +dependencies = [ + "digest 0.10.7", ] [[package]] @@ -1721,6 +2105,17 @@ dependencies = [ "digest 0.11.3", ] +[[package]] +name = "http" +version = "0.2.12" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "601cbb57e577e2f5ef5be8e7b83f0f63994f25aa94d673e54a92d5c516d101f1" +dependencies = [ + "bytes", + "fnv", + "itoa", +] + [[package]] name = "http" version = "1.5.0" @@ -1731,6 +2126,17 @@ dependencies = [ "itoa", ] +[[package]] +name = "http-body" +version = "0.4.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7ceab25649e9960c0311ea418d17bee82c0dcec1bd053b5f9a66e265a693bed2" +dependencies = [ + "bytes", + "http 0.2.12", + "pin-project-lite", +] + [[package]] name = "http-body" version = "1.1.0" @@ -1738,7 +2144,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "ca2a8f2913ee65f60facd6a5905613afaa448497a0230cc41ce022d93290bc2c" dependencies = [ "bytes", - "http", + "http 1.5.0", ] [[package]] @@ -1749,8 +2155,8 @@ checksum = "23169fe34a5fbcdd3f3862e78fb9b6fccd5f02a6dc6f732547005d45631ce71c" dependencies = [ "bytes", "futures-core", - "http", - "http-body", + "http 1.5.0", + "http-body 1.1.0", "pin-project-lite", ] @@ -1775,6 +2181,30 @@ dependencies = [ "typenum", ] +[[package]] +name = "hyper" +version = "0.14.32" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "41dfc780fdec9373c01bae43289ea34c972e40ee3c9f6b3c8801a35f35586ce7" +dependencies = [ + "bytes", + "futures-channel", + "futures-core", + "futures-util", + "h2", + "http 0.2.12", + "http-body 0.4.6", + "httparse", + "httpdate", + "itoa", + "pin-project-lite", + "socket2 0.5.10", + "tokio", + "tower-service", + "tracing", + "want", +] + [[package]] name = "hyper" version = "1.11.0" @@ -1785,8 +2215,8 @@ dependencies = [ "bytes", "futures-channel", "futures-core", - "http", - "http-body", + "http 1.5.0", + "http-body 1.1.0", "httparse", "httpdate", "itoa", @@ -1802,9 +2232,9 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "96547c2556ec9d12fb1578c4eaf448b04993e7fb79cbaad930a656880a6bdfa0" dependencies = [ "bytes", - "http", - "http-body", - "hyper", + "http 1.5.0", + "http-body 1.1.0", + "hyper 1.11.0", "pin-project-lite", "tokio", "tower-service", @@ -1944,6 +2374,24 @@ dependencies = [ "icu_properties", ] +[[package]] +name = "igd" +version = "0.12.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "556b5a75cd4adb7c4ea21c64af1c48cefb2ce7d43dc4352c720a1fe47c21f355" +dependencies = [ + "attohttpc", + "bytes", + "futures", + "http 0.2.12", + "hyper 0.14.32", + "log", + "rand 0.8.7", + "tokio", + "url", + "xmltree", +] + [[package]] name = "indexmap" version = "2.14.0" @@ -1965,6 +2413,24 @@ dependencies = [ "rustversion", ] +[[package]] +name = "inout" +version = "0.1.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "879f10e63c20629ecabbb64a8010319738c66a5cd0c29b02d63d272b03751d01" +dependencies = [ + "generic-array", +] + +[[package]] +name = "inout" +version = "0.2.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "4250ce6452e92010fdf7268ccc5d14faa80bb12fc741938534c58f16804e03c7" +dependencies = [ + "hybrid-array", +] + [[package]] name = "is_terminal_polyfill" version = "1.70.2" @@ -2034,10 +2500,10 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "881733cbc631fc9e472e24447ce32a64bedf2da498d6d8570b08edc87de71f65" dependencies = [ "aws-lc-rs", - "base64", + "base64 0.22.1", "getrandom 0.2.17", "js-sys", - "pem", + "pem 3.0.6", "serde", "serde_json", "signature", @@ -2152,6 +2618,17 @@ version = "0.8.3" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "47d9d19d1d6efa0109d2f65ff4c85cddd50bd572e5a00127ab10987290bcefae" +[[package]] +name = "local-ip-address" +version = "0.6.13" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "aa08fb2b1ec3ea84575e94b489d06d4ce0cbf052d12acd515838f50e3c3d63e3" +dependencies = [ + "libc", + "neli", + "windows-sys 0.61.2", +] + [[package]] name = "lock_api" version = "0.4.14" @@ -2246,6 +2723,12 @@ dependencies = [ "unicase", ] +[[package]] +name = "minimal-lexical" +version = "0.2.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "68354c5c6bd36d73ff3feceb05efa59b6acb7626617f4962be322a825e61f79a" + [[package]] name = "miniz_oxide" version = "0.8.9" @@ -2264,7 +2747,7 @@ checksum = "30d65c71f1ce40ab09135ce117d742b9f8a19ff91a41a8b57ed50bc2de59c427" dependencies = [ "libc", "wasi", - "windows-sys", + "windows-sys 0.61.2", ] [[package]] @@ -2276,7 +2759,7 @@ dependencies = [ "bytes", "encoding_rs", "futures-util", - "http", + "http 1.5.0", "httparse", "memchr", "mime", @@ -2284,6 +2767,35 @@ dependencies = [ "version_check", ] +[[package]] +name = "neli" +version = "0.7.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "22f9786d56d972959e1408b6a93be6af13b9c1392036c5c1fafa08a1b0c6ee87" +dependencies = [ + "bitflags", + "byteorder", + "derive_builder", + "getset", + "libc", + "log", + "neli-proc-macros", + "parking_lot", +] + +[[package]] +name = "neli-proc-macros" +version = "0.2.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "05d8d08c6e98f20a62417478ebf7be8e1425ec9acecc6f63e22da633f6b71609" +dependencies = [ + "either", + "proc-macro2", + "quote", + "serde", + "syn 2.0.119", +] + [[package]] name = "nix" version = "0.29.0" @@ -2297,13 +2809,23 @@ dependencies = [ "memoffset", ] +[[package]] +name = "nom" +version = "7.1.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d273983c5a657a70a3e8f2a01329822f3b8c8172b73826411a55751e404a0a4a" +dependencies = [ + "memchr", + "minimal-lexical", +] + [[package]] name = "nu-ansi-term" version = "0.50.3" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "7957b9740744892f114936ab4a57b3f487491bbeafaf8083688b16841a4240e5" dependencies = [ - "windows-sys", + "windows-sys 0.61.2", ] [[package]] @@ -2350,6 +2872,15 @@ dependencies = [ "libm", ] +[[package]] +name = "oid-registry" +version = "0.8.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "12f40cff3dde1b6087cc5d5f5d4d65712f34016a03ed60e9c08dcc392736b5b7" +dependencies = [ + "asn1-rs", +] + [[package]] name = "once_cell" version = "1.21.4" @@ -2368,6 +2899,12 @@ version = "11.1.5" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "d6790f58c7ff633d8771f42965289203411a5e5c68388703c06e14f24770b41e" +[[package]] +name = "opaque-debug" +version = "0.3.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c08d65885ee38876c4f86fa503fb49d7b507c2b62552df7c70b2fce627e06381" + [[package]] name = "ordered-float" version = "4.6.0" @@ -2430,6 +2967,7 @@ dependencies = [ "log", "migration", "parking_lot", + "rustrtc", "sea-orm", "serde", "serde_json", @@ -2448,6 +2986,18 @@ dependencies = [ "validator", ] +[[package]] +name = "p256" +version = "0.13.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c9863ad85fa8f4460f9c48cb909d38a0d689dba1f6f6988a5e3e0d31071bcd4b" +dependencies = [ + "ecdsa", + "elliptic-curve", + "primeorder", + "sha2 0.10.9", +] + [[package]] name = "page_size" version = "0.6.0" @@ -2509,10 +3059,29 @@ version = "3.0.6" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "1d30c53c26bc5b31a98cd02d20f25a7c8567146caf63ed593a9d87b2775291be" dependencies = [ - "base64", + "base64 0.22.1", "serde_core", ] +[[package]] +name = "pem" +version = "4.0.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d354a98a3d1251555de99e8fdd8afda05573c31b82f59063a7b0a29b5527f120" +dependencies = [ + "base64 0.23.1", + "serde_core", +] + +[[package]] +name = "pem-rfc7468" +version = "0.7.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "88b39c9bfcfc231068454382784bb460aae594343fb030d46e9f50a645418412" +dependencies = [ + "base64ct", +] + [[package]] name = "percent-encoding" version = "2.3.2" @@ -2604,6 +3173,16 @@ dependencies = [ "futures-io", ] +[[package]] +name = "pkcs8" +version = "0.10.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f950b2377845cebe5cf8b5165cb3cc1a5e0fa5cfa3e1f7f55707d8fd82e0a7b7" +dependencies = [ + "der", + "spki", +] + [[package]] name = "pkg-config" version = "0.3.34" @@ -2659,7 +3238,19 @@ dependencies = [ "hermit-abi", "pin-project-lite", "rustix", - "windows-sys", + "windows-sys 0.61.2", +] + +[[package]] +name = "polyval" +version = "0.6.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9d1fe60d06143b2430aa532c94cfe9e29783047f06c0d7fd359a9a51b729fa25" +dependencies = [ + "cfg-if", + "cpufeatures 0.2.17", + "opaque-debug", + "universal-hash", ] [[package]] @@ -2686,6 +3277,15 @@ dependencies = [ "zerocopy", ] +[[package]] +name = "primeorder" +version = "0.13.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "353e1ca18966c16d9deb1c69278edbc5f194139612772bd9537af60ac231e1e6" +dependencies = [ + "elliptic-curve", +] + [[package]] name = "proc-macro-crate" version = "3.5.0" @@ -2882,6 +3482,20 @@ dependencies = [ "crossbeam-utils", ] +[[package]] +name = "rcgen" +version = "0.14.10" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8774e05a7d0de114588e6a28fe7e71694b82614ed569d86d8b389dfbc98b8ad8" +dependencies = [ + "pem 4.0.0", + "ring", + "rustls-pki-types", + "time", + "x509-parser", + "yasna", +] + [[package]] name = "redox_syscall" version = "0.5.18" @@ -2929,6 +3543,30 @@ dependencies = [ "bytecheck", ] +[[package]] +name = "rfc6979" +version = "0.4.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f8dd2a808d456c4a54e300a23e9f5a67e122c3024119acbfd73e3bf664491cb2" +dependencies = [ + "hmac 0.12.1", + "subtle", +] + +[[package]] +name = "ring" +version = "0.17.14" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a4689e6c2294d81e88dc6261c768b63bc4fcdb852be6d1352498b114f61383b7" +dependencies = [ + "cc", + "cfg-if", + "getrandom 0.2.17", + "libc", + "untrusted 0.9.0", + "windows-sys 0.52.0", +] + [[package]] name = "rkyv" version = "0.7.46" @@ -3043,6 +3681,15 @@ dependencies = [ "semver", ] +[[package]] +name = "rusticata-macros" +version = "4.1.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "faf0c4a6ece9950b9abdb62b1cfcf2a68b3b67a10ba445b3bb85be2a293d0632" +dependencies = [ + "nom", +] + [[package]] name = "rustix" version = "1.1.4" @@ -3053,7 +3700,50 @@ dependencies = [ "errno", "libc", "linux-raw-sys", - "windows-sys", + "windows-sys 0.61.2", +] + +[[package]] +name = "rustls-pki-types" +version = "1.15.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "2f4925028c7eb5d1fcdaf196971378ed9d2c1c4efc7dc5d011256f76c99c0a96" +dependencies = [ + "zeroize", +] + +[[package]] +name = "rustrtc" +version = "0.3.133" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "6d8307edee45813b075a3a7dd237ab68f41d7220c53cf2752802431f68c055db" +dependencies = [ + "aes 0.9.3", + "aes-gcm", + "anyhow", + "async-trait", + "base64 0.22.1", + "bytes", + "crc32c", + "crc32fast", + "ctr 0.10.1", + "futures", + "hmac 0.13.0", + "igd", + "local-ip-address", + "md-5", + "p256", + "parking_lot", + "rand 0.10.2", + "rcgen", + "serde", + "serde_json", + "sha1 0.11.0", + "sha2 0.11.0", + "thiserror", + "tokio", + "tracing", + "x509-parser", ] [[package]] @@ -3262,6 +3952,20 @@ version = "4.1.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "1c107b6f4780854c8b126e228ea8869f4d7b71260f962fefb57b996b8959ba6b" +[[package]] +name = "sec1" +version = "0.7.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d3e97a565f76233a6003f9f5c54be1d9c5bdfa3eccfb189469f11ec4901c47dc" +dependencies = [ + "base16ct", + "der", + "generic-array", + "pkcs8", + "subtle", + "zeroize", +] + [[package]] name = "semver" version = "1.0.28" @@ -3436,6 +4140,7 @@ version = "2.2.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "77549399552de45a898a580c1b41d445bf730df867cc44e6c0233bbc4b8329de" dependencies = [ + "digest 0.10.7", "rand_core 0.6.4", ] @@ -3478,6 +4183,16 @@ dependencies = [ "serde", ] +[[package]] +name = "socket2" +version = "0.5.10" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e22376abed350d73dd1cd119b57ffccad95b4e585a7cda43e286245ce23c0678" +dependencies = [ + "libc", + "windows-sys 0.52.0", +] + [[package]] name = "socket2" version = "0.6.5" @@ -3485,7 +4200,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "c3d1e2c7f27f8d4cb10542a02c49005dbd6e93095799d6f3be745fae9f8fedd4" dependencies = [ "libc", - "windows-sys", + "windows-sys 0.61.2", ] [[package]] @@ -3497,6 +4212,16 @@ dependencies = [ "lock_api", ] +[[package]] +name = "spki" +version = "0.7.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d91ed6c858b01f942cd56b37a94b3e0a1798290327d1236e4d9cf4eaca44d29d" +dependencies = [ + "base64ct", + "der", +] + [[package]] name = "sqlx" version = "0.9.0" @@ -3516,7 +4241,7 @@ version = "0.9.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "05b44e85bf579a8eeb4ceaa77a3a523baf2bf0e9bac7e40f405d537b5d2d5ccb" dependencies = [ - "base64", + "base64 0.22.1", "bytes", "cfg-if", "chrono", @@ -3623,7 +4348,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "87a2bdd6e83f6b3ea525ca9fee568030508b58355a43d0b2c1674d5f79dcd65e" dependencies = [ "atoi", - "base64", + "base64 0.22.1", "bitflags", "byteorder", "chrono", @@ -3634,8 +4359,8 @@ dependencies = [ "futures-core", "futures-util", "hex", - "hkdf", - "hmac", + "hkdf 0.13.0", + "hmac 0.13.0", "itoa", "log", "md-5", @@ -3717,6 +4442,12 @@ version = "0.28.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "9628de9b8791db39ceda2b119bbe13134770b56c138ec1d3af810d045c04f9bd" +[[package]] +name = "subtle" +version = "2.6.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "13c2bddecc57b384dee18652358fb23172facb8a2c51ccc10d74c157bdea3292" + [[package]] name = "syn" version = "1.0.109" @@ -3888,9 +4619,9 @@ dependencies = [ "parking_lot", "pin-project-lite", "signal-hook-registry", - "socket2", + "socket2 0.6.5", "tokio-macros", - "windows-sys", + "windows-sys 0.61.2", ] [[package]] @@ -3928,10 +4659,24 @@ dependencies = [ ] [[package]] -name = "toml" -version = "1.1.4+spec-1.1.0" +name = "tokio-util" +version = "0.7.19" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "3aace63f4bbcdfc2c965b059de67119c89c4017a70d633be6c104910f67056f5" +checksum = "494815d09bf52b5548659851081238f0ca39ff638363907596da739561c62c52" +dependencies = [ + "bytes", + "futures-core", + "futures-sink", + "libc", + "pin-project-lite", + "tokio", +] + +[[package]] +name = "toml" +version = "1.1.6+spec-1.1.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "920602543f0911ab71da12c50d59701da54c196d1a2bf5cb4b75667f137a406a" dependencies = [ "indexmap", "serde_core", @@ -3996,15 +4741,15 @@ dependencies = [ [[package]] name = "tower-http" -version = "0.7.0" +version = "0.7.1" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "b11f75e912b0c2be01b63d8cf8057b8c3f97cf34abb3d431a3a4c8675498e233" +checksum = "08a05a66a4fdd61cbbe0a1d755ffe0ca6aba159dd4820936a0ff8a8278245b9c" dependencies = [ "bitflags", "bytes", "futures-util", - "http", - "http-body", + "http 1.5.0", + "http-body 1.1.0", "http-body-util", "percent-encoding", "pin-project-lite", @@ -4088,6 +4833,12 @@ dependencies = [ "tracing-log", ] +[[package]] +name = "try-lock" +version = "0.2.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e421abadd41a4225275504ea4d6566923418b7f05506fbc9c0fe86ba7396114b" + [[package]] name = "tungstenite" version = "0.29.0" @@ -4096,7 +4847,7 @@ checksum = "6c01152af293afb9c7c2a57e4b559c5620b421f6d133261c60dd2d0cdb38e6b8" dependencies = [ "bytes", "data-encoding", - "http", + "http 1.5.0", "httparse", "log", "rand 0.9.5", @@ -4167,12 +4918,28 @@ version = "0.2.6" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "ebc1c04c71510c7f702b52b7c350734c9ff1295c464a03335b00bb84fc54f853" +[[package]] +name = "universal-hash" +version = "0.5.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "fc1de2c688dc15305988b563c3854064043356019f97a4b46276fe734c4f07ea" +dependencies = [ + "crypto-common 0.1.6", + "subtle", +] + [[package]] name = "untrusted" version = "0.7.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "a156c684c91ea7d62626509bce3cb4e1d9ed5c4d978f7b4352658f96a4c26b4a" +[[package]] +name = "untrusted" +version = "0.9.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8ecb6da28b8a351d773b68d5825ac39017e680750f980f3a1a85cd8dd28a47c1" + [[package]] name = "url" version = "2.5.8" @@ -4228,7 +4995,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "d047458f1b5b65237c2f6dc6db136945667f40a7668627b3490b9513a3d43a55" dependencies = [ "axum", - "base64", + "base64 0.22.1", "mime_guess", "regex", "rust-embed", @@ -4241,9 +5008,9 @@ dependencies = [ [[package]] name = "uuid" -version = "1.26.0" +version = "1.26.1" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "b5772d71c9be8a8a6ac2117d949c5b224c1b72241bb611d9a3012edcf8af7812" +checksum = "2ef6dac1e96601b4fb3acccccff2139741fcb757cb9a36089bf5be91cfb285ce" dependencies = [ "getrandom 0.4.3", "js-sys", @@ -4315,6 +5082,15 @@ dependencies = [ "winapi-util", ] +[[package]] +name = "want" +version = "0.3.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "bfa7760aed19e106de2c7c0b581b509f2f25d3dacaf737cb82ac61bc6d760b0e" +dependencies = [ + "try-lock", +] + [[package]] name = "wasi" version = "0.11.1+wasi-snapshot-preview1" @@ -4412,6 +5188,12 @@ version = "2.1.3" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "626c4bac6755d76ffc12cb01b2eac751db1996b9e0041de9aa02c8c211ddc82c" +[[package]] +name = "wildmatch" +version = "1.1.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7f44b95f62d34113cf558c93511ac93027e03e9c29a60dd0fd70e6e025c7270a" + [[package]] name = "winapi" version = "0.3.9" @@ -4434,7 +5216,7 @@ version = "0.1.11" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "c2a7b1c03c876122aa43f3020e6c3c3ee5c05081c9a00739faf7503aeba10d22" dependencies = [ - "windows-sys", + "windows-sys 0.61.2", ] [[package]] @@ -4502,6 +5284,15 @@ dependencies = [ "windows-link", ] +[[package]] +name = "windows-sys" +version = "0.52.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "282be5f36a8ce781fad8c8ae18fa3f9beff57ec1b52cb3de0789201425d9a33d" +dependencies = [ + "windows-targets", +] + [[package]] name = "windows-sys" version = "0.61.2" @@ -4511,6 +5302,70 @@ dependencies = [ "windows-link", ] +[[package]] +name = "windows-targets" +version = "0.52.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9b724f72796e036ab90c1021d4780d4d3d648aca59e491e6b98e725b84e99973" +dependencies = [ + "windows_aarch64_gnullvm", + "windows_aarch64_msvc", + "windows_i686_gnu", + "windows_i686_gnullvm", + "windows_i686_msvc", + "windows_x86_64_gnu", + "windows_x86_64_gnullvm", + "windows_x86_64_msvc", +] + +[[package]] +name = "windows_aarch64_gnullvm" +version = "0.52.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "32a4622180e7a0ec044bb555404c800bc9fd9ec262ec147edd5989ccd0c02cd3" + +[[package]] +name = "windows_aarch64_msvc" +version = "0.52.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "09ec2a7bb152e2252b53fa7803150007879548bc709c039df7627cabbd05d469" + +[[package]] +name = "windows_i686_gnu" +version = "0.52.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8e9b5ad5ab802e97eb8e295ac6720e509ee4c243f69d781394014ebfe8bbfa0b" + +[[package]] +name = "windows_i686_gnullvm" +version = "0.52.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0eee52d38c090b3caa76c563b86c3a4bd71ef1a819287c19d586d7334ae8ed66" + +[[package]] +name = "windows_i686_msvc" +version = "0.52.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "240948bc05c5e7c6dabba28bf89d89ffce3e303022809e73deaefe4f6ec56c66" + +[[package]] +name = "windows_x86_64_gnu" +version = "0.52.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "147a5c80aabfbf0c7d901cb5895d1de30ef2907eb21fbbab29ca94c5b08b1a78" + +[[package]] +name = "windows_x86_64_gnullvm" +version = "0.52.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "24d5b23dc417412679681396f2b49f3de8c1473deb516bd34410872eff51ed0d" + +[[package]] +name = "windows_x86_64_msvc" +version = "0.52.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "589f6da84c646204747d1270a2a5661ea66ed1cced2631d546fdfb155959f9ec" + [[package]] name = "winnow" version = "1.0.4" @@ -4541,6 +5396,39 @@ dependencies = [ "tap", ] +[[package]] +name = "x509-parser" +version = "0.18.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d43b0f71ce057da06bc0851b23ee24f3f86190b07203dd8f567d0b706a185202" +dependencies = [ + "asn1-rs", + "data-encoding", + "der-parser", + "lazy_static", + "nom", + "oid-registry", + "ring", + "rusticata-macros", + "thiserror", + "time", +] + +[[package]] +name = "xml-rs" +version = "0.8.29" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e450f9b2ed1dff33c94c12589a87338689467b9c4f5d8a5710bd09a847d2c8a7" + +[[package]] +name = "xmltree" +version = "0.10.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d7d8a75eaf6557bb84a65ace8609883db44a29951042ada9b393151532e41fcb" +dependencies = [ + "xml-rs", +] + [[package]] name = "yaml-rust2" version = "0.11.0" @@ -4558,6 +5446,16 @@ version = "1.0.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "cfe53a6657fd280eaa890a3bc59152892ffa3e30101319d168b781ed6529b049" +[[package]] +name = "yasna" +version = "0.6.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b5f6765e852b9b4dc8e2a76843e4d64d1cea8e79bcde0b6901aea8e7c7f08282" +dependencies = [ + "bit-vec", + "time", +] + [[package]] name = "yoke" version = "0.8.3" diff --git a/Cargo.toml b/Cargo.toml index 32c0e6f..ea89137 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -21,19 +21,19 @@ event_bus = { path = "event_bus" } parking_lot = "0.12.5" serde = "1.0.229" serde_json = "1.0.151" -toml = "1.1.4" -uuid = { version = "1.26.0", features = ["v4", "v7", "fast-rng", "serde"] } +toml = "1.1.6" +uuid = { version = "1.26.1", features = ["v4", "v7", "fast-rng", "serde"] } tracing = "0.1.44" tracing-subscriber = { version = "0.3", features = ["env-filter", "fmt", "time"] } thiserror = "2" utoipa = { version = "5", features = ["uuid", "chrono"] } utoipa-swagger-ui = { version = "9", features = ["axum"] } log = "0.4" -bitflags = "2.13.1" +bitflags = "2.13.2" argon2 = { version = "0.6.0", features = ["password-hash"] } jsonwebtoken = { version = "11.0.0", features = ["aws_lc_rs"] } tower = { version = "0.5", features = ["util"] } -tower-http = { version = "0.7.0", features = ["catch-panic", "cors", "trace"] } +tower-http = { version = "0.7.1", features = ["catch-panic", "cors", "trace"] } chrono = "0.4.45" validator = { version = "0.21.0", features = ["derive"] } async-trait = "0.1.92" @@ -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" diff --git a/event_bus/Cargo.toml b/event_bus/Cargo.toml index a092a5a..3f17cbe 100644 --- a/event_bus/Cargo.toml +++ b/event_bus/Cargo.toml @@ -16,7 +16,7 @@ harness = false tokio = { version = "1.53.1", default-features = false, features = ["rt", "sync"] } parking_lot = "0.12.5" tracing = "0.1" -uuid = { version = "1.26.0", features = ["v4"] } +uuid = { version = "1.26.1", features = ["v4"] } [dev-dependencies] tokio = { version = "1.53.1", default-features = false, features = ["rt", "rt-multi-thread", "macros", "time", "sync"] } diff --git a/src/config.rs b/src/config.rs index e8b49b3..1af1bd7 100644 --- a/src/config.rs +++ b/src/config.rs @@ -48,7 +48,7 @@ host = "0.0.0.0" # TCP and UDP port can be the same # HTTP port tcp_port = 8080 -# Voice/Video port +# WebRTC ICE/Media UDP multiplexing port udp_port = 8080 [database] diff --git a/src/core/mod.rs b/src/core/mod.rs index 0289708..18e100b 100644 --- a/src/core/mod.rs +++ b/src/core/mod.rs @@ -7,7 +7,7 @@ use crate::metrics::{AppMetrics, reporter}; use crate::repositories::Repositories; use crate::routes::gateway::{GatewayManager, RealtimeRouter}; use crate::services::Services; -use crate::udp::server::UdpServer; +use crate::voice::VoiceService; use event_bus::EventBus; use migration::{Migrator, MigratorTrait}; pub use state::AppState; @@ -65,6 +65,8 @@ impl App { }; let metrics = AppMetrics::new(); + let voice_metrics = Arc::clone(&metrics.voice); + let voice = Arc::new(VoiceService::new(&config.network, voice_metrics)); let services = Arc::new(Services::new(repositories.clone(), event_bus.clone())); services.permission_sync.start_listen_event().await; @@ -90,6 +92,7 @@ impl App { gateway, event_bus, services, + voice, }; Ok(Self { state }) @@ -116,28 +119,20 @@ impl App { // Initialize HTTP Server let (http_server, http_shutdown_tx) = HttpServer::new(&config.network, self.state.clone()); - // Initialize UDP service - let udp_metrics = Arc::clone(&self.state.metrics.udp); - let (udp_server, udp_shutdown_tx) = UdpServer::new(&config.network, udp_metrics); - // Lance le reporter central de métriques toutes les 30 secondes reporter::spawn_reporter( Arc::new(self.state.metrics.clone()), Duration::from_secs(30), ); - // On lance les serveurs dans des tâches séparées + // On lance le serveur HTTP dans une tâche séparée let mut http_handle = tokio::spawn(http_server.run()); - let mut udp_handle = tokio::spawn(udp_server.run()); // On arbitre : soit un signal arrive, soit une tâche se termine (erreur/crash) tokio::select! { res = &mut http_handle => { tracing::error!("HTTP server stopped unexpectedly: {:?}", res); } - res = &mut udp_handle => { - tracing::error!("UDP server stopped unexpectedly: {:?}", res); - } _ = Self::shutdown_signal() => { tracing::info!("Shutdown signal received, initiating graceful shutdown..."); } @@ -145,11 +140,9 @@ impl App { // Dans tous les cas (Ctrl-C ou crash d'un service), on demande l'arrêt global let _ = http_shutdown_tx.send(()); - let _ = udp_shutdown_tx.send(()); - // On attend que tout le monde ait fini de nettoyer - // (Note: join! supporte les handles déjà terminés ou annulés) - let _ = tokio::join!(http_handle, udp_handle); + // On attend que la tâche HTTP termine + let _ = http_handle.await; Database::checkpoint_wal(&self.state.db).await?; Database::close(&self.state.db).await?; diff --git a/src/core/state.rs b/src/core/state.rs index 0484830..c8e2a75 100644 --- a/src/core/state.rs +++ b/src/core/state.rs @@ -4,6 +4,7 @@ use crate::models::server; use crate::repositories::Repositories; use crate::routes::gateway::GatewayManager; use crate::services::Services; +use crate::voice::VoiceService; use event_bus::EventBus; use sea_orm::DatabaseConnection; use std::sync::{Arc, RwLock}; @@ -19,6 +20,7 @@ pub struct AppState { pub gateway: Arc, pub event_bus: Arc, pub services: Arc, + pub voice: Arc, } impl AppState {} diff --git a/src/lib.rs b/src/lib.rs index 9015fd0..0d300da 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -7,7 +7,7 @@ pub mod core; pub mod permissions; pub mod repositories; pub mod routes; -pub mod udp; +pub mod voice; pub mod auth; pub mod metrics; diff --git a/src/metrics/mod.rs b/src/metrics/mod.rs index 03b8726..3f59ea8 100644 --- a/src/metrics/mod.rs +++ b/src/metrics/mod.rs @@ -4,7 +4,7 @@ use std::sync::Arc; use std::time::Instant; use crate::http::metrics::HttpMetrics; -use crate::udp::metrics::UdpMetrics; +use crate::voice::metrics::VoiceMetrics; /// Contrat minimal pour un jeu de compteurs métriques. pub trait Metrics { @@ -26,14 +26,14 @@ pub trait MetricsSnapshot: Clone { #[derive(Debug, Clone)] pub struct AppMetrics { pub http: Arc, - pub udp: Arc, + pub voice: Arc, } impl AppMetrics { pub fn new() -> Self { Self { http: HttpMetrics::new(), - udp: UdpMetrics::new(), + voice: VoiceMetrics::new(), } } } diff --git a/src/metrics/reporter.rs b/src/metrics/reporter.rs index 733e049..94af361 100644 --- a/src/metrics/reporter.rs +++ b/src/metrics/reporter.rs @@ -8,23 +8,23 @@ use crate::metrics::AppMetrics; /// Lance une tâche tokio unique qui reporte toutes les métriques à intervalle régulier. pub fn spawn_reporter(metrics: Arc, interval: Duration) { let metrics_http = Arc::clone(&metrics.http); - let metrics_udp = Arc::clone(&metrics.udp); + let metrics_voice = Arc::clone(&metrics.voice); tokio::spawn(async move { let mut ticker = tokio::time::interval(interval); ticker.tick().await; let mut prev_http = metrics_http.snapshot(); - let mut prev_udp = metrics_udp.snapshot(); + let mut prev_voice = metrics_voice.snapshot(); loop { ticker.tick().await; let current_http = metrics_http.snapshot(); - let current_udp = metrics_udp.snapshot(); + let current_voice = metrics_voice.snapshot(); let http_rates = current_http.rates_since(&prev_http); - let udp_rates = current_udp.rates_since(&prev_udp); + let voice_rates = current_voice.rates_since(&prev_voice); tracing::info!( // ── HTTP ── @@ -34,17 +34,17 @@ pub fn spawn_reporter(metrics: Arc, interval: Duration) { http_responses_5xx = current_http.responses_5xx, http_req_per_sec = format_args!("{:.2}", http_rates.requests_per_sec), http_avg_latency_ms = format_args!("{:.1}", http_rates.avg_latency_ms), - // ── UDP ── - udp_pkts_rx = current_udp.packets_received, - udp_pkts_tx = current_udp.packets_sent, - udp_pkts_dropped = current_udp.packets_dropped, - udp_pkts_rx_s = format_args!("{:.1}", udp_rates.packets_received_per_sec), - udp_pkts_tx_s = format_args!("{:.1}", udp_rates.packets_sent_per_sec), + // ── Voice ── + voice_pkts_rx = current_voice.packets_received, + voice_pkts_tx = current_voice.packets_sent, + voice_pkts_dropped = current_voice.packets_dropped, + voice_pkts_rx_s = format_args!("{:.1}", voice_rates.packets_received_per_sec), + voice_pkts_tx_s = format_args!("{:.1}", voice_rates.packets_sent_per_sec), "App metrics" ); prev_http = current_http; - prev_udp = current_udp; + prev_voice = current_voice; } }); } diff --git a/src/routes/gateway/handlers.rs b/src/routes/gateway/handlers.rs index b9ba05d..d635478 100644 --- a/src/routes/gateway/handlers.rs +++ b/src/routes/gateway/handlers.rs @@ -56,9 +56,10 @@ async fn handle_socket(socket: WebSocket, state: AppState, user: User) { // Task pour recevoir les messages du WebSocket let client_clone = 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).await; + client_clone.on_message(message, &state_clone).await; } }); @@ -70,5 +71,5 @@ async fn handle_socket(socket: WebSocket, state: AppState, user: User) { state.gateway.remove_client(&client); // // Déconnexion (Disconnect) - client.on_disconnect().await; + client.on_disconnect(&state).await; } diff --git a/src/routes/gateway/mod.rs b/src/routes/gateway/mod.rs index 11ec38f..72d9374 100644 --- a/src/routes/gateway/mod.rs +++ b/src/routes/gateway/mod.rs @@ -1,3 +1,4 @@ +use crate::core::AppState; use crate::domain::events::channel::{ ChannelCreatedEvent, ChannelDeletedEvent, ChannelUpdatedEvent, }; @@ -472,16 +473,18 @@ impl GatewayClient { } } - async fn on_connect(&mut self) { + pub async fn on_connect(&mut self) { tracing::info!(user_id = %self.user.id, "Client connected"); } - async fn on_disconnect(&mut self) { + + pub async fn on_disconnect(&mut self, state: &AppState) { tracing::info!(user_id = %self.user.id, "Client disconnected"); + state.voice.leave_all(self.user.id).await; } - async fn on_message(&self, message: Message) { + pub async fn on_message(&self, message: Message, _state: &AppState) { if let Message::Text(content) = message { - tracing::info!(user_id = %self.user.id, "Received text message: {}", content); + tracing::debug!(user_id = %self.user.id, "Received text message: {}", content); } } } diff --git a/src/routes/mod.rs b/src/routes/mod.rs index 53cd712..5327d05 100644 --- a/src/routes/mod.rs +++ b/src/routes/mod.rs @@ -18,6 +18,7 @@ pub mod role; pub mod server; pub mod server_item_order; pub mod user; +pub mod voice; pub fn router() -> OxRouter { // Routes nécessitant une authentification @@ -41,7 +42,9 @@ pub fn router() -> OxRouter { .merge(core::routes::router()); let public_attachment_routes = attachment::routes::public_router(); - let ws_routes = Router::new().merge(gateway::routes::router()); + let ws_routes = Router::new() + .merge(gateway::routes::router()) + .merge(voice::routes::router()); Router::new() .nest("/api", api_routes) diff --git a/src/routes/voice/handlers.rs b/src/routes/voice/handlers.rs new file mode 100644 index 0000000..71000df --- /dev/null +++ b/src/routes/voice/handlers.rs @@ -0,0 +1,94 @@ +use super::messages::{VoiceClientMessage, VoiceServerMessage}; +use crate::{core::AppState, http::context::CurrentUser}; +use axum::{ + extract::{ + State, + ws::{Message, WebSocket, WebSocketUpgrade}, + }, + response::IntoResponse, +}; +use futures_util::{SinkExt, StreamExt}; +use tokio::sync::mpsc; + +pub async fn ws_handler( + ws: WebSocketUpgrade, + State(state): State, + CurrentUser(user): CurrentUser, +) -> impl IntoResponse { + ws.on_upgrade(move |socket| handle_socket(socket, state, user)) +} + +async fn handle_socket(socket: WebSocket, state: AppState, user: crate::models::user::Model) { + let (mut sender, mut receiver) = socket.split(); + let (tx, mut rx) = mpsc::unbounded_channel::(); + let send_task = tokio::spawn(async move { + while let Some(message) = rx.recv().await { + if sender.send(message).await.is_err() { + break; + } + } + }); + while let Some(Ok(message)) = receiver.next().await { + let Message::Text(text) = message else { + continue; + }; + let parsed = match serde_json::from_str::(&text) { + Ok(message) => message, + Err(error) => { + send_error(&tx, format!("Invalid voice message: {error}")); + continue; + } + }; + match parsed { + VoiceClientMessage::Offer { channel_id, sdp } => { + match state + .voice + .handle_offer( + user.id, + channel_id, + sdp, + &state.repositories, + Some(tx.clone()), + ) + .await + { + Ok(answer) => send( + &tx, + VoiceServerMessage::Answer { + channel_id, + sdp: answer, + }, + ), + Err(error) => send_error(&tx, error.to_string()), + } + } + VoiceClientMessage::IceCandidate { + channel_id, + candidate, + } => { + if let Err(error) = state + .voice + .handle_ice_candidate(user.id, channel_id, candidate) + .await + { + send_error(&tx, error.to_string()); + } + } + VoiceClientMessage::Leave { channel_id } => { + state.voice.leave(user.id, channel_id).await + } + } + } + state.voice.leave_all(user.id).await; + send_task.abort(); +} + +fn send(tx: &mpsc::UnboundedSender, message: VoiceServerMessage) { + if let Ok(json) = serde_json::to_string(&message) { + let _ = tx.send(Message::Text(json.into())); + } +} + +fn send_error(tx: &mpsc::UnboundedSender, message: String) { + send(tx, VoiceServerMessage::Error { message }); +} diff --git a/src/routes/voice/messages.rs b/src/routes/voice/messages.rs new file mode 100644 index 0000000..4183c97 --- /dev/null +++ b/src/routes/voice/messages.rs @@ -0,0 +1,18 @@ +use serde::{Deserialize, Serialize}; +use uuid::Uuid; + +#[derive(Debug, Deserialize)] +#[serde(tag = "action", rename_all = "kebab-case")] +pub enum VoiceClientMessage { + Offer { channel_id: Uuid, sdp: String }, + IceCandidate { channel_id: Uuid, candidate: String }, + Leave { channel_id: Uuid }, +} + +#[derive(Debug, Serialize)] +#[serde(tag = "action", rename_all = "kebab-case")] +pub enum VoiceServerMessage { + Answer { channel_id: Uuid, sdp: String }, + IceCandidate { channel_id: Uuid, candidate: String }, + Error { message: String }, +} diff --git a/src/routes/voice/mod.rs b/src/routes/voice/mod.rs new file mode 100644 index 0000000..dd74b37 --- /dev/null +++ b/src/routes/voice/mod.rs @@ -0,0 +1,3 @@ +pub mod handlers; +pub mod messages; +pub mod routes; diff --git a/src/routes/voice/routes.rs b/src/routes/voice/routes.rs new file mode 100644 index 0000000..d8be868 --- /dev/null +++ b/src/routes/voice/routes.rs @@ -0,0 +1,7 @@ +use super::handlers; +use crate::core::AppState; +use axum::{Router, routing::get}; + +pub fn router() -> Router { + Router::new().route("/voice", get(handlers::ws_handler)) +} diff --git a/src/udp/metrics.rs b/src/udp/metrics.rs deleted file mode 100644 index b5e6a02..0000000 --- a/src/udp/metrics.rs +++ /dev/null @@ -1,244 +0,0 @@ -//! Métrologie du serveur UDP. -//! -//! Ce module expose : -//! - [`UdpMetrics`] : compteurs atomiques lock-free (pas de contention dans la -//! boucle de routage). -//! - [`UdpMetricsSnapshot`] : lecture cohérente de tous les compteurs à un -//! instant T, utilisable pour calculer des deltas. -//! - [`UdpRates`] : taux moyens par seconde calculés entre deux snapshots. -//! - [`spawn_reporter`] : tâche tokio de reporting périodique via `tracing`. -//! -//! # Métriques collectées -//! -//! | Compteur | Description | -//! |--------------------|-----------------------------------------------| -//! | `packets_received` | Datagrammes reçus | -//! | `bytes_received` | Octets reçus (payload uniquement) | -//! | `packets_sent` | Datagrammes retransmis vers des abonnés | -//! | `bytes_sent` | Octets retransmis | -//! | `packets_dropped` | Paquets ignorés (canal sans abonnés) | -//! | `send_errors` | Échecs `send_to` | -//! | `recv_errors` | Échecs `recv_from` (avant erreur fatale) | -//! -//! Chaque métrique est également disponible en taux moyen par seconde via -//! [`UdpMetricsSnapshot::rates_since`]. - -use std::sync::Arc; -use std::sync::atomic::{AtomicU64, Ordering}; -use std::time::{Duration, Instant}; - -use crate::metrics::{Metrics, MetricsSnapshot}; - -// ── Compteurs ──────────────────────────────────────────────────────────────── - -/// Compteurs atomiques du serveur UDP. -/// -/// Partagé via [`Arc`] entre la boucle de routage et le reporter périodique. -/// Tous les accès utilisent [`Ordering::Relaxed`] : on accepte que les lectures -/// voient des valeurs légèrement décalées entre compteurs (suffisant pour de la -/// métrologie), ce qui évite tout overhead de synchronisation. -#[derive(Debug, Default)] -pub struct UdpMetrics { - /// Nombre total de datagrammes reçus. - pub packets_received: AtomicU64, - /// Volume total d'octets reçus (payload des datagrammes). - pub bytes_received: AtomicU64, - /// Nombre total de datagrammes retransmis (somme sur tous les abonnés). - pub packets_sent: AtomicU64, - /// Volume total d'octets retransmis. - pub bytes_sent: AtomicU64, - /// Paquets ignorés car le canal ne possède aucun abonné. - pub packets_dropped: AtomicU64, - /// Nombre d'erreurs `send_to` (non fatales). - pub send_errors: AtomicU64, - /// Nombre d'erreurs `recv_from` enregistrées avant arrêt du serveur. - pub recv_errors: AtomicU64, -} - -impl UdpMetrics { - /// Crée un jeu de métriques vide enroulé dans un [`Arc`]. - pub fn new() -> Arc { - Arc::new(Self::default()) - } - - /// Enregistre la réception d'un datagramme 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 datagramme de `bytes` octets vers un client. - #[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é (canal sans abonnés). - #[inline] - pub fn inc_dropped(&self) { - self.packets_dropped.fetch_add(1, Ordering::Relaxed); - } - - /// Enregistre un échec `send_to` non fatal. - #[inline] - pub fn inc_send_error(&self) { - self.send_errors.fetch_add(1, Ordering::Relaxed); - } - - /// Enregistre un échec `recv_from`. - #[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) -> UdpMetricsSnapshot { - UdpMetricsSnapshot { - 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 UdpMetrics { - type Snapshot = UdpMetricsSnapshot; - - fn snapshot(&self) -> UdpMetricsSnapshot { - self.snapshot() - } -} - -// ── Snapshot ───────────────────────────────────────────────────────────────── - -/// Lecture cohérente de l'ensemble des compteurs à un instant T. -/// -/// Permet de calculer des deltas et des taux entre deux points dans le temps -/// sans bloquer la boucle de routage. -#[derive(Debug, Clone, Copy)] -pub struct UdpMetricsSnapshot { - 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 UdpMetricsSnapshot { - /// Calcule les taux moyens par seconde depuis un snapshot précédent. - pub fn rates_since(&self, previous: &Self) -> UdpRates { - let secs = self - .taken_at - .duration_since(previous.taken_at) - .as_secs_f64() - .max(f64::EPSILON); - - UdpRates { - 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 UdpMetricsSnapshot { - fn taken_at(&self) -> Instant { - self.taken_at - } -} - -// ── Taux ───────────────────────────────────────────────────────────────────── - -/// Taux moyens par seconde calculés entre deux [`UdpMetricsSnapshot`]. -#[derive(Debug, Clone, Copy)] -pub struct UdpRates { - /// Paquets reçus par seconde. - pub packets_received_per_sec: f64, - /// Octets reçus par seconde. - pub bytes_received_per_sec: f64, - /// Paquets envoyés par seconde. - pub packets_sent_per_sec: f64, - /// Octets envoyés par seconde. - pub bytes_sent_per_sec: f64, - /// Paquets ignorés par seconde. - pub packets_dropped_per_sec: f64, -} - -// ── Reporter périodique ─────────────────────────────────────────────────────── - -/// Lance une tâche Tokio qui logue les métriques toutes les `interval`. -/// -/// Chaque rapport inclut les compteurs cumulatifs **et** les taux moyens sur -/// la fenêtre écoulée depuis le rapport précédent. -/// -/// # Exemple -/// ```no_run -/// use std::time::Duration; -/// use std::sync::Arc; -/// use oxspeak_server_lib::udp::metrics::{UdpMetrics, spawn_reporter}; -/// -/// #[tokio::main] -/// async fn main() { -/// let metrics = UdpMetrics::new(); -/// spawn_reporter(Arc::clone(&metrics), Duration::from_secs(5)); -/// } -/// ``` -pub fn spawn_reporter(metrics: Arc, interval: Duration) { - tokio::spawn(async move { - let mut ticker = tokio::time::interval(interval); - // Le premier tick est immédiat ; on le consomme pour démarrer à t=0. - 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!( - // ── Cumulatifs ── - 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, - // ── Taux / s ── - 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), - "UDP metrics" - ); - - prev_snapshot = current; - } - }); -} diff --git a/src/udp/mod.rs b/src/udp/mod.rs deleted file mode 100644 index 7ab1abf..0000000 --- a/src/udp/mod.rs +++ /dev/null @@ -1,3 +0,0 @@ -pub mod metrics; -pub mod router; -pub mod server; diff --git a/src/udp/router.rs b/src/udp/router.rs deleted file mode 100644 index 7780037..0000000 --- a/src/udp/router.rs +++ /dev/null @@ -1,61 +0,0 @@ -use std::collections::HashMap; -use std::net::SocketAddr; - -/// Identifiant d'un canal de routage (ex: room ID, channel name…). -pub type ChannelId = String; - -/// Table de routage UDP. -/// -/// Associe un identifiant de canal à la liste des clients (adresses IP/port) -/// actuellement abonnés à ce canal. Le serveur UDP utilise cette table pour -/// décider où retransmettre les paquets entrants. -/// -/// # Note future -/// Ce type est intentionnellement simple pour démarrer. La prochaine étape -/// sera d'y associer une logique de dispatch (ex: forwarding sélectif, -/// authentification du client, etc.). -#[derive(Debug, Default)] -pub struct RoutingTable { - channels: HashMap>, -} - -impl RoutingTable { - /// Crée une table de routage vide. - pub fn new() -> Self { - Self::default() - } - - /// Inscrit un client dans un canal. - /// - /// Si le canal n'existe pas encore, il est créé automatiquement. - /// Si le client est déjà inscrit dans ce canal, l'appel est sans effet. - pub fn join(&mut self, channel: impl Into, client: SocketAddr) { - let clients = self.channels.entry(channel.into()).or_default(); - if !clients.contains(&client) { - clients.push(client); - } - } - - /// Retire un client d'un canal. - /// - /// Si le canal devient vide après le retrait, il est supprimé de la table. - pub fn leave(&mut self, channel: &str, client: &SocketAddr) { - if let Some(clients) = self.channels.get_mut(channel) { - clients.retain(|c| c != client); - if clients.is_empty() { - self.channels.remove(channel); - } - } - } - - /// Retourne la liste des clients abonnés à un canal, ou `None` si le - /// canal n'existe pas. - pub fn subscribers(&self, channel: &str) -> Option<&[SocketAddr]> { - self.channels.get(channel).map(Vec::as_slice) - } - - /// Retourne tous les canaux connus et leurs abonnés. - pub fn channels(&self) -> &HashMap> { - &self.channels - } -} diff --git a/src/udp/server.rs b/src/udp/server.rs deleted file mode 100644 index b89d39f..0000000 --- a/src/udp/server.rs +++ /dev/null @@ -1,178 +0,0 @@ -use std::net::SocketAddr; -use std::sync::Arc; -use tokio::net::UdpSocket; -use tokio::sync::broadcast; - -use super::metrics::UdpMetrics; -use super::router::RoutingTable; -use crate::config::NetworkConfig; - -/// Taille du buffer de lecture pour chaque datagramme UDP entrant. -/// -/// La RFC 768 limite les datagrammes UDP à 65 507 octets (payload max avec -/// en-têtes IP+UDP). Pour de la voix/vidéo compressée, les paquets réels -/// seront bien plus petits, mais on alloue le maximum une seule fois pour -/// éviter toute troncature silencieuse. -const UDP_READ_BUFFER_SIZE: usize = 65_507; - -/// Erreurs pouvant survenir pendant l'opération du serveur UDP. -#[derive(Debug, thiserror::Error)] -pub enum UdpServerError { - #[error("failed to bind UDP socket to {addr}: {source}")] - Bind { - addr: SocketAddr, - #[source] - source: std::io::Error, - }, - #[error("I/O error: {0}")] - Io(#[from] std::io::Error), -} - -/// Serveur UDP asynchrone agissant comme routeur de paquets. -/// -/// Reçoit des datagrammes entrants et les route vers les clients inscrits -/// dans les canaux correspondants via une [`RoutingTable`]. -/// -/// La configuration réseau est fournie par [`NetworkConfig`] (issue de -/// [`AppConfig`][crate::config::AppConfig]), qui centralise la lecture du -/// fichier TOML. -/// -/// Les métriques sont collectées dans un [`UdpMetrics`] partageable via -/// [`Arc`] — passez-le à [`metrics::spawn_reporter`][super::metrics::spawn_reporter] -/// pour un reporting périodique automatique. -/// -/// # Exemple -/// ```no_run -/// use std::time::Duration; -/// use oxspeak_server_lib::config::{AppConfig, NetworkConfig}; -/// use oxspeak_server_lib::udp::server::UdpServer; -/// use oxspeak_server_lib::udp::metrics::{UdpMetrics, spawn_reporter}; -/// -/// #[tokio::main] -/// async fn main() { -/// let config = AppConfig::load().unwrap(); -/// let metrics = UdpMetrics::new(); -/// spawn_reporter(metrics.clone(), Duration::from_secs(10)); -/// let (server, _shutdown_tx) = UdpServer::new(config.network, metrics); -/// server.run().await.unwrap(); -/// } -/// ``` -pub struct UdpServer { - bind_addr: SocketAddr, - routing_table: RoutingTable, - metrics: Arc, - shutdown_rx: broadcast::Receiver<()>, -} - -impl UdpServer { - /// Crée un nouveau [`UdpServer`] depuis la configuration réseau globale. - /// - /// Retourne le serveur et un [`broadcast::Sender`] pour déclencher le - /// shutdown gracieux. - pub fn new(network: &NetworkConfig, metrics: Arc) -> (Self, broadcast::Sender<()>) { - let bind_addr = SocketAddr::new(network.host.into(), network.udp_port); - let (shutdown_tx, shutdown_rx) = broadcast::channel(1); - ( - Self { - bind_addr, - routing_table: RoutingTable::new(), - metrics, - shutdown_rx, - }, - shutdown_tx, - ) - } - - /// Expose la table de routage de façon mutable pour y inscrire des - /// clients avant ou pendant l'exécution (via partage d'état ou messages). - pub fn routing_table_mut(&mut self) -> &mut RoutingTable { - &mut self.routing_table - } - - /// Retourne une référence aux métriques du serveur. - pub fn metrics(&self) -> &Arc { - &self.metrics - } - - /// Bind le socket et démarre la boucle de routage. - /// - /// Pour chaque datagramme reçu, le paquet est retransmis inline (sans - /// spawn de tâche) vers tous les clients abonnés au canal identifié. - /// La future se résout lorsqu'un signal de shutdown est reçu ou qu'une - /// erreur I/O fatale survient. - pub async fn run(mut self) -> Result<(), UdpServerError> { - let socket = - UdpSocket::bind(self.bind_addr) - .await - .map_err(|source| UdpServerError::Bind { - addr: self.bind_addr, - source, - })?; - - tracing::info!(addr = %self.bind_addr, "UDP server listening"); - - let mut buf = vec![0u8; UDP_READ_BUFFER_SIZE]; - - loop { - tokio::select! { - result = socket.recv_from(&mut buf) => { - match result { - Ok((len, peer)) => { - self.metrics.inc_received(len as u64); - self.route_packet(&socket, &buf[..len], peer).await; - } - Err(err) => { - self.metrics.inc_recv_error(); - tracing::error!(%err, "recv_from failed"); - return Err(UdpServerError::Io(err)); - } - } - } - _ = self.shutdown_rx.recv() => { - tracing::info!("UDP server shutting down"); - break; - } - } - } - - Ok(()) - } - - /// Route un paquet entrant vers les abonnés du canal correspondant. - /// - /// # Logique actuelle (placeholder) - /// En l'absence de protocole applicatif défini, tous les paquets reçus - /// sont logués. Branche ici la logique d'identification du canal - /// (ex: lire un header de paquet pour extraire le `channel_id`). - async fn route_packet(&self, socket: &UdpSocket, data: &[u8], sender: SocketAddr) { - tracing::debug!(%sender, bytes = data.len(), "datagram received"); - - // TODO: extraire le channel_id depuis le header du paquet applicatif. - // Pour l'instant on utilise un canal de démonstration statique. - let channel_id = "default"; - - match self.routing_table.subscribers(channel_id) { - Some(clients) => { - for &client in clients { - // Ne pas renvoyer au sender lui-même. - if client == sender { - continue; - } - match socket.send_to(data, client).await { - Ok(_) => { - self.metrics.inc_sent(data.len() as u64); - } - Err(err) => { - self.metrics.inc_send_error(); - tracing::warn!(%client, %err, "failed to forward packet"); - } - } - } - } - None => { - self.metrics.inc_dropped(); - tracing::debug!(%sender, channel = channel_id, "no subscribers, packet dropped"); - } - } - } -} diff --git a/src/voice/metrics.rs b/src/voice/metrics.rs new file mode 100644 index 0000000..8eb1d33 --- /dev/null +++ b/src/voice/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 { + 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, 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; + } + }); +} diff --git a/src/voice/mod.rs b/src/voice/mod.rs new file mode 100644 index 0000000..68214f0 --- /dev/null +++ b/src/voice/mod.rs @@ -0,0 +1,12 @@ +pub mod metrics; +pub mod room; +pub mod service; + +pub use metrics::VoiceMetrics; +pub use room::VoiceRoom; +pub use service::VoiceService; + +/// Nom générique du moteur média. `VoiceService` reste exporté pour compatibilité +/// avec le code de démarrage existant, mais ce service est destiné à l'audio, +/// la vidéo et les autres pistes WebRTC. +pub type MediaService = VoiceService; diff --git a/src/voice/room.rs b/src/voice/room.rs new file mode 100644 index 0000000..8397e4c --- /dev/null +++ b/src/voice/room.rs @@ -0,0 +1,27 @@ +use rustrtc::media::track::MediaRelay; +use rustrtc::peer_connection::PeerConnection; +use std::collections::HashMap; +use std::sync::Arc; +use uuid::Uuid; + +/// Représente une salle vocale correspondant à un canal vocal (`ChannelType::Voice`). +#[derive(Clone)] +pub struct VoiceRoom { + pub channel_id: Uuid, + pub peers: HashMap>, + pub relays: HashMap>, +} + +impl VoiceRoom { + pub fn new(channel_id: Uuid) -> Self { + Self { + channel_id, + peers: HashMap::new(), + relays: HashMap::new(), + } + } + + pub fn is_empty(&self) -> bool { + self.peers.is_empty() + } +} diff --git a/src/voice/service.rs b/src/voice/service.rs new file mode 100644 index 0000000..2130f3f --- /dev/null +++ b/src/voice/service.rs @@ -0,0 +1,431 @@ +use std::collections::HashMap; +use std::fmt; +use std::sync::Arc; +use tokio::sync::{RwLock, mpsc}; +use uuid::Uuid; + +use axum::extract::ws::Message; +use rustrtc::config::{RtcConfiguration, RtcConfigurationBuilder}; +use rustrtc::media::track::MediaRelay; +use rustrtc::peer_connection::{PeerConnection, PeerConnectionEvent, RtpCodecParameters}; +use rustrtc::sdp::{SdpType, SessionDescription}; +use rustrtc::transports::ice::IceCandidate; + +use crate::config::NetworkConfig; +use crate::models::{channel::ChannelType, channel_user, computed_permission::PermissionScopeType}; +use crate::permissions::ChannelPermission; +use crate::repositories::Repositories; +use crate::routes::voice::messages::VoiceServerMessage; +use crate::voice::metrics::VoiceMetrics; +use crate::voice::room::VoiceRoom; + +/// Paramètres de codec audio par défaut (Opus). +pub fn opus_codec() -> RtpCodecParameters { + RtpCodecParameters { + payload_type: 111, + name: "opus".to_string(), + clock_rate: 48000, + channels: 2, + } +} + +/// Moteur média WebRTC central (SFU). +/// +/// Le WebSocket ne fait que transporter l'offre SDP, les réponses et les +/// candidats ICE. Toute la vie des PeerConnections et des pistes média est +/// gérée ici, ce qui permettra d'ajouter la vidéo sans mélanger les protocoles. +#[derive(Clone)] +pub struct VoiceService { + pub config: RtcConfiguration, + pub rooms: Arc>>, + pub metrics: Arc, +} + +impl fmt::Debug for VoiceService { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + f.debug_struct("VoiceService") + .field("config", &self.config) + .field("metrics", &self.metrics) + .finish() + } +} + +impl VoiceService { + /// Crée une nouvelle instance de [`VoiceService`] configurée avec les paramètres réseau. + pub fn new(network: &NetworkConfig, metrics: Arc) -> 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, + rooms: Arc::new(RwLock::new(HashMap::new())), + metrics, + } + } + + /// Récupère les métriques de la voix. + pub fn metrics(&self) -> &Arc { + &self.metrics + } + + /// Vérifie si l'utilisateur possède la permission de rejoindre le canal vocal. + pub async fn check_permission( + repositories: &Repositories, + user_id: Uuid, + channel_id: Uuid, + ) -> anyhow::Result { + let channel = repositories + .channel + .get_by_id(channel_id) + .await? + .ok_or_else(|| anyhow::anyhow!("Channel not found"))?; + + if channel.channel_type == ChannelType::DM { + use sea_orm::{ColumnTrait, EntityTrait, QueryFilter}; + let is_member = channel_user::Entity::find() + .filter(channel_user::Column::ChannelId.eq(channel_id)) + .filter(channel_user::Column::UserId.eq(user_id)) + .one(&repositories.channel.context.db) + .await? + .is_some(); + return Ok(is_member); + } + + if channel.channel_type != ChannelType::Voice { + return Ok(false); + } + + let permissions = repositories.computed_permission.get_all().await?; + for perm in permissions { + if perm.user_id == user_id + && perm.scope_type == PermissionScopeType::Channel + && perm.resource_id == channel_id + { + let chan_perm = ChannelPermission::from_bits_retain(perm.permissions as u64); + if chan_perm.contains(ChannelPermission::JOIN_VOICE) { + return Ok(true); + } + } + } + + Ok(false) + } + + /// Traite une offre SDP entrante d'un client pour un canal donné. + pub async fn handle_offer( + &self, + user_id: Uuid, + channel_id: Uuid, + offer_sdp: String, + repositories: &Repositories, + ice_sender: Option>, + ) -> anyhow::Result { + let allowed = Self::check_permission(repositories, user_id, channel_id).await?; + if !allowed { + anyhow::bail!("Permission denied: cannot join voice channel"); + } + + let pc = Arc::new(PeerConnection::new(self.config.clone())); + + // Abonner la nouvelle PeerConnection à tous les flux déjà présents dans la room + let existing_relays = { + let rooms = self.rooms.read().await; + rooms + .get(&channel_id) + .map(|r| r.relays.clone()) + .unwrap_or_default() + }; + + for (&other_user_id, relay) in &existing_relays { + if other_user_id != user_id { + let sub_track = relay.subscribe(); + if let Err(err) = pc.add_track(sub_track, opus_codec()) { + tracing::warn!(%err, "Failed to subscribe new peer to existing track"); + } + } + } + + let offer_desc = SessionDescription::parse(SdpType::Offer, &offer_sdp)?; + pc.set_remote_description(offer_desc).await?; + + let answer_desc = pc.create_answer().await?; + let answer_sdp = answer_desc.to_sdp_string(); + pc.set_local_description(answer_desc)?; + + // Enregistrer la peer connection dans la room + { + let mut rooms = self.rooms.write().await; + let room = rooms + .entry(channel_id) + .or_insert_with(|| VoiceRoom::new(channel_id)); + if let Some(old_pc) = room.peers.insert(user_id, Arc::clone(&pc)) { + old_pc.close(); + } + } + + // Relayer les candidats ICE locaux vers le client via le WebSocket + if let Some(sender) = ice_sender { + let mut ice_rx = pc.subscribe_ice_candidates(); + tokio::spawn(async move { + while let Ok(candidate) = ice_rx.recv().await { + let cand_sdp = candidate.to_sdp(); + let event = VoiceServerMessage::IceCandidate { + channel_id, + candidate: cand_sdp, + }; + if let Ok(json) = serde_json::to_string(&event) { + if sender.send(Message::Text(json.into())).is_err() { + break; + } + } + } + }); + } + + self.start_peer_event_loop(pc, user_id, channel_id); + + self.metrics.inc_received(1); + Ok(answer_sdp) + } + + /// Starts the media event loop for one WebRTC connection. + /// + /// ```text + /// Client microphone/camera + /// | + /// | SRTP/RTP packets over ICE/UDP + /// v + /// rustrtc internal transport <- raw packets are read here + /// | + /// | PeerConnectionEvent::Track + /// v + /// this function (the server's media entry point) + /// | + /// | MediaRelay::subscribe() for every other peer + /// v + /// other PeerConnections -> their clients + /// ``` + /// + /// There is deliberately no `UdpSocket::recv` in this service. `rustrtc` + /// owns the UDP mux, ICE, SRTP decryption and RTP parsing. `recv()` below + /// receives the resulting high-level WebRTC events, especially tracks. + fn start_peer_event_loop(&self, pc: Arc, user_id: Uuid, channel_id: Uuid) { + let rooms = Arc::clone(&self.rooms); + let metrics = Arc::clone(&self.metrics); + + tokio::spawn(async move { + while let Some(event) = pc.recv().await { + match event { + PeerConnectionEvent::Track(transceiver) => { + Self::forward_incoming_track( + rooms.as_ref(), + metrics.as_ref(), + user_id, + channel_id, + transceiver, + ) + .await; + } + PeerConnectionEvent::DataChannel(_) => {} + } + } + }); + } + + /// Converts one received track into a relay and subscribes every other + /// peer in the same room to it. This is the SFU fan-out operation. + async fn forward_incoming_track( + rooms: &RwLock>, + metrics: &VoiceMetrics, + user_id: Uuid, + channel_id: Uuid, + transceiver: Arc, + ) { + let Some(receiver) = transceiver.receiver() else { return }; + let relay = Arc::new(MediaRelay::new(receiver.track())); + let mut rooms = rooms.write().await; + let Some(room) = rooms.get_mut(&channel_id) else { return }; + room.relays.insert(user_id, Arc::clone(&relay)); + for (&other_user_id, other_pc) in &room.peers { + if other_user_id == user_id { continue; } + if let Err(error) = other_pc.add_track(relay.subscribe(), opus_codec()) { + tracing::warn!(%error, "Failed to fan-out incoming media track"); + metrics.inc_send_error(); + } else { + metrics.inc_sent(1); + } + } + } + + /// Traite un candidat ICE reçu d'un client. + pub async fn handle_ice_candidate( + &self, + user_id: Uuid, + channel_id: Uuid, + candidate_sdp: String, + ) -> anyhow::Result<()> { + let pc = { + let rooms = self.rooms.read().await; + rooms + .get(&channel_id) + .and_then(|r| r.peers.get(&user_id)) + .cloned() + }; + + if let Some(pc) = pc { + let candidate = IceCandidate::from_sdp(&candidate_sdp)?; + pc.add_ice_candidate(candidate)?; + self.metrics.inc_received(1); + } + Ok(()) + } + + /// Quitte un canal vocal pour un utilisateur. + pub async fn leave(&self, user_id: Uuid, channel_id: Uuid) { + let pc_to_close = { + let mut rooms = self.rooms.write().await; + if let Some(room) = rooms.get_mut(&channel_id) { + let pc = room.peers.remove(&user_id); + room.relays.remove(&user_id); + if room.is_empty() { + rooms.remove(&channel_id); + } + pc + } else { + None + } + }; + + if let Some(pc) = pc_to_close { + pc.close(); + } + } + + /// Quitte tous les canaux vocaux pour un utilisateur (ex. déconnexion WebSocket). + pub async fn leave_all(&self, user_id: Uuid) { + let mut pcs_to_close = Vec::new(); + { + let mut rooms = self.rooms.write().await; + let mut empty_rooms = Vec::new(); + for (&channel_id, room) in rooms.iter_mut() { + if let Some(pc) = room.peers.remove(&user_id) { + pcs_to_close.push(pc); + room.relays.remove(&user_id); + } + if room.is_empty() { + empty_rooms.push(channel_id); + } + } + for channel_id in empty_rooms { + rooms.remove(&channel_id); + } + } + for pc in pcs_to_close { + pc.close(); + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + use std::net::Ipv4Addr; + + fn test_network_config() -> NetworkConfig { + NetworkConfig { + host: Ipv4Addr::new(127, 0, 0, 1), + hostv6: None, + tcp_port: 8080, + udp_port: 9000, + } + } + + #[tokio::test] + async fn test_voice_service_creation_and_room_lifecycle() { + let network = test_network_config(); + let metrics = VoiceMetrics::new(); + let service = VoiceService::new(&network, metrics); + + let user1 = Uuid::new_v4(); + let user2 = Uuid::new_v4(); + let channel_id = Uuid::new_v4(); + + // Check room creation and empty initially + assert!(service.rooms.read().await.is_empty()); + + // Insert peer connections into room + let pc1 = Arc::new(PeerConnection::new(service.config.clone())); + let pc2 = Arc::new(PeerConnection::new(service.config.clone())); + + { + let mut rooms = service.rooms.write().await; + let room = rooms + .entry(channel_id) + .or_insert_with(|| VoiceRoom::new(channel_id)); + room.peers.insert(user1, Arc::clone(&pc1)); + room.peers.insert(user2, Arc::clone(&pc2)); + } + + assert_eq!(service.rooms.read().await.len(), 1); + assert_eq!( + service + .rooms + .read() + .await + .get(&channel_id) + .unwrap() + .peers + .len(), + 2 + ); + + // User 1 leaves + service.leave(user1, channel_id).await; + assert_eq!( + service + .rooms + .read() + .await + .get(&channel_id) + .unwrap() + .peers + .len(), + 1 + ); + + // User 2 leaves via leave_all + service.leave_all(user2).await; + assert!(service.rooms.read().await.is_empty()); + } + + #[test] + fn test_opus_codec_parameters() { + let codec = opus_codec(); + assert_eq!(codec.name, "opus"); + assert_eq!(codec.clock_rate, 48000); + assert_eq!(codec.channels, 2); + } + + #[test] + fn test_voice_metrics_counters_and_rates() { + let metrics = VoiceMetrics::new(); + metrics.inc_received(500); + metrics.inc_sent(1000); + metrics.inc_dropped(); + metrics.inc_send_error(); + metrics.inc_recv_error(); + + let snap = metrics.snapshot(); + assert_eq!(snap.packets_received, 1); + assert_eq!(snap.bytes_received, 500); + assert_eq!(snap.packets_sent, 1); + assert_eq!(snap.bytes_sent, 1000); + assert_eq!(snap.packets_dropped, 1); + assert_eq!(snap.send_errors, 1); + assert_eq!(snap.recv_errors, 1); + } +}