+7

.env.example

Reviewed

··· 54 54 # forward it to the container. If unset, mediasoup picks ephemeral ports. 55 55 SFU_RTC_MIN_PORT=40000 56 56 SFU_RTC_MAX_PORT=40100 57 57 + 58 58 + # STUN/TURN servers handed to clients for ICE, as a JSON array of RTCIceServer 59 59 + # objects. A TURN server is what lets clients on restrictive/symmetric-NAT 60 60 + # networks (and some browsers) connect when a direct path to the media ports 61 61 + # fails. Unset means no ICE servers (direct connection only). Example: 62 62 + # SFU_ICE_SERVERS=[{"urls":["turn:turn.example.com:3478"],"username":"user","credential":"pass"}] 63 63 + SFU_ICE_SERVERS=

+5

docker-compose.dev.yml

Reviewed

··· 10 10 TAP_HOSTNAME: tap:2480 11 11 ROCKET_ADDRESS: 0.0.0.0 12 12 ROCKET_PORT: 8000 13 13 + SFU_ANNOUNCED_IP: 127.0.0.1 14 14 + SFU_RTC_MIN_PORT: 40000 15 15 + SFU_RTC_MAX_PORT: 40019 13 16 ports: 14 17 - "8000:8000" 18 18 + - "40000-40019:40000-40019/udp" 19 19 + - "40000-40019:40000-40019/tcp" 15 20 depends_on: 16 21 postgres: 17 22 condition: service_healthy

+3

docker-compose.yml

Reviewed

··· 7 7 environment: 8 8 ROCKET_ADDRESS: 0.0.0.0 9 9 ROCKET_PORT: 8000 10 10 + ports: 11 11 + - "${SFU_RTC_MIN_PORT:-40000}-${SFU_RTC_MAX_PORT:-40100}:${SFU_RTC_MIN_PORT:-40000}-${SFU_RTC_MAX_PORT:-40100}/udp" 12 12 + - "${SFU_RTC_MIN_PORT:-40000}-${SFU_RTC_MAX_PORT:-40100}:${SFU_RTC_MIN_PORT:-40000}-${SFU_RTC_MAX_PORT:-40100}/tcp" 10 13 depends_on: 11 14 postgres: 12 15 condition: service_healthy

+2

migration/src/lib.rs

Reviewed

··· 10 10 mod m20260624_120000_create_dismissed_applications; 11 11 mod m20260625_000000_create_push_subscriptions; 12 12 mod m20260628_120000_record_data_data_to_jsonb; 13 13 + mod m20260709_000000_add_vc_state_to_user_states; 13 14 14 15 pub struct Migrator; 15 16 ··· 27 28 Box::new(m20260624_120000_create_dismissed_applications::Migration), 28 29 Box::new(m20260625_000000_create_push_subscriptions::Migration), 29 30 Box::new(m20260628_120000_record_data_data_to_jsonb::Migration), 31 31 + Box::new(m20260709_000000_add_vc_state_to_user_states::Migration), 30 32 ] 31 33 } 32 34 }

+31

migration/src/m20260709_000000_add_vc_state_to_user_states.rs

Reviewed

··· 1 1 + use sea_orm_migration::{prelude::*, schema::*}; 2 2 + 3 3 + #[derive(DeriveMigrationName)] 4 4 + pub struct Migration; 5 5 + 6 6 + #[async_trait::async_trait] 7 7 + impl MigrationTrait for Migration { 8 8 + async fn up(&self, manager: &SchemaManager) -> Result<(), DbErr> { 9 9 + manager 10 10 + .alter_table( 11 11 + Table::alter() 12 12 + .table("user_states") 13 13 + .add_column_if_not_exists(boolean_null("vc_muted")) 14 14 + .add_column_if_not_exists(boolean_null("vc_deafened")) 15 15 + .to_owned(), 16 16 + ) 17 17 + .await 18 18 + } 19 19 + 20 20 + async fn down(&self, manager: &SchemaManager) -> Result<(), DbErr> { 21 21 + manager 22 22 + .alter_table( 23 23 + Table::alter() 24 24 + .table("user_states") 25 25 + .drop_column_if_exists("vc_muted") 26 26 + .drop_column_if_exists("vc_deafened") 27 27 + .to_owned(), 28 28 + ) 29 29 + .await 30 30 + } 31 31 + }

+3

rust-toolchain.toml

Reviewed

··· 1 1 + [toolchain] 2 2 + channel = "1.93" 3 3 + components = ["rustfmt"]

+19

src/lib/events.rs

Reviewed

··· 264 264 } 265 265 266 266 #[derive(Serialize, Deserialize, Debug, Clone)] 267 267 + pub struct VoiceStateEventData { 268 268 + pub channel: String, 269 269 + pub did: String, 270 270 + pub muted: bool, 271 271 + pub deafened: bool, 272 272 + } 273 273 + 274 274 + #[derive(Serialize, Deserialize, Debug, Clone)] 267 275 pub struct NotificationEventMessage { 268 276 pub text: String, 269 277 #[serde(default)] ··· 375 383 User(UserEventData), 376 384 Typing(TypingEventData), 377 385 VoicePresence(VoicePresenceEventData), 386 386 + VoiceState(VoiceStateEventData), 378 387 Notification(NotificationEventData), 379 388 Seen(SeenEventData), 380 389 Mute(MuteEventData), ··· 411 420 Typing(TypingEventData), 412 421 #[serde(rename = "voice_presence_event")] 413 422 VoicePresence(VoicePresenceEventData), 423 423 + #[serde(rename = "voice_state_event")] 424 424 + VoiceState(VoiceStateEventData), 414 425 } 415 426 416 427 /// An off-protocol event relayed between AppViews. Authenticated via ··· 433 444 pub struct VoiceChannelData { 434 445 pub channel: String, 435 446 pub community: String, 447 447 + } 448 448 + 449 449 + #[derive(Serialize, Deserialize, Debug, Clone)] 450 450 + pub struct VoiceStateData { 451 451 + pub channel: String, 452 452 + pub community: String, 453 453 + pub muted: bool, 454 454 + pub deafened: bool, 436 455 } 437 456 438 457 #[derive(Serialize, Deserialize, Debug, Clone)]

