1use std::collections::BTreeMap;
2
3use o_sfu_protocol::{
4 host::NegotiationKind,
5 wire::{
6 ClientResponse, DownloadStates, NegotiationUploadEncoding, NegotiationUploadSlot,
7 RequestId, ServerEnvelope, ServerRequest, SessionDescriptionPayload, StreamType, UserId,
8 },
9};
10use tracing::{Span, field, instrument, warn};
11
12use super::{User, UserError, UserOutput};
13use crate::{
14 application::stream_catalog::{DiscussStream, source_publish_intent_for_stream_type},
15 core::prelude::{
16 Bitrate, MediaSession, NegotiationOffer, SessionError, SfuCoreError,
17 SourceDeactivateIntent, SourcePublishIntent,
18 },
19 runtime::telemetry::schema::event as telemetry_event,
20};
21
22impl User {
23 pub(super) async fn complete_negotiation(
24 &mut self,
25 response_to: RequestId,
26 response: ClientResponse,
27 ) -> Result<UserOutput, UserError> {
28 let result = self.media.answer(response_to, response).await;
29 result.map_err(|e| self.answer_error(e))
30 }
31
32 #[instrument(
33 name = "transport.renegotiate",
34 skip_all,
35 fields(
36 room_id = %self.room_id(),
37 user_id = ?self.user_id(),
38 connection_id = ?self.connection_id()
39 )
40 )]
41 pub(super) async fn renegotiate(&mut self) -> Result<UserOutput, UserError> {
42 self.reject_stale_connection().await?;
43 let result = self.media.renegotiate().await;
44 result.map_err(|e| self.negotiation_error(NegotiationKind::Renegotiate, None, e))
45 }
46
47 #[instrument(
48 name = "publish.intent",
49 skip_all,
50 fields(
51 room_id = %self.room_id(),
52 user_id = ?self.user_id(),
53 connection_id = ?self.connection_id(),
54 ?stream_type,
55 active
56 )
57 )]
58 pub(super) async fn set_publication_active(
59 &mut self,
60 stream_type: StreamType,
61 active: bool,
62 ) -> Result<UserOutput, UserError> {
63 if active {
64 let intent = source_publish_intent_for_stream_type(stream_type);
65 return self
66 .media
67 .publish(intent)
68 .await
69 .map_err(|error| self.publish_error(stream_type, error));
70 }
71 let intent = DiscussStream::for_type(stream_type).deactivate_intent();
72 Ok(self.media.deactivate_publication(intent).await)
73 }
74
75 #[instrument(
76 name = "subscribe.intent",
77 skip_all,
78 fields(
79 room_id = %self.room_id(),
80 user_id = ?self.user_id(),
81 connection_id = ?self.connection_id(),
82 target_session_id = field::Empty,
83 source_count = field::Empty
84 )
85 )]
86 pub(super) async fn subscribe(
87 &self,
88 target_user_id: UserId,
89 states: DownloadStates,
90 ) -> Result<UserOutput, UserError> {
91 let target_user_id = target_user_id.normalized_for_runtime();
92 let span = Span::current();
93 span.record("target_session_id", field::debug(&target_user_id));
94 let source_intents = DiscussStream::all()
95 .filter_map(|stream| stream.subscription_intent_if_requested(&states))
96 .collect::<BTreeMap<_, _>>();
97 span.record("source_count", source_intents.len());
98 let session = self.media.session();
99 let result = session.subscribe(&target_user_id, &source_intents).await;
100 result.map_err(|error| self.subscribe_error(&target_user_id, error))?;
101 Ok(UserOutput::new())
102 }
103
104 #[instrument(
105 name = "transport.offer.create",
106 skip_all,
107 fields(
108 room_id = %self.room_id(),
109 user_id = ?self.user_id(),
110 connection_id = ?self.connection_id()
111 )
112 )]
113 pub(super) async fn run_initial_offer(&mut self) -> Result<UserOutput, UserError> {
114 let result = self.media.establish().await;
115 result.map_err(|e| self.negotiation_error(NegotiationKind::Offer, None, e))
116 }
117}
118
119pub(super) struct ServerMediaNegotiation {
120 session: MediaSession,
121 next: u64,
122 pending: Option<(RequestId, NegotiationKind)>,
123}
124
125enum AnswerError {
126 Protocol(RequestId, &'static str),
127 Session(NegotiationKind, RequestId, SessionError),
128}
129
130const EMPTY_SDP_ANSWER_LOG: &str = "received empty SDP answer for negotiation request";
131const UNKNOWN_ANSWER_LOG: &str = "received negotiation answer for an unknown or stale request";
132
133impl ServerMediaNegotiation {
134 pub(super) fn new(session: MediaSession) -> Self {
135 Self {
136 session,
137 next: 0,
138 pending: None,
139 }
140 }
141
142 pub(super) const fn session(&self) -> &MediaSession {
143 &self.session
144 }
145
146 async fn establish(&mut self) -> Result<UserOutput, SessionError> {
147 let offer = self.session.establish().await?;
148 Ok(self.issue(NegotiationKind::Offer, offer))
149 }
150
151 async fn renegotiate(&mut self) -> Result<UserOutput, SessionError> {
152 let offer = self.session.renegotiate().await?;
153 Ok(self.issue(NegotiationKind::Renegotiate, offer))
154 }
155
156 async fn publish(&mut self, intent: SourcePublishIntent) -> Result<UserOutput, SessionError> {
157 let offer = self.session.publish(intent).await?;
158 Ok(self.issue(NegotiationKind::Renegotiate, offer))
159 }
160
161 async fn deactivate_publication(&mut self, intent: SourceDeactivateIntent) -> UserOutput {
162 self.session.deactivate_publication(intent).await;
163 UserOutput::new()
164 }
165
166 async fn answer(
167 &mut self,
168 response_to: RequestId,
169 response: ClientResponse,
170 ) -> Result<UserOutput, AnswerError> {
171 let (kind, answer) = match response {
172 ClientResponse::Offer(answer) => (NegotiationKind::Offer, answer),
173 ClientResponse::Renegotiate(answer) => (NegotiationKind::Renegotiate, answer),
174 };
175 if answer.sdp.is_empty() {
176 return Err(AnswerError::Protocol(response_to, EMPTY_SDP_ANSWER_LOG));
177 }
178 if !self.expects(&response_to, kind) {
179 return Err(AnswerError::Protocol(response_to, UNKNOWN_ANSWER_LOG));
180 }
181 let offer = match self.session.answer(&answer.sdp).await {
182 Ok(offer) => offer,
183 Err(error) => return Err(AnswerError::Session(kind, response_to, error)),
184 };
185 self.pending = None;
186 Ok(self.issue(NegotiationKind::Renegotiate, offer))
187 }
188
189 pub(super) async fn close(&mut self) {
190 self.session.close().await;
191 }
192
193 fn issue(&mut self, kind: NegotiationKind, offer: Option<NegotiationOffer>) -> UserOutput {
194 let Some(offer) = offer else {
195 return UserOutput::new();
196 };
197 let req_id = RequestId::new(format!("server-{}", self.next));
198 self.next = self.next.saturating_add(1);
199 let payload = session_description_payload(offer);
200 let request = match kind {
201 NegotiationKind::Offer => ServerRequest::Offer(payload),
202 NegotiationKind::Renegotiate => ServerRequest::Renegotiate(payload),
203 };
204 self.pending = Some((req_id.clone(), kind));
205 vec![ServerEnvelope::Request {
206 request_id: req_id,
207 request,
208 }]
209 }
210
211 fn expects(&self, id: &RequestId, kind: NegotiationKind) -> bool {
212 matches!(self.pending.as_ref(), Some((req_id, pending_kind)) if req_id == id && *pending_kind == kind)
213 }
214}
215
216fn session_description_payload(offer: NegotiationOffer) -> SessionDescriptionPayload {
217 SessionDescriptionPayload {
218 sdp: offer.sdp,
219 upload_slots: offer
220 .upload_slots
221 .into_iter()
222 .map(|slot| NegotiationUploadSlot {
223 mid: slot.mid,
224 kind: slot.kind,
225 codecs: slot.codecs,
226 simulcast_encodings: slot
227 .simulcast_encodings
228 .into_iter()
229 .map(|encoding| NegotiationUploadEncoding {
230 rid: encoding.rid,
231 max_bitrate: encoding.max_bitrate.map(Bitrate::as_bps),
232 resolution_scale: encoding.resolution_scale,
233 max_framerate: encoding.max_framerate,
234 })
235 .collect(),
236 })
237 .collect(),
238 }
239}
240
241impl User {
242 fn answer_error(&self, error: AnswerError) -> UserError {
243 match error {
244 AnswerError::Protocol(response_to, message) => {
245 warn!(
246 user_id = ?self.user_id(),
247 connection_id = ?self.connection_id(),
248 remote_address = self.remote_address.as_ref(),
249 ?response_to,
250 "{message}"
251 );
252 UserError::ProtocolViolation
253 }
254 AnswerError::Session(kind, response_to, error) => {
255 self.negotiation_error(kind, Some(&response_to), error)
256 }
257 }
258 }
259
260 fn negotiation_error(
261 &self,
262 kind: NegotiationKind,
263 response_to: Option<&RequestId>,
264 error: SessionError,
265 ) -> UserError {
266 let operation = match kind {
267 NegotiationKind::Offer => "initial_offer_create",
268 NegotiationKind::Renegotiate => "renegotiation_offer_create",
269 };
270 let outcome = match error {
271 SessionError::NoPendingRequest => "no_pending_media_request",
272 SessionError::Core(error) if error.is_client_error() => "client_negotiation_error",
273 SessionError::Core(_) => "transport_error",
274 };
275 warn!(
276 event = telemetry_event::NEGOTIATION_FAILED,
277 operation,
278 outcome,
279 user_id = ?self.user_id(),
280 connection_id = ?self.connection_id(),
281 remote_address = self.remote_address.as_ref(),
282 response_to = ?response_to,
283 ?error,
284 "media session command failed"
285 );
286 user_error(error)
287 }
288
289 fn publish_error(&self, stream_type: StreamType, error: SessionError) -> UserError {
290 warn!(
291 event = telemetry_event::PUBLISH_ABORTED,
292 operation = "publish_intent",
293 outcome = "publish_rejected",
294 user_id = ?self.user_id(),
295 connection_id = ?self.connection_id(),
296 remote_address = self.remote_address.as_ref(),
297 ?stream_type,
298 ?error,
299 "media session command failed"
300 );
301 user_error(error)
302 }
303
304 fn subscribe_error(&self, target_user_id: &UserId, error: SessionError) -> UserError {
305 let outcome = match error {
306 SessionError::Core(SfuCoreError::SubscriptionUpdateRejected) => "stale_connection",
307 SessionError::NoPendingRequest | SessionError::Core(_) => "subscription_failed",
308 };
309 warn!(
310 event = telemetry_event::SUBSCRIBE_REJECTED,
311 operation = "consume_prepare",
312 outcome,
313 user_id = ?self.user_id(),
314 connection_id = ?self.connection_id(),
315 remote_address = self.remote_address.as_ref(),
316 ?target_user_id,
317 ?error,
318 "media session command failed"
319 );
320 user_error(error)
321 }
322}
323
324fn user_error(error: SessionError) -> UserError {
325 if error.is_client_error() {
326 UserError::ProtocolViolation
327 } else {
328 UserError::InternalError
329 }
330}