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_arc = {
265 let guard = crate::get_source_manager().lock().unwrap();
266 guard.as_ref().unwrap().clone()
267 };
268 let player_config = sm_arc.player_config.clone();
269 let playable_track = match load_and_resolve_first_result(&sm_arc, &url).await {
270 Ok(pt) => pt,
271 Err(e) => {
272 events.send("error", json!({ "message": e }));
273 return Ok(());
274 }
275 };
276 let mixer_clone = self.mixer.clone();
277 let track_handle_lock = self.track_handle.clone();
278 let events_clone = events.clone();
279 tokio::spawn(async move {
280 println!("Starting play task in background for identifier: {url}");
281 let (frame_rx, cmd_tx, err_rx) = playable_track.start_decoding(player_config.clone());
282 let (handle, audio_state, vol, pos, is_buffering) =
283 TrackHandle::new(cmd_tx, Arc::new(AtomicBool::new(false)));
284 {
285 let mut mixer_guard = mixer_clone.lock().await;
286 mixer_guard.add_track(
287 frame_rx,
288 audio_state,
289 vol,
290 pos,
291 is_buffering,
292 player_config.clone(),
293 );
294 }
295 {
296 let mut handle_guard = track_handle_lock.lock().await;
297 *handle_guard = Some(handle);
298 }
299 events_clone.send("trackStart", json!({}));
300 let err_rx = err_rx;
301 tokio::spawn(async move {
302 if let Ok(err) = err_rx.recv_async().await {
303 events_clone.send("error", json!({ "message": err }));
304 }
305 });
306 });
307 let cancel_token = CancellationToken::new();
308 {
309 *self.voice_gateway_cancel.lock().await = Some(cancel_token.clone());
310 }
311 let track_handle_clone = self.track_handle.clone();
312 let events_position = events.clone();
313 let cancel_token_clone = cancel_token.clone();
314 tokio::spawn(async move {
315 let mut interval = tokio::time::interval(std::time::Duration::from_millis(100));
316 let mut ticks = 0;
317 while !cancel_token_clone.is_cancelled() {
318 interval.tick().await;
319 ticks += 1;
320 let handle_guard = track_handle_clone.lock().await;
321 if let Some(handle) = &*handle_guard {
322 let state = handle.get_state();
323 if state == PlayState::Stopped {
324 events_position.send("trackEnd", json!({ "reason": "FINISHED" }));
325 break;
326 }
327 if state == PlayState::Playing {
328 if ticks >= 10 {
329 let pos = handle.get_position();
330 events_position.send("position", json!({ "position": pos }));
331 ticks = 0;
332 }
333 }
334 }
335 }
336 });
337 let gateway_config = VoiceGatewayConfig {
338 guild_id: GuildId(self.guild_id.clone()),
339 user_id: UserId(user_id.parse().unwrap_or(0)),
340 channel_id: ChannelId(channel_id.parse().unwrap_or(0)),
341 session_id: SessionId(session_id.clone()),
342 token: token.clone(),
343 endpoint: endpoint.clone(),
344 mixer: self.mixer.clone(),
345 filter_chain: self.filter_chain.clone(),
346 ping: Arc::new(std::sync::atomic::AtomicI64::new(-1)),
347 event_tx: None,
348 frames_sent: Arc::new(std::sync::atomic::AtomicU64::new(0)),
349 frames_nulled: Arc::new(std::sync::atomic::AtomicU64::new(0)),
350 cancel_token: cancel_token.clone(),
351 };
352 let voice_gateway = VoiceGateway::new(gateway_config);
353 tokio::spawn(async move {
354 if let Err(e) = voice_gateway.run().await {
355 events.send(
356 "error",
357 json!({ "message": format!("VoiceGateway error: {e}") }),
358 );
359 }
360 });
361 Ok(())
362 }
363 pub async fn pause(&self) {
364 self.paused.store(true, Ordering::Release);
365 if let Some(handle) = &*self.track_handle.lock().await {
366 handle.pause();
367 }
368 let guard = self.event_sender.lock().await;
369 if let Some(ref e) = *guard {
370 e.send("paused", json!({}));
371 }
372 }
373 pub async fn resume(&self) {
374 self.paused.store(false, Ordering::Release);
375 if let Some(handle) = &*self.track_handle.lock().await {
376 handle.play();
377 }
378 let guard = self.event_sender.lock().await;
379 if let Some(ref e) = *guard {
380 e.send("resumed", json!({}));
381 }
382 }
383 pub async fn stop(&self) {
384 {
385 let mut cancel_guard = self.voice_gateway_cancel.lock().await;
386 if let Some(cancel) = cancel_guard.take() {
387 cancel.cancel();
388 }
389 }
390 if let Some(handle) = &*self.track_handle.lock().await {
391 handle.stop();
392 }
393 let mut mixer_guard = self.mixer.lock().await;
394 mixer_guard.stop_all();
395 }
396 pub async fn seek(&self, position_ms: i64) {
397 if let Some(handle) = &*self.track_handle.lock().await {
398 handle.seek(position_ms.max(0) as u64);
399 }
400 }
401 pub async fn set_volume(&self, volume: f64) {
402 let vol_f = volume as f32;
403 self.volume.store(vol_f.to_bits(), Ordering::Relaxed);
404 if let Some(handle) = &*self.track_handle.lock().await {
405 handle.set_volume(vol_f);
406 }
407 let guard = self.event_sender.lock().await;
408 if let Some(ref e) = *guard {
409 e.send("volume", json!({ "volume": volume }));
410 }
411 }
412 pub fn get_position(&self) -> i64 {
413 let handle_guard = futures::executor::block_on(self.track_handle.lock());
414 if let Some(handle) = &*handle_guard {
415 handle.get_position() as i64
416 } else {
417 0
418 }
419 }
420 pub fn is_paused(&self) -> bool {
421 let handle_guard = futures::executor::block_on(self.track_handle.lock());
422 if let Some(handle) = &*handle_guard {
423 matches!(handle.get_state(), PlayState::Paused)
424 } else {
425 self.paused.load(Ordering::Acquire)
426 }
427 }
428 pub async fn set_filters(&self, filters_json: String) -> Result<(), String> {
429 let filters: Filters = serde_json::from_str(&filters_json)
430 .map_err(|e| format!("Invalid filters JSON: {e}"))?;
431 let new_chain = FilterChain::from_config(&filters);
432 {
433 let mut filter_chain_guard = self.filter_chain.lock().await;
434 *filter_chain_guard = new_chain;
435 }
436 Ok(())
437 }
438}
439async fn load_and_resolve_first_result(
440 sm: &crate::sources::manager::SourceManager,
441 url: &str,
442) -> Result<crate::sources::playable_track::BoxedTrack, String> {
443 match sm.load(url, None).await {
444 crate::protocol::tracks::LoadResult::Track(track) => {
445 sm.resolve_track(&track.info, None).await
446 }
447 crate::protocol::tracks::LoadResult::Search(tracks) => {
448 if tracks.is_empty() {
449 return Err("Search returned no results".to_string());
450 }
451 sm.resolve_track(&tracks[0].info, None).await
452 }
453 crate::protocol::tracks::LoadResult::Playlist(playlist) => {
454 if playlist.tracks.is_empty() {
455 return Err("Playlist is empty".to_string());
456 }
457 sm.resolve_track(&playlist.tracks[0].info, None).await
458 }
459 _ => Err(format!("Failed to load track or query: {url}")),
460 }
461}