+2

src/lib/get_state.rs

Reviewed

··· 61 61 channel: None, 62 62 vc: None, 63 63 vc_community: None, 64 64 + vc_muted: None, 65 65 + vc_deafened: None, 64 66 }]]) 65 67 .into_connection(); 66 68

+22 -1

src/lib/hum_client.rs

Reviewed

··· 34 34 use crate::lib::event_scope::SharedScopedEvent; 35 35 use crate::lib::events::{ 36 36 HumEnvelope, HumEvent, TypingEventData, UserEventData, UserEventProfile, UserEventStatus, 37 37 - VoicePresenceEventData, 37 37 + VoicePresenceEventData, VoiceStateEventData, 38 38 }; 39 39 use crate::lib::get_atproto_record::get_atproto_record; 40 40 use crate::lib::get_state::get_state; ··· 83 83 did: String, 84 84 channel: String, 85 85 event: String, 86 86 + }, 87 87 + VoiceState { 88 88 + did: String, 89 89 + channel: String, 90 90 + muted: bool, 91 91 + deafened: bool, 86 92 }, 87 93 } 88 94 ··· 194 200 event, 195 201 channel, 196 202 did: did.clone(), 203 203 + }); 204 204 + Some((did, payload, vec![(community_uri, hub)])) 205 205 + } 206 206 + OutboundHum::VoiceState { 207 207 + did, 208 208 + channel, 209 209 + muted, 210 210 + deafened, 211 211 + } => { 212 212 + let (community_uri, hub) = channel_target(db, &channel).await?; 213 213 + let payload = HumEvent::VoiceState(VoiceStateEventData { 214 214 + channel, 215 215 + did: did.clone(), 216 216 + muted, 217 217 + deafened, 197 218 }); 198 219 Some((did, payload, vec![(community_uri, hub)])) 199 220 }

+37

src/lib/state.rs

Reviewed

