1use std::{
2 collections::{BTreeMap, BTreeSet},
3 mem,
4 sync::Arc,
5};
6
7use o_sfu_router::rtp::MediaCapabilities;
8use tracing::{debug, error, warn};
9
10use super::{
11 super::{
12 BroadcastPayload, BroadcastPayloadError, RoomEventMessage, RoomJoinError,
13 RoomUserPermissions, RouterPlacement, UserCloseReason,
14 effects::transport::RoomTransportPlan,
15 media_graph::{
16 CommittedTransportReceipt, SessionPlacementCommit, SessionPlacementRejection,
17 },
18 outbound::{MessageFanout, OutboundSender, VersionedRemoteTrackSnapshot, fanout_all},
19 },
20 UserJoinedFanout,
21 shared::{ActiveUser, RoomState},
22};
23#[cfg(test)]
24use crate::engine::MediaWorkerId;
25use crate::engine::{ConnectionId, UserId, UserInfo, media_transport::TransportTeardown};
26
27#[cfg(test)]
28#[allow(non_snake_case, reason = "test modules map to local TESTS directories")]
29mod TESTS;
30
31#[derive(Debug, Default)]
32pub struct LifecycleEffects {
33 pub close_requests: Vec<UserCloseRequest>,
34 pub fanouts: Vec<MessageFanout>,
35 pub track_snapshots: Vec<(OutboundSender, VersionedRemoteTrackSnapshot)>,
36}
37
38impl LifecycleEffects {
39 fn push_fanout(&mut self, fanout: Option<MessageFanout>) {
40 if let Some(fanout) = fanout {
41 self.fanouts.push(fanout);
42 }
43 }
44
45 fn push_close_request(&mut self, request: Option<UserCloseRequest>) {
46 if let Some(request) = request {
47 self.close_requests.push(request);
48 }
49 }
50}
51
52#[derive(Debug)]
53pub struct UserCloseRequest {
54 pub sender: OutboundSender,
55 pub reason: UserCloseReason,
56}
57
58type RuntimeUserRemoval = (ActiveUser, RoomTransportPlan);
59
60#[derive(Debug)]
61pub struct PresenceCommit {
62 pub fanout: MessageFanout,
63}
64
65#[derive(Debug)]
66pub struct JoinCommit {
67 pub effects: LifecycleEffects,
68 pub receipt: CommittedTransportReceipt,
69 pub transport_plan: RoomTransportPlan,
70}
71
72#[allow(
73 clippy::large_enum_variant,
74 reason = "connection close is cold and boxing the room transport plan would add allocation without simplifying ownership"
75)]
76#[derive(Debug)]
77pub enum ConnectionCloseCommit {
78 Current {
79 user_id: UserId,
80 connection_id: ConnectionId,
81 session_teardown: Option<TransportTeardown>,
82 effects: LifecycleEffects,
83 transport_plan: RoomTransportPlan,
84 },
85 StalePlacement {
86 session_teardown: TransportTeardown,
87 },
88}
89
90#[derive(Debug)]
91pub struct DisconnectCommit {
92 pub session_teardowns: Vec<TransportTeardown>,
93 pub effects: LifecycleEffects,
94 pub transport_plan: RoomTransportPlan,
95}
96
97impl RoomState {
98 pub fn fanout_all(&self, message: &RoomEventMessage) -> MessageFanout {
99 fanout_all(self.users.values().map(|user| user.sender.clone()), message)
100 }
101
102 pub fn fanout_all_except(
103 &self,
104 message: &RoomEventMessage,
105 excluded_user_id: &UserId,
106 ) -> MessageFanout {
107 fanout_all(
108 self.users
109 .iter()
110 .filter(|(user_id, _session)| excluded_user_id != *user_id)
111 .map(|(_user_id, user)| user.sender.clone()),
112 message,
113 )
114 }
115
116 fn apply_join_routing(
117 &mut self,
118 user_id: &UserId,
119 connection_id: ConnectionId,
120 is_new: bool,
121 home_placement: RouterPlacement,
122 ) -> Result<SessionPlacementCommit, RoomJoinError> {
123 let previous_connection = if is_new {
124 None
125 } else {
126 let Some(previous_connection) = self.users.get(user_id).map(|user| user.connection_id)
127 else {
128 error!(
129 ?user_id,
130 "missing previous room user for replacement join routing"
131 );
132 return Err(RoomJoinError::RouterState);
133 };
134 Some(previous_connection)
135 };
136 self.topology
137 .commit_session_placement(user_id, connection_id, previous_connection, home_placement)
138 .map_err(|rejection| {
139 match rejection {
140 SessionPlacementRejection::MissingPreviousSession {
141 previous_connection,
142 } => {
143 error!(
144 ?user_id,
145 connection_id = ?previous_connection,
146 "missing committed routing session for replacement join"
147 );
148 }
149 SessionPlacementRejection::Router(error) => {
150 error!(
151 ?user_id,
152 ?error,
153 "failed to mirror user join into room router"
154 );
155 }
156 }
157 RoomJoinError::RouterState
158 })
159 }
160
161 #[cfg(test)]
162 fn fallback_join_placement(&self) -> RouterPlacement {
163 RouterPlacement {
164 router: self.topology.router().placement_snapshot().primary(),
165 media_worker: MediaWorkerId::from_raw(0),
166 }
167 }
168
169 fn install_joined_session(
170 &mut self,
171 user_id: &UserId,
172 permissions: RoomUserPermissions,
173 sender: OutboundSender,
174 connection_id: ConnectionId,
175 ) -> Option<OutboundSender> {
176 if let Some(user) = self.users.get_mut(user_id) {
177 let old_sender = mem::replace(&mut user.sender, sender);
178 user.permissions = permissions;
179 user.reset_presentation();
180 user.parsed_client_rtp_capabilities = None;
181 user.connection_id = connection_id;
182 return Some(old_sender);
183 }
184 self.users.insert(
185 user_id.clone(),
186 ActiveUser {
187 user_id: Arc::new(user_id.clone()),
188 permissions,
189 info: UserInfo::default(),
190 server_featured: None,
191 parsed_client_rtp_capabilities: None,
192 connection_id,
193 sender,
194 },
195 );
196 None
197 }
198
199 #[cfg(test)]
200 pub fn apply_join(
201 &mut self,
202 user_id: &UserId,
203 permissions: impl Into<RoomUserPermissions>,
204 sender: OutboundSender,
205 ) -> Result<JoinCommit, RoomJoinError> {
206 self.apply_join_on_placement(
207 user_id,
208 permissions,
209 sender,
210 UserJoinedFanout::Suppress,
211 self.fallback_join_placement(),
212 )
213 }
214
215 pub fn apply_join_on_placement(
216 &mut self,
217 user_id: &UserId,
218 permissions: impl Into<RoomUserPermissions>,
219 sender: OutboundSender,
220 joined_fanout: UserJoinedFanout,
221 home_placement: RouterPlacement,
222 ) -> Result<JoinCommit, RoomJoinError> {
223 let permissions = permissions.into();
224 let previous_connection = self.users.get(user_id).map(|user| user.connection_id);
225 let is_new = previous_connection.is_none();
226 if is_new && self.users.len() >= self.admission_policy.max_sessions {
227 return Err(RoomJoinError::RoomFull);
228 }
229 let connection_id = ConnectionId::allocate(&mut self.next_connection_id);
230 let mut source_recipients = if previous_connection.is_some() {
231 self.topology
232 .committed_consumer_user_ids_for_owner_sources(user_id)
233 } else {
234 BTreeSet::new()
235 };
236 source_recipients.remove(user_id);
237 let placement = self.apply_join_routing(user_id, connection_id, is_new, home_placement)?;
238 let receipt = placement.receipt;
239 let mut transport_plan = placement.replacement_transport_plan;
240 if let Some(previous_connection) = previous_connection {
241 transport_plan.extend_teardown(
242 self.staged_publishes
243 .take_teardowns_for_connection(user_id, previous_connection),
244 );
245 }
246
247 let previous_sender =
248 self.install_joined_session(user_id, permissions, sender, connection_id);
249 let had_previous_sender = previous_sender.is_some();
250
251 let mut effects = LifecycleEffects::default();
252 effects.push_close_request(previous_sender.map(|sender| UserCloseRequest {
253 sender,
254 reason: UserCloseReason::Replaced,
255 }));
256 effects
257 .track_snapshots
258 .extend(self.remote_track_snapshots_for_users(source_recipients, true));
259 effects.push_fanout(had_previous_sender.then(|| {
260 self.fanout_all_except(
261 &RoomEventMessage::UserDeparted {
262 user_id: user_id.clone(),
263 },
264 user_id,
265 )
266 }));
267 effects.push_fanout(if joined_fanout == UserJoinedFanout::Emit {
268 self.user_info_snapshot(user_id)
269 .map(|(joined_user_id, info)| {
270 self.fanout_all_except(
271 &RoomEventMessage::UserJoined {
272 user_id: joined_user_id,
273 info,
274 },
275 user_id,
276 )
277 })
278 } else {
279 None
280 });
281 Ok(JoinCommit {
282 effects,
283 receipt,
284 transport_plan,
285 })
286 }
287
288 fn remove_runtime_user(&mut self, user_id: &UserId) -> Option<RuntimeUserRemoval> {
289 let user = self.users.remove(user_id)?;
290 let mut transport_plan = self.topology.remove_session(user_id);
291 transport_plan.extend_teardown(
292 self.staged_publishes
293 .take_teardowns_for_connection(user_id, user.connection_id),
294 );
295 Some((user, transport_plan))
296 }
297
298 pub fn close_connection(
299 &mut self,
300 user_id: &UserId,
301 connection_id: ConnectionId,
302 ) -> Option<ConnectionCloseCommit> {
303 if self
304 .users
305 .get(user_id)
306 .is_none_or(|user| user.connection_id != connection_id)
307 {
308 let session_key = self
309 .topology
310 .retire_committed_placement(user_id, connection_id)?;
311 return Some(ConnectionCloseCommit::StalePlacement {
312 session_teardown: TransportTeardown::CloseSession { session_key },
313 });
314 }
315 let session_teardown = self
316 .committed_transport_user_key(user_id, connection_id)
317 .map(|session_key| TransportTeardown::CloseSession { session_key });
318 let mut source_recipients = self
319 .topology
320 .committed_consumer_user_ids_for_owner_sources(user_id);
321 source_recipients.remove(user_id);
322 let (user, transport_plan) = self.remove_runtime_user(user_id)?;
323 Some(ConnectionCloseCommit::Current {
324 user_id: user_id.clone(),
325 connection_id,
326 session_teardown,
327 effects: LifecycleEffects {
328 close_requests: vec![UserCloseRequest {
329 sender: user.sender,
330 reason: UserCloseReason::RemovedByRuntime,
331 }],
332 fanouts: vec![self.fanout_all(&RoomEventMessage::UserDeparted {
333 user_id: user_id.clone(),
334 })],
335 track_snapshots: self.remote_track_snapshots_for_users(source_recipients, true),
336 },
337 transport_plan,
338 })
339 }
340
341 pub fn apply_presence_update(
342 &mut self,
343 user_id: &UserId,
344 connection_id: ConnectionId,
345 info: &UserInfo,
346 ) -> Option<PresenceCommit> {
347 let Some(current_user) = self.users.get(user_id) else {
348 warn!(
349 ?user_id,
350 connection_id = ?connection_id,
351 ?info,
352 "discarding user presence update because the user is missing"
353 );
354 return None;
355 };
356 if current_user.connection_id != connection_id {
357 warn!(
358 ?user_id,
359 connection_id = ?connection_id,
360 current_connection_id = ?current_user.connection_id,
361 ?info,
362 "discarding user presence update because the connection is stale"
363 );
364 return None;
365 }
366 {
367 let user = self.user_mut_for_connection(user_id, connection_id)?;
368 user.apply_info_update(info);
369 }
370 let snapshot = BTreeMap::from([self.user_info_snapshot(user_id)?]);
371 debug!(
372 ?user_id,
373 connection_id = ?connection_id,
374 ?info,
375 snapshot_len = snapshot.len(),
376 "applied user presence update and staged user info fanout"
377 );
378 Some(PresenceCommit {
379 fanout: self.fanout_all(&RoomEventMessage::UserInfoChanged(snapshot)),
380 })
381 }
382
383 pub fn set_user_negotiated(
384 &mut self,
385 user_id: &UserId,
386 connection_id: ConnectionId,
387 capabilities: MediaCapabilities,
388 ) -> Option<bool> {
389 let user = self.user_mut_for_connection(user_id, connection_id)?;
390 let became_ready = user.parsed_client_rtp_capabilities.is_none();
391 user.parsed_client_rtp_capabilities = Some(capabilities);
392 Some(became_ready)
393 }
394
395 pub fn apply_disconnect_users(&mut self, user_ids: &[UserId]) -> DisconnectCommit {
396 let mut source_recipients = BTreeSet::new();
397 for user_id in user_ids {
398 source_recipients.extend(
399 self.topology
400 .committed_consumer_user_ids_for_owner_sources(user_id),
401 );
402 }
403 for user_id in user_ids {
404 source_recipients.remove(user_id);
405 }
406 let mut close_requests = Vec::new();
407 let mut session_teardowns = Vec::new();
408 let mut fanouts = Vec::new();
409 let mut transport_plan = RoomTransportPlan::default();
410 for user_id in user_ids {
411 let Some(connection_id) = self.users.get(user_id).map(|user| user.connection_id) else {
412 continue;
413 };
414 let session_teardown = TransportTeardown::CloseSession {
415 session_key: self.transport_user_key(user_id, connection_id),
416 };
417 let Some((user, user_transport_plan)) = self.remove_runtime_user(user_id) else {
418 continue;
419 };
420 transport_plan.extend(user_transport_plan);
421 session_teardowns.push(session_teardown);
422 close_requests.push(UserCloseRequest {
423 sender: user.sender,
424 reason: UserCloseReason::RemovedByRuntime,
425 });
426 fanouts.push(self.fanout_all(&RoomEventMessage::UserDeparted {
427 user_id: user_id.clone(),
428 }));
429 }
430 DisconnectCommit {
431 session_teardowns,
432 effects: LifecycleEffects {
433 close_requests,
434 fanouts,
435 track_snapshots: self.remote_track_snapshots_for_users(source_recipients, true),
436 },
437 transport_plan,
438 }
439 }
440
441 pub fn broadcast_fanout(
453 &self,
454 user_id: &UserId,
455 connection_id: ConnectionId,
456 message: serde_json::Value,
457 ) -> Result<Option<MessageFanout>, BroadcastPayloadError> {
458 if self.user_for_connection(user_id, connection_id).is_none() {
459 return Ok(None);
460 }
461 let message = BroadcastPayload::try_new(message)?;
462 Ok(Some(self.fanout_all_except(
463 &RoomEventMessage::Broadcast {
464 sender_id: user_id.clone(),
465 message,
466 },
467 user_id,
468 )))
469 }
470}