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 = 5;
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::Skip);
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 return Ok(());
653 }
654 let has_ts = {
655 if let Some(mut filters) = try_lock_yield!(self.config.filter_chain) {
656 filters.process(pcm);
657 filters.has_timescale()
658 } else {
659 false
660 }
661 };
662 if !has_ts {
663 if has_input {
664 self.config.frames_sent.fetch_add(1, Ordering::Relaxed);
665 } else {
666 self.config.frames_nulled.fetch_add(1, Ordering::Relaxed);
667 }
668 if self.speaking_holdoff {
669 self.speaking_holdoff = false;
670 self.send_silence().await?;
671 }
672 return self.send_pcm(encoder, pcm, opus).await;
673 }
674 let filled_on_silence = {
675 if let Some(mut filters) = try_lock_yield!(self.config.filter_chain) {
676 !has_input && filters.fill_frame(ts_pcm)
677 } else {
678 false
679 }
680 };
681 if !has_input && !filled_on_silence {
682 break;
683 }
684 }
685 Ok(())
686 }
687 fn set_speaking(&mut self, speaking: bool) {
688 if speaking != self.is_speaking {
689 self.is_speaking = speaking;
690 let _ = self.config.speaking_tx.send(speaking);
691 if speaking {
692 self.speaking_holdoff = true;
693 }
694 }
695 }
696 async fn send_pcm(
697 &mut self,
698 encoder: &mut Encoder,
699 pcm: &[i16],
700 opus: &mut [u8],
701 ) -> Result<(), GatewayError> {
702 let size = encoder.encode(pcm, opus).unwrap_or_else(|e| {
703 error!("Opus encode failed: {e}");
704 0
705 });
706 if size > 0 {
707 self.send_raw(&opus[..size]).await?;
708 } else {
709 self.send_silence().await?;
710 }
711 Ok(())
712 }
713 async fn send_silence(&mut self) -> Result<(), GatewayError> {
714 self.config.frames_nulled.fetch_add(1, Ordering::Relaxed);
715 self.send_raw(&SILENCE_FRAME).await
716 }
717 async fn send_raw(&mut self, data: &[u8]) -> Result<(), GatewayError> {
718 let mut dave = self.config.dave.lock().await;
719 let encrypted = dave
720 .encrypt_opus(data)
721 .map_err(|e| GatewayError::Encryption(e.to_string()))?;
722 drop(dave);
723 self.transport.transmit_opus(&encrypted).await?;
724 self.last_tx_time = Instant::now();
725 Ok(())
726 }
727 fn reset_timers(&mut self) {
728 self.active_silence = MAX_SILENCE_FRAMES;
729 }
730 }
731 }
732 pub mod handler {
733 use super::{
734 VoiceGateway,
735 backoff::Backoff,
736 heartbeat::HeartbeatTracker,
737 protocol::{GatewayPayload, OpCode},
738 types::{GatewayError, PersistentSessionState, SessionOutcome},
739 voice::{SpeakConfig, discover_ip, speak_loop},
740 };
741 use crate::{
742 common::types::{Shared, UserId},
743 gateway::{
744 DaveHandler,
745 constants::{DAVE_INITIAL_VERSION, DEFAULT_VOICE_MODE},
746 },
747 };
748 use serde_json::Value;
749 use std::{
750 collections::HashSet,
751 net::SocketAddr,
752 sync::{
753 Arc,
754 atomic::{AtomicI64, Ordering},
755 },
756 };
757 use tokio::sync::mpsc::UnboundedSender;
758 use tokio_tungstenite::tungstenite::protocol::Message;
759 use tokio_util::sync::CancellationToken;
760 use tracing::{debug, error, trace, warn};
761 use uuid::Uuid;
762 pub struct SessionState<'a> {
763 gateway: &'a VoiceGateway,
764 tx: UnboundedSender<Message>,
765 seq_ack: Arc<AtomicI64>,
766 ssrc: u32,
767 udp_addr: Option<SocketAddr>,
768 selected_mode: String,
769 connected_users: HashSet<UserId>,
770 udp_socket: Arc<tokio::net::UdpSocket>,
771 dave: Shared<DaveHandler>,
772 heartbeat: HeartbeatTracker,
773 heartbeat_handle: Option<tokio::task::JoinHandle<()>>,
774 conn_token: CancellationToken,
775 speaking_tx: Option<UnboundedSender<bool>>,
776 session_key: Option<[u8; 32]>,
777 speak_task: Option<tokio::task::JoinHandle<()>>,
778 persistent_state: Arc<tokio::sync::Mutex<PersistentSessionState>>,
779 backoff: &'a mut Backoff,
780 }
781 impl<'a> SessionState<'a> {
782 pub async fn new(
783 gateway: &'a VoiceGateway,
784 tx: UnboundedSender<Message>,
785 seq_ack: Arc<AtomicI64>,
786 conn_token: CancellationToken,
787 persistent_state: Arc<tokio::sync::Mutex<PersistentSessionState>>,
788 backoff: &'a mut Backoff,
789 ) -> Result<Self, GatewayError> {
790 let mut socket_guard = gateway.udp_socket.lock().await;
791 let udp_socket = if let Some(existing) = &*socket_guard {
792 existing.clone()
793 } else {
794 let udp = std::net::UdpSocket::bind("0.0.0.0:0")?;
795 udp.set_nonblocking(true)?;
796 let socket = Arc::new(tokio::net::UdpSocket::from_std(udp)?);
797 *socket_guard = Some(socket.clone());
798 socket
799 };
800 Ok(Self {
801 gateway,
802 tx,
803 seq_ack,
804 ssrc: 0,
805 udp_addr: None,
806 selected_mode: DEFAULT_VOICE_MODE.to_string(),
807 connected_users: HashSet::from([gateway.user_id]),
808 udp_socket,
809 dave: gateway.dave.clone(),
810 heartbeat: HeartbeatTracker::new(),
811 heartbeat_handle: None,
812 conn_token,
813 speaking_tx: None,
814 session_key: None,
815 speak_task: None,
816 persistent_state,
817 backoff,
818 })
819 }
820 pub fn set_speaking_tx(&mut self, tx: UnboundedSender<bool>) {
821 self.speaking_tx = Some(tx);
822 }
823 pub fn ssrc(&self) -> u32 {
824 self.ssrc
825 }
826 pub fn tx(&self) -> &UnboundedSender<Message> {
827 &self.tx
828 }
829 pub fn attempt(&self) -> u32 {
830 self.backoff.attempt()
831 }
832 pub fn has_heartbeat(&self) -> bool {
833 self.heartbeat_handle.is_some()
834 }
835 pub async fn handle_text(&mut self, text: String) -> Option<SessionOutcome> {
836 let payload: GatewayPayload = match serde_json::from_str(&text) {
837 Ok(p) => p,
838 Err(e) => {
839 warn!("[{}] JSON Parse error: {e}", self.gateway.guild_id);
840 return None;
841 }
842 };
843 if let Some(seq) = payload.seq {
844 self.seq_ack.store(seq as i64, Ordering::Relaxed);
845 }
846 let op = OpCode::from(payload.op);
847 trace!(
848 "[{}] RX OP: {:?} (op={})",
849 self.gateway.guild_id, op, payload.op
850 );
851 match op {
852 OpCode::Hello => self.on_hello(payload.d),
853 OpCode::Ready => self.on_ready(payload.d).await,
854 OpCode::SessionDescription => self.on_session_description(payload.d).await,
855 OpCode::HeartbeatAck => self.on_heartbeat_ack(payload.d),
856 OpCode::Resumed => self.on_resumed().await,
857 OpCode::ClientConnect => self.on_user_connect(payload.d).await,
858 OpCode::ClientDisconnect => self.on_user_disconnect(payload.d).await,
859 OpCode::VoiceBackendVersion => {
860 debug!(
861 "[{}] Voice Backend Version: {:?}",
862 self.gateway.guild_id, payload.d
863 );
864 None
865 }
866 OpCode::MediaSinkWants => {
867 debug!(
868 "[{}] Media Sink Wants: {:?}",
869 self.gateway.guild_id, payload.d
870 );
871 None
872 }
873 OpCode::DavePrepareTransition => {
874 self.on_dave_prepare_transition(payload.d).await
875 }
876 OpCode::DaveExecuteTransition => {
877 self.on_dave_execute_transition(payload.d).await
878 }
879 OpCode::DavePrepareEpoch => self.on_dave_prepare_epoch(payload.d).await,
880 OpCode::MlsAnnounceCommitTransition => self.on_mls_transition(payload.d).await,
881 OpCode::MlsInvalidCommitWelcome => {
882 warn!(
883 "[{}] DAVE MLS Invalid Commit Welcome received, resetting session",
884 self.gateway.guild_id
885 );
886 self.reset_dave(0).await;
887 None
888 }
889 OpCode::NoRoute => {
890 warn!(
891 "[{}] No Route received: {:?}",
892 self.gateway.guild_id, payload.d
893 );
894 None
895 }
896 OpCode::Speaking
897 | OpCode::Video
898 | OpCode::Codecs
899 | OpCode::UserFlags
900 | OpCode::VoicePlatform => None,
901 _ => None,
902 }
903 }
904 pub async fn handle_binary(&mut self, bin: Vec<u8>) {
905 if bin.len() < 3 {
906 return;
907 }
908 let seq = u16::from_be_bytes([bin[0], bin[1]]);
909 let op = bin[2];
910 let data = &bin[3..];
911 self.seq_ack.store(seq as i64, Ordering::Relaxed);
912 let mut dave = self.dave.lock().await;
913 match op {
914 25 => {
915 if let Ok(res) = dave.process_external_sender(data) {
916 for r in res {
917 self.send_binary(28, &r);
918 }
919 }
920 }
921 27 => match dave.process_proposals(data) {
922 Ok(Some(cw)) => self.send_binary(28, &cw),
923 Err(e) => {
924 warn!("[{}] DAVE proposals failed: {e}", self.gateway.guild_id);
925 self.reset_dave_locked(&mut dave, 0).await;
926 }
927 _ => {}
928 },
929 29 | 30 => {
930 let res = if op == 30 {
931 dave.process_welcome(data)
932 } else {
933 dave.process_commit(data)
934 };
935 match res {
936 Ok(tid) if tid != 0 => {
937 self.send_json(23, serde_json::json!({ "transition_id": tid }))
938 }
939 Err(e) => {
940 let tid = if data.len() >= 2 {
941 u16::from_be_bytes([data[0], data[1]])
942 } else {
943 0
944 };
945 warn!(
946 "[{}] DAVE {} failed (tid {tid}): {e}",
947 self.gateway.guild_id,
948 if op == 30 { "welcome" } else { "commit" }
949 );
950 self.reset_dave_locked(&mut dave, tid).await;
951 }
952 _ => {}
953 }
954 }
955 _ => {}
956 }
957 }
958 fn on_hello(&mut self, d: Value) -> Option<SessionOutcome> {
959 let interval = d["heartbeat_interval"].as_u64().unwrap_or(30_000);
960 if let Some(h) = self.heartbeat_handle.take() {
961 debug!(
962 "[{}] Restarting heartbeat on HELLO (was already running)",
963 self.gateway.guild_id
964 );
965 h.abort();
966 }
967 trace!(
968 "[{}] Heartbeat interval: {interval}ms",
969 self.gateway.guild_id
970 );
971 self.heartbeat_handle = Some(self.heartbeat.spawn(
972 self.tx.clone(),
973 self.seq_ack.clone(),
974 self.conn_token.clone(),
975 interval,
976 ));
977 None
978 }
979 async fn on_ready(&mut self, d: Value) -> Option<SessionOutcome> {
980 let ssrc = d["ssrc"].as_u64();
981 let ip = d["ip"].as_str();
982 let port = d["port"].as_u64();
983 match (ssrc, ip, port) {
984 (Some(ssrc), Some(ip), Some(port)) if port <= 65535 => {
985 self.ssrc = ssrc as u32;
986 let addr_str = format!("{ip}:{port}");
987 match addr_str.parse::<SocketAddr>() {
988 Ok(addr) => self.udp_addr = Some(addr),
989 Err(_) => {
990 error!(
991 "[{}] Invalid READY address: {addr_str}",
992 self.gateway.guild_id
993 );
994 return Some(SessionOutcome::Reconnect);
995 }
996 }
997 }
998 _ => {
999 error!("[{}] Malformed READY payload", self.gateway.guild_id);
1000 return Some(SessionOutcome::Reconnect);
1001 }
1002 }
1003 if let Some(modes) = d["modes"].as_array() {
1004 let pref = ["aead_aes256_gcm_rtpsize", "xsalsa20_poly1305"];
1005 if let Some(m) = pref
1006 .iter()
1007 .find(|&&p| modes.iter().any(|m| m.as_str() == Some(p)))
1008 {
1009 self.selected_mode = (*m).to_owned();
1010 }
1011 }
1012 debug!(
1013 "[{}] Ready: ssrc={}, mode={}",
1014 self.gateway.guild_id, self.ssrc, self.selected_mode
1015 );
1016 {
1017 let mut state = self.persistent_state.lock().await;
1018 state.ssrc = self.ssrc;
1019 state.selected_mode = Some(self.selected_mode.clone());
1020 }
1021 if self.gateway.channel_id.0 > 0 {
1022 let ver = d["dave_protocol_version"]
1023 .as_u64()
1024 .unwrap_or(DAVE_INITIAL_VERSION as u64)
1025 as u16;
1026 let mut dave = self.dave.lock().await;
1027 if ver == 0 {
1028 dave.reset();
1029 } else {
1030 dave.set_protocol_version(ver);
1031 if let Ok(kp) = dave.setup_session(ver) {
1032 self.send_binary(26, &kp);
1033 }
1034 }
1035 }
1036 let target_addr = match self.udp_addr {
1037 Some(a) => a,
1038 None => return Some(SessionOutcome::Reconnect),
1039 };
1040 match discover_ip(&self.udp_socket, target_addr, self.ssrc).await {
1041 Ok((my_ip, my_port)) => {
1042 self.send_json(OpCode::SelectProtocol as u8, serde_json::json!({
1043 "protocol": "udp",
1044 "rtc_connection_id": Uuid::new_v4().to_string(),
1045 "codecs": [{"name": "opus", "type": "audio", "priority": 1000, "payload_type": 120}],
1046 "data": { "address": my_ip, "port": my_port, "mode": self.selected_mode },
1047 "address": my_ip,
1048 "port": my_port,
1049 "mode": self.selected_mode
1050 }));
1051 self.send_json(
1052 OpCode::Video as u8,
1053 serde_json::json!({"audio_ssrc": self.ssrc, "video_ssrc": 0, "rtx_ssrc": 0}),
1054 );
1055 self.send_json(
1056 OpCode::Speaking as u8,
1057 serde_json::json!({"speaking": 0, "delay": 0, "ssrc": self.ssrc}),
1058 );
1059 }
1060 Err(e) => {
1061 error!("[{}] IP discovery failed: {e}", self.gateway.guild_id);
1062 return Some(SessionOutcome::Reconnect);
1063 }
1064 }
1065 self.backoff.reset();
1066 None
1067 }
1068 async fn on_session_description(&mut self, d: Value) -> Option<SessionOutcome> {
1069 let ka = match d["secret_key"].as_array() {
1070 Some(a) if a.len() == 32 => a,
1071 _ => {
1072 error!(
1073 "[{}] Invalid or missing secret_key in VOICE_READY",
1074 self.gateway.guild_id
1075 );
1076 return Some(SessionOutcome::Reconnect);
1077 }
1078 };
1079 let mut key = [0u8; 32];
1080 for (i, v) in ka.iter().enumerate() {
1081 if let Some(val) = v.as_u64()
1082 && val <= 255
1083 {
1084 key[i] = val as u8;
1085 continue;
1086 }
1087 error!(
1088 "[{}] Invalid secret_key byte at index {i}",
1089 self.gateway.guild_id
1090 );
1091 return Some(SessionOutcome::Reconnect);
1092 }
1093 self.session_key = Some(key);
1094 let addr = match self.udp_addr {
1095 Some(a) => a,
1096 None => return Some(SessionOutcome::Reconnect),
1097 };
1098 {
1099 let mut state = self.persistent_state.lock().await;
1100 state.udp_addr = Some(addr);
1101 state.session_key = Some(key);
1102 state.ssrc = self.ssrc;
1103 state.selected_mode = Some(self.selected_mode.clone());
1104 }
1105 self.start_voice(addr, key).await;
1106 if self.gateway.channel_id.0 > 0 {
1107 let protocol_version = d["dave_protocol_version"]
1108 .as_u64()
1109 .unwrap_or(DAVE_INITIAL_VERSION as u64)
1110 as u16;
1111 let mls_group_id = d["mls_group_id"].as_u64().unwrap_or(0);
1112 let mut dave = self.dave.lock().await;
1113 if protocol_version > 0 {
1114 dave.set_protocol_version(protocol_version);
1115 if let Ok(kp) = dave.setup_session(protocol_version) {
1116 self.send_binary(26, &kp);
1117 }
1118 } else {
1119 dave.reset();
1120 }
1121 debug!(
1122 "DAVE setup context: protocol_version={}, mls_group_id={}",
1123 protocol_version, mls_group_id
1124 );
1125 }
1126 self.backoff.reset();
1127 None
1128 }
1129 async fn on_resumed(&mut self) -> Option<SessionOutcome> {
1130 debug!("[{}] Resumed", self.gateway.guild_id);
1131 let (addr, key, ssrc, mode) = {
1132 let state = self.persistent_state.lock().await;
1133 (
1134 state.udp_addr,
1135 state.session_key,
1136 state.ssrc,
1137 state.selected_mode.clone(),
1138 )
1139 };
1140 match (addr, key) {
1141 (Some(addr), Some(key)) => {
1142 self.udp_addr = Some(addr);
1143 self.session_key = Some(key);
1144 self.ssrc = ssrc;
1145 if let Some(m) = mode {
1146 self.selected_mode = m;
1147 }
1148 if let Some(task) = &self.speak_task
1149 && task.is_finished()
1150 {
1151 self.speak_task = None;
1152 }
1153 if self.speak_task.is_some() {
1154 debug!(
1155 "[{}] Keeping existing voice loop alive across resume",
1156 self.gateway.guild_id
1157 );
1158 } else {
1159 debug!(
1160 "[{}] Starting voice loop after resume (task was dead or missing)",
1161 self.gateway.guild_id
1162 );
1163 self.start_voice(addr, key).await;
1164 }
1165 }
1166 _ => {
1167 warn!(
1168 "[{}] Resume failed: missing persistent state",
1169 self.gateway.guild_id
1170 );
1171 return Some(SessionOutcome::Identify);
1172 }
1173 }
1174 None
1175 }
1176 fn on_heartbeat_ack(&self, d: Value) -> Option<SessionOutcome> {
1177 let nonce = d["t"].as_u64().unwrap_or(0);
1178 if let Some(rtt) = self.heartbeat.validate_ack(nonce) {
1179 self.gateway.ping.store(rtt as i64, Ordering::Relaxed);
1180 self.heartbeat.missed_acks.store(0, Ordering::Relaxed);
1181 }
1182 None
1183 }
1184 async fn on_user_connect(&mut self, d: Value) -> Option<SessionOutcome> {
1185 if let Some(ids) = d["user_ids"].as_array() {
1186 let mut uids = Vec::new();
1187 for id in ids {
1188 if let Some(uid) = id.as_str().and_then(|s| s.parse::<u64>().ok()) {
1189 self.connected_users.insert(UserId(uid));
1190 uids.push(uid);
1191 }
1192 }
1193 if !uids.is_empty() {
1194 self.dave.lock().await.add_users(&uids);
1195 }
1196 }
1197 None
1198 }
1199 async fn on_user_disconnect(&mut self, d: Value) -> Option<SessionOutcome> {
1200 if let Some(uid) = d["user_id"].as_str().and_then(|s| s.parse::<u64>().ok()) {
1201 self.connected_users.remove(&UserId(uid));
1202 self.dave.lock().await.remove_user(uid);
1203 }
1204 None
1205 }
1206 async fn on_dave_prepare_transition(&mut self, d: Value) -> Option<SessionOutcome> {
1207 let tid = d["transition_id"].as_u64().unwrap_or(0) as u16;
1208 let ver = d["protocol_version"].as_u64().unwrap_or(0) as u16;
1209 debug!(
1210 "[{}] DAVE Prepare Transition: id={}, version={}",
1211 self.gateway.guild_id, tid, ver
1212 );
1213 if self.dave.lock().await.prepare_transition(tid, ver) {
1214 debug!(
1215 "[{}] DAVE Transition Ready (tid={})",
1216 self.gateway.guild_id, tid
1217 );
1218 self.send_json(23, serde_json::json!({ "transition_id": tid }));
1219 }
1220 None
1221 }
1222 async fn on_dave_execute_transition(&mut self, d: Value) -> Option<SessionOutcome> {
1223 let tid = d["transition_id"].as_u64().unwrap_or(0) as u16;
1224 debug!(
1225 "[{}] DAVE Execute Transition: id={}",
1226 self.gateway.guild_id, tid
1227 );
1228 self.dave.lock().await.execute_transition(tid);
1229 None
1230 }
1231 async fn on_dave_prepare_epoch(&mut self, d: Value) -> Option<SessionOutcome> {
1232 let epoch = d["epoch"].as_u64().unwrap_or(0);
1233 let ver = d["protocol_version"].as_u64().unwrap_or(0) as u16;
1234 debug!(
1235 "[{}] DAVE Prepare Epoch: epoch={}, version={}",
1236 self.gateway.guild_id, epoch, ver
1237 );
1238 if let Some(kp) = self.dave.lock().await.prepare_epoch(epoch, ver) {
1239 self.send_binary(26, &kp);
1240 }
1241 None
1242 }
1243 async fn on_mls_transition(&mut self, d: Value) -> Option<SessionOutcome> {
1244 let tid = d["transition_id"].as_u64().unwrap_or(0) as u16;
1245 debug!(
1246 "[{}] DAVE MLS Announce Commit Transition: tid={}",
1247 self.gateway.guild_id, tid
1248 );
1249 let ver = d["protocol_version"].as_u64().map(|v| v as u16);
1250 if let Some(v) = ver {
1251 let mut dave = self.dave.lock().await;
1252 if dave.prepare_transition(tid, v) && tid != 0 {
1253 self.send_json(23, serde_json::json!({ "transition_id": tid }));
1254 }
1255 }
1256 None
1257 }
1258 async fn start_voice(&mut self, addr: SocketAddr, key: [u8; 32]) {
1259 if let Some(t) = self.speak_task.take() {
1260 t.abort();
1261 }
1262 let speaking_tx = if let Some(tx) = &self.speaking_tx {
1263 tx.clone()
1264 } else {
1265 error!(
1266 "[{}] speaking_tx is missing, cannot start voice",
1267 self.gateway.guild_id
1268 );
1269 return;
1270 };
1271 let config = SpeakConfig {
1272 mixer: self.gateway.mixer.clone(),
1273 socket: self.udp_socket.clone(),
1274 addr,
1275 ssrc: self.ssrc,
1276 key,
1277 mode: self.selected_mode.clone(),
1278 dave: self.dave.clone(),
1279 filter_chain: self.gateway.filter_chain.clone(),
1280 frames_sent: self.gateway.frames_sent.clone(),
1281 frames_nulled: self.gateway.frames_nulled.clone(),
1282 cancel_token: self.conn_token.clone(),
1283 speaking_tx,
1284 persistent_state: self.persistent_state.clone(),
1285 };
1286 let guild_id = self.gateway.guild_id.clone();
1287 let conn_token = self.conn_token.clone();
1288 self.speak_task = Some(tokio::spawn(async move {
1289 if let Err(e) = speak_loop(config).await {
1290 error!("[{guild_id}] speak_loop failed: {e}");
1291 conn_token.cancel();
1292 }
1293 }));
1294 self.send_json(
1295 OpCode::Video as u8,
1296 serde_json::json!({"audio_ssrc": self.ssrc, "video_ssrc": 0, "rtx_ssrc": 0}),
1297 );
1298 self.send_json(
1299 OpCode::Speaking as u8,
1300 serde_json::json!({"speaking": 0, "delay": 0, "ssrc": self.ssrc}),
1301 );
1302 }
1303 async fn reset_dave(&self, tid: u16) {
1304 let mut dave = self.dave.lock().await;
1305 self.reset_dave_locked(&mut dave, tid).await;
1306 }
1307 async fn reset_dave_locked(&self, dave: &mut DaveHandler, tid: u16) {
1308 dave.reset();
1309 self.send_json(31, serde_json::json!({ "transition_id": tid }));
1310 if let Ok(kp) = dave.setup_session(DAVE_INITIAL_VERSION) {
1311 self.send_binary(26, &kp);
1312 }
1313 }
1314 fn send_json(&self, op: u8, d: Value) {
1315 match serde_json::to_string(&GatewayPayload { op, seq: None, d }) {
1316 Ok(json) => {
1317 let _ = self.tx.send(Message::Text(json.into()));
1318 }
1319 Err(e) => {
1320 warn!("[{}] JSON serialization failed: {e}", self.gateway.guild_id);
1321 }
1322 }
1323 }
1324 fn send_binary(&self, op: u8, payload: &[u8]) {
1325 let mut b = vec![op];
1326 b.extend_from_slice(payload);
1327 let _ = self.tx.send(Message::Binary(b.into()));
1328 }
1329 }
1330 impl<'a> Drop for SessionState<'a> {
1331 fn drop(&mut self) {
1332 if let Some(h) = self.heartbeat_handle.take() {
1333 h.abort();
1334 }
1335 if let Some(t) = self.speak_task.take() {
1336 t.abort();
1337 }
1338 }
1339 }
1340 }
1341 use self::{
1342 backoff::Backoff,
1343 policy::FailurePolicy,
1344 types::{GatewayError, PersistentSessionState, SessionOutcome},
1345 };
1346 use crate::{
1347 audio::{Mixer, filters::FilterChain},
1348 common::types::{ChannelId, GuildId, SessionId, Shared, UserId},
1349 gateway::constants::VOICE_GATEWAY_VERSION,
1350 protocol::RustalinkEvent,
1351 };
1352 use futures::{SinkExt, StreamExt};
1353 use std::sync::{
1354 Arc,
1355 atomic::{AtomicI64, Ordering},
1356 };
1357 use tokio::sync::mpsc::{UnboundedSender, unbounded_channel};
1358 use tokio_tungstenite::tungstenite::protocol::{Message, WebSocketConfig};
1359 use tokio_util::sync::CancellationToken;
1360 use tracing::{debug, error, warn};
1361 pub struct VoiceGateway {
1362 pub guild_id: GuildId,
1363 pub user_id: UserId,
1364 pub channel_id: ChannelId,
1365 session_id: SessionId,
1366 token: String,
1367 endpoint: String,
1368 pub mixer: Shared<Mixer>,
1369 pub filter_chain: Shared<FilterChain>,
1370 pub ping: Arc<AtomicI64>,
1371 event_tx: Option<UnboundedSender<RustalinkEvent>>,
1372 pub frames_sent: Arc<std::sync::atomic::AtomicU64>,
1373 pub frames_nulled: Arc<std::sync::atomic::AtomicU64>,
1374 pub udp_socket: Shared<Option<Arc<tokio::net::UdpSocket>>>,
1375 pub dave: Shared<crate::gateway::DaveHandler>,
1376 outer_token: CancellationToken,
1377 policy: FailurePolicy,
1378 }
1379 pub struct VoiceGatewayConfig {
1380 pub guild_id: GuildId,
1381 pub user_id: UserId,
1382 pub channel_id: ChannelId,
1383 pub session_id: SessionId,
1384 pub token: String,
1385 pub endpoint: String,
1386 pub mixer: Shared<Mixer>,
1387 pub filter_chain: Shared<FilterChain>,
1388 pub ping: Arc<AtomicI64>,
1389 pub event_tx: Option<UnboundedSender<RustalinkEvent>>,
1390 pub frames_sent: Arc<std::sync::atomic::AtomicU64>,
1391 pub frames_nulled: Arc<std::sync::atomic::AtomicU64>,
1392 }
1393 impl VoiceGateway {
1394 pub fn new(config: VoiceGatewayConfig) -> Self {
1395 Self {
1396 guild_id: config.guild_id,
1397 user_id: config.user_id,
1398 channel_id: config.channel_id,
1399 session_id: config.session_id,
1400 token: config.token,
1401 endpoint: config.endpoint,
1402 mixer: config.mixer,
1403 filter_chain: config.filter_chain,
1404 ping: config.ping,
1405 event_tx: config.event_tx,
1406 frames_sent: config.frames_sent,
1407 frames_nulled: config.frames_nulled,
1408 udp_socket: Arc::new(tokio::sync::Mutex::new(None)),
1409 dave: Arc::new(tokio::sync::Mutex::new(crate::gateway::DaveHandler::new(
1410 config.user_id,
1411 config.channel_id,
1412 ))),
1413 outer_token: CancellationToken::new(),
1414 policy: FailurePolicy::new(3),
1415 }
1416 }
1417 pub async fn run(self) -> Result<(), GatewayError> {
1418 let mut backoff = Backoff::new();
1419 let mut is_resume = false;
1420 let seq_ack = Arc::new(AtomicI64::new(-1));
1421 let persistent_state =
1422 Arc::new(tokio::sync::Mutex::new(PersistentSessionState::default()));
1423 while !self.outer_token.is_cancelled() {
1424 let attempt = backoff.attempt();
1425 match self
1426 .connect(
1427 is_resume,
1428 seq_ack.clone(),
1429 persistent_state.clone(),
1430 &mut backoff,
1431 )
1432 .await
1433 {
1434 Ok(SessionOutcome::Shutdown) => break,
1435 Ok(outcome) => {
1436 if backoff.is_exhausted() {
1437 warn!("[{}] Max attempts reached ({})", self.guild_id, attempt);
1438 break;
1439 }
1440 let delay = backoff.next_delay();
1441 is_resume = matches!(outcome, SessionOutcome::Reconnect);
1442 if !is_resume {
1443 seq_ack.store(-1, Ordering::Relaxed);
1444 *persistent_state.lock().await = PersistentSessionState::default();
1445 *self.udp_socket.lock().await = None;
1446 }
1447 debug!(
1448 "[{}] Retrying ({:?}) in {:?}",
1449 self.guild_id, outcome, delay
1450 );
1451 tokio::time::sleep(delay).await;
1452 }
1453 Err(e) => {
1454 if backoff.is_exhausted() {
1455 error!("[{}] Fatal connection error: {e}", self.guild_id);
1456 break;
1457 }
1458 let delay = backoff.next_delay();
1459 warn!(
1460 "[{}] Connection error: {e}. Retrying in {:?}",
1461 self.guild_id, delay
1462 );
1463 tokio::time::sleep(delay).await;
1464 is_resume = false;
1465 }
1466 }
1467 }
1468 Ok(())
1469 }
1470 async fn connect(
1471 &self,
1472 is_resume: bool,
1473 seq_ack: Arc<AtomicI64>,
1474 persistent_state: Arc<tokio::sync::Mutex<PersistentSessionState>>,
1475 backoff: &mut Backoff,
1476 ) -> Result<SessionOutcome, GatewayError> {
1477 let endpoint = if self.endpoint.ends_with(":80") {
1478 &self.endpoint[..self.endpoint.len() - 3]
1479 } else {
1480 &self.endpoint
1481 };
1482 let url = format!("wss://{}/?v={}", endpoint, VOICE_GATEWAY_VERSION);
1483 let mut config = WebSocketConfig::default();
1484 config.max_message_size = None;
1485 config.max_frame_size = None;
1486 let (ws_stream, _) =
1487 tokio_tungstenite::connect_async_with_config(&url, Some(config), true).await?;
1488 let (mut write, mut read) = ws_stream.split();
1489 let conn_token = CancellationToken::new();
1490 let write_token = conn_token.clone();
1491 let (ws_tx, mut ws_rx) = unbounded_channel::<Message>();
1492 let writer_handle = tokio::spawn(async move {
1493 while let Some(msg) = tokio::select! {
1494 biased;
1495 _ = write_token.cancelled() => None,
1496 msg = ws_rx.recv() => msg,
1497 } {
1498 if write.send(msg).await.is_err() {
1499 break;
1500 }
1501 }
1502 });
1503 let mut state = handler::SessionState::new(
1504 self,
1505 ws_tx.clone(),
1506 seq_ack.clone(),
1507 conn_token.clone(),
1508 persistent_state,
1509 backoff,
1510 )
1511 .await
1512 .inspect_err(|_e| {
1513 conn_token.cancel();
1514 })?;
1515 let outcome = match read.next().await {
1516 Some(Ok(m)) => self.handle_message(&mut state, m).await,
1517 _ => Some(SessionOutcome::Reconnect),
1518 };
1519 if let Some(out) = outcome {
1520 conn_token.cancel();
1521 writer_handle.abort();
1522 let _ = writer_handle.await;
1523 return Ok(out);
1524 }
1525 if !state.has_heartbeat() {
1526 conn_token.cancel();
1527 writer_handle.abort();
1528 let _ = writer_handle.await;
1529 return Ok(SessionOutcome::Reconnect);
1530 }
1531 let handshake = if is_resume {
1532 debug!(
1533 "[{}] Sending Resume with seq_ack={}",
1534 self.guild_id,
1535 seq_ack.load(Ordering::Relaxed)
1536 );
1537 protocol::builders::resume(
1538 self.guild_id.to_string(),
1539 self.session_id.to_string(),
1540 self.token.clone(),
1541 seq_ack.load(Ordering::Relaxed),
1542 )
1543 } else {
1544 protocol::builders::identify(
1545 self.guild_id.to_string(),
1546 self.user_id.0.to_string(),
1547 self.session_id.to_string(),
1548 self.token.clone(),
1549 1,
1550 )
1551 };
1552 let _ = ws_tx.send(Message::Text(
1553 serde_json::to_string(&handshake).unwrap().into(),
1554 ));
1555 let (speaking_tx, mut speaking_rx) = unbounded_channel::<bool>();
1556 state.set_speaking_tx(speaking_tx);
1557 let outcome = loop {
1558 tokio::select! {
1559 biased;
1560 _ = self.outer_token.cancelled() => break SessionOutcome::Shutdown,
1561 _ = conn_token.cancelled() => break SessionOutcome::Reconnect,
1562 Some(speaking) = speaking_rx.recv() => {
1563 self.notify_speaking(&ws_tx, state.ssrc(), speaking);
1564 }
1565 msg = read.next() => match msg {
1566 Some(Ok(m)) => if let Some(out) = self.handle_message(&mut state, m).await {
1567 break out;
1568 },
1569 Some(Err(_)) => break SessionOutcome::Reconnect,
1570 None => break SessionOutcome::Reconnect,
1571 }
1572 }
1573 };
1574 conn_token.cancel();
1575 writer_handle.abort();
1576 let _ = writer_handle.await;
1577 Ok(outcome)
1578 }
1579 async fn handle_message(
1580 &self,
1581 state: &mut handler::SessionState<'_>,
1582 msg: Message,
1583 ) -> Option<SessionOutcome> {
1584 match msg {
1585 Message::Text(text) => state.handle_text(text.to_string()).await,
1586 Message::Binary(bin) => {
1587 state.handle_binary(bin.to_vec()).await;
1588 None
1589 }
1590 Message::Close(frame) => {
1591 let code = frame.as_ref().map(|f| f.code.into()).unwrap_or(1000u16);
1592 let reason = frame.map(|f| f.reason.to_string()).unwrap_or_default();
1593 let attempt = state.attempt();
1594 debug!("[{}] Gateway closed: {} ({})", self.guild_id, code, reason);
1595 if !self.policy.is_retryable(code, attempt) {
1596 self.emit_close(code, reason);
1597 }
1598 Some(self.policy.classify(code))
1599 }
1600 Message::Ping(p) => {
1601 let _ = state.tx().send(Message::Pong(p));
1602 None
1603 }
1604 _ => None,
1605 }
1606 }
1607 fn notify_speaking(&self, tx: &UnboundedSender<Message>, ssrc: u32, speaking: bool) {
1608 let msg = protocol::GatewayPayload {
1609 op: protocol::OpCode::Speaking as u8,
1610 seq: None,
1611 d: serde_json::json!({
1612 "speaking": if speaking { 1 } else { 0 },
1613 "delay": 0,
1614 "ssrc": ssrc
1615 }),
1616 };
1617 if let Ok(json) = serde_json::to_string(&msg) {
1618 let _ = tx.send(Message::Text(json.into()));
1619 }
1620 }
1621 fn emit_close(&self, code: u16, reason: String) {
1622 if let Some(tx) = &self.event_tx {
1623 let _ = tx.send(RustalinkEvent::WebSocketClosed {
1624 guild_id: self.guild_id.clone(),
1625 code,
1626 reason,
1627 by_remote: true,
1628 });
1629 }
1630 }
1631 }
1632}
1633pub mod udp_link {
1634 use crate::{
1635 common::types::AnyResult,
1636 gateway::{
1637 constants::{
1638 RTP_OPUS_PAYLOAD_TYPE, RTP_TIMESTAMP_STEP, RTP_VERSION_BYTE,
1639 UDP_PACKET_BUF_CAPACITY,
1640 },
1641 session::types::map_boxed_err,
1642 },
1643 };
1644 use davey::{AeadInPlace, Aes256Gcm, KeyInit};
1645 use serde::{Deserialize, Serialize};
1646 use std::{net::SocketAddr, sync::Arc};
1647 use tokio::net::UdpSocket;
1648 use xsalsa20poly1305::XSalsa20Poly1305;
1649 pub struct UDPVoiceTransport {
1650 socket: Arc<UdpSocket>,
1651 address: SocketAddr,
1652 pub ssrc: u32,
1653 pub crypto: CryptoBackend,
1654 pub rtp: RtpState,
1655 pub buffer: Vec<u8>,
1656 }
1657 pub enum CryptoBackend {
1658 XSalsa20Poly1305(Box<XSalsa20Poly1305>),
1659 Aes256Gcm(Box<Aes256Gcm>),
1660 }
1661 #[derive(Debug, Clone, Copy, Serialize, Deserialize)]
1662 pub struct RtpState {
1663 pub sequence: u16,
1664 pub timestamp: u32,
1665 pub nonce: u32,
1666 }
1667 impl UDPVoiceTransport {
1668 pub fn new(
1669 socket: Arc<UdpSocket>,
1670 address: SocketAddr,
1671 ssrc: u32,
1672 secret_key: [u8; 32],
1673 mode: &str,
1674 rtp_state: Option<RtpState>,
1675 ) -> AnyResult<Self> {
1676 let crypto = match mode {
1677 "aead_aes256_gcm_rtpsize" => {
1678 CryptoBackend::Aes256Gcm(Box::new(Aes256Gcm::new(&secret_key.into())))
1679 }
1680 _ => CryptoBackend::XSalsa20Poly1305(Box::new(XSalsa20Poly1305::new(
1681 &secret_key.into(),
1682 ))),
1683 };
1684 Ok(Self {
1685 socket,
1686 address,
1687 ssrc,
1688 crypto,
1689 rtp: rtp_state.unwrap_or_else(RtpState::randomize),
1690 buffer: Vec::with_capacity(UDP_PACKET_BUF_CAPACITY),
1691 })
1692 }
1693 pub async fn send_keepalive(&self, counter: u32) -> AnyResult<()> {
1694 let payload = counter.to_be_bytes();
1695 self.socket.send_to(&payload, self.address).await?;
1696 Ok(())
1697 }
1698 pub async fn transmit_opus(&mut self, opus_data: &[u8]) -> AnyResult<()> {
1699 let (seq, ts, nonce_val) = self.rtp.next();
1700 let mut header = [0u8; 12];
1701 header[0] = RTP_VERSION_BYTE;
1702 header[1] = RTP_OPUS_PAYLOAD_TYPE;
1703 header[2..4].copy_from_slice(&seq.to_be_bytes());
1704 header[4..8].copy_from_slice(&ts.to_be_bytes());
1705 header[8..12].copy_from_slice(&self.ssrc.to_be_bytes());
1706 self.buffer.clear();
1707 self.buffer.extend_from_slice(&header);
1708 self.buffer.extend_from_slice(opus_data);
1709 match &self.crypto {
1710 CryptoBackend::XSalsa20Poly1305(cipher) => {
1711 let mut nonce = [0u8; 24];
1712 nonce[0..12].copy_from_slice(&header);
1713 let tag = cipher
1714 .encrypt_in_place_detached(&nonce.into(), &header, &mut self.buffer[12..])
1715 .map_err(|e| map_boxed_err(format!("XSalsa20 error: {e:?}")))?;
1716 self.buffer.extend_from_slice(&tag);
1717 }
1718 CryptoBackend::Aes256Gcm(cipher) => {
1719 let mut nonce = [0u8; 12];
1720 nonce[0..4].copy_from_slice(&nonce_val.to_be_bytes());
1721 let tag = cipher
1722 .encrypt_in_place_detached(&nonce.into(), &header, &mut self.buffer[12..])
1723 .map_err(|e| map_boxed_err(format!("AES-GCM error: {e:?}")))?;
1724 self.buffer.extend_from_slice(&tag);
1725 self.buffer.extend_from_slice(&nonce_val.to_be_bytes());
1726 }
1727 }
1728 self.socket.send_to(&self.buffer, self.address).await?;
1729 Ok(())
1730 }
1731 }
1732 impl RtpState {
1733 fn randomize() -> Self {
1734 Self {
1735 sequence: rand::random(),
1736 timestamp: rand::random(),
1737 nonce: rand::random(),
1738 }
1739 }
1740 fn next(&mut self) -> (u16, u32, u32) {
1741 let seq = self.sequence;
1742 let ts = self.timestamp;
1743 let n = self.nonce;
1744 self.sequence = self.sequence.wrapping_add(1);
1745 self.timestamp = self.timestamp.wrapping_add(RTP_TIMESTAMP_STEP);
1746 self.nonce = self.nonce.wrapping_add(1);
1747 (seq, ts, n)
1748 }
1749 }
1750}
1751pub mod encryption {
1752 use crate::{
1753 common::types::{AnyError, AnyResult, ChannelId, UserId},
1754 gateway::{
1755 constants::{DAVE_INITIAL_VERSION, MAX_PENDING_PROPOSALS, SILENCE_FRAME},
1756 session::types::map_boxed_err,
1757 },
1758 };
1759 use davey::{DaveSession, ProposalsOperationType};
1760 use std::{
1761 collections::{HashMap, HashSet},
1762 num::NonZeroU16,
1763 };
1764 use tracing::{debug, trace, warn};
1765 const DAVE_MIN_VERSION: NonZeroU16 = match NonZeroU16::new(DAVE_INITIAL_VERSION) {
1766 Some(v) => v,
1767 None => unreachable!(),
1768 };
1769 pub struct DaveHandler {
1770 session: Option<DaveSession>,
1771 user_id: UserId,
1772 channel_id: ChannelId,
1773 protocol_version: u16,
1774 pending_transitions: HashMap<u16, u16>,
1775 external_sender_set: bool,
1776 saved_external_sender: Option<Vec<u8>>,
1777 pending_proposals: Vec<Vec<u8>>,
1778 pending_handshake: Vec<(Vec<u8>, bool)>,
1779 was_ready: bool,
1780 recognized_users: HashSet<UserId>,
1781 cached_user_ids: Vec<u64>,
1782 }
1783 impl DaveHandler {
1784 pub fn new(user_id: UserId, channel_id: ChannelId) -> Self {
1785 let mut recognized_users = HashSet::new();
1786 recognized_users.insert(user_id);
1787 Self {
1788 session: None,
1789 user_id,
1790 channel_id,
1791 protocol_version: 0,
1792 pending_transitions: HashMap::new(),
1793 external_sender_set: false,
1794 saved_external_sender: None,
1795 pending_proposals: Vec::new(),
1796 pending_handshake: Vec::new(),
1797 was_ready: false,
1798 recognized_users,
1799 cached_user_ids: vec![user_id.0],
1800 }
1801 }
1802 pub fn add_users(&mut self, uids: &[u64]) {
1803 for &uid in uids {
1804 self.recognized_users.insert(UserId(uid));
1805 }
1806 self.update_user_cache();
1807 debug!("DAVE adding users: {:?}", uids);
1808 }
1809 pub fn remove_user(&mut self, uid: u64) {
1810 if self.recognized_users.remove(&UserId(uid)) {
1811 self.update_user_cache();
1812 }
1813 debug!("DAVE removing user: {}", uid);
1814 }
1815 fn update_user_cache(&mut self) {
1816 self.cached_user_ids.clear();
1817 self.cached_user_ids
1818 .extend(self.recognized_users.iter().map(|u| u.0));
1819 self.cached_user_ids.sort_unstable();
1820 }
1821 pub fn protocol_version(&self) -> u16 {
1822 self.protocol_version
1823 }
1824 pub fn set_protocol_version(&mut self, version: u16) {
1825 self.protocol_version = version;
1826 }
1827 pub fn setup_session(&mut self, version: u16) -> AnyResult<Vec<u8>> {
1828 if version == 0 {
1829 self.reset();
1830 return Ok(Vec::new());
1831 }
1832 let nz_version = NonZeroU16::new(version).unwrap_or(DAVE_MIN_VERSION);
1833 if let Some(s) = &mut self.session {
1834 s.reinit(nz_version, self.user_id.0, self.channel_id.0, None)
1835 .map_err(map_boxed_err)?;
1836 } else {
1837 let session = DaveSession::new(nz_version, self.user_id.0, self.channel_id.0, None)
1838 .map_err(map_boxed_err)?;
1839 self.session = Some(session);
1840 }
1841 let session = self
1842 .session
1843 .as_mut()
1844 .ok_or_else(|| map_boxed_err("DAVE session initialization failed"))?;
1845 self.protocol_version = version;
1846 self.external_sender_set = false;
1847 self.pending_proposals.clear();
1848 self.pending_handshake.clear();
1849 self.was_ready = false;
1850 debug!("DAVE session setup (v{})", version);
1851 let key_package = session.create_key_package().map_err(map_boxed_err)?;
1852 if let Some(saved) = self.saved_external_sender.as_deref()
1853 && let Some(sess) = &mut self.session
1854 {
1855 match sess.set_external_sender(saved) {
1856 Ok(()) => {
1857 self.external_sender_set = true;
1858 debug!("DAVE re-applied saved external sender after epoch reset");
1859 }
1860 Err(e) => {
1861 warn!("DAVE failed to re-apply saved external sender: {e}");
1862 self.saved_external_sender = None;
1863 }
1864 }
1865 }
1866 Ok(key_package)
1867 }
1868 pub fn reset(&mut self) {
1869 self.protocol_version = 0;
1870 self.pending_transitions.clear();
1871 self.external_sender_set = false;
1872 self.saved_external_sender = None;
1873 self.pending_proposals.clear();
1874 self.pending_handshake.clear();
1875 self.was_ready = false;
1876 self.session = None;
1877 debug!("DAVE session reset to plaintext");
1878 }
1879 pub fn prepare_transition(&mut self, transition_id: u16, protocol_version: u16) -> bool {
1880 self.pending_transitions
1881 .insert(transition_id, protocol_version);
1882 if transition_id == 0 {
1883 self.execute_transition(0);
1884 return false;
1885 }
1886 true
1887 }
1888 pub fn execute_transition(&mut self, transition_id: u16) {
1889 if let Some(next_version) = self.pending_transitions.remove(&transition_id) {
1890 self.protocol_version = next_version;
1891 trace!(
1892 "DAVE transition {} executed (v{})",
1893 transition_id, next_version
1894 );
1895 }
1896 }
1897 pub fn prepare_epoch(&mut self, epoch: u64, protocol_version: u16) -> Option<Vec<u8>> {
1898 if epoch == 1 {
1899 match self.setup_session(protocol_version) {
1900 Ok(kp) => return Some(kp),
1901 Err(e) => warn!("DAVE prepare_epoch setup failed: {e}"),
1902 }
1903 }
1904 None
1905 }
1906 pub fn process_external_sender(&mut self, data: &[u8]) -> AnyResult<Vec<Vec<u8>>> {
1907 let mut responses = Vec::new();
1908 if let Some(session) = &mut self.session {
1909 session.set_external_sender(data).map_err(map_boxed_err)?;
1910 self.external_sender_set = true;
1911 self.saved_external_sender = Some(data.to_vec());
1912 if !self.pending_proposals.is_empty() {
1913 debug!(
1914 "DAVE processing {} buffered proposals",
1915 self.pending_proposals.len()
1916 );
1917 for prop_data in std::mem::take(&mut self.pending_proposals) {
1918 if let Ok(Some(res)) =
1919 Self::do_process_proposals(session, &prop_data, &self.cached_user_ids)
1920 {
1921 responses.push(res);
1922 }
1923 }
1924 }
1925 if !self.pending_handshake.is_empty() {
1926 debug!(
1927 "DAVE processing {} buffered handshake messages",
1928 self.pending_handshake.len()
1929 );
1930 for (handshake_data, is_welcome) in std::mem::take(&mut self.pending_handshake)
1931 {
1932 if let Err(e) = self.do_process_handshake(&handshake_data, is_welcome) {
1933 warn!("DAVE buffered handshake processing failed: {e}");
1934 }
1935 }
1936 }
1937 }
1938 Ok(responses)
1939 }
1940 pub fn process_welcome(&mut self, data: &[u8]) -> AnyResult<u16> {
1941 self.process_handshake_message(data, true)
1942 }
1943 pub fn process_commit(&mut self, data: &[u8]) -> AnyResult<u16> {
1944 self.process_handshake_message(data, false)
1945 }
1946 fn process_handshake_message(&mut self, data: &[u8], is_welcome: bool) -> AnyResult<u16> {
1947 let tag = if is_welcome { "welcome" } else { "commit" };
1948 if data.len() < 2 {
1949 let msg = if is_welcome {
1950 "DAVE welcome"
1951 } else {
1952 "DAVE commit"
1953 };
1954 return Err(short_payload_err(msg));
1955 }
1956 let transition_id = u16::from_be_bytes([data[0], data[1]]);
1957 if !self.external_sender_set {
1958 if self.pending_handshake.len() < MAX_PENDING_PROPOSALS {
1959 debug!("DAVE buffering {tag} — external sender not set");
1960 self.pending_handshake.push((data.to_vec(), is_welcome));
1961 } else {
1962 warn!("DAVE handshake buffer full, dropping {tag}");
1963 }
1964 return Ok(transition_id);
1965 }
1966 self.do_process_handshake(data, is_welcome)?;
1967 Ok(transition_id)
1968 }
1969 fn do_process_handshake(&mut self, data: &[u8], is_welcome: bool) -> AnyResult<()> {
1970 let transition_id = u16::from_be_bytes([data[0], data[1]]);
1971 if let Some(session) = &mut self.session {
1972 if is_welcome {
1973 session.process_welcome(&data[2..]).map_err(map_boxed_err)?;
1974 } else {
1975 session.process_commit(&data[2..]).map_err(map_boxed_err)?;
1976 }
1977 if transition_id != 0 {
1978 self.pending_transitions
1979 .insert(transition_id, self.protocol_version);
1980 }
1981 debug!(
1982 "DAVE {} processed (tid {})",
1983 if is_welcome { "welcome" } else { "commit" },
1984 transition_id
1985 );
1986 }
1987 Ok(())
1988 }
1989 pub fn process_proposals(&mut self, data: &[u8]) -> AnyResult<Option<Vec<u8>>> {
1990 if data.is_empty() {
1991 return Err(short_payload_err("DAVE proposals"));
1992 }
1993 if !self.external_sender_set {
1994 if self.pending_proposals.len() < MAX_PENDING_PROPOSALS {
1995 debug!("DAVE buffering proposal — external sender not set");
1996 self.pending_proposals.push(data.to_vec());
1997 } else {
1998 warn!("DAVE proposal buffer full, dropping proposal");
1999 }
2000 return Ok(None);
2001 }
2002 let session = match &mut self.session {
2003 Some(s) => s,
2004 None => return Ok(None),
2005 };
2006 Self::do_process_proposals(session, data, &self.cached_user_ids)
2007 }
2008 fn do_process_proposals(
2009 session: &mut DaveSession,
2010 data: &[u8],
2011 user_ids: &[u64],
2012 ) -> AnyResult<Option<Vec<u8>>> {
2013 let op_type = match data[0] {
2014 0 => ProposalsOperationType::APPEND,
2015 1 => ProposalsOperationType::REVOKE,
2016 raw => return Err(map_boxed_err(format!("Unknown DAVE proposals op: {raw}"))),
2017 };
2018 let result = session
2019 .process_proposals(op_type, &data[1..], Some(user_ids))
2020 .map_err(map_boxed_err)?;
2021 if let Some(cw) = result {
2022 let mut out = cw.commit;
2023 if let Some(w) = cw.welcome {
2024 out.extend_from_slice(&w);
2025 }
2026 return Ok(Some(out));
2027 }
2028 Ok(None)
2029 }
2030 pub fn encrypt_opus(&mut self, packet: &[u8]) -> AnyResult<Vec<u8>> {
2031 if packet == SILENCE_FRAME || self.protocol_version == 0 {
2032 return Ok(packet.to_vec());
2033 }
2034 if let Some(session) = &mut self.session {
2035 let is_ready = session.is_ready();
2036 if is_ready != self.was_ready {
2037 if is_ready {
2038 debug!("DAVE session (v{}) is READY", self.protocol_version);
2039 } else {
2040 warn!("DAVE session (v{}) LOST readiness", self.protocol_version);
2041 }
2042 self.was_ready = is_ready;
2043 }
2044 if is_ready {
2045 return session
2046 .encrypt_opus(packet)
2047 .map(|c| c.into_owned())
2048 .map_err(map_boxed_err);
2049 }
2050 }
2051 Ok(packet.to_vec())
2052 }
2053 pub fn voice_privacy_code(&self) -> Option<String> {
2054 self.session
2055 .as_ref()
2056 .and_then(|s| s.voice_privacy_code().map(|c| c.to_string()))
2057 }
2058 }
2059 #[inline]
2060 fn short_payload_err(context: &str) -> AnyError {
2061 map_boxed_err(format!("Invalid {context} payload: too short"))
2062 }
2063}
2064pub mod engine {
2065 use crate::{audio::Mixer, common::types::Shared, gateway::constants::DEFAULT_SAMPLE_RATE};
2066 use tokio::sync::Mutex;
2067 pub struct VoiceEngine {
2068 pub mixer: Shared<Mixer>,
2069 pub dave: Option<Shared<crate::gateway::DaveHandler>>,
2070 }
2071 impl VoiceEngine {
2072 pub fn new() -> Self {
2073 Self {
2074 mixer: Shared::new(Mutex::new(Mixer::new(DEFAULT_SAMPLE_RATE))),
2075 dave: None,
2076 }
2077 }
2078 }
2079 impl Default for VoiceEngine {
2080 fn default() -> Self {
2081 Self::new()
2082 }
2083 }
2084}
2085pub use encryption::DaveHandler;
2086pub use engine::VoiceEngine;
2087pub use session::{VoiceGateway, VoiceGatewayConfig};