··· 97 97 user_states::Column::VcCommunity, 98 98 Expr::value(Option::<String>::None), 99 99 ) 100 100 + .col_expr( 101 101 + user_states::Column::VcMuted, 102 102 + Expr::value(Option::<bool>::None), 103 103 + ) 104 104 + .col_expr( 105 105 + user_states::Column::VcDeafened, 106 106 + Expr::value(Option::<bool>::None), 107 107 + ) 108 108 + .filter(user_states::Column::Did.eq(did)) 109 109 + .exec(db) 110 110 + .await; 111 111 + } 112 112 + 113 113 + pub async fn reset_all_presence(db: &DatabaseConnection) { 114 114 + let _ = UserStates::update_many() 115 115 + .col_expr(user_states::Column::State, Expr::value("offline")) 116 116 + .col_expr(user_states::Column::Vc, Expr::value(Option::<String>::None)) 117 117 + .col_expr( 118 118 + user_states::Column::VcCommunity, 119 119 + Expr::value(Option::<String>::None), 120 120 + ) 121 121 + .col_expr( 122 122 + user_states::Column::VcMuted, 123 123 + Expr::value(Option::<bool>::None), 124 124 + ) 125 125 + .col_expr( 126 126 + user_states::Column::VcDeafened, 127 127 + Expr::value(Option::<bool>::None), 128 128 + ) 129 129 + .exec(db) 130 130 + .await; 131 131 + } 132 132 + 133 133 + pub async fn set_vc_state(did: String, muted: bool, deafened: bool, db: &DatabaseConnection) { 134 134 + let _ = UserStates::update_many() 135 135 + .col_expr(user_states::Column::VcMuted, Expr::value(muted)) 136 136 + .col_expr(user_states::Column::VcDeafened, Expr::value(deafened)) 100 137 .filter(user_states::Column::Did.eq(did)) 101 138 .exec(db) 102 139 .await;

+2

src/main.rs

Reviewed

··· 178 178 .await 179 179 .expect("Unable to apply SeaORM migrations"); 180 180 181 181 + crate::lib::state::reset_all_presence(&db).await; 182 182 + 181 183 // Supervise the tap connection: connect, run until the socket drops, then 182 184 // reconnect after a short backoff. The broadcast senders are cloned into 183 185 // each connection; subscribers' receivers persist across reconnects.

+2

src/models/user_states.rs

Reviewed

··· 17 17 pub vc_community: Option<String>, 18 18 #[sea_orm(column_type = "Text")] 19 19 pub channel: Option<String>, 20 20 + pub vc_muted: Option<bool>, 21 21 + pub vc_deafened: Option<bool>, 20 22 } 21 23 22 24 impl ActiveModelBehavior for ActiveModel {}

+26 -8

src/sfu.rs

Reviewed

··· 2 2 mod backend { 3 3 use std::collections::HashMap; 4 4 use std::net::IpAddr; 5 5 - use std::num::{NonZeroU8, NonZeroU32}; 5 5 + use std::num::{NonZeroU8, NonZeroU16, NonZeroU32}; 6 6 use std::sync::Arc; 7 7 use std::sync::Mutex; 8 8 use std::sync::atomic::{AtomicUsize, Ordering}; ··· 47 47 did: String, 48 48 producer_id: ProducerId, 49 49 kind: MediaKind, 50 50 + source: String, 50 51 }, 51 52 ProducerRemoved { 52 53 did: String, ··· 62 63 pub struct ProducerInfo { 63 64 pub did: String, 64 65 pub kind: MediaKind, 66 66 + pub source: String, 65 67 } 66 68 67 69 fn worker_count() -> usize { ··· 97 99 _ => None, 98 100 }; 99 101 100 100 - WebRtcTransportOptions::new(WebRtcTransportListenInfos::new(ListenInfo { 101 101 - protocol: Protocol::Udp, 102 102 + let listen_info = |protocol| ListenInfo { 103 103 + protocol, 102 104 ip: listen_ip, 103 103 - announced_address, 105 105 + announced_address: announced_address.clone(), 104 106 expose_internal_ip: false, 105 107 port: None, 106 106 - port_range, 108 108 + port_range: port_range.clone(), 107 109 flags: None, 108 110 send_buffer_size: None, 109 111 recv_buffer_size: None, 110 110 - })) 112 112 + }; 113 113 + 114 114 + WebRtcTransportOptions::new( 115 115 + WebRtcTransportListenInfos::new(listen_info(Protocol::Udp)) 116 116 + .insert(listen_info(Protocol::Tcp)), 117 117 + ) 111 118 } 112 119 113 120 pub struct ChannelSfu { ··· 142 149 .collect() 143 150 } 144 151 145 145 - pub fn add_producer(&self, producer_id: ProducerId, did: String, kind: MediaKind) { 152 152 + pub fn add_producer( 153 153 + &self, 154 154 + producer_id: ProducerId, 155 155 + did: String, 156 156 + kind: MediaKind, 157 157 + source: String, 158 158 + ) { 146 159 self.producers.lock().unwrap().insert( 147 160 producer_id, 148 161 ProducerInfo { 149 162 did: did.clone(), 150 163 kind, 164 164 + source: source.clone(), 151 165 }, 152 166 ); 153 167 let _ = self.events.send(RoomEvent::ProducerAdded { 154 168 did, 155 169 producer_id, 156 170 kind, 171 171 + source, 157 172 }); 158 173 } 159 174 ··· 227 242 .create_router(RouterOptions::new(media_codecs())) 228 243 .await 229 244 .map_err(|e| format!("create_router: {e}"))?; 245 245 + let mut audio_observer_options = AudioLevelObserverOptions::default(); 246 246 + audio_observer_options.max_entries = NonZeroU16::new(20).unwrap(); 247 247 + audio_observer_options.interval = 400; 230 248 let audio_observer = router 231 231 - .create_audio_level_observer(AudioLevelObserverOptions::default()) 249 249 + .create_audio_level_observer(audio_observer_options) 232 250 .await 233 251 .map_err(|e| format!("create_audio_level_observer: {e}"))?; 234 252

+15 -2

src/xrpc/social/colibri/community/reads/get_data_handler.rs

Reviewed

··· 212 212 mut colibri_profiles, 213 213 mut actor_data, 214 214 mut states, 215 215 + mut vc, 216 216 + mut vc_muted, 217 217 + mut vc_deafened, 215 218 mut handles, 216 219 mut member_roles, 217 220 } = raw.member_aggregate; ··· 224 227 let colibri_profile = colibri_profiles.remove(&did); 225 228 let data = actor_data.remove(&did); 226 229 let state = states.remove(&did); 230 230 + let member_vc = vc.remove(&did); 231 231 + let member_vc_muted = vc_muted.remove(&did); 232 232 + let member_vc_deafened = vc_deafened.remove(&did); 227 233 let role_rkeys = member_roles.remove(&did).unwrap_or_default(); 228 228 - build_member( 234 234 + let mut member = build_member( 229 235 did, 230 236 handle, 231 237 profile, ··· 234 240 state, 235 241 &community.authority, 236 242 role_rkeys, 237 237 - ) 243 243 + ); 244 244 + member.vc = member_vc; 245 245 + member.vc_muted = member_vc_muted; 246 246 + member.vc_deafened = member_vc_deafened; 247 247 + member 238 248 }) 239 249 .collect(); 240 250 ··· 337 347 colibri_profiles: HashMap::new(), 338 348 actor_data: HashMap::new(), 339 349 states: HashMap::new(), 350 350 + vc: HashMap::new(), 351 351 + vc_muted: HashMap::new(), 352 352 + vc_deafened: HashMap::new(), 340 353 handles: HashMap::from([( 341 354 String::from("did:plc:alice"), 342 355 String::from("alice.test"),

+48 -6

src/xrpc/social/colibri/community/reads/list_members_handler.rs

Reviewed

··· 21 21 pub handle: String, 22 22 pub roles: Vec<String>, 23 23 pub data: ActorData, 24 24 + #[serde(skip_serializing_if = "Option::is_none")] 25 25 + pub vc: Option<String>, 26 26 + #[serde(rename = "vcMuted", skip_serializing_if = "Option::is_none")] 27 27 + pub vc_muted: Option<bool>, 28 28 + #[serde(rename = "vcDeafened", skip_serializing_if = "Option::is_none")] 29 29 + pub vc_deafened: Option<bool>, 24 30 } 25 31 26 32 #[derive(Serialize, Deserialize, Debug)] ··· 37 43 pub colibri_profiles: HashMap<String, ColibriActorProfile>, 38 44 pub actor_data: HashMap<String, ColibriActorData>, 39 45 pub states: HashMap<String, String>, 46 46 + pub vc: HashMap<String, String>, 47 47 + pub vc_muted: HashMap<String, bool>, 48 48 + pub vc_deafened: HashMap<String, bool>, 40 49 pub handles: HashMap<String, String>, 41 50 /// Role rkeys assigned to each member, keyed by member DID. These are 42 51 /// raw record keys — callers must expand them to full AT-URIs. ··· 91 100 colibri_profiles: HashMap::new(), 92 101 actor_data: HashMap::new(), 93 102 states: HashMap::new(), 103 103 + vc: HashMap::new(), 104 104 + vc_muted: HashMap::new(), 105 105 + vc_deafened: HashMap::new(), 94 106 handles: HashMap::new(), 95 107 member_roles: HashMap::new(), 96 108 }); ··· 131 143 .filter(user_states::Column::Did.is_in(member_dids.clone())) 132 144 .all(db) 133 145 .await?; 134 134 - let states: HashMap<String, String> = state_rows 135 135 - .into_iter() 136 136 - .map(|row| (row.did, row.state)) 137 137 - .collect(); 146 146 + let mut states: HashMap<String, String> = HashMap::new(); 147 147 + let mut vc: HashMap<String, String> = HashMap::new(); 148 148 + let mut vc_muted: HashMap<String, bool> = HashMap::new(); 149 149 + let mut vc_deafened: HashMap<String, bool> = HashMap::new(); 150 150 + for row in state_rows { 151 151 + if let Some(channel) = row.vc { 152 152 + vc.insert(row.did.clone(), channel); 153 153 + vc_muted.insert(row.did.clone(), row.vc_muted.unwrap_or(false)); 154 154 + vc_deafened.insert(row.did.clone(), row.vc_deafened.unwrap_or(false)); 155 155 + } 156 156 + states.insert(row.did, row.state); 157 157 + } 138 158 139 159 let repo_rows = repos::Entity::find() 140 160 .filter(repos::Column::Did.is_in(member_dids.clone())) ··· 151 171 colibri_profiles, 152 172 actor_data, 153 173 states, 174 174 + vc, 175 175 + vc_muted, 176 176 + vc_deafened, 154 177 handles, 155 178 member_roles, 156 179 }) ··· 198 221 roles, 199 222 data: actor_data_from_effective(effective, is_bot, &handle, online_state, status), 200 223 handle, 224 224 + vc: None, 225 225 + vc_muted: None, 226 226 + vc_deafened: None, 201 227 } 202 228 } 203 229 ··· 224 250 mut colibri_profiles, 225 251 mut actor_data, 226 252 mut states, 253 253 + mut vc, 254 254 + mut vc_muted, 255 255 + mut vc_deafened, 227 256 mut handles, 228 257 mut member_roles, 229 258 } = aggregate; ··· 236 265 let colibri_profile = colibri_profiles.remove(&did); 237 266 let data = actor_data.remove(&did); 238 267 let state = states.remove(&did); 268 268 + let member_vc = vc.remove(&did); 269 269 + let member_vc_muted = vc_muted.remove(&did); 270 270 + let member_vc_deafened = vc_deafened.remove(&did); 239 271 let role_rkeys = member_roles.remove(&did).unwrap_or_default(); 240 240 - build_member( 272 272 + let mut member = build_member( 241 273 did, 242 274 handle, 243 275 profile, ··· 246 278 state, 247 279 &community.authority, 248 280 role_rkeys, 249 249 - ) 281 281 + ); 282 282 + member.vc = member_vc; 283 283 + member.vc_muted = member_vc_muted; 284 284 + member.vc_deafened = member_vc_deafened; 285 285 + member 250 286 }) 251 287 .collect(); 252 288 ··· 346 382 String::from("did:plc:alice"), 347 383 String::from("online"), 348 384 )]), 385 385 + vc: HashMap::new(), 386 386 + vc_muted: HashMap::new(), 387 387 + vc_deafened: HashMap::new(), 349 388 handles: HashMap::from([ 350 389 (String::from("did:plc:alice"), String::from("alice.test")), 351 390 (String::from("did:plc:bob"), String::from("bob.test")), ··· 394 433 colibri_profiles: HashMap::new(), 395 434 actor_data: HashMap::new(), 396 435 states: HashMap::new(), 436 436 + vc: HashMap::new(), 437 437 + vc_muted: HashMap::new(), 438 438 + vc_deafened: HashMap::new(), 397 439 handles: HashMap::new(), 398 440 member_roles: HashMap::from([( 399 441 String::from("did:plc:alice"),

+17 -1

src/xrpc/social/colibri/sync/send_hum_handler.rs

Reviewed

··· 39 39 use crate::lib::event_scope::{EventScope, ScopedEvent, SharedScopedEvent}; 40 40 use crate::lib::events::{ 41 41 ColibriServerEvent, ColibriServerEventData, HumEnvelope, HumEvent, UserEventStatus, 42 42 - VoicePresenceEventData, 42 42 + VoicePresenceEventData, VoiceStateEventData, 43 43 }; 44 44 use crate::lib::get_atproto_record::get_atproto_record; 45 45 use crate::lib::map_tap_event::build_presence_user_event; ··· 225 225 match event { 226 226 HumEvent::Typing(t) => Some(&t.channel), 227 227 HumEvent::VoicePresence(v) => Some(&v.channel), 228 228 + HumEvent::VoiceState(v) => Some(&v.channel), 228 229 HumEvent::User(_) => None, 229 230 } 230 231 } ··· 281 282 did: subject.to_string(), 282 283 }, 283 284 )), 285 285 + }, 286 286 + ); 287 287 + } 288 288 + HumEvent::VoiceState(voice) => { 289 289 + broadcast_community( 290 290 + broadcast_tx, 291 291 + community_did, 292 292 + ColibriServerEvent { 293 293 + event_type: String::from("voice_state_event"), 294 294 + data: Some(ColibriServerEventData::VoiceState(VoiceStateEventData { 295 295 + channel: voice.channel.clone(), 296 296 + did: subject.to_string(), 297 297 + muted: voice.muted, 298 298 + deafened: voice.deafened, 299 299 + })), 284 300 }, 285 301 ); 286 302 }

