1pub mod constants {
2 pub const VOICE_GATEWAY_VERSION: u8 = 8;
3 pub const DEFAULT_SAMPLE_RATE: u32 = 48_000;
4 pub const DAVE_INITIAL_VERSION: u16 = 1;
5 pub const DEFAULT_VOICE_MODE: &str = "xsalsa20_poly1305";
6 pub const MAX_RECONNECT_ATTEMPTS: u32 = 15;
7 pub const BACKOFF_BASE_MS: u64 = 1_000;
8 pub const RECONNECT_DELAY_FRESH_MS: u64 = 3000;
9 pub const UDP_KEEPALIVE_GAP_MS: u64 = 5000;
10 pub const WRITE_TASK_SHUTDOWN_MS: u64 = 500;
11 pub const RTP_VERSION_BYTE: u8 = 0x80;
12 pub const RTP_OPUS_PAYLOAD_TYPE: u8 = 0x78;
13 pub const RTP_TIMESTAMP_STEP: u32 = 960;
14 pub const FRAME_DURATION_MS: u64 = 20;
15 pub const PCM_FRAME_SAMPLES: usize = 960;
16 pub const MAX_OPUS_FRAME_SIZE: usize = 4000;
17 pub const SILENCE_FRAME: [u8; 3] = [0xf8, 0xff, 0xfe];
18 pub const MAX_SILENCE_FRAMES: u32 = 25;
19 pub const UDP_PACKET_BUF_CAPACITY: usize = 1500;
20 pub const DISCOVERY_PACKET_SIZE: usize = 74;
21 pub const IP_DISCOVERY_TIMEOUT_SECS: u64 = 2;
22 pub const IP_DISCOVERY_RETRIES: u32 = 10;
23 pub const IP_DISCOVERY_RETRY_INTERVAL_MS: u64 = 1000;
24 pub const OP_HEARTBEAT: u8 = 3;
25 pub const MAX_PENDING_PROPOSALS: usize = 64;
26}
27pub mod protocol {
28 use serde::{Deserialize, Serialize};
29 use serde_json::Value;
30 #[derive(Serialize, Deserialize, Debug, Clone)]
31 pub struct GatewayPayload {
32 pub op: u8,
33 #[serde(skip_serializing_if = "Option::is_none")]
34 pub seq: Option<u32>,
35 pub d: Value,
36 }
37 #[derive(Debug, Clone, Copy, PartialEq, Eq)]
38 #[repr(u8)]
39 pub enum OpCode {
40 Identify = 0,
41 SelectProtocol = 1,
42 Ready = 2,
43 Heartbeat = 3,
44 SessionDescription = 4,
45 Speaking = 5,
46 HeartbeatAck = 6,
47 Resume = 7,
48 Hello = 8,
49 Resumed = 9,
50 ClientConnect = 11,
51 Video = 12,
52 ClientDisconnect = 13,
53 Codecs = 14,
54 MediaSinkWants = 15,
55 VoiceBackendVersion = 16,
56 UserFlags = 18,
57 VoicePlatform = 20,
58 Unknown = 255,
59 }
60 impl From<u8> for OpCode {
61 fn from(op: u8) -> Self {
62 match op {
63 0 => Self::Identify,
64 1 => Self::SelectProtocol,
65 2 => Self::Ready,
66 3 => Self::Heartbeat,
67 4 => Self::SessionDescription,
68 5 => Self::Speaking,
69 6 => Self::HeartbeatAck,
70 7 => Self::Resume,
71 8 => Self::Hello,
72 9 => Self::Resumed,
73 11 => Self::ClientConnect,
74 12 => Self::Video,
75 13 => Self::ClientDisconnect,
76 14 => Self::Codecs,
77 15 => Self::MediaSinkWants,
78 16 => Self::VoiceBackendVersion,
79 18 => Self::UserFlags,
80 20 => Self::VoicePlatform,
81 _ => Self::Unknown,
82 }
83 }
84 }
85 pub mod builders {
86 use super::*;
87 use serde_json::json;
88 pub fn identify(
89 guild_id: String,
90 user_id: String,
91 session_id: String,
92 token: String,
93 ) -> GatewayPayload {
94 GatewayPayload {
95 op: OpCode::Identify as u8,
96 seq: None,
97 d: json!({
98 "server_id": guild_id,
99 "user_id": user_id,
100 "session_id": session_id,
101 "token": token,
102 "video": true,
103 }),
104 }
105 }
106 pub fn resume(
107 guild_id: String,
108 session_id: String,
109 token: String,
110 seq_ack: i64,
111 ) -> GatewayPayload {
112 let seq_ack = seq_ack.max(0);
113 GatewayPayload {
114 op: OpCode::Resume as u8,
115 seq: None,
116 d: json!({
117 "server_id": guild_id,
118 "session_id": session_id,
119 "token": token,
120 "video": true,
121 "seq_ack": seq_ack,
122 }),
123 }
124 }
125 }
126}
127pub mod session {
128 pub mod types {
129 use serde::{Deserialize, Serialize};
130 use thiserror::Error;
131 #[derive(Error, Debug)]
132 pub enum GatewayError {
133 #[error("WebSocket error: {0}")]
134 WebSocket(#[from] tokio_tungstenite::tungstenite::Error),
135 #[error("IO error: {0}")]
136 Io(#[from] std::io::Error),
137 #[error("JSON error: {0}")]
138 Json(#[from] serde_json::Error),
139 #[error("Discovery failed: {0}")]
140 Discovery(String),
141 #[error("Encoding error: {0}")]
142 Encoding(String),
143 #[error("Encryption error: {0}")]
144 Encryption(String),
145 #[error("Internal error: {0}")]
146 Internal(String),
147 #[error("Other error: {0}")]
148 Other(#[from] Box<dyn std::error::Error + Send + Sync>),
149 }
150 pub fn map_boxed_err<E: std::fmt::Display>(e: E) -> crate::common::types::AnyError {
151 Box::new(std::io::Error::other(e.to_string()))
152 }
153 #[derive(Serialize, Deserialize, Debug, Clone, Copy, PartialEq, Eq)]
154 pub enum SessionOutcome {
155 Reconnect,
156 Identify,
157 Shutdown,
158 }
159 #[derive(Default, Debug)]
160 pub struct PersistentSessionState {
161 pub ssrc: u32,
162 pub udp_addr: Option<std::net::SocketAddr>,
163 pub session_key: Option<[u8; 32]>,
164 pub rtp_state: Option<crate::gateway::udp_link::RtpState>,
165 pub selected_mode: Option<String>,
166 }
167 }
168 pub mod protocol {
169 use serde::{Deserialize, Serialize};
170 use serde_json::Value;
171 #[derive(Serialize, Deserialize, Debug, Clone)]
172 pub struct GatewayPayload {
173 pub op: u8,
174 #[serde(skip_serializing_if = "Option::is_none")]
175 pub seq: Option<u32>,
176 pub d: Value,
177 }
178 #[derive(Debug, Clone, Copy, PartialEq, Eq)]
179 #[repr(u8)]
180 pub enum OpCode {
181 Identify = 0,
182 SelectProtocol = 1,
183 Ready = 2,
184 Heartbeat = 3,
185 SessionDescription = 4,
186 Speaking = 5,
187 HeartbeatAck = 6,
188 Resume = 7,
189 Hello = 8,
190 Resumed = 9,
191 ClientConnect = 11,
192 Video = 12,
193 ClientDisconnect = 13,
194 Codecs = 14,
195 MediaSinkWants = 15,
196 VoiceBackendVersion = 16,
197 UserFlags = 18,
198 VoicePlatform = 20,
199 DavePrepareTransition = 21,
200 DaveExecuteTransition = 22,
201 DaveTransitionReady = 23,
202 DavePrepareEpoch = 24,
203 MlsExternalSender = 25,
204 MlsProposals = 27,
205 MlsAnnounceCommitTransition = 29,
206 MlsWelcome = 30,
207 MlsInvalidCommitWelcome = 31,
208 NoRoute = 32,
209 Unknown = 255,
210 }
211 impl From<u8> for OpCode {
212 fn from(op: u8) -> Self {
213 match op {
214 0 => Self::Identify,
215 1 => Self::SelectProtocol,
216 2 => Self::Ready,
217 3 => Self::Heartbeat,
218 4 => Self::SessionDescription,
219 5 => Self::Speaking,
220 6 => Self::HeartbeatAck,
221 7 => Self::Resume,
222 8 => Self::Hello,
223 9 => Self::Resumed,
224 11 => Self::ClientConnect,
225 12 => Self::Video,
226 13 => Self::ClientDisconnect,
227 14 => Self::Codecs,
228 15 => Self::MediaSinkWants,
229 16 => Self::VoiceBackendVersion,
230 18 => Self::UserFlags,
231 20 => Self::VoicePlatform,
232 21 => Self::DavePrepareTransition,
233 22 => Self::DaveExecuteTransition,
234 23 => Self::DaveTransitionReady,
235 24 => Self::DavePrepareEpoch,
236 25 => Self::MlsExternalSender,
237 27 => Self::MlsProposals,
238 29 => Self::MlsAnnounceCommitTransition,
239 30 => Self::MlsWelcome,
240 31 => Self::MlsInvalidCommitWelcome,
241 32 => Self::NoRoute,
242 _ => Self::Unknown,
243 }
244 }
245 }
246 pub mod builders {
247 use super::*;
248 use serde_json::json;
249 pub fn identify(
250 guild_id: String,
251 user_id: String,
252 session_id: String,
253 token: String,
254 dave_version: u16,
255 ) -> GatewayPayload {
256 GatewayPayload {
257 op: OpCode::Identify as u8,
258 seq: None,
259 d: json!({
260 "server_id": guild_id,
261 "user_id": user_id,
262 "session_id": session_id,
263 "token": token,
264 "video": true,
265 "max_dave_protocol_version": dave_version,
266 }),
267 }
268 }
269 pub fn resume(
270 guild_id: String,
271 session_id: String,
272 token: String,
273 seq_ack: i64,
274 ) -> GatewayPayload {
275 let seq_ack = seq_ack.max(0);
276 GatewayPayload {
277 op: OpCode::Resume as u8,
278 seq: None,
279 d: json!({
280 "server_id": guild_id,
281 "session_id": session_id,
282 "token": token,
283 "video": true,
284 "seq_ack": seq_ack,
285 }),
286 }
287 }
288 }
289 }
290 pub mod backoff {
291 use crate::gateway::constants::{BACKOFF_BASE_MS, MAX_RECONNECT_ATTEMPTS};
292 use std::time::Duration;
293 #[derive(Debug, Clone, Default)]
294 pub struct Backoff {
295 attempt: u32,
296 }
297 impl Backoff {
298 pub const fn new() -> Self {
299 Self { attempt: 0 }
300 }
301 pub fn next_delay(&mut self) -> Duration {
302 let exponent = self.attempt.min(3);
303 let ms = BACKOFF_BASE_MS * 2u64.pow(exponent);
304 self.attempt += 1;
305 Duration::from_millis(ms)
306 }
307 #[inline]
308 pub const fn is_exhausted(&self) -> bool {
309 self.attempt >= MAX_RECONNECT_ATTEMPTS
310 }
311 #[inline]
312 pub fn reset(&mut self) {
313 self.attempt = 0;
314 }
315 #[inline]
316 pub const fn attempt(&self) -> u32 {
317 self.attempt
318 }
319 }
320 }
321 pub mod policy {
322 use super::types::SessionOutcome;
323 pub struct FailurePolicy {
324 max_retries: u32,
325 }
326 impl FailurePolicy {
327 pub const fn new(max_retries: u32) -> Self {
328 Self { max_retries }
329 }
330 pub const fn is_retryable(&self, code: u16, attempt: u32) -> bool {
331 if attempt >= self.max_retries {
332 return false;
333 }
334 matches!(
335 code,
336 1000 | 1001
337 | 1006
338 | 4000
339 | 4001
340 | 4002
341 | 4003
342 | 4005
343 | 4006
344 | 4009
345 | 4012
346 | 4015
347 | 4016
348 | 4020
349 | 4900
350 )
351 }
352 pub fn classify(&self, code: u16) -> SessionOutcome {
353 match code {
354 4004 | 4011 | 4014 | 4021 | 4022 => SessionOutcome::Shutdown,
355 4006 | 4009 => SessionOutcome::Identify,
356 _ => SessionOutcome::Reconnect,
357 }
358 }
359 }
360 }
361 pub mod heartbeat {
362 use super::protocol::{GatewayPayload, OpCode};
363 use crate::common::utils::now_ms;
364 use std::sync::{
365 Arc,
366 atomic::{AtomicI64, AtomicU32, AtomicU64, Ordering},
367 };
368 use tokio::sync::mpsc::UnboundedSender;
369 use tokio_tungstenite::tungstenite::protocol::Message;
370 use tokio_util::sync::CancellationToken;
371 use tracing::warn;
372 pub struct HeartbeatTracker {
373 pub last_nonce: Arc<AtomicU64>,
374 pub sent_at: Arc<AtomicU64>,
375 pub missed_acks: Arc<AtomicU32>,
376 }
377 impl Default for HeartbeatTracker {
378 fn default() -> Self {
379 Self {
380 last_nonce: Arc::new(AtomicU64::new(0)),
381 sent_at: Arc::new(AtomicU64::new(0)),
382 missed_acks: Arc::new(AtomicU32::new(0)),
383 }
384 }
385 }
386 impl HeartbeatTracker {
387 pub fn new() -> Self {
388 Self::default()
389 }
390 pub fn validate_ack(&self, acked_nonce: u64) -> Option<u64> {
391 let expected = self.last_nonce.load(Ordering::Relaxed);
392 if expected != acked_nonce {
393 warn!("Heartbeat mismatch: sent={expected} got={acked_nonce}");
394 return None;
395 }
396 Some(now_ms().saturating_sub(self.sent_at.load(Ordering::Relaxed)))
397 }
398 pub fn spawn(
399 &self,
400 tx: UnboundedSender<Message>,
401 seq_ack: Arc<AtomicI64>,
402 conn_token: CancellationToken,
403 interval_ms: u64,
404 ) -> tokio::task::JoinHandle<()> {
405 let last_nonce = self.last_nonce.clone();
406 let sent_at = self.sent_at.clone();
407 let missed_acks = self.missed_acks.clone();
408 tokio::spawn(async move {
409 let mut ticker =
410 tokio::time::interval(tokio::time::Duration::from_millis(interval_ms));
411 ticker.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip);
412 loop {
413 ticker.tick().await;
414 let missed = missed_acks.fetch_add(1, Ordering::Relaxed);
415 if missed >= 2 {
416 warn!("Heartbeat timeout: {missed} missed ACKs.");
417 conn_token.cancel();
418 break;
419 }
420 let nonce = now_ms();
421 last_nonce.store(nonce, Ordering::Relaxed);
422 sent_at.store(nonce, Ordering::Relaxed);
423 let hb = GatewayPayload {
424 op: OpCode::Heartbeat as u8,
425 seq: None,
426 d: serde_json::json!({
427 "t": nonce,
428 "seq_ack": seq_ack.load(Ordering::Relaxed)
429 }),
430 };
431 if let Ok(json) = serde_json::to_string(&hb)
432 && tx.send(Message::Text(json.into())).is_err()
433 {
434 break;
435 }
436 }
437 })
438 }
439 }
440 }
441 pub mod voice {
442 use super::types::GatewayError;
443 use crate::{
444 audio::{Mixer, engine::Encoder, filters::FilterChain},
445 common::types::Shared,
446 gateway::{
447 DaveHandler,
448 constants::{
449 DISCOVERY_PACKET_SIZE, FRAME_DURATION_MS, IP_DISCOVERY_RETRIES,
450 IP_DISCOVERY_RETRY_INTERVAL_MS, IP_DISCOVERY_TIMEOUT_SECS, MAX_OPUS_FRAME_SIZE,
451 MAX_SILENCE_FRAMES, PCM_FRAME_SAMPLES, SILENCE_FRAME, UDP_KEEPALIVE_GAP_MS,
452 },
453 udp_link::UDPVoiceTransport,
454 },
455 };
456 use std::{
457 net::SocketAddr,
458 sync::{Arc, atomic::Ordering},
459 time::{Duration, Instant},
460 };
461 use tokio::sync::mpsc::UnboundedSender;
462 use tokio_util::sync::CancellationToken;
463 use tracing::error;
464 pub async fn discover_ip(
465 socket: &tokio::net::UdpSocket,
466 addr: SocketAddr,
467 ssrc: u32,
468 ) -> Result<(String, u16), GatewayError> {
469 let mut packet = [0u8; DISCOVERY_PACKET_SIZE];
470 packet[0..2].copy_from_slice(&1u16.to_be_bytes());
471 packet[2..4].copy_from_slice(&70u16.to_be_bytes());
472 packet[4..8].copy_from_slice(&ssrc.to_be_bytes());
473 for attempt in 1..=IP_DISCOVERY_RETRIES {
474 if attempt > 1 {
475 tokio::time::sleep(Duration::from_millis(IP_DISCOVERY_RETRY_INTERVAL_MS)).await;
476 }
477 if let Err(e) = socket.send_to(&packet, addr).await {
478 if attempt == IP_DISCOVERY_RETRIES {
479 return Err(GatewayError::Discovery(e.to_string()));
480 }
481 continue;
482 }
483 let mut client_buf = [0u8; DISCOVERY_PACKET_SIZE];
484 match tokio::time::timeout(
485 Duration::from_secs(IP_DISCOVERY_TIMEOUT_SECS),
486 socket.recv_from(&mut client_buf),
487 )
488 .await
489 {
490 Ok(Ok((n, peer))) if n >= DISCOVERY_PACKET_SIZE => {
491 if peer != addr {
492 continue;
493 }
494 let ip = std::str::from_utf8(&client_buf[8..72])
495 .map_err(|e| GatewayError::Discovery(e.to_string()))?
496 .trim_end_matches('\0')
497 .to_owned();
498 let port = u16::from_be_bytes([client_buf[72], client_buf[73]]);
499 return Ok((ip, port));
500 }
501 _ => {
502 if attempt == IP_DISCOVERY_RETRIES {
503 return Err(GatewayError::Discovery("Timed out".into()));
504 }
505 }
506 }
507 }
508 Err(GatewayError::Discovery("Exhausted".into()))
509 }
510 pub struct SpeakConfig {
511 pub mixer: Shared<Mixer>,
512 pub socket: Arc<tokio::net::UdpSocket>,
513 pub addr: SocketAddr,
514 pub ssrc: u32,
515 pub key: [u8; 32],
516 pub mode: String,
517 pub dave: Shared<DaveHandler>,
518 pub filter_chain: Shared<FilterChain>,
519 pub frames_sent: Arc<std::sync::atomic::AtomicU64>,
520 pub frames_nulled: Arc<std::sync::atomic::AtomicU64>,
521 pub cancel_token: CancellationToken,
522 pub speaking_tx: UnboundedSender<bool>,
523 pub persistent_state: Arc<tokio::sync::Mutex<super::types::PersistentSessionState>>,
524 }
525 pub async fn speak_loop(config: SpeakConfig) -> Result<(), GatewayError> {
526 let rtp_state = { config.persistent_state.lock().await.rtp_state };
527 let transport = UDPVoiceTransport::new(
528 config.socket.clone(),
529 config.addr,
530 config.ssrc,
531 config.key,
532 &config.mode,
533 rtp_state,
534 )?;
535 let mut encoder = Encoder::new().map_err(|e| GatewayError::Encoding(e.to_string()))?;
536 let mut session = VoiceSession::new(config, transport);
537 session.run(&mut encoder).await
538 }
539 struct VoiceSession {
540 config: SpeakConfig,
541 transport: UDPVoiceTransport,
542 is_speaking: bool,
543 speaking_holdoff: bool,
544 last_tx_time: Instant,
545 active_silence: u32,
546 }
547 impl VoiceSession {
548 fn new(config: SpeakConfig, transport: UDPVoiceTransport) -> Self {
549 Self {
550 config,
551 transport,
552 is_speaking: false,
553 speaking_holdoff: false,
554 last_tx_time: Instant::now(),
555 active_silence: 0,
556 }
557 }
558 async fn run(&mut self, encoder: &mut Encoder) -> Result<(), GatewayError> {
559 let mut interval = tokio::time::interval(Duration::from_millis(FRAME_DURATION_MS));
560 interval.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Burst);
561 let mut pcm = vec![0i16; PCM_FRAME_SAMPLES * 2];
562 let mut opus = vec![0u8; MAX_OPUS_FRAME_SIZE];
563 let mut ts_pcm = vec![0i16; PCM_FRAME_SAMPLES * 2];
564 while !self.config.cancel_token.is_cancelled() {
565 interval.tick().await;
566 self.tick(encoder, &mut pcm, &mut opus, &mut ts_pcm).await?;
567 if self
568 .config
569 .frames_sent
570 .load(Ordering::Relaxed)
571 .is_multiple_of(100)
572 {
573 self.config.persistent_state.lock().await.rtp_state =
574 Some(self.transport.rtp);
575 }
576 }
577 self.config.persistent_state.lock().await.rtp_state = Some(self.transport.rtp);
578 Ok(())
579 }
580 async fn tick(
581 &mut self,
582 encoder: &mut Encoder,
583 pcm: &mut [i16],
584 opus: &mut [u8],
585 ts_pcm: &mut [i16],
586 ) -> Result<(), GatewayError> {
587 macro_rules! try_lock_yield {
588 ($mutex:expr) => {{
589 let mut guard = None;
590 for _ in 0..10 {
591 if let Ok(g) = $mutex.try_lock() {
592 guard = Some(g);
593 break;
594 }
595 tokio::task::yield_now().await;
596 }
597 guard
598 }};
599 }
600 let mut loop_count = 0;
601 while loop_count < 10 {
602 loop_count += 1;
603 let ready_from_ts = {
604 if let Some(mut filters) = try_lock_yield!(self.config.filter_chain) {
605 filters.has_timescale() && filters.fill_frame(ts_pcm)
606 } else {
607 false
608 }
609 };
610 if ready_from_ts {
611 self.set_speaking(true);
612 self.config.frames_sent.fetch_add(1, Ordering::Relaxed);
613 if self.speaking_holdoff {
614 self.speaking_holdoff = false;
615 self.send_silence().await?;
616 }
617 return self.send_pcm(encoder, ts_pcm, opus).await;
618 }
619 let mut has_input = false;
620 let mut opus_data = None;
621 if let Some(mut mixer) = try_lock_yield!(self.config.mixer) {
622 if let Some(data) = mixer.take_opus_frame() {
623 opus_data = Some(data);
624 } else {
625 has_input = mixer.mix(pcm);
626 }
627 }
628 if let Some(data) = opus_data {
629 self.reset_timers();
630 self.set_speaking(true);
631 self.config.frames_sent.fetch_add(1, Ordering::Relaxed);
632 if self.speaking_holdoff {
633 self.speaking_holdoff = false;
634 self.send_silence().await?;
635 }
636 return self.send_raw(&data).await;
637 }
638 if has_input {
639 self.reset_timers();
640 self.set_speaking(true);
641 } else if self.active_silence > 0 {
642 self.active_silence -= 1;
643 pcm.fill(0);
644 self.set_speaking(true);
645 } else {
646 self.set_speaking(false);
647 if self.last_tx_time.elapsed()
648 >= Duration::from_millis(UDP_KEEPALIVE_GAP_MS)
649 {
650 return self.send_silence().await;
651 }
652 self.transport.rtp.timestamp = self
653 .transport
654 .rtp
655 .timestamp
656 .wrapping_add(crate::gateway::constants::RTP_TIMESTAMP_STEP);
657 return Ok(());
658 }
659 let has_ts = {
660 if let Some(mut filters) = try_lock_yield!(self.config.filter_chain) {
661 filters.process(pcm);
662 filters.has_timescale()
663 } else {
664 false
665 }
666 };
667 if !has_ts {
668 if has_input {
669 self.config.frames_sent.fetch_add(1, Ordering::Relaxed);
670 } else {
671 self.config.frames_nulled.fetch_add(1, Ordering::Relaxed);
672 }
673 if self.speaking_holdoff {
674 self.speaking_holdoff = false;
675 self.send_silence().await?;
676 }
677 return self.send_pcm(encoder, pcm, opus).await;
678 }
679 let filled_on_silence = {
680 if let Some(mut filters) = try_lock_yield!(self.config.filter_chain) {
681 !has_input && filters.fill_frame(ts_pcm)
682 } else {
683 false
684 }
685 };
686 if !has_input && !filled_on_silence {
687 break;
688 }
689 }
690 Ok(())
691 }
692 fn set_speaking(&mut self, speaking: bool) {
693 if speaking != self.is_speaking {
694 self.is_speaking = speaking;
695 let _ = self.config.speaking_tx.send(speaking);
696 if speaking {
697 self.speaking_holdoff = true;
698 }
699 }
700 }
701 async fn send_pcm(
702 &mut self,
703 encoder: &mut Encoder,
704 pcm: &[i16],
705 opus: &mut [u8],
706 ) -> Result<(), GatewayError> {
707 let size = encoder.encode(pcm, opus).unwrap_or_else(|e| {
708 error!("Opus encode failed: {e}");
709 0
710 });
711 if size > 0 {
712 self.send_raw(&opus[..size]).await?;
713 } else {
714 self.send_silence().await?;
715 }
716 Ok(())
717 }
718 async fn send_silence(&mut self) -> Result<(), GatewayError> {
719 self.config.frames_nulled.fetch_add(1, Ordering::Relaxed);
720 self.send_raw(&SILENCE_FRAME).await
721 }
722 async fn send_raw(&mut self, data: &[u8]) -> Result<(), GatewayError> {
723 let mut dave = self.config.dave.lock().await;
724 let encrypted = dave
725 .encrypt_opus(data)
726 .map_err(|e| GatewayError::Encryption(e.to_string()))?;
727 drop(dave);
728 self.transport.transmit_opus(&encrypted).await?;
729 self.last_tx_time = Instant::now();
730 Ok(())
731 }
732 fn reset_timers(&mut self) {
733 self.active_silence = MAX_SILENCE_FRAMES;
734 }
735 }
736 }
737 pub mod handler {
738 use super::{
739 VoiceGateway,
740 backoff::Backoff,
741 heartbeat::HeartbeatTracker,
742 protocol::{GatewayPayload, OpCode},
743 types::{GatewayError, PersistentSessionState, SessionOutcome},
744 voice::{SpeakConfig, discover_ip, speak_loop},
745 };
746 use crate::{
747 common::types::{Shared, UserId},
748 gateway::{
749 DaveHandler,
750 constants::{DAVE_INITIAL_VERSION, DEFAULT_VOICE_MODE},
751 },
752 };
753 use serde_json::Value;
754 use std::{
755 collections::HashSet,
756 net::SocketAddr,
757 sync::{
758 Arc,
759 atomic::{AtomicI64, Ordering},
760 },
761 };
762 use tokio::sync::mpsc::UnboundedSender;
763 use tokio_tungstenite::tungstenite::protocol::Message;
764 use tokio_util::sync::CancellationToken;
765 use tracing::{debug, error, trace, warn};
766 use uuid::Uuid;
767 pub struct SessionState<'a> {
768 gateway: &'a VoiceGateway,
769 tx: UnboundedSender<Message>,
770 seq_ack: Arc<AtomicI64>,
771 ssrc: u32,
772 udp_addr: Option<SocketAddr>,
773 selected_mode: String,
774 connected_users: HashSet<UserId>,
775 udp_socket: Arc<tokio::net::UdpSocket>,
776 dave: Shared<DaveHandler>,
777 heartbeat: HeartbeatTracker,
778 heartbeat_handle: Option<tokio::task::JoinHandle<()>>,
779 conn_token: CancellationToken,
780 speaking_tx: Option<UnboundedSender<bool>>,
781 session_key: Option<[u8; 32]>,
782 speak_task: Option<tokio::task::JoinHandle<()>>,
783 persistent_state: Arc<tokio::sync::Mutex<PersistentSessionState>>,
784 backoff: &'a mut Backoff,
785 }
786 impl<'a> SessionState<'a> {
787 pub async fn new(
788 gateway: &'a VoiceGateway,
789 tx: UnboundedSender<Message>,
790 seq_ack: Arc<AtomicI64>,
791 conn_token: CancellationToken,
792 persistent_state: Arc<tokio::sync::Mutex<PersistentSessionState>>,
793 backoff: &'a mut Backoff,
794 ) -> Result<Self, GatewayError> {
795 let mut socket_guard = gateway.udp_socket.lock().await;
796 let udp_socket = if let Some(existing) = &*socket_guard {
797 existing.clone()
798 } else {
799 let udp = std::net::UdpSocket::bind("0.0.0.0:0")?;
800 udp.set_nonblocking(true)?;
801 let socket = Arc::new(tokio::net::UdpSocket::from_std(udp)?);
802 *socket_guard = Some(socket.clone());
803 socket
804 };
805 Ok(Self {
806 gateway,
807 tx,
808 seq_ack,
809 ssrc: 0,
810 udp_addr: None,
811 selected_mode: DEFAULT_VOICE_MODE.to_string(),
812 connected_users: HashSet::from([gateway.user_id]),
813 udp_socket,
814 dave: gateway.dave.clone(),
815 heartbeat: HeartbeatTracker::new(),
816 heartbeat_handle: None,
817 conn_token,
818 speaking_tx: None,
819 session_key: None,
820 speak_task: None,
821 persistent_state,
822 backoff,
823 })
824 }
825 pub fn set_speaking_tx(&mut self, tx: UnboundedSender<bool>) {
826 self.speaking_tx = Some(tx);
827 }
828 pub fn ssrc(&self) -> u32 {
829 self.ssrc
830 }
831 pub fn tx(&self) -> &UnboundedSender<Message> {
832 &self.tx
833 }
834 pub fn attempt(&self) -> u32 {
835 self.backoff.attempt()
836 }
837 pub fn has_heartbeat(&self) -> bool {
838 self.heartbeat_handle.is_some()
839 }
840 pub async fn handle_text(&mut self, text: String) -> Option<SessionOutcome> {
841 let payload: GatewayPayload = match serde_json::from_str(&text) {
842 Ok(p) => p,
843 Err(e) => {
844 warn!("[{}] JSON Parse error: {e}", self.gateway.guild_id);
845 return None;
846 }
847 };
848 if let Some(seq) = payload.seq {
849 self.seq_ack.store(seq as i64, Ordering::Relaxed);
850 }
851 let op = OpCode::from(payload.op);
852 trace!(
853 "[{}] RX OP: {:?} (op={})",
854 self.gateway.guild_id, op, payload.op
855 );
856 match op {
857 OpCode::Hello => self.on_hello(payload.d),
858 OpCode::Ready => self.on_ready(payload.d).await,
859 OpCode::SessionDescription => self.on_session_description(payload.d).await,
860 OpCode::HeartbeatAck => self.on_heartbeat_ack(payload.d),
861 OpCode::Resumed => self.on_resumed().await,
862 OpCode::ClientConnect => self.on_user_connect(payload.d).await,
863 OpCode::ClientDisconnect => self.on_user_disconnect(payload.d).await,
864 OpCode::VoiceBackendVersion => {
865 debug!(
866 "[{}] Voice Backend Version: {:?}",
867 self.gateway.guild_id, payload.d
868 );
869 None
870 }
871 OpCode::MediaSinkWants => {
872 debug!(
873 "[{}] Media Sink Wants: {:?}",
874 self.gateway.guild_id, payload.d
875 );
876 None
877 }
878 OpCode::DavePrepareTransition => {
879 self.on_dave_prepare_transition(payload.d).await
880 }
881 OpCode::DaveExecuteTransition => {
882 self.on_dave_execute_transition(payload.d).await
883 }
884 OpCode::DavePrepareEpoch => self.on_dave_prepare_epoch(payload.d).await,
885 OpCode::MlsAnnounceCommitTransition => self.on_mls_transition(payload.d).await,
886 OpCode::MlsInvalidCommitWelcome => {
887 warn!(
888 "[{}] DAVE MLS Invalid Commit Welcome received, resetting session",
889 self.gateway.guild_id
890 );
891 self.reset_dave(0).await;
892 None
893 }
894 OpCode::NoRoute => {
895 warn!(
896 "[{}] No Route received: {:?}",
897 self.gateway.guild_id, payload.d
898 );
899 None
900 }
901 OpCode::Speaking
902 | OpCode::Video
903 | OpCode::Codecs
904 | OpCode::UserFlags
905 | OpCode::VoicePlatform => None,
906 _ => None,
907 }
908 }
909 pub async fn handle_binary(&mut self, bin: Vec<u8>) {
910 if bin.len() < 3 {
911 return;
912 }
913 let seq = u16::from_be_bytes([bin[0], bin[1]]);
914 let op = bin[2];
915 let data = &bin[3..];
916 self.seq_ack.store(seq as i64, Ordering::Relaxed);
917 let mut dave = self.dave.lock().await;
918 match op {
919 25 => {
920 if let Ok(res) = dave.process_external_sender(data) {
921 for r in res {
922 self.send_binary(28, &r);
923 }
924 }
925 }
926 27 => match dave.process_proposals(data) {
927 Ok(Some(cw)) => self.send_binary(28, &cw),
928 Err(e) => {
929 warn!("[{}] DAVE proposals failed: {e}", self.gateway.guild_id);
930 self.reset_dave_locked(&mut dave, 0).await;
931 }
932 _ => {}
933 },
934 29 | 30 => {
935 let res = if op == 30 {
936 dave.process_welcome(data)
937 } else {
938 dave.process_commit(data)
939 };
940 match res {
941 Ok(tid) if tid != 0 => {
942 self.send_json(23, serde_json::json!({ "transition_id": tid }))
943 }
944 Err(e) => {
945 let tid = if data.len() >= 2 {
946 u16::from_be_bytes([data[0], data[1]])
947 } else {
948 0
949 };
950 warn!(
951 "[{}] DAVE {} failed (tid {tid}): {e}",
952 self.gateway.guild_id,
953 if op == 30 { "welcome" } else { "commit" }
954 );
955 self.reset_dave_locked(&mut dave, tid).await;
956 }
957 _ => {}
958 }
959 }
960 _ => {}
961 }
962 }
963 fn on_hello(&mut self, d: Value) -> Option<SessionOutcome> {
964 let interval = d["heartbeat_interval"].as_u64().unwrap_or(30_000);
965 if let Some(h) = self.heartbeat_handle.take() {
966 debug!(
967 "[{}] Restarting heartbeat on HELLO (was already running)",
968 self.gateway.guild_id
969 );
970 h.abort();
971 }
972 trace!(
973 "[{}] Heartbeat interval: {interval}ms",
974 self.gateway.guild_id
975 );
976 self.heartbeat_handle = Some(self.heartbeat.spawn(
977 self.tx.clone(),
978 self.seq_ack.clone(),
979 self.conn_token.clone(),
980 interval,
981 ));
982 None
983 }
984 async fn on_ready(&mut self, d: Value) -> Option<SessionOutcome> {
985 let ssrc = d["ssrc"].as_u64();
986 let ip = d["ip"].as_str();
987 let port = d["port"].as_u64();
988 match (ssrc, ip, port) {
989 (Some(ssrc), Some(ip), Some(port)) if port <= 65535 => {
990 self.ssrc = ssrc as u32;
991 let addr_str = format!("{ip}:{port}");
992 match addr_str.parse::<SocketAddr>() {
993 Ok(addr) => self.udp_addr = Some(addr),
994 Err(_) => {
995 error!(
996 "[{}] Invalid READY address: {addr_str}",
997 self.gateway.guild_id
998 );
999 return Some(SessionOutcome::Reconnect);
1000 }
1001 }
1002 }
1003 _ => {
1004 error!("[{}] Malformed READY payload", self.gateway.guild_id);
1005 return Some(SessionOutcome::Reconnect);
1006 }
1007 }
1008 if let Some(modes) = d["modes"].as_array() {
1009 let pref = ["aead_aes256_gcm_rtpsize", "xsalsa20_poly1305"];
1010 if let Some(m) = pref
1011 .iter()
1012 .find(|&&p| modes.iter().any(|m| m.as_str() == Some(p)))
1013 {
1014 self.selected_mode = (*m).to_owned();
1015 }
1016 }
1017 debug!(
1018 "[{}] Ready: ssrc={}, mode={}",
1019 self.gateway.guild_id, self.ssrc, self.selected_mode
1020 );
1021 {
1022 let mut state = self.persistent_state.lock().await;
1023 state.ssrc = self.ssrc;
1024 state.selected_mode = Some(self.selected_mode.clone());
1025 }
1026 if self.gateway.channel_id.0 > 0 {
1027 let ver = d["dave_protocol_version"]
1028 .as_u64()
1029 .unwrap_or(DAVE_INITIAL_VERSION as u64)
1030 as u16;
1031 let mut dave = self.dave.lock().await;
1032 if ver == 0 {
1033 dave.reset();
1034 } else {
1035 dave.set_protocol_version(ver);
1036 if let Ok(kp) = dave.setup_session(ver) {
1037 self.send_binary(26, &kp);
1038 }
1039 }
1040 }
1041 let target_addr = match self.udp_addr {
1042 Some(a) => a,
1043 None => return Some(SessionOutcome::Reconnect),
1044 };
1045 match discover_ip(&self.udp_socket, target_addr, self.ssrc).await {
1046 Ok((my_ip, my_port)) => {
1047 self.send_json(OpCode::SelectProtocol as u8, serde_json::json!({
1048 "protocol": "udp",
1049 "rtc_connection_id": Uuid::new_v4().to_string(),
1050 "codecs": [{"name": "opus", "type": "audio", "priority": 1000, "payload_type": 120}],
1051 "data": { "address": my_ip, "port": my_port, "mode": self.selected_mode },
1052 "address": my_ip,
1053 "port": my_port,
1054 "mode": self.selected_mode
1055 }));
1056 self.send_json(
1057 OpCode::Video as u8,
1058 serde_json::json!({"audio_ssrc": self.ssrc, "video_ssrc": 0, "rtx_ssrc": 0}),
1059 );
1060 self.send_json(
1061 OpCode::Speaking as u8,
1062 serde_json::json!({"speaking": 0, "delay": 0, "ssrc": self.ssrc}),
1063 );
1064 }
1065 Err(e) => {
1066 error!("[{}] IP discovery failed: {e}", self.gateway.guild_id);
1067 return Some(SessionOutcome::Reconnect);
1068 }
1069 }
1070 self.backoff.reset();
1071 None
1072 }
1073 async fn on_session_description(&mut self, d: Value) -> Option<SessionOutcome> {
1074 let ka = match d["secret_key"].as_array() {
1075 Some(a) if a.len() == 32 => a,
1076 _ => {
1077 error!(
1078 "[{}] Invalid or missing secret_key in VOICE_READY",
1079 self.gateway.guild_id
1080 );
1081 return Some(SessionOutcome::Reconnect);
1082 }
1083 };
1084 let mut key = [0u8; 32];
1085 for (i, v) in ka.iter().enumerate() {
1086 if let Some(val) = v.as_u64()
1087 && val <= 255
1088 {
1089 key[i] = val as u8;
1090 continue;
1091 }
1092 error!(
1093 "[{}] Invalid secret_key byte at index {i}",
1094 self.gateway.guild_id
1095 );
1096 return Some(SessionOutcome::Reconnect);
1097 }
1098 self.session_key = Some(key);
1099 let addr = match self.udp_addr {
1100 Some(a) => a,
1101 None => return Some(SessionOutcome::Reconnect),
1102 };
1103 {
1104 let mut state = self.persistent_state.lock().await;
1105 state.udp_addr = Some(addr);
1106 state.session_key = Some(key);
1107 state.ssrc = self.ssrc;
1108 state.selected_mode = Some(self.selected_mode.clone());
1109 }
1110 self.start_voice(addr, key).await;
1111 if self.gateway.channel_id.0 > 0 {
1112 let protocol_version = d["dave_protocol_version"]
1113 .as_u64()
1114 .unwrap_or(DAVE_INITIAL_VERSION as u64)
1115 as u16;
1116 let mls_group_id = d["mls_group_id"].as_u64().unwrap_or(0);
1117 let mut dave = self.dave.lock().await;
1118 if protocol_version > 0 {
1119 dave.set_protocol_version(protocol_version);
1120 if let Ok(kp) = dave.setup_session(protocol_version) {
1121 self.send_binary(26, &kp);
1122 }
1123 } else {
1124 dave.reset();
1125 }
1126 debug!(
1127 "DAVE setup context: protocol_version={}, mls_group_id={}",
1128 protocol_version, mls_group_id
1129 );
1130 }
1131 self.backoff.reset();
1132 None
1133 }
1134 async fn on_resumed(&mut self) -> Option<SessionOutcome> {
1135 debug!("[{}] Resumed", self.gateway.guild_id);
1136 let (addr, key, ssrc, mode) = {
1137 let state = self.persistent_state.lock().await;
1138 (
1139 state.udp_addr,
1140 state.session_key,
1141 state.ssrc,
1142 state.selected_mode.clone(),
1143 )
1144 };
1145 match (addr, key) {
1146 (Some(addr), Some(key)) => {
1147 self.udp_addr = Some(addr);
1148 self.session_key = Some(key);
1149 self.ssrc = ssrc;
1150 if let Some(m) = mode {
1151 self.selected_mode = m;
1152 }
1153 if let Some(task) = &self.speak_task
1154 && task.is_finished()
1155 {
1156 self.speak_task = None;
1157 }
1158 if self.speak_task.is_some() {
1159 debug!(
1160 "[{}] Keeping existing voice loop alive across resume",
1161 self.gateway.guild_id
1162 );
1163 } else {
1164 debug!(
1165 "[{}] Starting voice loop after resume (task was dead or missing)",
1166 self.gateway.guild_id
1167 );
1168 self.start_voice(addr, key).await;
1169 }
1170 }
1171 _ => {
1172 warn!(
1173 "[{}] Resume failed: missing persistent state",
1174 self.gateway.guild_id
1175 );
1176 return Some(SessionOutcome::Identify);
1177 }
1178 }
1179 None
1180 }
1181 fn on_heartbeat_ack(&self, d: Value) -> Option<SessionOutcome> {
1182 let nonce = d["t"].as_u64().unwrap_or(0);
1183 if let Some(rtt) = self.heartbeat.validate_ack(nonce) {
1184 self.gateway.ping.store(rtt as i64, Ordering::Relaxed);
1185 self.heartbeat.missed_acks.store(0, Ordering::Relaxed);
1186 }
1187 None
1188 }
1189 async fn on_user_connect(&mut self, d: Value) -> Option<SessionOutcome> {
1190 if let Some(ids) = d["user_ids"].as_array() {
1191 let mut uids = Vec::new();
1192 for id in ids {
1193 if let Some(uid) = id.as_str().and_then(|s| s.parse::<u64>().ok()) {
1194 self.connected_users.insert(UserId(uid));
1195 uids.push(uid);
1196 }
1197 }
1198 if !uids.is_empty() {
1199 self.dave.lock().await.add_users(&uids);
1200 }
1201 }
1202 None
1203 }
1204 async fn on_user_disconnect(&mut self, d: Value) -> Option<SessionOutcome> {
1205 if let Some(uid) = d["user_id"].as_str().and_then(|s| s.parse::<u64>().ok()) {
1206 self.connected_users.remove(&UserId(uid));
1207 self.dave.lock().await.remove_user(uid);
1208 }
1209 None
1210 }
1211 async fn on_dave_prepare_transition(&mut self, d: Value) -> Option<SessionOutcome> {
1212 let tid = d["transition_id"].as_u64().unwrap_or(0) as u16;
1213 let ver = d["protocol_version"].as_u64().unwrap_or(0) as u16;
1214 debug!(
1215 "[{}] DAVE Prepare Transition: id={}, version={}",
1216 self.gateway.guild_id, tid, ver
1217 );
1218 if self.dave.lock().await.prepare_transition(tid, ver) {
1219 debug!(
1220 "[{}] DAVE Transition Ready (tid={})",
1221 self.gateway.guild_id, tid
1222 );
1223 self.send_json(23, serde_json::json!({ "transition_id": tid }));
1224 }
1225 None
1226 }
1227 async fn on_dave_execute_transition(&mut self, d: Value) -> Option<SessionOutcome> {
1228 let tid = d["transition_id"].as_u64().unwrap_or(0) as u16;
1229 debug!(
1230 "[{}] DAVE Execute Transition: id={}",
1231 self.gateway.guild_id, tid
1232 );
1233 self.dave.lock().await.execute_transition(tid);
1234 None
1235 }
1236 async fn on_dave_prepare_epoch(&mut self, d: Value) -> Option<SessionOutcome> {
1237 let epoch = d["epoch"].as_u64().unwrap_or(0);
1238 let ver = d["protocol_version"].as_u64().unwrap_or(0) as u16;
1239 debug!(
1240 "[{}] DAVE Prepare Epoch: epoch={}, version={}",
1241 self.gateway.guild_id, epoch, ver
1242 );
1243 if let Some(kp) = self.dave.lock().await.prepare_epoch(epoch, ver) {
1244 self.send_binary(26, &kp);
1245 }
1246 None
1247 }
1248 async fn on_mls_transition(&mut self, d: Value) -> Option<SessionOutcome> {
1249 let tid = d["transition_id"].as_u64().unwrap_or(0) as u16;
1250 debug!(
1251 "[{}] DAVE MLS Announce Commit Transition: tid={}",
1252 self.gateway.guild_id, tid
1253 );
1254 let ver = d["protocol_version"].as_u64().map(|v| v as u16);
1255 if let Some(v) = ver {
1256 let mut dave = self.dave.lock().await;
1257 if dave.prepare_transition(tid, v) && tid != 0 {
1258 self.send_json(23, serde_json::json!({ "transition_id": tid }));
1259 }
1260 }
1261 None
1262 }
1263 async fn start_voice(&mut self, addr: SocketAddr, key: [u8; 32]) {
1264 if let Some(t) = self.speak_task.take() {
1265 t.abort();
1266 }
1267 let speaking_tx = if let Some(tx) = &self.speaking_tx {
1268 tx.clone()
1269 } else {
1270 error!(
1271 "[{}] speaking_tx is missing, cannot start voice",
1272 self.gateway.guild_id
1273 );
1274 return;
1275 };
1276 let config = SpeakConfig {
1277 mixer: self.gateway.mixer.clone(),
1278 socket: self.udp_socket.clone(),
1279 addr,
1280 ssrc: self.ssrc,
1281 key,
1282 mode: self.selected_mode.clone(),
1283 dave: self.dave.clone(),
1284 filter_chain: self.gateway.filter_chain.clone(),
1285 frames_sent: self.gateway.frames_sent.clone(),
1286 frames_nulled: self.gateway.frames_nulled.clone(),
1287 cancel_token: self.conn_token.clone(),
1288 speaking_tx,
1289 persistent_state: self.persistent_state.clone(),
1290 };
1291 let guild_id = self.gateway.guild_id.clone();
1292 let conn_token = self.conn_token.clone();
1293 self.speak_task = Some(tokio::spawn(async move {
1294 if let Err(e) = speak_loop(config).await {
1295 error!("[{guild_id}] speak_loop failed: {e}");
1296 conn_token.cancel();
1297 }
1298 }));
1299 self.send_json(
1300 OpCode::Video as u8,
1301 serde_json::json!({"audio_ssrc": self.ssrc, "video_ssrc": 0, "rtx_ssrc": 0}),
1302 );
1303 self.send_json(
1304 OpCode::Speaking as u8,
1305 serde_json::json!({"speaking": 0, "delay": 0, "ssrc": self.ssrc}),
1306 );
1307 }
1308 async fn reset_dave(&self, tid: u16) {
1309 let mut dave = self.dave.lock().await;
1310 self.reset_dave_locked(&mut dave, tid).await;
1311 }
1312 async fn reset_dave_locked(&self, dave: &mut DaveHandler, tid: u16) {
1313 dave.reset();
1314 self.send_json(31, serde_json::json!({ "transition_id": tid }));
1315 if let Ok(kp) = dave.setup_session(DAVE_INITIAL_VERSION) {
1316 self.send_binary(26, &kp);
1317 }
1318 }
1319 fn send_json(&self, op: u8, d: Value) {
1320 match serde_json::to_string(&GatewayPayload { op, seq: None, d }) {
1321 Ok(json) => {
1322 let _ = self.tx.send(Message::Text(json.into()));
1323 }
1324 Err(e) => {
1325 warn!("[{}] JSON serialization failed: {e}", self.gateway.guild_id);
1326 }
1327 }
1328 }
1329 fn send_binary(&self, op: u8, payload: &[u8]) {
1330 let mut b = vec![op];
1331 b.extend_from_slice(payload);
1332 let _ = self.tx.send(Message::Binary(b.into()));
1333 }
1334 }
1335 impl<'a> Drop for SessionState<'a> {
1336 fn drop(&mut self) {
1337 if let Some(h) = self.heartbeat_handle.take() {
1338 h.abort();
1339 }
1340 if let Some(t) = self.speak_task.take() {
1341 t.abort();
1342 }
1343 }
1344 }
1345 }
1346 use self::{
1347 backoff::Backoff,
1348 policy::FailurePolicy,
1349 types::{GatewayError, PersistentSessionState, SessionOutcome},
1350 };
1351 use crate::{
1352 audio::{Mixer, filters::FilterChain},
1353 common::types::{ChannelId, GuildId, SessionId, Shared, UserId},
1354 gateway::constants::VOICE_GATEWAY_VERSION,
1355 protocol::LavendeEvent,
1356 };
1357 use futures::{SinkExt, StreamExt};
1358 use std::sync::{
1359 Arc,
1360 atomic::{AtomicI64, Ordering},
1361 };
1362 use tokio::sync::mpsc::{UnboundedSender, unbounded_channel};
1363 use tokio_tungstenite::tungstenite::protocol::{Message, WebSocketConfig};
1364 use tokio_util::sync::CancellationToken;
1365 use tracing::{debug, error, warn};
1366 pub struct VoiceGateway {
1367 pub guild_id: GuildId,
1368 pub user_id: UserId,
1369 pub channel_id: ChannelId,
1370 session_id: SessionId,
1371 token: String,
1372 endpoint: String,
1373 pub mixer: Shared<Mixer>,
1374 pub filter_chain: Shared<FilterChain>,
1375 pub ping: Arc<AtomicI64>,
1376 event_tx: Option<UnboundedSender<LavendeEvent>>,
1377 pub frames_sent: Arc<std::sync::atomic::AtomicU64>,
1378 pub frames_nulled: Arc<std::sync::atomic::AtomicU64>,
1379 pub udp_socket: Shared<Option<Arc<tokio::net::UdpSocket>>>,
1380 pub dave: Shared<crate::gateway::DaveHandler>,
1381 outer_token: CancellationToken,
1382 policy: FailurePolicy,
1383 }
1384 pub struct VoiceGatewayConfig {
1385 pub guild_id: GuildId,
1386 pub user_id: UserId,
1387 pub channel_id: ChannelId,
1388 pub session_id: SessionId,
1389 pub token: String,
1390 pub endpoint: String,
1391 pub mixer: Shared<Mixer>,
1392 pub filter_chain: Shared<FilterChain>,
1393 pub ping: Arc<AtomicI64>,
1394 pub event_tx: Option<UnboundedSender<LavendeEvent>>,
1395 pub frames_sent: Arc<std::sync::atomic::AtomicU64>,
1396 pub frames_nulled: Arc<std::sync::atomic::AtomicU64>,
1397 pub cancel_token: CancellationToken,
1398 }
1399 impl VoiceGateway {
1400 pub fn new(config: VoiceGatewayConfig) -> Self {
1401 Self {
1402 guild_id: config.guild_id,
1403 user_id: config.user_id,
1404 channel_id: config.channel_id,
1405 session_id: config.session_id,
1406 token: config.token,
1407 endpoint: config.endpoint,
1408 mixer: config.mixer,
1409 filter_chain: config.filter_chain,
1410 ping: config.ping,
1411 event_tx: config.event_tx,
1412 frames_sent: config.frames_sent,
1413 frames_nulled: config.frames_nulled,
1414 udp_socket: Arc::new(tokio::sync::Mutex::new(None)),
1415 dave: Arc::new(tokio::sync::Mutex::new(crate::gateway::DaveHandler::new(
1416 config.user_id,
1417 config.channel_id,
1418 ))),
1419 outer_token: config.cancel_token,
1420 policy: FailurePolicy::new(3),
1421 }
1422 }
1423 pub async fn run(self) -> Result<(), GatewayError> {
1424 let mut backoff = Backoff::new();
1425 let mut is_resume = false;
1426 let seq_ack = Arc::new(AtomicI64::new(-1));
1427 let persistent_state =
1428 Arc::new(tokio::sync::Mutex::new(PersistentSessionState::default()));
1429 while !self.outer_token.is_cancelled() {
1430 let attempt = backoff.attempt();
1431 match self
1432 .connect(
1433 is_resume,
1434 seq_ack.clone(),
1435 persistent_state.clone(),
1436 &mut backoff,
1437 )
1438 .await
1439 {
1440 Ok(SessionOutcome::Shutdown) => break,
1441 Ok(outcome) => {
1442 if backoff.is_exhausted() {
1443 warn!("[{}] Max attempts reached ({})", self.guild_id, attempt);
1444 break;
1445 }
1446 let delay = backoff.next_delay();
1447 is_resume = matches!(outcome, SessionOutcome::Reconnect);
1448 if !is_resume {
1449 seq_ack.store(-1, Ordering::Relaxed);
1450 *persistent_state.lock().await = PersistentSessionState::default();
1451 *self.udp_socket.lock().await = None;
1452 }
1453 debug!(
1454 "[{}] Retrying ({:?}) in {:?}",
1455 self.guild_id, outcome, delay
1456 );
1457 tokio::time::sleep(delay).await;
1458 }
1459 Err(e) => {
1460 if backoff.is_exhausted() {
1461 error!("[{}] Fatal connection error: {e}", self.guild_id);
1462 break;
1463 }
1464 let delay = backoff.next_delay();
1465 warn!(
1466 "[{}] Connection error: {e}. Retrying in {:?}",
1467 self.guild_id, delay
1468 );
1469 tokio::time::sleep(delay).await;
1470 is_resume = false;
1471 }
1472 }
1473 }
1474 Ok(())
1475 }
1476 async fn connect(
1477 &self,
1478 is_resume: bool,
1479 seq_ack: Arc<AtomicI64>,
1480 persistent_state: Arc<tokio::sync::Mutex<PersistentSessionState>>,
1481 backoff: &mut Backoff,
1482 ) -> Result<SessionOutcome, GatewayError> {
1483 let endpoint = if self.endpoint.ends_with(":80") {
1484 &self.endpoint[..self.endpoint.len() - 3]
1485 } else {
1486 &self.endpoint
1487 };
1488 let url = format!("wss://{}/?v={}", endpoint, VOICE_GATEWAY_VERSION);
1489 let mut config = WebSocketConfig::default();
1490 config.max_message_size = None;
1491 config.max_frame_size = None;
1492 let (ws_stream, _) =
1493 tokio_tungstenite::connect_async_with_config(&url, Some(config), true).await?;
1494 let (mut write, mut read) = ws_stream.split();
1495 let conn_token = CancellationToken::new();
1496 let write_token = conn_token.clone();
1497 let (ws_tx, mut ws_rx) = unbounded_channel::<Message>();
1498 let writer_handle = tokio::spawn(async move {
1499 while let Some(msg) = tokio::select! {
1500 biased;
1501 _ = write_token.cancelled() => None,
1502 msg = ws_rx.recv() => msg,
1503 } {
1504 if write.send(msg).await.is_err() {
1505 break;
1506 }
1507 }
1508 });
1509 let mut state = handler::SessionState::new(
1510 self,
1511 ws_tx.clone(),
1512 seq_ack.clone(),
1513 conn_token.clone(),
1514 persistent_state,
1515 backoff,
1516 )
1517 .await
1518 .inspect_err(|_e| {
1519 conn_token.cancel();
1520 })?;
1521 let outcome = match read.next().await {
1522 Some(Ok(m)) => self.handle_message(&mut state, m).await,
1523 _ => Some(SessionOutcome::Reconnect),
1524 };
1525 if let Some(out) = outcome {
1526 conn_token.cancel();
1527 writer_handle.abort();
1528 let _ = writer_handle.await;
1529 return Ok(out);
1530 }
1531 if !state.has_heartbeat() {
1532 conn_token.cancel();
1533 writer_handle.abort();
1534 let _ = writer_handle.await;
1535 return Ok(SessionOutcome::Reconnect);
1536 }
1537 let handshake = if is_resume {
1538 debug!(
1539 "[{}] Sending Resume with seq_ack={}",
1540 self.guild_id,
1541 seq_ack.load(Ordering::Relaxed)
1542 );
1543 protocol::builders::resume(
1544 self.guild_id.to_string(),
1545 self.session_id.to_string(),
1546 self.token.clone(),
1547 seq_ack.load(Ordering::Relaxed),
1548 )
1549 } else {
1550 protocol::builders::identify(
1551 self.guild_id.to_string(),
1552 self.user_id.0.to_string(),
1553 self.session_id.to_string(),
1554 self.token.clone(),
1555 1,
1556 )
1557 };
1558 let _ = ws_tx.send(Message::Text(
1559 serde_json::to_string(&handshake).unwrap().into(),
1560 ));
1561 let (speaking_tx, mut speaking_rx) = unbounded_channel::<bool>();
1562 state.set_speaking_tx(speaking_tx);
1563 let outcome = loop {
1564 tokio::select! {
1565 biased;
1566 _ = self.outer_token.cancelled() => break SessionOutcome::Shutdown,
1567 _ = conn_token.cancelled() => break SessionOutcome::Reconnect,
1568 Some(speaking) = speaking_rx.recv() => {
1569 self.notify_speaking(&ws_tx, state.ssrc(), speaking);
1570 }
1571 msg = read.next() => match msg {
1572 Some(Ok(m)) => if let Some(out) = self.handle_message(&mut state, m).await {
1573 break out;
1574 },
1575 Some(Err(_)) => break SessionOutcome::Reconnect,
1576 None => break SessionOutcome::Reconnect,
1577 }
1578 }
1579 };
1580 conn_token.cancel();
1581 writer_handle.abort();
1582 let _ = writer_handle.await;
1583 Ok(outcome)
1584 }
1585 async fn handle_message(
1586 &self,
1587 state: &mut handler::SessionState<'_>,
1588 msg: Message,
1589 ) -> Option<SessionOutcome> {
1590 match msg {
1591 Message::Text(text) => state.handle_text(text.to_string()).await,
1592 Message::Binary(bin) => {
1593 state.handle_binary(bin.to_vec()).await;
1594 None
1595 }
1596 Message::Close(frame) => {
1597 let code = frame.as_ref().map(|f| f.code.into()).unwrap_or(1000u16);
1598 let reason = frame.map(|f| f.reason.to_string()).unwrap_or_default();
1599 let attempt = state.attempt();
1600 debug!("[{}] Gateway closed: {} ({})", self.guild_id, code, reason);
1601 if !self.policy.is_retryable(code, attempt) {
1602 self.emit_close(code, reason);
1603 }
1604 Some(self.policy.classify(code))
1605 }
1606 Message::Ping(p) => {
1607 let _ = state.tx().send(Message::Pong(p));
1608 None
1609 }
1610 _ => None,
1611 }
1612 }
1613 fn notify_speaking(&self, tx: &UnboundedSender<Message>, ssrc: u32, speaking: bool) {
1614 let msg = protocol::GatewayPayload {
1615 op: protocol::OpCode::Speaking as u8,
1616 seq: None,
1617 d: serde_json::json!({
1618 "speaking": if speaking { 1 } else { 0 },
1619 "delay": 0,
1620 "ssrc": ssrc
1621 }),
1622 };
1623 if let Ok(json) = serde_json::to_string(&msg) {
1624 let _ = tx.send(Message::Text(json.into()));
1625 }
1626 }
1627 fn emit_close(&self, code: u16, reason: String) {
1628 if let Some(tx) = &self.event_tx {
1629 let _ = tx.send(LavendeEvent::WebSocketClosed {
1630 guild_id: self.guild_id.clone(),
1631 code,
1632 reason,
1633 by_remote: true,
1634 });
1635 }
1636 }
1637 }
1638}
1639pub mod udp_link {
1640 use crate::{
1641 common::types::AnyResult,
1642 gateway::{
1643 constants::{
1644 RTP_OPUS_PAYLOAD_TYPE, RTP_TIMESTAMP_STEP, RTP_VERSION_BYTE,
1645 UDP_PACKET_BUF_CAPACITY,
1646 },
1647 session::types::map_boxed_err,
1648 },
1649 };
1650 use davey::{AeadInPlace, Aes256Gcm, KeyInit};
1651 use serde::{Deserialize, Serialize};
1652 use std::{net::SocketAddr, sync::Arc};
1653 use tokio::net::UdpSocket;
1654 use xsalsa20poly1305::XSalsa20Poly1305;
1655 pub struct UDPVoiceTransport {
1656 socket: Arc<UdpSocket>,
1657 address: SocketAddr,
1658 pub ssrc: u32,
1659 pub crypto: CryptoBackend,
1660 pub rtp: RtpState,
1661 pub buffer: Vec<u8>,
1662 }
1663 pub enum CryptoBackend {
1664 XSalsa20Poly1305(Box<XSalsa20Poly1305>),
1665 Aes256Gcm(Box<Aes256Gcm>),
1666 }
1667 #[derive(Debug, Clone, Copy, Serialize, Deserialize)]
1668 pub struct RtpState {
1669 pub sequence: u16,
1670 pub timestamp: u32,
1671 pub nonce: u32,
1672 }
1673 impl UDPVoiceTransport {
1674 pub fn new(
1675 socket: Arc<UdpSocket>,
1676 address: SocketAddr,
1677 ssrc: u32,
1678 secret_key: [u8; 32],
1679 mode: &str,
1680 rtp_state: Option<RtpState>,
1681 ) -> AnyResult<Self> {
1682 let crypto = match mode {
1683 "aead_aes256_gcm_rtpsize" => {
1684 CryptoBackend::Aes256Gcm(Box::new(Aes256Gcm::new(&secret_key.into())))
1685 }
1686 _ => CryptoBackend::XSalsa20Poly1305(Box::new(XSalsa20Poly1305::new(
1687 &secret_key.into(),
1688 ))),
1689 };
1690 Ok(Self {
1691 socket,
1692 address,
1693 ssrc,
1694 crypto,
1695 rtp: rtp_state.unwrap_or_else(RtpState::randomize),
1696 buffer: Vec::with_capacity(UDP_PACKET_BUF_CAPACITY),
1697 })
1698 }
1699 pub async fn send_keepalive(&self, counter: u32) -> AnyResult<()> {
1700 let payload = counter.to_be_bytes();
1701 self.socket.send_to(&payload, self.address).await?;
1702 Ok(())
1703 }
1704 pub async fn transmit_opus(&mut self, opus_data: &[u8]) -> AnyResult<()> {
1705 let (seq, ts, nonce_val) = self.rtp.next();
1706 let mut header = [0u8; 12];
1707 header[0] = RTP_VERSION_BYTE;
1708 header[1] = RTP_OPUS_PAYLOAD_TYPE;
1709 header[2..4].copy_from_slice(&seq.to_be_bytes());
1710 header[4..8].copy_from_slice(&ts.to_be_bytes());
1711 header[8..12].copy_from_slice(&self.ssrc.to_be_bytes());
1712 self.buffer.clear();
1713 self.buffer.extend_from_slice(&header);
1714 self.buffer.extend_from_slice(opus_data);
1715 match &self.crypto {
1716 CryptoBackend::XSalsa20Poly1305(cipher) => {
1717 let mut nonce = [0u8; 24];
1718 nonce[0..12].copy_from_slice(&header);
1719 let tag = cipher
1720 .encrypt_in_place_detached(&nonce.into(), &header, &mut self.buffer[12..])
1721 .map_err(|e| map_boxed_err(format!("XSalsa20 error: {e:?}")))?;
1722 self.buffer.extend_from_slice(&tag);
1723 }
1724 CryptoBackend::Aes256Gcm(cipher) => {
1725 let mut nonce = [0u8; 12];
1726 nonce[0..4].copy_from_slice(&nonce_val.to_be_bytes());
1727 let tag = cipher
1728 .encrypt_in_place_detached(&nonce.into(), &header, &mut self.buffer[12..])
1729 .map_err(|e| map_boxed_err(format!("AES-GCM error: {e:?}")))?;
1730 self.buffer.extend_from_slice(&tag);
1731 self.buffer.extend_from_slice(&nonce_val.to_be_bytes());
1732 }
1733 }
1734 self.socket.send_to(&self.buffer, self.address).await?;
1735 Ok(())
1736 }
1737 }
1738 impl RtpState {
1739 fn randomize() -> Self {
1740 Self {
1741 sequence: rand::random(),
1742 timestamp: rand::random(),
1743 nonce: rand::random(),
1744 }
1745 }
1746 fn next(&mut self) -> (u16, u32, u32) {
1747 let seq = self.sequence;
1748 let ts = self.timestamp;
1749 let n = self.nonce;
1750 self.sequence = self.sequence.wrapping_add(1);
1751 self.timestamp = self.timestamp.wrapping_add(RTP_TIMESTAMP_STEP);
1752 self.nonce = self.nonce.wrapping_add(1);
1753 (seq, ts, n)
1754 }
1755 }
1756}
1757pub mod encryption {
1758 use crate::{
1759 common::types::{AnyError, AnyResult, ChannelId, UserId},
1760 gateway::{
1761 constants::{DAVE_INITIAL_VERSION, MAX_PENDING_PROPOSALS, SILENCE_FRAME},
1762 session::types::map_boxed_err,
1763 },
1764 };
1765 use davey::{DaveSession, ProposalsOperationType};
1766 use std::{
1767 collections::{HashMap, HashSet},
1768 num::NonZeroU16,
1769 };
1770 use tracing::{debug, trace, warn};
1771 const DAVE_MIN_VERSION: NonZeroU16 = match NonZeroU16::new(DAVE_INITIAL_VERSION) {
1772 Some(v) => v,
1773 None => unreachable!(),
1774 };
1775 pub struct DaveHandler {
1776 session: Option<DaveSession>,
1777 user_id: UserId,
1778 channel_id: ChannelId,
1779 protocol_version: u16,
1780 pending_transitions: HashMap<u16, u16>,
1781 external_sender_set: bool,
1782 saved_external_sender: Option<Vec<u8>>,
1783 pending_proposals: Vec<Vec<u8>>,
1784 pending_handshake: Vec<(Vec<u8>, bool)>,
1785 was_ready: bool,
1786 recognized_users: HashSet<UserId>,
1787 cached_user_ids: Vec<u64>,
1788 }
1789 impl DaveHandler {
1790 pub fn new(user_id: UserId, channel_id: ChannelId) -> Self {
1791 let mut recognized_users = HashSet::new();
1792 recognized_users.insert(user_id);
1793 Self {
1794 session: None,
1795 user_id,
1796 channel_id,
1797 protocol_version: 0,
1798 pending_transitions: HashMap::new(),
1799 external_sender_set: false,
1800 saved_external_sender: None,
1801 pending_proposals: Vec::new(),
1802 pending_handshake: Vec::new(),
1803 was_ready: false,
1804 recognized_users,
1805 cached_user_ids: vec![user_id.0],
1806 }
1807 }
1808 pub fn add_users(&mut self, uids: &[u64]) {
1809 for &uid in uids {
1810 self.recognized_users.insert(UserId(uid));
1811 }
1812 self.update_user_cache();
1813 debug!("DAVE adding users: {:?}", uids);
1814 }
1815 pub fn remove_user(&mut self, uid: u64) {
1816 if self.recognized_users.remove(&UserId(uid)) {
1817 self.update_user_cache();
1818 }
1819 debug!("DAVE removing user: {}", uid);
1820 }
1821 fn update_user_cache(&mut self) {
1822 self.cached_user_ids.clear();
1823 self.cached_user_ids
1824 .extend(self.recognized_users.iter().map(|u| u.0));
1825 self.cached_user_ids.sort_unstable();
1826 }
1827 pub fn protocol_version(&self) -> u16 {
1828 self.protocol_version
1829 }
1830 pub fn set_protocol_version(&mut self, version: u16) {
1831 self.protocol_version = version;
1832 }
1833 pub fn setup_session(&mut self, version: u16) -> AnyResult<Vec<u8>> {
1834 if version == 0 {
1835 self.reset();
1836 return Ok(Vec::new());
1837 }
1838 let nz_version = NonZeroU16::new(version).unwrap_or(DAVE_MIN_VERSION);
1839 if let Some(s) = &mut self.session {
1840 s.reinit(nz_version, self.user_id.0, self.channel_id.0, None)
1841 .map_err(map_boxed_err)?;
1842 } else {
1843 let session = DaveSession::new(nz_version, self.user_id.0, self.channel_id.0, None)
1844 .map_err(map_boxed_err)?;
1845 self.session = Some(session);
1846 }
1847 let session = self
1848 .session
1849 .as_mut()
1850 .ok_or_else(|| map_boxed_err("DAVE session initialization failed"))?;
1851 self.protocol_version = version;
1852 self.external_sender_set = false;
1853 self.pending_proposals.clear();
1854 self.pending_handshake.clear();
1855 self.was_ready = false;
1856 debug!("DAVE session setup (v{})", version);
1857 let key_package = session.create_key_package().map_err(map_boxed_err)?;
1858 if let Some(saved) = self.saved_external_sender.as_deref()
1859 && let Some(sess) = &mut self.session
1860 {
1861 match sess.set_external_sender(saved) {
1862 Ok(()) => {
1863 self.external_sender_set = true;
1864 debug!("DAVE re-applied saved external sender after epoch reset");
1865 }
1866 Err(e) => {
1867 warn!("DAVE failed to re-apply saved external sender: {e}");
1868 self.saved_external_sender = None;
1869 }
1870 }
1871 }
1872 Ok(key_package)
1873 }
1874 pub fn reset(&mut self) {
1875 self.protocol_version = 0;
1876 self.pending_transitions.clear();
1877 self.external_sender_set = false;
1878 self.saved_external_sender = None;
1879 self.pending_proposals.clear();
1880 self.pending_handshake.clear();
1881 self.was_ready = false;
1882 self.session = None;
1883 debug!("DAVE session reset to plaintext");
1884 }
1885 pub fn prepare_transition(&mut self, transition_id: u16, protocol_version: u16) -> bool {
1886 self.pending_transitions
1887 .insert(transition_id, protocol_version);
1888 if transition_id == 0 {
1889 self.execute_transition(0);
1890 return false;
1891 }
1892 true
1893 }
1894 pub fn execute_transition(&mut self, transition_id: u16) {
1895 if let Some(next_version) = self.pending_transitions.remove(&transition_id) {
1896 self.protocol_version = next_version;
1897 trace!(
1898 "DAVE transition {} executed (v{})",
1899 transition_id, next_version
1900 );
1901 }
1902 }
1903 pub fn prepare_epoch(&mut self, epoch: u64, protocol_version: u16) -> Option<Vec<u8>> {
1904 if epoch == 1 {
1905 match self.setup_session(protocol_version) {
1906 Ok(kp) => return Some(kp),
1907 Err(e) => warn!("DAVE prepare_epoch setup failed: {e}"),
1908 }
1909 }
1910 None
1911 }
1912 pub fn process_external_sender(&mut self, data: &[u8]) -> AnyResult<Vec<Vec<u8>>> {
1913 let mut responses = Vec::new();
1914 if let Some(session) = &mut self.session {
1915 session.set_external_sender(data).map_err(map_boxed_err)?;
1916 self.external_sender_set = true;
1917 self.saved_external_sender = Some(data.to_vec());
1918 if !self.pending_proposals.is_empty() {
1919 debug!(
1920 "DAVE processing {} buffered proposals",
1921 self.pending_proposals.len()
1922 );
1923 for prop_data in std::mem::take(&mut self.pending_proposals) {
1924 if let Ok(Some(res)) =
1925 Self::do_process_proposals(session, &prop_data, &self.cached_user_ids)
1926 {
1927 responses.push(res);
1928 }
1929 }
1930 }
1931 if !self.pending_handshake.is_empty() {
1932 debug!(
1933 "DAVE processing {} buffered handshake messages",
1934 self.pending_handshake.len()
1935 );
1936 for (handshake_data, is_welcome) in std::mem::take(&mut self.pending_handshake)
1937 {
1938 if let Err(e) = self.do_process_handshake(&handshake_data, is_welcome) {
1939 warn!("DAVE buffered handshake processing failed: {e}");
1940 }
1941 }
1942 }
1943 }
1944 Ok(responses)
1945 }
1946 pub fn process_welcome(&mut self, data: &[u8]) -> AnyResult<u16> {
1947 self.process_handshake_message(data, true)
1948 }
1949 pub fn process_commit(&mut self, data: &[u8]) -> AnyResult<u16> {
1950 self.process_handshake_message(data, false)
1951 }
1952 fn process_handshake_message(&mut self, data: &[u8], is_welcome: bool) -> AnyResult<u16> {
1953 let tag = if is_welcome { "welcome" } else { "commit" };
1954 if data.len() < 2 {
1955 let msg = if is_welcome {
1956 "DAVE welcome"
1957 } else {
1958 "DAVE commit"
1959 };
1960 return Err(short_payload_err(msg));
1961 }
1962 let transition_id = u16::from_be_bytes([data[0], data[1]]);
1963 if !self.external_sender_set {
1964 if self.pending_handshake.len() < MAX_PENDING_PROPOSALS {
1965 debug!("DAVE buffering {tag} — external sender not set");
1966 self.pending_handshake.push((data.to_vec(), is_welcome));
1967 } else {
1968 warn!("DAVE handshake buffer full, dropping {tag}");
1969 }
1970 return Ok(transition_id);
1971 }
1972 self.do_process_handshake(data, is_welcome)?;
1973 Ok(transition_id)
1974 }
1975 fn do_process_handshake(&mut self, data: &[u8], is_welcome: bool) -> AnyResult<()> {
1976 let transition_id = u16::from_be_bytes([data[0], data[1]]);
1977 if let Some(session) = &mut self.session {
1978 if is_welcome {
1979 session.process_welcome(&data[2..]).map_err(map_boxed_err)?;
1980 } else {
1981 session.process_commit(&data[2..]).map_err(map_boxed_err)?;
1982 }
1983 if transition_id != 0 {
1984 self.pending_transitions
1985 .insert(transition_id, self.protocol_version);
1986 }
1987 debug!(
1988 "DAVE {} processed (tid {})",
1989 if is_welcome { "welcome" } else { "commit" },
1990 transition_id
1991 );
1992 }
1993 Ok(())
1994 }
1995 pub fn process_proposals(&mut self, data: &[u8]) -> AnyResult<Option<Vec<u8>>> {
1996 if data.is_empty() {
1997 return Err(short_payload_err("DAVE proposals"));
1998 }
1999 if !self.external_sender_set {
2000 if self.pending_proposals.len() < MAX_PENDING_PROPOSALS {
2001 debug!("DAVE buffering proposal — external sender not set");
2002 self.pending_proposals.push(data.to_vec());
2003 } else {
2004 warn!("DAVE proposal buffer full, dropping proposal");
2005 }
2006 return Ok(None);
2007 }
2008 let session = match &mut self.session {
2009 Some(s) => s,
2010 None => return Ok(None),
2011 };
2012 Self::do_process_proposals(session, data, &self.cached_user_ids)
2013 }
2014 fn do_process_proposals(
2015 session: &mut DaveSession,
2016 data: &[u8],
2017 user_ids: &[u64],
2018 ) -> AnyResult<Option<Vec<u8>>> {
2019 let op_type = match data[0] {
2020 0 => ProposalsOperationType::APPEND,
2021 1 => ProposalsOperationType::REVOKE,
2022 raw => return Err(map_boxed_err(format!("Unknown DAVE proposals op: {raw}"))),
2023 };
2024 let result = session
2025 .process_proposals(op_type, &data[1..], Some(user_ids))
2026 .map_err(map_boxed_err)?;
2027 if let Some(cw) = result {
2028 let mut out = cw.commit;
2029 if let Some(w) = cw.welcome {
2030 out.extend_from_slice(&w);
2031 }
2032 return Ok(Some(out));
2033 }
2034 Ok(None)
2035 }
2036 pub fn encrypt_opus(&mut self, packet: &[u8]) -> AnyResult<Vec<u8>> {
2037 if packet == SILENCE_FRAME || self.protocol_version == 0 {
2038 return Ok(packet.to_vec());
2039 }
2040 if let Some(session) = &mut self.session {
2041 let is_ready = session.is_ready();
2042 if is_ready != self.was_ready {
2043 if is_ready {
2044 debug!("DAVE session (v{}) is READY", self.protocol_version);
2045 } else {
2046 warn!("DAVE session (v{}) LOST readiness", self.protocol_version);
2047 }
2048 self.was_ready = is_ready;
2049 }
2050 if is_ready {
2051 return session
2052 .encrypt_opus(packet)
2053 .map(|c| c.into_owned())
2054 .map_err(map_boxed_err);
2055 }
2056 }
2057 Ok(packet.to_vec())
2058 }
2059 pub fn voice_privacy_code(&self) -> Option<String> {
2060 self.session
2061 .as_ref()
2062 .and_then(|s| s.voice_privacy_code().map(|c| c.to_string()))
2063 }
2064 }
2065 #[inline]
2066 fn short_payload_err(context: &str) -> AnyError {
2067 map_boxed_err(format!("Invalid {context} payload: too short"))
2068 }
2069}
2070pub mod engine {
2071 use crate::{audio::Mixer, common::types::Shared, gateway::constants::DEFAULT_SAMPLE_RATE};
2072 use tokio::sync::Mutex;
2073 pub struct VoiceEngine {
2074 pub mixer: Shared<Mixer>,
2075 pub dave: Option<Shared<crate::gateway::DaveHandler>>,
2076 }
2077 impl VoiceEngine {
2078 pub fn new() -> Self {
2079 Self {
2080 mixer: Shared::new(Mutex::new(Mixer::new(DEFAULT_SAMPLE_RATE))),
2081 dave: None,
2082 }
2083 }
2084 }
2085 impl Default for VoiceEngine {
2086 fn default() -> Self {
2087 Self::new()
2088 }
2089 }
2090}
2091pub use encryption::DaveHandler;
2092pub use engine::VoiceEngine;
2093pub use session::{VoiceGateway, VoiceGatewayConfig};