1use std::collections::{HashMap, HashSet};
2
3use serde_json::{Value, json};
4
5use super::common::{
6 download_main_stat, route_state_color, route_state_label, stream_id_color, stream_id_label,
7 transport_health_label,
8};
9use crate::diagnostics::types::{
10 DiagnosticsIncomingBitrate, DiagnosticsRoomDetail, DiagnosticsRouteState, DiagnosticsSource,
11 DiagnosticsSubscription, DiagnosticsUserView,
12};
13
14const ARC_RATIO_SCALE: u32 = 1_000_000;
15
16fn room_node(detail: &DiagnosticsRoomDetail) -> Value {
17 let room_uuid = detail.summary.uuid.as_str();
18 let short_uuid = if room_uuid.len() > 8 {
19 &room_uuid[..8]
20 } else {
21 room_uuid
22 };
23
24 json!({
25 "id": format!("room:{}", room_uuid),
26 "title": short_uuid,
27 "subtitle": "room",
28 "mainStat": format!("{} sessions", detail.summary.user_count),
29 "secondaryStat": format!("{} pub / {} sub", detail.summary.publication_count, detail.summary.subscription_count),
30 "detail__recording": format!("{:?}", detail.summary.recording_state.recording),
31 "detail__worker": detail.summary.media_worker_id,
32 "detail__transport": format!("{} conn / {} disc / {} unk", detail.summary.transport.connected, detail.summary.transport.disconnected, detail.summary.transport.unknown),
33 })
34}
35
36fn source_ids(detail: &DiagnosticsRoomDetail) -> HashSet<u64> {
37 detail
38 .sources
39 .iter()
40 .map(|source| source.source_id)
41 .collect()
42}
43
44fn download_counts(detail: &DiagnosticsRoomDetail) -> HashMap<u64, usize> {
45 let mut download_counts: HashMap<u64, usize> = HashMap::new();
46 for user in &detail.users {
47 for sub in &user.subscriptions {
48 *download_counts.entry(sub.source_id).or_insert(0) += 1;
49 }
50 }
51 download_counts
52}
53
54fn bitrate_share(part: u64, total: u64) -> Option<f64> {
55 if total == 0 {
56 return None;
57 }
58
59 let scaled = (u128::from(part.min(total)) * u128::from(ARC_RATIO_SCALE)) / u128::from(total);
60 let scaled = u32::try_from(scaled).map_or(ARC_RATIO_SCALE, |value| value);
61 Some(f64::from(scaled) / f64::from(ARC_RATIO_SCALE))
62}
63
64fn add_bitrate_arcs(node: &mut Value, bitrate: &DiagnosticsIncomingBitrate) {
65 if let Some(obj) = node.as_object_mut() {
66 for (stream_id, bps) in &bitrate.by_stream_bps {
67 if let Some(share) = bitrate_share(*bps, bitrate.total) {
68 let field_name = stream_id
69 .chars()
70 .map(|ch| if ch.is_ascii_alphanumeric() { ch } else { '_' })
71 .collect::<String>();
72 obj.insert(format!("arc__stream_{field_name}"), json!(share));
73 }
74 }
75 }
76}
77
78fn session_node(room_uuid: &str, user: &DiagnosticsUserView) -> Value {
79 let session_id = user.user_id.path_segment();
80 let health = transport_health_label(user.transport.health.as_ref());
81 let bitrate = &user.transport.quality_summary.current_incoming_bitrate;
82 let mut node = json!({
83 "id": format!("session:{}:{}", room_uuid, session_id),
84 "title": session_id,
85 "subtitle": health,
86 "mainStat": format!("{} bps", bitrate.total),
87 "secondaryStat": format!("{} pub / {} sub", user.publications.len(), user.subscriptions.len()),
88 "detail__connection": user.transport.connection_id.to_string(),
89 "detail__worker": user.transport.media_worker_id,
90 });
91
92 add_bitrate_arcs(&mut node, bitrate);
93 node
94}
95
96fn push_session_entries(
97 nodes: &mut Vec<Value>,
98 edges: &mut Vec<Value>,
99 detail: &DiagnosticsRoomDetail,
100) {
101 let room_uuid = detail.summary.uuid.as_str();
102 for user in &detail.users {
103 let session_id = user.user_id.path_segment();
104 let health = transport_health_label(user.transport.health.as_ref());
105 let edge_color = match health {
106 "connected" => "green",
107 "disconnected" => "red",
108 _ => "gray",
109 };
110 nodes.push(session_node(room_uuid, user));
111 edges.push(json!({
112 "id": format!("member:{}:{}", room_uuid, session_id),
113 "source": format!("room:{}", room_uuid),
114 "target": format!("session:{}:{}", room_uuid, session_id),
115 "mainStat": health,
116 "color": edge_color,
117 }));
118 }
119}
120
121fn encoding_detail_lists(source: &DiagnosticsSource) -> (String, String, String, String, String) {
122 let enc_ids: Vec<String> = source
123 .encodings
124 .iter()
125 .map(|e| e.encoding_id.to_string())
126 .collect();
127 let rids: Vec<String> = source
128 .encodings
129 .iter()
130 .filter_map(|e| e.rid.clone())
131 .collect();
132 let max_bitrates: Vec<String> = source
133 .encodings
134 .iter()
135 .filter_map(|e| e.max_bitrate_bps.map(|b| b.to_string()))
136 .collect();
137 let primary_ssrcs: Vec<String> = source
138 .encodings
139 .iter()
140 .filter_map(|e| e.primary_ssrc.map(|s| s.to_string()))
141 .collect();
142 let repair_ssrcs: Vec<String> = source
143 .encodings
144 .iter()
145 .filter_map(|e| e.repair_ssrc.map(|s| s.to_string()))
146 .collect();
147
148 (
149 enc_ids.join(", "),
150 rids.join(", "),
151 max_bitrates.join(", "),
152 primary_ssrcs.join(", "),
153 repair_ssrcs.join(", "),
154 )
155}
156
157fn push_source_entries(
158 nodes: &mut Vec<Value>,
159 edges: &mut Vec<Value>,
160 detail: &DiagnosticsRoomDetail,
161 download_counts: &HashMap<u64, usize>,
162) {
163 let room_uuid = detail.summary.uuid.as_str();
164 for source in &detail.sources {
165 let stream_id = stream_id_label(&source.stream_id);
166 let media_kind_str = format!("{:?}", source.media_kind).to_lowercase();
167 let active_str = if source.active { "active" } else { "inactive" };
168 let downloads = download_counts.get(&source.source_id).copied().unwrap_or(0);
169 let (enc_ids, rids, max_bitrates, primary_ssrcs, repair_ssrcs) =
170 encoding_detail_lists(source);
171
172 nodes.push(json!({
173 "id": format!("source:{}:{}", room_uuid, source.source_id),
174 "title": format!("{} #{}", stream_id, source.source_id),
175 "subtitle": format!("{} {}", active_str, media_kind_str),
176 "mainStat": format!("{} bps", source.current_incoming_bitrate_bps),
177 "secondaryStat": format!("{} encodings / {} downloads", source.encodings.len(), downloads),
178 "detail__owner_session_id": source.owner_user_id.path_segment(),
179 "detail__stream_id": stream_id,
180 "detail__media_kind": media_kind_str,
181 "detail__transport_media_id": source.transport_media_id,
182 "detail__mid": source.mid,
183 "detail__encoding_ids": enc_ids,
184 "detail__rids": rids,
185 "detail__max_bitrates_bps": max_bitrates,
186 "detail__primary_ssrcs": primary_ssrcs,
187 "detail__repair_ssrcs": repair_ssrcs,
188 }));
189
190 let thickness = if source.current_incoming_bitrate_bps > 0 {
191 2.0
192 } else {
193 1.0
194 };
195 let color = stream_id_color(&source.stream_id);
196
197 edges.push(json!({
198 "id": format!("publish:{}:{}", room_uuid, source.source_id),
199 "source": format!("session:{}:{}", room_uuid, source.owner_user_id.path_segment()),
200 "target": format!("source:{}:{}", room_uuid, source.source_id),
201 "mainStat": format!("{} upload", stream_id),
202 "secondaryStat": format!("{} bps", source.current_incoming_bitrate_bps),
203 "thickness": thickness,
204 "color": color,
205 "detail__source_id": source.source_id,
206 "detail__encoding_ids": enc_ids,
207 "detail__rids": rids,
208 "detail__transport_media_id": source.transport_media_id,
209 }));
210 }
211}
212
213fn download_edge(
214 room_uuid: &str,
215 user: &DiagnosticsUserView,
216 sub: &DiagnosticsSubscription,
217) -> Value {
218 let session_id = user.user_id.path_segment();
219 let mut edge = json!({
220 "id": format!("download:{}:{}:{}", room_uuid, sub.source_id, session_id),
221 "source": format!("source:{}:{}", room_uuid, sub.source_id),
222 "target": format!("session:{}:{}", room_uuid, session_id),
223 "mainStat": download_main_stat(sub),
224 "secondaryStat": route_state_label(&sub.state),
225 "color": route_state_color(&sub.state),
226 "detail__source_id": sub.source_id,
227 "detail__producer_session_id": sub.producer_user_id.path_segment(),
228 "detail__stream_id": &sub.stream_id,
229 "detail__selector": format!("{:?}", sub.selection.selector),
230 "detail__selection_reason": format!("{:?}", sub.selection.selection_reason),
231 "detail__selection_active": sub.selection.active,
232 "detail__pressure_observations": sub.selection.pressure_observations,
233 "detail__upgrade_observations": sub.selection.upgrade_observations,
234 "detail__source_transport_media_id": sub.source_transport_media_id,
235 "detail__consumer_transport_media_id": sub.consumer_transport_media_id,
236 });
237
238 if let Some(obj) = edge.as_object_mut() {
239 if sub.state != DiagnosticsRouteState::Active {
240 obj.insert("strokeDasharray".to_string(), json!("5, 5"));
241 }
242 if let Some(enc_id) = sub.selection.selected_encoding_id {
243 obj.insert("detail__selected_encoding_id".to_string(), json!(enc_id));
244 }
245 if let Some(rid) = &sub.selection.selected_rid {
246 obj.insert("detail__selected_rid".to_string(), json!(rid));
247 }
248 }
249
250 edge
251}
252
253fn push_download_entries(
254 nodes: &mut Vec<Value>,
255 edges: &mut Vec<Value>,
256 detail: &DiagnosticsRoomDetail,
257 source_ids: &HashSet<u64>,
258) {
259 let room_uuid = detail.summary.uuid.as_str();
260 for user in &detail.users {
261 for sub in &user.subscriptions {
262 if !source_ids.contains(&sub.source_id) {
263 nodes.push(json!({
264 "id": format!("source:{}:{}", room_uuid, sub.source_id),
265 "title": format!("missing #{}", sub.source_id),
266 "subtitle": "not found",
267 }));
268 }
269 edges.push(download_edge(room_uuid, user, sub));
270 }
271 }
272}
273
274#[must_use]
286pub fn build_graph(detail: &DiagnosticsRoomDetail) -> Value {
287 let mut nodes = Vec::new();
288 let mut edges = Vec::new();
289 let source_ids = source_ids(detail);
290 let download_counts = download_counts(detail);
291
292 nodes.push(room_node(detail));
293 push_session_entries(&mut nodes, &mut edges, detail);
294 push_source_entries(&mut nodes, &mut edges, detail, &download_counts);
295 push_download_entries(&mut nodes, &mut edges, detail, &source_ids);
296
297 json!({
298 "nodes": nodes,
299 "edges": edges,
300 })
301}