+141 -20

src/xrpc/social/colibri/sync/subscribe_events_handler.rs

Reviewed

··· 1 1 use crate::EventNotification; 2 2 use crate::lib::at_uri::AtUri; 3 3 - use crate::lib::event_scope::{EventScope, SharedScopedEvent}; 3 3 + use crate::lib::event_scope::{EventScope, ScopedEvent, SharedScopedEvent}; 4 4 use crate::lib::events::{ 5 5 ColibriServerEventData, CommunityCreationProgressEvent, MuteEvent, NotificationEventData, 6 6 NotificationEventMessage, SeenEvent, TypingEventData, TypingMessageData, ViewData, 7 7 - VoiceChannelData, 7 7 + VoiceChannelData, VoicePresenceEventData, VoiceStateData, VoiceStateEventData, 8 8 }; 9 9 use crate::lib::get_state::get_did_states; 10 10 use crate::lib::hum_client::{self, OutboundHum}; 11 11 use crate::lib::notifications::IndexedNotification; 12 12 - use crate::lib::state::{broadcast_state_change, join_vc, leave_vc, view_channel}; 12 12 + use crate::lib::state::{broadcast_state_change, join_vc, leave_vc, set_vc_state, view_channel}; 13 13 use crate::lib::tap::CommsBridge; 14 14 use crate::xrpc::social::colibri::actor::list_communities_handler::get_authorized_communities; 15 15 use crate::{ ··· 32 32 use rocket_ws::{Message as WsMessage, WebSocket, stream::DuplexStream}; 33 33 use sea_orm::DatabaseConnection; 34 34 use std::collections::HashSet; 35 35 + use std::sync::Arc; 35 36 36 37 type WsSink = futures_util::stream::SplitSink<DuplexStream, WsMessage>; 37 38 ··· 44 45 text: &str, 45 46 did: String, 46 47 to_c2c_broadcast: &Sender<EventNotification>, 48 48 + to_tap_broadcast: &Sender<SharedScopedEvent>, 47 49 hum_outbox: &mpsc::Sender<OutboundHum>, 48 50 db: &DatabaseConnection, 49 51 ) -> Option<String> { ··· 97 99 }, 98 100 ); 99 101 102 102 + if let Some(uri) = AtUri::parse(&vc_msg_data.community) { 103 103 + broadcast_voice_presence( 104 104 + to_tap_broadcast, 105 105 + &uri.authority, 106 106 + &vc_msg_data.channel, 107 107 + &did, 108 108 + "join", 109 109 + ); 110 110 + } 111 111 + 100 112 join_vc(did, vc_msg_data.channel, vc_msg_data.community, db).await; 101 113 102 114 None 103 115 } 104 116 "voice_leave" => { 105 105 - // Resolve the channel being left before we clear it, so the outbound 106 106 - // Hum can carry it (the leave event has no payload of its own). 107 107 - if let Ok(states) = get_did_states(did.clone(), db).await 108 108 - && let Some(channel) = states.vc 109 109 - { 110 110 - hum_client::enqueue( 111 111 - hum_outbox, 112 112 - OutboundHum::Voice { 113 113 - did: did.clone(), 114 114 - channel, 115 115 - event: String::from("leave"), 116 116 - }, 117 117 + leave_vc_and_broadcast(&did, to_tap_broadcast, hum_outbox, db).await; 118 118 + None 119 119 + } 120 120 + "voice_state" => { 121 121 + let state_data: VoiceStateData = serde_json::from_value(user_message.data?).ok()?; 122 122 + 123 123 + hum_client::enqueue( 124 124 + hum_outbox, 125 125 + OutboundHum::VoiceState { 126 126 + did: did.clone(), 127 127 + channel: state_data.channel.clone(), 128 128 + muted: state_data.muted, 129 129 + deafened: state_data.deafened, 130 130 + }, 131 131 + ); 132 132 + 133 133 + if let Some(uri) = AtUri::parse(&state_data.community) { 134 134 + broadcast_voice_state( 135 135 + to_tap_broadcast, 136 136 + &uri.authority, 137 137 + &state_data.channel, 138 138 + &did, 139 139 + state_data.muted, 140 140 + state_data.deafened, 117 141 ); 118 142 } 119 143 120 120 - leave_vc(did, db).await; 144 144 + set_vc_state(did, state_data.muted, state_data.deafened, db).await; 145 145 + 121 146 None 122 147 } 123 148 _ => None, 124 149 } 125 150 } 126 151 152 152 + fn broadcast_voice_state( 153 153 + to_tap_broadcast: &Sender<SharedScopedEvent>, 154 154 + community_did: &str, 155 155 + channel: &str, 156 156 + did: &str, 157 157 + muted: bool, 158 158 + deafened: bool, 159 159 + ) { 160 160 + let server_event = ColibriServerEvent { 161 161 + event_type: String::from("voice_state_event"), 162 162 + data: Some(ColibriServerEventData::VoiceState(VoiceStateEventData { 163 163 + channel: channel.to_string(), 164 164 + did: did.to_string(), 165 165 + muted, 166 166 + deafened, 167 167 + })), 168 168 + }; 169 169 + let scoped = Arc::new(ScopedEvent { 170 170 + scope: EventScope::Community(community_did.to_string()), 171 171 + payload: server_event.serialize(), 172 172 + }); 173 173 + let _ = to_tap_broadcast.send(scoped); 174 174 + } 175 175 + 176 176 + async fn leave_vc_and_broadcast( 177 177 + did: &str, 178 178 + to_tap_broadcast: &Sender<SharedScopedEvent>, 179 179 + hum_outbox: &mpsc::Sender<OutboundHum>, 180 180 + db: &DatabaseConnection, 181 181 + ) { 182 182 + if let Ok(states) = get_did_states(did.to_string(), db).await 183 183 + && let Some(channel) = states.vc 184 184 + { 185 185 + hum_client::enqueue( 186 186 + hum_outbox, 187 187 + OutboundHum::Voice { 188 188 + did: did.to_string(), 189 189 + channel: channel.clone(), 190 190 + event: String::from("leave"), 191 191 + }, 192 192 + ); 193 193 + 194 194 + if let Some(community) = states.vc_community 195 195 + && let Some(uri) = AtUri::parse(&community) 196 196 + { 197 197 + broadcast_voice_presence(to_tap_broadcast, &uri.authority, &channel, did, "leave"); 198 198 + } 199 199 + 200 200 + leave_vc(did.to_string(), db).await; 201 201 + } 202 202 + } 203 203 + 204 204 + fn broadcast_voice_presence( 205 205 + to_tap_broadcast: &Sender<SharedScopedEvent>, 206 206 + community_did: &str, 207 207 + channel: &str, 208 208 + did: &str, 209 209 + event: &str, 210 210 + ) { 211 211 + let server_event = ColibriServerEvent { 212 212 + event_type: String::from("voice_presence_event"), 213 213 + data: Some(ColibriServerEventData::VoicePresence( 214 214 + VoicePresenceEventData { 215 215 + event: event.to_string(), 216 216 + channel: channel.to_string(), 217 217 + did: did.to_string(), 218 218 + }, 219 219 + )), 220 220 + }; 221 221 + let scoped = Arc::new(ScopedEvent { 222 222 + scope: EventScope::Community(community_did.to_string()), 223 223 + payload: server_event.serialize(), 224 224 + }); 225 225 + let _ = to_tap_broadcast.send(scoped); 226 226 + } 227 227 + 127 228 async fn serialize_typing_broadcast( 128 229 msg: EventNotification, 129 230 did: String, ··· 245 346 msg: Option<Result<WsMessage, rocket_ws::result::Error>>, 246 347 did: String, 247 348 to_c2c_broadcast: &Sender<EventNotification>, 349 349 + to_tap_broadcast: &Sender<SharedScopedEvent>, 248 350 hum_outbox: &mpsc::Sender<OutboundHum>, 249 351 db: &DatabaseConnection, 250 352 ) -> bool { 251 353 match msg { 252 354 Some(Ok(WsMessage::Close(_))) | None => false, 253 355 Some(Ok(WsMessage::Text(text))) => { 254 254 - if let Some(serialized_ack_res) = 255 255 - parse_client_event(&text, did, to_c2c_broadcast, hum_outbox, db).await 356 356 + if let Some(serialized_ack_res) = parse_client_event( 357 357 + &text, 358 358 + did, 359 359 + to_c2c_broadcast, 360 360 + to_tap_broadcast, 361 361 + hum_outbox, 362 362 + db, 363 363 + ) 364 364 + .await 256 365 { 257 366 let _ = ws_sink.send(WsMessage::Text(serialized_ack_res)).await; 258 367 } ··· 515 624 loop { 516 625 let connected = tokio::select! { 517 626 msg = from_tap.recv() => handle_tap_message(&mut ws_sink, msg, &did, &mut communities, &db).await, 518 518 - msg = ws_source.next() => handle_client_message(&mut ws_sink, msg, did.clone(), &to_c2c_broadcast, &hum_outbox, &db).await, 627 627 + msg = ws_source.next() => handle_client_message(&mut ws_sink, msg, did.clone(), &to_c2c_broadcast, &to_tap_broadcast, &hum_outbox, &db).await, 519 628 msg = from_c2c_broadcast.recv() => handle_client_broadcast_msg(&mut ws_sink, msg, did.clone(), &db).await, 520 629 msg = from_notifications.recv() => handle_notification_msg(&mut ws_sink, msg, &did).await, 521 630 msg = from_applications.recv() => handle_application_msg(&mut ws_sink, msg, &communities).await, ··· 529 638 } 530 639 } 531 640 641 641 + leave_vc_and_broadcast(&did, &to_tap_broadcast, &hum_outbox, &db).await; 532 642 save_state(&db, did.clone(), String::from("offline")).await; 533 643 broadcast_state_change(&to_tap_broadcast, &did, &db, &hum_outbox).await; 534 644 } ··· 677 787 scope_matches, serialize_typing_broadcast, 678 788 }; 679 789 use crate::EventNotification; 680 680 - use crate::lib::event_scope::EventScope; 790 790 + use crate::lib::event_scope::{EventScope, SharedScopedEvent}; 681 791 use crate::lib::events::{ApplicationEventData, ColibriServerEvent, ColibriServerEventData}; 682 792 use crate::lib::hum_client::OutboundHum; 683 793 use crate::lib::test_fixtures::mock_db; ··· 694 804 mpsc::channel::<OutboundHum>(4).0 695 805 } 696 806 807 807 + fn tap_broadcast() -> broadcast::Sender<SharedScopedEvent> { 808 808 + broadcast::channel::<SharedScopedEvent>(4).0 809 809 + } 810 810 + 697 811 #[test] 698 812 fn scope_matches_routes_each_scope() { 699 813 let did = "did:plc:me"; ··· 778 892 r#"{"type":"heartbeat","data":null}"#, 779 893 String::from("did:plc:me"), 780 894 &tx, 895 895 + &tap_broadcast(), 781 896 &hum_outbox(), 782 897 &db, 783 898 ) ··· 795 910 r#"{"type":"typing","data":{"channel":"community-1"}}"#, 796 911 String::from("did:plc:me"), 797 912 &tx, 913 913 + &tap_broadcast(), 798 914 &hum_outbox(), 799 915 &db, 800 916 ) ··· 814 930 r#"{"type":"typing","data":null}"#, 815 931 String::from("did:plc:me"), 816 932 &tx, 933 933 + &tap_broadcast(), 817 934 &hum_outbox(), 818 935 &db, 819 936 ) ··· 843 960 r#"{"type":"view","data":{"channel":"community-1"}}"#, 844 961 String::from("did:plc:me"), 845 962 &tx, 963 963 + &tap_broadcast(), 846 964 &hum_outbox(), 847 965 &db, 848 966 ) ··· 865 983 r#"{"type":"unknown","data":null}"#, 866 984 String::from("did:plc:me"), 867 985 &tx, 986 986 + &tap_broadcast(), 868 987 &hum_outbox(), 869 988 &db, 870 989 ) ··· 881 1000 vc: None, 882 1001 vc_community: None, 883 1002 channel: Some(String::from("community-1")), 1003 1003 + vc_muted: None, 1004 1004 + vc_deafened: None, 884 1005 }]]) 885 1006 .into_connection(); 886 1007 let payload = serialize_typing_broadcast(

+6

src/xrpc/social/colibri/voice/messages.rs

Reviewed

··· 27 27 router_rtp_capabilities: RtpCapabilitiesFinalized, 28 28 producer_transport_options: TransportOptions, 29 29 consumer_transport_options: TransportOptions, 30 30 + ice_servers: serde_json::Value, 30 31 }, 31 32 ConnectedProducerTransport, 32 33 #[serde(rename_all = "camelCase")] ··· 46 47 did: String, 47 48 producer_id: ProducerId, 48 49 kind: MediaKind, 50 50 + source: String, 49 51 }, 50 52 #[serde(rename_all = "camelCase")] 51 53 ProducerRemoved { ··· 73 75 Produce { 74 76 kind: MediaKind, 75 77 rtp_parameters: RtpParameters, 78 78 + #[serde(default)] 79 79 + source: String, 76 80 }, 77 81 #[serde(rename_all = "camelCase")] 78 82 ConnectConsumerTransport { dtls_parameters: DtlsParameters }, ··· 80 84 Consume { producer_id: ProducerId }, 81 85 #[serde(rename_all = "camelCase")] 82 86 ConsumerResume { id: ConsumerId }, 87 87 + #[serde(rename_all = "camelCase")] 88 88 + CloseProducer { producer_id: ProducerId }, 83 89 }

