livekit/room/participant/
mod.rs

1// Copyright 2025 LiveKit, Inc.
2//
3// Licensed under the Apache License, Version 2.0 (the "License");
4// you may not use this file except in compliance with the License.
5// You may obtain a copy of the License at
6//
7//     http://www.apache.org/licenses/LICENSE-2.0
8//
9// Unless required by applicable law or agreed to in writing, software
10// distributed under the License is distributed on an "AS IS" BASIS,
11// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12// See the License for the specific language governing permissions and
13// limitations under the License.
14
15use std::{collections::HashMap, fmt::Debug, sync::Arc};
16
17use libwebrtc::enum_dispatch;
18use livekit_protocol as proto;
19use parking_lot::{Mutex, RwLock};
20
21use crate::{prelude::*, rtc_engine::RtcEngine};
22
23mod local_participant;
24mod remote_participant;
25mod rpc;
26
27pub use local_participant::*;
28pub use remote_participant::*;
29pub use rpc::*;
30
31#[derive(Debug, Clone, Copy, Eq, PartialEq)]
32pub enum ConnectionQuality {
33    Excellent,
34    Good,
35    Poor,
36    Lost,
37}
38
39#[derive(Debug, Clone, Copy, Eq, PartialEq)]
40pub enum ParticipantState {
41    Joining,
42    Joined,
43    Active,
44    Disconnected,
45}
46
47#[derive(Debug, Clone, Copy, Eq, PartialEq)]
48pub enum ParticipantKind {
49    Standard,
50    Ingress,
51    Egress,
52    Sip,
53    Agent,
54    Connector,
55    Bridge,
56}
57
58#[derive(Debug, Clone, Copy, Eq, PartialEq)]
59pub enum ParticipantKindDetail {
60    CloudAgent,
61    Forwarded,
62    ConnectorWhatsapp,
63    ConnectorTwilio,
64    BridgeRtsp,
65}
66
67#[derive(Debug, Clone, Copy, Eq, PartialEq)]
68pub enum DisconnectReason {
69    UnknownReason,
70    ClientInitiated,
71    DuplicateIdentity,
72    ServerShutdown,
73    ParticipantRemoved,
74    RoomDeleted,
75    StateMismatch,
76    JoinFailure,
77    Migration,
78    SignalClose,
79    RoomClosed,
80    UserUnavailable,
81    UserRejected,
82    SipTrunkFailure,
83    ConnectionTimeout,
84    MediaFailure,
85    AgentError,
86}
87
88#[derive(Debug, Clone)]
89pub enum Participant {
90    Local(LocalParticipant),
91    Remote(RemoteParticipant),
92}
93
94impl Participant {
95    enum_dispatch!(
96        [Local, Remote];
97        pub fn sid(self: &Self) -> ParticipantSid;
98        pub fn identity(self: &Self) -> ParticipantIdentity;
99        pub fn name(self: &Self) -> String;
100        pub fn state(self: &Self) -> ParticipantState;
101        pub fn metadata(self: &Self) -> String;
102        pub fn attributes(self: &Self) -> HashMap<String, String>;
103        pub fn is_speaking(self: &Self) -> bool;
104        pub fn audio_level(self: &Self) -> f32;
105        pub fn connection_quality(self: &Self) -> ConnectionQuality;
106        pub fn kind(self: &Self) -> ParticipantKind;
107        pub fn kind_details(self: &Self) -> Vec<ParticipantKindDetail>;
108        pub fn disconnect_reason(self: &Self) -> DisconnectReason;
109        pub fn joined_at(self: &Self) -> i64;
110        pub fn is_encrypted(self: &Self) -> bool;
111        pub fn permission(self: &Self) -> Option<proto::ParticipantPermission>;
112        pub fn client_protocol(self: &Self) -> i32;
113
114        pub(crate) fn update_info(self: &Self, info: proto::ParticipantInfo) -> ();
115
116        // Internal functions called by the Room when receiving the associated signal messages
117        pub(crate) fn set_speaking(self: &Self, speaking: bool) -> ();
118        pub(crate) fn set_audio_level(self: &Self, level: f32) -> ();
119        pub(crate) fn set_connection_quality(self: &Self, quality: ConnectionQuality) -> ();
120        pub(crate) fn add_publication(self: &Self, publication: TrackPublication) -> ();
121        pub(crate) fn remove_publication(self: &Self, sid: &TrackSid) -> Option<TrackPublication>;
122        pub(crate) fn update_data_encryption_status(self: &Self, is_encrypted: bool) -> ();
123    );
124
125    pub fn track_publications(&self) -> HashMap<TrackSid, TrackPublication> {
126        match self {
127            Participant::Local(p) => p.internal_track_publications(),
128            Participant::Remote(p) => p.internal_track_publications(),
129        }
130    }
131}
132
133struct ParticipantInfo {
134    pub sid: ParticipantSid,
135    pub identity: ParticipantIdentity,
136    pub name: String,
137    pub state: ParticipantState,
138    pub metadata: String,
139    pub attributes: HashMap<String, String>,
140    pub speaking: bool,
141    pub audio_level: f32,
142    pub connection_quality: ConnectionQuality,
143    pub kind: ParticipantKind,
144    pub kind_details: Vec<ParticipantKindDetail>,
145    pub disconnect_reason: DisconnectReason,
146    pub joined_at: i64,
147    pub permission: Option<proto::ParticipantPermission>,
148    pub client_protocol: i32,
149}
150
151type TrackMutedHandler = Box<dyn Fn(Participant, TrackPublication) + Send>;
152type TrackUnmutedHandler = Box<dyn Fn(Participant, TrackPublication) + Send>;
153type MetadataChangedHandler = Box<dyn Fn(Participant, String, String) + Send>;
154type AttributesChangedHandler = Box<dyn Fn(Participant, HashMap<String, String>) + Send>;
155type NameChangedHandler = Box<dyn Fn(Participant, String, String) + Send>;
156type EncryptionStatusChangedHandler = Box<dyn Fn(Participant, bool) + Send>;
157type PermissionChangedHandler =
158    Box<dyn Fn(Participant, Option<proto::ParticipantPermission>) + Send>;
159
160#[derive(Default)]
161struct ParticipantEvents {
162    track_muted: Mutex<Option<TrackMutedHandler>>,
163    track_unmuted: Mutex<Option<TrackUnmutedHandler>>,
164    metadata_changed: Mutex<Option<MetadataChangedHandler>>,
165    attributes_changed: Mutex<Option<AttributesChangedHandler>>,
166    name_changed: Mutex<Option<NameChangedHandler>>,
167    encryption_status_changed: Mutex<Option<EncryptionStatusChangedHandler>>,
168    permission_changed: Mutex<Option<PermissionChangedHandler>>,
169}
170
171pub(super) struct ParticipantInner {
172    rtc_engine: Arc<RtcEngine>,
173    info: RwLock<ParticipantInfo>,
174    track_publications: RwLock<HashMap<TrackSid, TrackPublication>>,
175    events: Arc<ParticipantEvents>,
176    is_encrypted: RwLock<bool>,
177    is_data_encrypted: RwLock<Option<bool>>,
178}
179
180#[derive(Clone)]
181pub struct ParticipantTrackPermission {
182    pub participant_identity: ParticipantIdentity,
183    pub allow_all: bool,
184    pub allowed_track_sids: Vec<TrackSid>,
185}
186
187pub(super) fn new_inner(
188    rtc_engine: Arc<RtcEngine>,
189    sid: ParticipantSid,
190    identity: ParticipantIdentity,
191    name: String,
192    state: ParticipantState,
193    metadata: String,
194    attributes: HashMap<String, String>,
195    kind: ParticipantKind,
196    kind_details: Vec<ParticipantKindDetail>,
197    joined_at: i64,
198    permission: Option<proto::ParticipantPermission>,
199    client_protocol: i32,
200) -> Arc<ParticipantInner> {
201    Arc::new(ParticipantInner {
202        rtc_engine,
203        info: RwLock::new(ParticipantInfo {
204            sid,
205            identity,
206            name,
207            state,
208            metadata,
209            attributes,
210            kind,
211            kind_details,
212            speaking: false,
213            audio_level: 0.0,
214            connection_quality: ConnectionQuality::Excellent,
215            disconnect_reason: DisconnectReason::UnknownReason,
216            joined_at,
217            permission,
218            client_protocol,
219        }),
220        track_publications: Default::default(),
221        events: Default::default(),
222        is_encrypted: RwLock::new(false),
223        is_data_encrypted: RwLock::new(None),
224    })
225}
226
227pub(super) fn update_info(
228    inner: &Arc<ParticipantInner>,
229    participant: &Participant,
230    new_info: proto::ParticipantInfo,
231) {
232    let mut info = inner.info.write();
233    info.state = new_info.state().into();
234    info.disconnect_reason = new_info.disconnect_reason().into();
235    info.kind = new_info.kind().into();
236    info.kind_details = crate::utils::convert_kind_details(&new_info.kind_details);
237    info.sid = new_info.sid.try_into().unwrap();
238    info.identity = new_info.identity.into();
239    info.joined_at = new_info.joined_at_ms;
240
241    let old_name = std::mem::replace(&mut info.name, new_info.name.clone());
242    if old_name != new_info.name {
243        if let Some(cb) = inner.events.name_changed.lock().as_ref() {
244            cb(participant.clone(), old_name, new_info.name);
245        }
246    }
247
248    let old_metadata = std::mem::replace(&mut info.metadata, new_info.metadata.clone());
249    if old_metadata != new_info.metadata {
250        if let Some(cb) = inner.events.metadata_changed.lock().as_ref() {
251            cb(participant.clone(), old_metadata, new_info.metadata);
252        }
253    }
254
255    let old_attributes = std::mem::replace(&mut info.attributes, new_info.attributes.clone());
256    let changed_attributes =
257        crate::utils::calculate_changed_attributes(old_attributes, new_info.attributes.clone());
258    if changed_attributes.len() != 0 {
259        if let Some(cb) = inner.events.attributes_changed.lock().as_ref() {
260            cb(participant.clone(), changed_attributes);
261        }
262    }
263
264    let old_permission = std::mem::replace(&mut info.permission, new_info.permission.clone());
265    if old_permission != new_info.permission {
266        if let Some(cb) = inner.events.permission_changed.lock().as_ref() {
267            cb(participant.clone(), new_info.permission.clone());
268        }
269    }
270
271    info.client_protocol = new_info.client_protocol;
272}
273
274pub(super) fn set_speaking(
275    inner: &Arc<ParticipantInner>,
276    _participant: &Participant,
277    speaking: bool,
278) {
279    inner.info.write().speaking = speaking;
280}
281
282pub(super) fn set_audio_level(
283    inner: &Arc<ParticipantInner>,
284    _participant: &Participant,
285    audio_level: f32,
286) {
287    inner.info.write().audio_level = audio_level;
288}
289
290pub(super) fn set_connection_quality(
291    inner: &Arc<ParticipantInner>,
292    _participant: &Participant,
293    quality: ConnectionQuality,
294) {
295    inner.info.write().connection_quality = quality;
296}
297
298pub(super) fn on_track_muted(
299    inner: &Arc<ParticipantInner>,
300    handler: impl Fn(Participant, TrackPublication) + Send + 'static,
301) {
302    *inner.events.track_muted.lock() = Some(Box::new(handler));
303}
304
305pub(super) fn on_track_unmuted(
306    inner: &Arc<ParticipantInner>,
307    handler: impl Fn(Participant, TrackPublication) + Send + 'static,
308) {
309    *inner.events.track_unmuted.lock() = Some(Box::new(handler));
310}
311
312pub(super) fn on_metadata_changed(
313    inner: &Arc<ParticipantInner>,
314    handler: impl Fn(Participant, String, String) + Send + 'static,
315) {
316    *inner.events.metadata_changed.lock() = Some(Box::new(handler));
317}
318
319pub(super) fn on_name_changed(
320    inner: &Arc<ParticipantInner>,
321    handler: impl Fn(Participant, String, String) + Send + 'static,
322) {
323    *inner.events.name_changed.lock() = Some(Box::new(handler));
324}
325
326pub(super) fn on_attributes_changed(
327    inner: &Arc<ParticipantInner>,
328    handler: impl Fn(Participant, HashMap<String, String>) + Send + 'static,
329) {
330    *inner.events.attributes_changed.lock() = Some(Box::new(handler));
331}
332
333pub(super) fn on_encryption_status_changed(
334    inner: &Arc<ParticipantInner>,
335    handler: impl Fn(Participant, bool) + Send + 'static,
336) {
337    *inner.events.encryption_status_changed.lock() = Some(Box::new(handler));
338}
339
340pub(super) fn on_permission_changed(
341    inner: &Arc<ParticipantInner>,
342    handler: impl Fn(Participant, Option<proto::ParticipantPermission>) + Send + 'static,
343) {
344    *inner.events.permission_changed.lock() = Some(Box::new(handler));
345}
346
347pub(super) fn update_encryption_status(inner: &Arc<ParticipantInner>, participant: &Participant) {
348    use crate::e2ee::EncryptionType;
349
350    let track_publications = inner.track_publications.read();
351    let data_encryption_status = inner.is_data_encrypted.read();
352
353    // Check if all track publications are encrypted
354    let tracks_encrypted = !track_publications.is_empty()
355        && track_publications.values().all(|pub_| pub_.encryption_type() != EncryptionType::None);
356
357    // Overall encryption status: both tracks and data must be encrypted (if data exists)
358    let is_encrypted = match *data_encryption_status {
359        Some(data_encrypted) => tracks_encrypted && data_encrypted,
360        None => tracks_encrypted, // No data messages yet, only consider tracks
361    };
362
363    let mut current_status = inner.is_encrypted.write();
364    if *current_status != is_encrypted {
365        *current_status = is_encrypted;
366        drop(current_status);
367        drop(track_publications);
368        drop(data_encryption_status);
369
370        if let Some(cb) = inner.events.encryption_status_changed.lock().as_ref() {
371            cb(participant.clone(), is_encrypted);
372        }
373    }
374}
375
376pub(super) fn update_data_encryption_status(
377    inner: &Arc<ParticipantInner>,
378    participant: &Participant,
379    is_encrypted: bool,
380) {
381    let mut data_encryption_status = inner.is_data_encrypted.write();
382    let previous_status = *data_encryption_status;
383
384    match previous_status {
385        Some(current) if current == is_encrypted => {
386            // No change needed
387            return;
388        }
389        Some(true) if !is_encrypted => {
390            // Data was encrypted, now unencrypted - update immediately
391            *data_encryption_status = Some(false);
392        }
393        Some(false) if is_encrypted => {
394            // Data was unencrypted, now encrypted - but we need to keep it false
395            // because once we've seen unencrypted data, participant is not fully encrypted
396            return;
397        }
398        None => {
399            // First data message - set the status
400            *data_encryption_status = Some(is_encrypted);
401        }
402        _ => return,
403    }
404
405    drop(data_encryption_status);
406
407    // Update overall encryption status
408    update_encryption_status(inner, participant);
409}
410
411pub(super) fn remove_publication(
412    inner: &Arc<ParticipantInner>,
413    participant: &Participant,
414    sid: &TrackSid,
415) -> Option<TrackPublication> {
416    let mut tracks = inner.track_publications.write();
417    let publication = tracks.remove(sid);
418    if let Some(publication) = publication.clone() {
419        // remove events
420        publication.on_muted(|_| {});
421        publication.on_unmuted(|_| {});
422    } else {
423        // shouldn't happen (internal)
424        log::warn!("could not find publication to remove: {:?}", sid);
425    }
426    drop(tracks);
427
428    // Update encryption status after removing publication
429    update_encryption_status(inner, participant);
430
431    publication
432}
433
434pub(super) fn add_publication(
435    inner: &Arc<ParticipantInner>,
436    participant: &Participant,
437    publication: TrackPublication,
438) {
439    let mut tracks = inner.track_publications.write();
440    tracks.insert(publication.sid(), publication.clone());
441
442    publication.on_muted({
443        let events = inner.events.clone();
444        let participant = participant.clone();
445        let rtc_engine = inner.rtc_engine.clone();
446        move |publication| {
447            if let Some(cb) = events.track_muted.lock().as_ref() {
448                if !publication.is_remote() {
449                    let rtc_engine = rtc_engine.clone();
450                    let publication_cloned = publication.clone();
451                    livekit_runtime::spawn(async move {
452                        let engine_request = rtc_engine
453                            .mute_track(proto::MuteTrackRequest {
454                                sid: publication_cloned.sid().to_string(),
455                                muted: true,
456                            })
457                            .await;
458                        if let Err(e) = engine_request {
459                            log::error!("could not mute track: {e:?}");
460                        }
461                    });
462                }
463                cb(participant.clone(), publication);
464            }
465        }
466    });
467
468    publication.on_unmuted({
469        let events = inner.events.clone();
470        let participant = participant.clone();
471        let rtc_engine = inner.rtc_engine.clone();
472        move |publication| {
473            if let Some(cb) = events.track_unmuted.lock().as_ref() {
474                if !publication.is_remote() {
475                    let rtc_engine = rtc_engine.clone();
476                    let publication_cloned = publication.clone();
477                    livekit_runtime::spawn(async move {
478                        let engine_request = rtc_engine
479                            .mute_track(proto::MuteTrackRequest {
480                                sid: publication_cloned.sid().to_string(),
481                                muted: false,
482                            })
483                            .await;
484                        if let Err(e) = engine_request {
485                            log::error!("could not unmute track: {e:?}");
486                        }
487                    });
488                }
489                cb(participant.clone(), publication);
490            }
491        }
492    });
493    drop(tracks);
494
495    // Update encryption status after adding publication
496    update_encryption_status(inner, participant);
497}
livekit/room/participant/mod.rs

livekit/room/participant/
mod.rs
