1use itertools::Itertools;
2use o_sfu_rfc::rtp::{Mid, Rid, Ssrc};
3use o_sfu_router::{
4 MediaKind as RouterMediaKind, RouterError,
5 rtp::{MediaFormat, MediaStream as RouterRtpParameters},
6};
7use tracing::{error, warn};
8
9use super::{
10 super::{
11 outbound::{OutboundSender, VersionedRemoteTrackSnapshot},
12 state::{PresenceCommit, RoomState},
13 },
14 ReceiverRouteWork,
15 source_index::PublishedSources,
16 subscription::ReceiverRouteScope,
17};
18use crate::{
19 Bitrate,
20 engine::{
21 ConnectionId, UserId, UserInfo,
22 media_transport::{
23 ProducerActivity, SessionUploadEncoding, SourceActivityUpdate, TransportMediaId,
24 TransportSessionKey, TransportSourceActivityEffect, TransportSourceKey,
25 },
26 source_model::{
27 PublishedSourceDescriptor, PublishedSourceDescriptorParts, PublishedSourceId,
28 PublishedSourceOwner, SourceEncodingDescriptor, SourceEncodingDescriptorParts,
29 SourceModelError, SourcePolicy, SourcePublishIntent, UploadLayerPolicyRole,
30 UserStreamId,
31 },
32 },
33};
34
35#[derive(Debug, Clone)]
36pub struct ValidatedPublish {
37 pub session_key: TransportSessionKey,
38 pub stream_id: UserStreamId,
39 pub media_kind: RouterMediaKind,
40 pub policy: SourcePolicy,
41 pub presence: Option<UserInfo>,
42}
43
44#[derive(Debug)]
45pub struct PublishCommit {
46 pub receiver_route_work: ReceiverRouteWork,
47 pub presence: Option<PresenceCommit>,
48}
49
50#[derive(Debug)]
51pub enum PublishIntentPlan {
52 Activate(ProducerActivityCommit),
53 Noop,
54 Queue,
55 Stage(ValidatedPublish),
56}
57
58#[derive(Debug)]
59pub struct ProducerActivityCommit {
60 pub source: TransportSourceKey,
61 pub stream_id: UserStreamId,
62 pub update: SourceActivityUpdate,
63 pub remote_activity_effects: Vec<TransportSourceActivityEffect>,
64 pub track_snapshots: Vec<(OutboundSender, VersionedRemoteTrackSnapshot)>,
65 pub presence: Option<PresenceCommit>,
66}
67
68#[derive(Debug)]
69pub enum ProducerActivityRejection {
70 MissingPublication,
71 StalePublication,
72}
73
74#[derive(Debug, thiserror::Error)]
75pub(in crate::engine::room) enum PublicationCommitError {
76 #[error(transparent)]
77 Source(#[from] SourceModelError),
78 #[error(transparent)]
79 Router(#[from] RouterError),
80}
81
82impl RoomState {
83 pub fn apply_publish_intent(
84 &mut self,
85 user_id: &UserId,
86 publisher_connection_id: ConnectionId,
87 intent: &SourcePublishIntent,
88 can_stage: bool,
89 ) -> PublishIntentPlan {
90 let Some(user) = self.users.get(user_id) else {
91 warn!(
92 ?user_id,
93 publisher_connection_id = ?publisher_connection_id,
94 stream_id = %intent.stream_id(),
95 "cannot start publish because the user is missing from room state"
96 );
97 return PublishIntentPlan::Noop;
98 };
99 if user.connection_id != publisher_connection_id {
100 warn!(
101 ?user_id,
102 publisher_connection_id = ?publisher_connection_id,
103 current_connection_id = ?user.connection_id,
104 stream_id = %intent.stream_id(),
105 "cannot start publish because the connection is stale"
106 );
107 return PublishIntentPlan::Noop;
108 }
109 if self
110 .published_source_id(user_id, publisher_connection_id, intent.stream_id())
111 .is_some()
112 {
113 return self
114 .apply_publication_activity(
115 user_id,
116 publisher_connection_id,
117 intent.stream_id(),
118 true,
119 intent.presence(),
120 )
121 .map_or_else(|_| PublishIntentPlan::Noop, PublishIntentPlan::Activate);
122 }
123 if !can_stage {
124 return PublishIntentPlan::Queue;
125 }
126 self.validate_publish(user_id, publisher_connection_id, intent)
127 .map_or(PublishIntentPlan::Noop, PublishIntentPlan::Stage)
128 }
129
130 pub fn validate_publish(
131 &self,
132 user_id: &UserId,
133 publisher_connection_id: ConnectionId,
134 intent: &SourcePublishIntent,
135 ) -> Option<ValidatedPublish> {
136 let Some(user) = self.users.get(user_id) else {
137 warn!(
138 ?user_id,
139 publisher_connection_id = ?publisher_connection_id,
140 stream_id = %intent.stream_id(),
141 "cannot prepare negotiated publish because the user is missing from room state"
142 );
143 return None;
144 };
145 if user.connection_id != publisher_connection_id {
146 warn!(
147 ?user_id,
148 publisher_connection_id = ?publisher_connection_id,
149 current_connection_id = ?user.connection_id,
150 stream_id = %intent.stream_id(),
151 "cannot prepare negotiated publish because the connection is stale"
152 );
153 return None;
154 }
155 if user.parsed_client_rtp_capabilities.is_none() {
156 warn!(
157 ?user_id,
158 publisher_connection_id = ?publisher_connection_id,
159 stream_id = %intent.stream_id(),
160 "cannot prepare negotiated publish because the user is not publish-ready"
161 );
162 return None;
163 }
164 Some(ValidatedPublish {
165 session_key: self.transport_user_key(user_id, publisher_connection_id),
166 stream_id: intent.stream_id().clone(),
167 media_kind: intent.media_kind(),
168 policy: intent.policy(),
169 presence: intent.presence().cloned(),
170 })
171 }
172
173 pub fn commit_publish_reservation(
174 &mut self,
175 publish: ValidatedPublish,
176 consumable_rtp_parameters: RouterRtpParameters,
177 upload_encodings: &[SessionUploadEncoding],
178 transport_media_id: TransportMediaId,
179 ) -> Option<PublishCommit> {
180 self.validate_publish_commit(&publish, transport_media_id)?;
181 let owner_user_id = publish.session_key.user_id().clone();
182 let owner_connection_id = publish.session_key.connection_id();
183 let stream_id = publish.stream_id.clone();
184 let presence = publish.presence.clone();
185 let source_id = match self.topology.commit_publication(
186 publish,
187 consumable_rtp_parameters,
188 upload_encodings,
189 transport_media_id,
190 ) {
191 Ok(source_id) => source_id,
192 Err(error) => {
193 error!(
194 user_id = ?owner_user_id,
195 ?owner_connection_id,
196 stream_id = %stream_id,
197 ?transport_media_id,
198 ?error,
199 "failed to commit negotiated publication"
200 );
201 return None;
202 }
203 };
204 let receiver_route_work =
205 self.plan_missing_receiver_routes(ReceiverRouteScope::Source(source_id));
206 let presence = presence.and_then(|info| {
207 self.apply_presence_update(&owner_user_id, owner_connection_id, &info)
208 });
209 Some(PublishCommit {
210 receiver_route_work,
211 presence,
212 })
213 }
214
215 pub(in crate::engine::room) fn validate_publish_commit(
216 &self,
217 publish: &ValidatedPublish,
218 transport_media_id: TransportMediaId,
219 ) -> Option<()> {
220 let user_id = publish.session_key.user_id();
221 let connection_id = publish.session_key.connection_id();
222 let Some(user) = self.users.get(user_id) else {
223 warn!(
224 ?user_id,
225 ?connection_id,
226 stream_id = %publish.stream_id,
227 ?transport_media_id,
228 "cannot commit negotiated publish because the user is missing from room state"
229 );
230 return None;
231 };
232 let publish_ready = user.parsed_client_rtp_capabilities.is_some();
233 if user.connection_id != connection_id || !publish_ready {
234 warn!(
235 ?user_id,
236 ?connection_id,
237 current_connection_id = ?user.connection_id,
238 publish_ready,
239 stream_id = %publish.stream_id,
240 ?transport_media_id,
241 "cannot commit negotiated publish because the user state changed before commit"
242 );
243 return None;
244 }
245 if self
246 .topology
247 .source_id_for_owner_stream(user_id, &publish.stream_id)
248 .is_some()
249 {
250 warn!(
251 ?user_id,
252 ?connection_id,
253 stream_id = %publish.stream_id,
254 ?transport_media_id,
255 "cannot commit negotiated publish because a source already exists for this stream"
256 );
257 return None;
258 }
259 Some(())
260 }
261
262 #[must_use]
263 pub fn producer_stream_id_for_transport_media_id(
264 &self,
265 transport_media_id: TransportMediaId,
266 ) -> Option<UserStreamId> {
267 self.topology
268 .source_for_transport_media(transport_media_id)
269 .map(|source| source.descriptor.stream_id().clone())
270 }
271
272 #[must_use]
273 pub fn published_source_id(
274 &self,
275 owner: &UserId,
276 connection: ConnectionId,
277 stream: &UserStreamId,
278 ) -> Option<PublishedSourceId> {
279 self.topology.published_source_id(owner, connection, stream)
280 }
281
282 #[cfg(any(test, feature = "testing-transport"))]
283 pub fn published_source_id_for_user(
284 &self,
285 user: &UserId,
286 stream: &UserStreamId,
287 ) -> Option<PublishedSourceId> {
288 self.published_source_id(user, self.user_connection_id(user)?, stream)
289 }
290
291 pub fn apply_publication_activity(
292 &mut self,
293 user_id: &UserId,
294 connection_id: ConnectionId,
295 stream_id: &UserStreamId,
296 active: bool,
297 presence: Option<&UserInfo>,
298 ) -> Result<ProducerActivityCommit, ProducerActivityRejection> {
299 let source_id = self
300 .published_source_id(user_id, connection_id, stream_id)
301 .ok_or(ProducerActivityRejection::MissingPublication)?;
302 let revision = self
303 .topology
304 .set_published_source_activity(source_id, connection_id, active)
305 .ok_or(ProducerActivityRejection::StalePublication)?;
306 let source_recipients = self
307 .topology
308 .committed_consumer_user_ids_for_source(source_id);
309 let source = self
310 .topology
311 .published_source(source_id)
312 .ok_or(ProducerActivityRejection::StalePublication)?
313 .transport
314 .clone();
315 let update = SourceActivityUpdate::new(ProducerActivity::from_active(active), revision);
316 let remote_activity_effects = self.topology.source_activity_effects(&source, update);
317 let presence =
318 presence.and_then(|info| self.apply_presence_update(user_id, connection_id, info));
319 Ok(ProducerActivityCommit {
320 source,
321 stream_id: stream_id.clone(),
322 update,
323 remote_activity_effects,
324 track_snapshots: self.remote_track_snapshots_for_users(source_recipients, false),
325 presence,
326 })
327 }
328}
329
330pub(super) fn allocate_source_descriptor(
331 sources: &mut PublishedSources,
332 publish: &ValidatedPublish,
333 consumable_rtp_parameters: &RouterRtpParameters,
334 upload_encodings: &[SessionUploadEncoding],
335) -> Result<PublishedSourceDescriptor, SourceModelError> {
336 let source_id = sources.allocate_id();
337 let encodings = consumable_rtp_parameters
338 .bindings()
339 .map(|binding| {
340 let upload_profile = upload_profile_for_rid(upload_encodings, binding.rid());
341 SourceEncodingDescriptor::new(SourceEncodingDescriptorParts {
342 encoding_id: sources.allocate_encoding_id(),
343 source_id,
344 rid: binding.rid().map(Rid::new),
345 primary_ssrc: binding.ssrc().map(Ssrc::new),
346 repair_ssrc: None,
347 max_bitrate: binding
348 .max_bitrate()
349 .map(Bitrate::from_bps)
350 .or_else(|| upload_profile.and_then(MatchedUploadEncoding::max_bitrate)),
351 resolution_scale: upload_profile.and_then(MatchedUploadEncoding::resolution_scale),
352 max_framerate: upload_profile.and_then(MatchedUploadEncoding::max_framerate),
353 policy_role: upload_profile
354 .map(|profile| upload_layer_policy_role_for_rank(profile.rank)),
355 negotiated_format: negotiated_format_for_binding(
356 consumable_rtp_parameters,
357 binding.payload_type(),
358 ),
359 })
360 })
361 .collect::<Vec<_>>();
362 PublishedSourceDescriptor::new(PublishedSourceDescriptorParts {
363 source_id,
364 owner: PublishedSourceOwner::new(publish.session_key.user_id().clone()),
365 stream_id: publish.stream_id.clone(),
366 media_kind: publish.media_kind,
367 policy: publish.policy,
368 mid: consumable_rtp_parameters.mid().map(Mid::new),
369 encodings,
370 })
371}
372
373fn upload_profile_for_rid<'a>(
374 upload_encodings: &'a [SessionUploadEncoding],
375 rid: Option<&str>,
376) -> Option<MatchedUploadEncoding<'a>> {
377 let rid = rid?;
378 upload_encodings
379 .iter()
380 .enumerate()
381 .find(|(_rank, encoding)| encoding.rid == rid)
382 .map(|(rank, encoding)| MatchedUploadEncoding { rank, encoding })
383}
384
385#[derive(Debug, Clone, Copy)]
386struct MatchedUploadEncoding<'a> {
387 rank: usize,
388 encoding: &'a SessionUploadEncoding,
389}
390
391impl MatchedUploadEncoding<'_> {
392 const fn max_bitrate(self) -> Option<Bitrate> {
393 self.encoding.max_bitrate
394 }
395
396 const fn resolution_scale(self) -> Option<u16> {
397 self.encoding.resolution_scale
398 }
399
400 const fn max_framerate(self) -> Option<u16> {
401 self.encoding.max_framerate
402 }
403}
404
405const fn upload_layer_policy_role_for_rank(rank: usize) -> UploadLayerPolicyRole {
406 if rank == 0 {
407 UploadLayerPolicyRole::Thumbnail
408 } else {
409 UploadLayerPolicyRole::Featured
410 }
411}
412
413fn negotiated_format_for_binding(
414 parameters: &RouterRtpParameters,
415 payload_type: Option<u8>,
416) -> Option<MediaFormat> {
417 if let Some(payload_type) = payload_type
418 && let Some(format) = parameters
419 .formats()
420 .find(|format| format.payload_type() == payload_type)
421 {
422 return Some(format.clone());
423 }
424 parameters
425 .formats()
426 .find_or_first(|format| !format.codec().is_rtx())
427 .cloned()
428}