+80 -2

src/xrpc/social/colibri/voice/signal_handler.rs

Reviewed

··· 6 6 use rocket::tokio::sync::broadcast::error::RecvError; 7 7 use rocket::{State, get, serde::json::Json, tokio}; 8 8 use rocket_ws::{Message as WsMessage, WebSocket, stream::DuplexStream}; 9 9 + use sea_orm::DatabaseConnection; 9 10 11 11 + use crate::lib::at_uri::AtUri; 12 12 + use crate::lib::colibri::ColibriChannel; 10 13 use crate::lib::responses::{ErrorBody, ErrorResponse}; 11 11 - use crate::lib::service_auth; 14 14 + use crate::lib::{channel_authz, community_authz, community_write, service_auth}; 12 15 use crate::sfu::{ChannelSfu, RoomEvent, Sfu}; 13 16 use crate::xrpc::social::colibri::sync::subscribe_events_handler::{ 14 17 AUTH_SUBPROTOCOL, ChannelWithProtocol, SubprotocolAuth, ··· 20 23 21 24 const LXM: &str = "social.colibri.voice.signal"; 22 25 26 26 + fn ice_servers() -> serde_json::Value { 27 27 + std::env::var("SFU_ICE_SERVERS") 28 28 + .ok() 29 29 + .filter(|v| !v.trim().is_empty()) 30 30 + .and_then(|v| serde_json::from_str(&v).ok()) 31 31 + .unwrap_or_else(|| serde_json::Value::Array(Vec::new())) 32 32 + } 33 33 + 34 34 + async fn authorize_channel_access( 35 35 + db: &DatabaseConnection, 36 36 + channel_uri: &str, 37 37 + did: &str, 38 38 + ) -> Result<(), ErrorResponse> { 39 39 + let Some(parsed) = AtUri::parse(channel_uri) else { 40 40 + return Ok(()); 41 41 + }; 42 42 + 43 43 + let chan_json = match community_write::read_cached( 44 44 + db, 45 45 + &parsed.authority, 46 46 + "social.colibri.channel", 47 47 + &parsed.rkey, 48 48 + ) 49 49 + .await 50 50 + { 51 51 + Ok(Some(json)) => json, 52 52 + Ok(None) => return Ok(()), 53 53 + Err(e) => { 54 54 + log::warn!("voice authz channel read failed for {channel_uri}: {e}; allowing"); 55 55 + return Ok(()); 56 56 + } 57 57 + }; 58 58 + 59 59 + let Ok(channel) = serde_json::from_value::<ColibriChannel>(chan_json) else { 60 60 + return Ok(()); 61 61 + }; 62 62 + 63 63 + let restricted = channel.owner_only == Some(true) 64 64 + || !channel.allowed_roles.is_empty() 65 65 + || !channel.allowed_members.is_empty(); 66 66 + if !restricted || did == parsed.authority { 67 67 + return Ok(()); 68 68 + } 69 69 + 70 70 + let community_uri = format!("at://{}/social.colibri.community/self", parsed.authority); 71 71 + match community_authz::load_actor_authz(db, &community_uri, did).await { 72 72 + Ok(authz) if !channel_authz::can_post(&channel, &authz, did) => Err(ErrorResponse { 73 73 + body: Json(ErrorBody { 74 74 + error: String::from("Forbidden"), 75 75 + message: String::from("You are not permitted to join this voice channel."), 76 76 + }), 77 77 + }), 78 78 + Ok(_) => Ok(()), 79 79 + Err(e) => { 80 80 + log::warn!("voice authz lookup failed for {did}: {e}; allowing"); 81 81 + Ok(()) 82 82 + } 83 83 + } 84 84 + } 85 85 + 23 86 async fn send(ws_sink: &mut WsSink, msg: &ServerMessage) -> bool { 24 87 let Ok(text) = serde_json::to_string(msg) else { 25 88 return true; ··· 79 142 ClientMessage::Produce { 80 143 kind, 81 144 rtp_parameters, 145 145 + source, 82 146 } => match producer_transport 83 147 .produce(ProducerOptions::new(kind, rtp_parameters)) 84 148 .await ··· 88 152 if kind == MediaKind::Audio { 89 153 channel.observe_audio(id).await; 90 154 } 91 91 - channel.add_producer(id, did.to_string(), kind); 155 155 + channel.add_producer(id, did.to_string(), kind, source); 92 156 my_producer_ids.push(id); 93 157 producers.push(producer); 94 158 send(ws_sink, &ServerMessage::Produced { id }).await; ··· 172 236 log::warn!("failed to resume consumer {id}: {e}"); 173 237 } 174 238 } 239 239 + ClientMessage::CloseProducer { producer_id } => { 240 240 + if let Some(index) = producers.iter().position(|p| p.id() == producer_id) { 241 241 + let _ = producers.remove(index); 242 242 + my_producer_ids.retain(|id| *id != producer_id); 243 243 + channel.remove_producer(&producer_id); 244 244 + } 245 245 + } 175 246 } 176 247 } 177 248 ··· 185 256 did, 186 257 producer_id, 187 258 kind, 259 259 + source, 188 260 } => { 189 261 if my_producer_ids.contains(&producer_id) { 190 262 return true; ··· 193 265 did, 194 266 producer_id, 195 267 kind, 268 268 + source, 196 269 } 197 270 } 198 271 RoomEvent::ProducerRemoved { did, producer_id } => { ··· 224 297 router_rtp_capabilities: channel.rtp_capabilities(), 225 298 producer_transport_options: transport_options(&producer_transport), 226 299 consumer_transport_options: transport_options(&consumer_transport), 300 300 + ice_servers: ice_servers(), 227 301 }; 228 302 if !send(&mut ws_sink, &init).await { 229 303 return; ··· 236 310 did: info.did, 237 311 producer_id, 238 312 kind: info.kind, 313 313 + source: info.source, 239 314 }, 240 315 ) 241 316 .await; ··· 301 376 subprotocol_auth: SubprotocolAuth, 302 377 ws: WebSocket, 303 378 sfu: &State<Arc<Sfu>>, 379 379 + db: &State<DatabaseConnection>, 304 380 ) -> Result<ChannelWithProtocol, ErrorResponse> { 305 381 let used_subprotocol = subprotocol_auth.token().is_some(); 306 382 let token = subprotocol_auth ··· 317 393 message: e.to_string(), 318 394 }), 319 395 })?; 396 396 + 397 397 + authorize_channel_access(db.inner(), &channel, &did).await?; 320 398 321 399 let channel_sfu = sfu 322 400 .get_or_create_channel(&channel)