1use std::sync::Arc;
2use std::sync::atomic::{AtomicBool, AtomicU32, Ordering};
3
4use crate::audio::{
5 Mixer,
6 filters::FilterChain,
7 playback::{TrackHandle, handle::PlaybackState as PlayState},
8};
9use crate::common::types::{ChannelId, GuildId, SessionId, Shared, UserId};
10use crate::events::EventSender;
11use crate::gateway::{VoiceGateway, VoiceGatewayConfig};
12use serde::{Deserialize, Serialize};
13use serde_json::json;
14use tokio_util::sync::CancellationToken;
15#[derive(Debug, Clone, Serialize, Deserialize)]
16pub struct EqBand {
17 pub band: u8,
18 pub gain: f32,
19}
20#[derive(Debug, Clone, Serialize, Deserialize)]
21#[serde(rename_all = "camelCase")]
22pub struct KaraokeFilter {
23 pub level: Option<f32>,
24 pub mono_level: Option<f32>,
25 pub filter_band: Option<f32>,
26 pub filter_width: Option<f32>,
27}
28#[derive(Debug, Clone, Serialize, Deserialize)]
29#[serde(rename_all = "camelCase")]
30pub struct TimescaleFilter {
31 pub speed: Option<f64>,
32 pub pitch: Option<f64>,
33 pub rate: Option<f64>,
34}
35#[derive(Debug, Clone, Serialize, Deserialize)]
36#[serde(rename_all = "camelCase")]
37pub struct TremoloFilter {
38 pub frequency: Option<f32>,
39 pub depth: Option<f32>,
40}
41#[derive(Debug, Clone, Serialize, Deserialize)]
42#[serde(rename_all = "camelCase")]
43pub struct VibratoFilter {
44 pub frequency: Option<f32>,
45 pub depth: Option<f32>,
46}
47#[derive(Debug, Clone, Serialize, Deserialize)]
48#[serde(rename_all = "camelCase")]
49pub struct DistortionFilter {
50 pub sin_offset: Option<f32>,
51 pub sin_scale: Option<f32>,
52 pub cos_offset: Option<f32>,
53 pub cos_scale: Option<f32>,
54 pub tan_offset: Option<f32>,
55 pub tan_scale: Option<f32>,
56 pub offset: Option<f32>,
57 pub scale: Option<f32>,
58}
59#[derive(Debug, Clone, Serialize, Deserialize)]
60#[serde(rename_all = "camelCase")]
61pub struct RotationFilter {
62 pub rotation_hz: Option<f64>,
63}
64#[derive(Debug, Clone, Serialize, Deserialize)]
65#[serde(rename_all = "camelCase")]
66pub struct ChannelMixFilter {
67 pub left_to_left: Option<f32>,
68 pub left_to_right: Option<f32>,
69 pub right_to_left: Option<f32>,
70 pub right_to_right: Option<f32>,
71}
72#[derive(Debug, Clone, Serialize, Deserialize)]
73#[serde(rename_all = "camelCase")]
74pub struct LowPassFilter {
75 pub smoothing: Option<f32>,
76}
77#[derive(Debug, Clone, Serialize, Deserialize)]
78#[serde(rename_all = "camelCase")]
79pub struct EchoFilter {
80 pub echo_length: Option<f32>,
81 pub decay: Option<f32>,
82}
83#[derive(Debug, Clone, Serialize, Deserialize)]
84#[serde(rename_all = "camelCase")]
85pub struct HighPassFilter {
86 pub cutoff_frequency: Option<i32>,
87 pub boost_factor: Option<f32>,
88}
89#[derive(Debug, Clone, Serialize, Deserialize)]
90#[serde(rename_all = "camelCase")]
91pub struct NormalizationFilter {
92 pub max_amplitude: Option<f32>,
93 pub adaptive: Option<bool>,
94}
95#[derive(Debug, Clone, Serialize, Deserialize)]
96#[serde(rename_all = "camelCase")]
97pub struct ChorusFilter {
98 pub rate: Option<f32>,
99 pub depth: Option<f32>,
100 pub delay: Option<f32>,
101 pub mix: Option<f32>,
102 pub feedback: Option<f32>,
103}
104#[derive(Debug, Clone, Serialize, Deserialize)]
105#[serde(rename_all = "camelCase")]
106pub struct CompressorFilter {
107 pub threshold: Option<f32>,
108 pub ratio: Option<f32>,
109 pub attack: Option<f32>,
110 pub release: Option<f32>,
111 pub makeup_gain: Option<f32>,
112}
113#[derive(Debug, Clone, Serialize, Deserialize)]
114#[serde(rename_all = "camelCase")]
115pub struct FlangerFilter {
116 pub rate: Option<f32>,
117 pub depth: Option<f32>,
118 pub feedback: Option<f32>,
119}
120#[derive(Debug, Clone, Serialize, Deserialize)]
121#[serde(rename_all = "camelCase")]
122pub struct PhaserFilter {
123 pub stages: Option<i32>,
124 pub rate: Option<f32>,
125 pub depth: Option<f32>,
126 pub feedback: Option<f32>,
127 pub mix: Option<f32>,
128 pub min_frequency: Option<f32>,
129 pub max_frequency: Option<f32>,
130}
131#[derive(Debug, Clone, Serialize, Deserialize)]
132#[serde(rename_all = "camelCase")]
133pub struct PhonographFilter {
134 pub frequency: Option<f32>,
135 pub depth: Option<f32>,
136 pub crackle: Option<f32>,
137 pub flutter: Option<f32>,
138 pub room: Option<f32>,
139 pub mic_agc: Option<f32>,
140 pub drive: Option<f32>,
141}
142#[derive(Debug, Clone, Serialize, Deserialize)]
143#[serde(rename_all = "camelCase")]
144pub struct ReverbFilter {
145 pub mix: Option<f32>,
146 pub room_size: Option<f32>,
147 pub damping: Option<f32>,
148 pub width: Option<f32>,
149}
150#[derive(Debug, Clone, Serialize, Deserialize)]
151#[serde(rename_all = "camelCase")]
152pub struct SpatialFilter {
153 pub depth: Option<f32>,
154 pub rate: Option<f32>,
155}
156macro_rules! define_filters {
157 ($($field:ident : $type:ty => $name:expr),* $(,)?) => {
158 #[derive(Debug, Clone, Default, Serialize, Deserialize)]
159 #[serde(rename_all = "camelCase")]
160 pub struct Filters {
161 $(
162 #[serde(skip_serializing_if = "Option::is_none")]
163 pub $field: Option<$type>,
164 )*
165 }
166 impl Filters {
167 pub fn names() -> Vec<String> {
168 vec![
169 $($name.into()),*
170 ]
171 }
172 pub fn merge_from(&mut self, incoming: Filters) {
173 $(
174 if incoming.$field.is_some() {
175 self.$field = incoming.$field;
176 }
177 )*
178 }
179 pub fn is_all_none(&self) -> bool {
180 $(
181 self.$field.is_none() &&
182 )* true
183 }
184 }
185 };
186}
187define_filters! {
188 volume: f32 => "volume",
189 equalizer: Vec<EqBand> => "equalizer",
190 karaoke: KaraokeFilter => "karaoke",
191 timescale: TimescaleFilter => "timescale",
192 tremolo: TremoloFilter => "tremolo",
193 vibrato: VibratoFilter => "vibrato",
194 distortion: DistortionFilter => "distortion",
195 rotation: RotationFilter => "rotation",
196 channel_mix: ChannelMixFilter => "channelMix",
197 low_pass: LowPassFilter => "lowPass",
198 echo: EchoFilter => "echo",
199 high_pass: HighPassFilter => "highPass",
200 normalization: NormalizationFilter => "normalization",
201 chorus: ChorusFilter => "chorus",
202 compressor: CompressorFilter => "compressor",
203 flanger: FlangerFilter => "flanger",
204 phaser: PhaserFilter => "phaser",
205 phonograph: PhonographFilter => "phonograph",
206 reverb: ReverbFilter => "reverb",
207 spatial: SpatialFilter => "spatial",
208 plugin_filters: std::collections::HashMap<String, serde_json::Value> => "pluginFilters",
209}
210pub struct Player {
211 pub guild_id: String,
212 pub paused: Arc<AtomicBool>,
213 pub volume: Arc<AtomicU32>,
214 pub mixer: Shared<Mixer>,
215 pub filter_chain: Shared<FilterChain>,
216 pub voice_gateway_cancel: Arc<tokio::sync::Mutex<Option<CancellationToken>>>,
217 pub track_handle: Arc<tokio::sync::Mutex<Option<TrackHandle>>>,
218 pub event_sender: Arc<tokio::sync::Mutex<Option<EventSender>>>,
219}
220
221impl Player {
222 pub fn new(guild_id: String) -> Self {
223 Self {
224 guild_id,
225 paused: Arc::new(AtomicBool::new(false)),
226 volume: Arc::new(AtomicU32::new(1.0f32.to_bits())),
227 mixer: Shared::new(tokio::sync::Mutex::new(Mixer::new(48000))),
228 filter_chain: Shared::new(tokio::sync::Mutex::new(FilterChain::from_config(
229 &Filters::default(),
230 ))),
231 voice_gateway_cancel: Arc::new(tokio::sync::Mutex::new(None)),
232 track_handle: Arc::new(tokio::sync::Mutex::new(None)),
233 event_sender: Arc::new(tokio::sync::Mutex::new(None)),
234 }
235 }
236 pub async fn play<F>(
237 &self,
238 user_id: String,
239 channel_id: String,
240 session_id: String,
241 token: String,
242 endpoint: String,
243 url: String,
244 callback: F,
245 ) -> Result<(), String>
246 where
247 F: Fn(&str, serde_json::Value) + Send + Sync + 'static,
248 {
249 let events = EventSender::new(callback);
250 {
251 *self.event_sender.lock().await = Some(events.clone());
252 }
253 {
254 let mut cancel_guard = self.voice_gateway_cancel.lock().await;
255 if let Some(cancel) = cancel_guard.take() {
256 cancel.cancel();
257 }
258 }
259 {
260 let mut mixer_guard = self.mixer.lock().await;
261 mixer_guard.stop_all();
262 }
263 self.paused.store(false, Ordering::Release);
264 let sm = crate::get_source_manager();
265 let player_config = sm.player_config.clone();
266 let playable_track = match load_and_resolve_first_result(sm, &url).await {
267 Ok(pt) => pt,
268 Err(e) => {
269 events.send("error", json!({ "message": e }));
270 return Ok(());
271 }
272 };
273 let mixer_clone = self.mixer.clone();
274 let track_handle_lock = self.track_handle.clone();
275 let events_clone = events.clone();
276 tokio::spawn(async move {
277 println!("Starting play task in background for identifier: {url}");
278 let (frame_rx, cmd_tx, err_rx) = playable_track.start_decoding(player_config.clone());
279 let (handle, audio_state, vol, pos, is_buffering) =
280 TrackHandle::new(cmd_tx, Arc::new(AtomicBool::new(false)));
281 {
282 let mut mixer_guard = mixer_clone.lock().await;
283 mixer_guard.add_track(
284 frame_rx,
285 audio_state,
286 vol,
287 pos,
288 is_buffering,
289 player_config.clone(),
290 );
291 }
292 {
293 let mut handle_guard = track_handle_lock.lock().await;
294 *handle_guard = Some(handle);
295 }
296 events_clone.send("trackStart", json!({}));
297 let err_rx = err_rx;
298 tokio::spawn(async move {
299 if let Ok(err) = err_rx.recv_async().await {
300 events_clone.send("error", json!({ "message": err }));
301 } else {
302 events_clone.send("trackEnd", json!({}));
303 }
304 });
305 });
306 let cancel_token = CancellationToken::new();
307 {
308 *self.voice_gateway_cancel.lock().await = Some(cancel_token.clone());
309 }
310 let track_handle_clone = self.track_handle.clone();
311 let events_position = events.clone();
312 let cancel_token_clone = cancel_token.clone();
313 tokio::spawn(async move {
314 let mut interval = tokio::time::interval(std::time::Duration::from_millis(1000));
315 while !cancel_token_clone.is_cancelled() {
316 interval.tick().await;
317 let handle_guard = track_handle_clone.lock().await;
318 if let Some(handle) = &*handle_guard {
319 let state = handle.get_state();
320 if state == PlayState::Playing {
321 let pos = handle.get_position();
322 events_position.send("position", json!({ "position": pos }));
323 }
324 }
325 }
326 });
327 let gateway_config = VoiceGatewayConfig {
328 guild_id: GuildId(self.guild_id.clone()),
329 user_id: UserId(user_id.parse().unwrap_or(0)),
330 channel_id: ChannelId(channel_id.parse().unwrap_or(0)),
331 session_id: SessionId(session_id.clone()),
332 token: token.clone(),
333 endpoint: endpoint.clone(),
334 mixer: self.mixer.clone(),
335 filter_chain: self.filter_chain.clone(),
336 ping: Arc::new(std::sync::atomic::AtomicI64::new(-1)),
337 event_tx: None,
338 frames_sent: Arc::new(std::sync::atomic::AtomicU64::new(0)),
339 frames_nulled: Arc::new(std::sync::atomic::AtomicU64::new(0)),
340 };
341 let voice_gateway = VoiceGateway::new(gateway_config);
342 tokio::spawn(async move {
343 if let Err(e) = voice_gateway.run().await {
344 events.send(
345 "error",
346 json!({ "message": format!("VoiceGateway error: {e}") }),
347 );
348 }
349 });
350 Ok(())
351 }
352 pub async fn pause(&self) {
353 self.paused.store(true, Ordering::Release);
354 if let Some(handle) = &*self.track_handle.lock().await {
355 handle.pause();
356 }
357 let guard = self.event_sender.lock().await;
358 if let Some(ref e) = *guard {
359 e.send("paused", json!({}));
360 }
361 }
362 pub async fn resume(&self) {
363 self.paused.store(false, Ordering::Release);
364 if let Some(handle) = &*self.track_handle.lock().await {
365 handle.play();
366 }
367 let guard = self.event_sender.lock().await;
368 if let Some(ref e) = *guard {
369 e.send("resumed", json!({}));
370 }
371 }
372 pub async fn stop(&self) {
373 {
374 let mut cancel_guard = self.voice_gateway_cancel.lock().await;
375 if let Some(cancel) = cancel_guard.take() {
376 cancel.cancel();
377 }
378 }
379 if let Some(handle) = &*self.track_handle.lock().await {
380 handle.stop();
381 }
382 let mut mixer_guard = self.mixer.lock().await;
383 mixer_guard.stop_all();
384 }
385 pub async fn seek(&self, position_ms: i64) {
386 if let Some(handle) = &*self.track_handle.lock().await {
387 handle.seek(position_ms.max(0) as u64);
388 }
389 }
390 pub async fn set_volume(&self, volume: f64) {
391 let vol_f = volume as f32;
392 self.volume.store(vol_f.to_bits(), Ordering::Relaxed);
393 if let Some(handle) = &*self.track_handle.lock().await {
394 handle.set_volume(vol_f);
395 }
396 let guard = self.event_sender.lock().await;
397 if let Some(ref e) = *guard {
398 e.send("volume", json!({ "volume": volume }));
399 }
400 }
401 pub fn get_position(&self) -> i64 {
402 let handle_guard = futures::executor::block_on(self.track_handle.lock());
403 if let Some(handle) = &*handle_guard {
404 handle.get_position() as i64
405 } else {
406 0
407 }
408 }
409 pub fn is_paused(&self) -> bool {
410 let handle_guard = futures::executor::block_on(self.track_handle.lock());
411 if let Some(handle) = &*handle_guard {
412 matches!(handle.get_state(), PlayState::Paused)
413 } else {
414 self.paused.load(Ordering::Acquire)
415 }
416 }
417 pub async fn set_filters(&self, filters_json: String) -> Result<(), String> {
418 let filters: Filters = serde_json::from_str(&filters_json)
419 .map_err(|e| format!("Invalid filters JSON: {e}"))?;
420 let new_chain = FilterChain::from_config(&filters);
421 {
422 let mut filter_chain_guard = self.filter_chain.lock().await;
423 *filter_chain_guard = new_chain;
424 }
425 Ok(())
426 }
427}
428async fn load_and_resolve_first_result(
429 sm: &crate::sources::manager::SourceManager,
430 url: &str,
431) -> Result<crate::sources::playable_track::BoxedTrack, String> {
432 match sm.load(url, None).await {
433 crate::protocol::tracks::LoadResult::Track(track) => {
434 sm.resolve_track(&track.info, None).await
435 }
436 crate::protocol::tracks::LoadResult::Search(tracks) => {
437 if tracks.is_empty() {
438 return Err("Search returned no results".to_string());
439 }
440 sm.resolve_track(&tracks[0].info, None).await
441 }
442 crate::protocol::tracks::LoadResult::Playlist(playlist) => {
443 if playlist.tracks.is_empty() {
444 return Err("Playlist is empty".to_string());
445 }
446 sm.resolve_track(&playlist.tracks[0].info, None).await
447 }
448 _ => Err(format!("Failed to load track or query: {url}")),
449 }
450}