o_sfu/runtime/http_server/rooms/
disconnect.rs

1//! Bulk disconnect with bounded body JWTs and normalized runtime user IDs.
2
3use std::{str, sync::Arc};
4
5use axum::{
6    body::Bytes,
7    extract::{DefaultBodyLimit, FromRef, FromRequest, Request, State},
8    http::StatusCode,
9    response::{IntoResponse, Response},
10    routing::{MethodRouter, post},
11};
12use secrecy::SecretString;
13use tokio::sync::Semaphore;
14use tokio_util::task::TaskTracker;
15use tracing::{Instrument, warn};
16
17use crate::runtime::{
18    MediaTransport, RuntimeMetrics, RuntimeState,
19    auth::{self, HttpDisconnectClaims},
20    room::RoomManager,
21    telemetry,
22};
23
24const MAX_BODY_BYTES: usize = 16 * 1024;
25
26pub(super) fn route() -> MethodRouter<RuntimeState> {
27    post(disconnect).layer(DefaultBodyLimit::max(MAX_BODY_BYTES))
28}
29
30#[derive(Debug, Clone)]
31struct Services {
32    room_manager: Arc<RoomManager>,
33    media_transport: MediaTransport,
34    metrics: Arc<RuntimeMetrics>,
35    mutations: TaskTracker,
36    admission: Arc<Semaphore>,
37}
38
39impl FromRef<RuntimeState> for Services {
40    fn from_ref(state: &RuntimeState) -> Self {
41        Self {
42            room_manager: Arc::clone(&state.room_manager),
43            media_transport: state.media_transport.clone(),
44            metrics: Arc::clone(&state.metrics),
45            mutations: state.session_tasks.clone(),
46            admission: Arc::clone(&state.http_mutation_admission),
47        }
48    }
49}
50
51/// Body JWT verified with the global auth key and runtime-normalized user IDs.
52///
53/// # Errors
54///
55/// Extraction returns a [`Response`]: `400 Bad Request` for non-UTF-8 bodies
56/// or `422 Unprocessable Entity` when JWT verification fails. Both record the
57/// corresponding disconnect rejection counter. Body buffering failures retain
58/// Axum's rejection response, including `413 Payload Too Large` for the route's
59/// body limit, without incrementing those counters.
60#[derive(Debug, Clone, PartialEq, Eq)]
61struct VerifiedDisconnectClaims(HttpDisconnectClaims);
62
63/// Bulk-disconnect endpoint used by Odoo to remove users from active rooms.
64///
65/// [`VerifiedDisconnectClaims`] owns JWT verification and request-body decoding.
66async fn disconnect(
67    State(services): State<Services>,
68    VerifiedDisconnectClaims(claims): VerifiedDisconnectClaims,
69) -> Response {
70    let permit = match services.admission.acquire_owned().await {
71        Ok(permit) => permit,
72        Err(error) => {
73            warn!(?error, "HTTP mutation admission closed unexpectedly");
74            return StatusCode::INTERNAL_SERVER_ERROR.into_response();
75        }
76    };
77    // Request loss must not cancel post-commit effects or free mutation capacity.
78    let task = services.mutations.spawn(
79        async move {
80            let _permit = permit;
81            for (room_id, user_ids) in &claims.user_ids_by_room {
82                services
83                    .room_manager
84                    .disconnect_users(room_id, user_ids, &services.media_transport)
85                    .await;
86            }
87            services.metrics.record_http_disconnect_success();
88            StatusCode::OK.into_response()
89        }
90        .instrument(telemetry::http_request_span("disconnect")),
91    );
92    match task.await {
93        Ok(response) => response,
94        Err(error) => {
95            warn!(?error, "HTTP disconnect task stopped unexpectedly");
96            StatusCode::INTERNAL_SERVER_ERROR.into_response()
97        }
98    }
99}
100
101impl FromRequest<RuntimeState> for VerifiedDisconnectClaims {
102    type Rejection = Response;
103
104    async fn from_request(req: Request, state: &RuntimeState) -> Result<Self, Self::Rejection> {
105        let body = Bytes::from_request(req, state)
106            .await
107            .map_err(IntoResponse::into_response)?;
108        let token = str::from_utf8(&body)
109            .map_err(|_error| record_rejection(state, StatusCode::BAD_REQUEST))?;
110        let token = SecretString::from(token);
111        let mut claims =
112            auth::verify_claims::<HttpDisconnectClaims>(&token, &state.config.auth.key)
113                .map_err(|_error| record_rejection(state, StatusCode::UNPROCESSABLE_ENTITY))?;
114        claims.normalize_runtime_user_ids();
115        Ok(Self(claims))
116    }
117}
118
119fn record_rejection(state: &RuntimeState, status: StatusCode) -> Response {
120    match status {
121        StatusCode::BAD_REQUEST => state.metrics.record_http_disconnect_bad_request(),
122        StatusCode::UNPROCESSABLE_ENTITY => {
123            state.metrics.record_http_disconnect_unprocessable_entity();
124        }
125        _ => {}
126    }
127    status.into_response()
128}
o_sfu/runtime/http_server/rooms/disconnect.rs

o_sfu/runtime/http_server/rooms/
disconnect.rs
