bramley_engine/
player.rs

1use std::sync::mpsc::{Receiver, RecvTimeoutError, Sender, channel};
2use std::time::{Duration, Instant};
3
4use anyhow::Result;
5use serde_json::{Value as J, json};
6
7use crate::{Engine, EngineConfig, redact};
8
9pub(crate) const PLAYER: &str = include_str!("../js/player.js");
10const MUSICKIT_URL: &str = "https://js-cdn.music.apple.com/musickit/v3/musickit.js";
11const PLAYER_STACK: usize = 8 * 1024 * 1024;
12const COMMAND_WAIT_MS: u64 = 5000;
13pub const QUEUE_SHOWN: usize = 50;
14pub const COMMAND_WAIT: Duration = Duration::from_millis(COMMAND_WAIT_MS);
15const HEARTBEAT: Duration = Duration::from_secs(1);
16#[cfg(feature = "diagnostics")]
17const ENGINE_STATUS_EVERY: Duration = Duration::from_secs(1);
18
19pub struct Tokens {
20    pub developer: String,
21    pub user: String,
22}
23
24#[derive(Clone)]
25#[non_exhaustive]
26pub struct PlayerConfig {
27    pub engine: EngineConfig,
28    pub snapshot_interval: Duration,
29    #[cfg(feature = "sabotage")]
30    pub truncate_musickit: bool,
31}
32
33impl Default for PlayerConfig {
34    fn default() -> Self {
35        PlayerConfig {
36            engine: EngineConfig::default(),
37            snapshot_interval: Duration::from_millis(250),
38            #[cfg(feature = "sabotage")]
39            truncate_musickit: false,
40        }
41    }
42}
43
44impl std::fmt::Debug for PlayerConfig {
45    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
46        f.debug_struct("PlayerConfig")
47            .field("engine", &self.engine)
48            .field("snapshot_interval", &self.snapshot_interval)
49            .finish_non_exhaustive()
50    }
51}
52
53#[derive(Clone, Debug, PartialEq)]
54pub enum Queue {
55    Songs(Vec<String>),
56    Album(String),
57    Playlist(String),
58}
59
60#[derive(Clone, Debug, PartialEq)]
61pub enum Command {
62    SetQueue { queue: Queue, start: usize },
63    Play,
64    Pause,
65    Seek(f64),
66    Next,
67    Previous,
68    Volume(f32),
69    PlayLater(Queue),
70    Shuffle(bool),
71    TopUp(Queue),
72    PlayNext(Queue),
73    JumpTo { index: usize, id: String },
74    Shutdown,
75}
76
77#[derive(Clone, Copy, Debug, Default, PartialEq)]
78pub enum PlayState {
79    #[default]
80    Stopped,
81    Loading,
82    Playing,
83    Paused,
84}
85
86#[derive(Clone, Debug, PartialEq)]
87pub enum PlayerError {
88    SignInNeeded(String),
89    Playback(String),
90    LoadFailed(String),
91}
92
93#[cfg(feature = "diagnostics")]
94#[derive(Clone, Debug, PartialEq)]
95pub struct Diagnostics {
96    pub js: J,
97    pub engine: Option<J>,
98}
99
100#[derive(Clone, Debug, Default, PartialEq)]
101pub struct QueueItem {
102    pub id: String,
103    pub title: Option<String>,
104    pub duration: Option<f64>,
105}
106
107#[derive(Clone, Debug, Default, PartialEq)]
108#[non_exhaustive]
109pub struct Snapshot {
110    pub item_id: Option<String>,
111    pub title: Option<String>,
112    pub artist: Option<String>,
113    pub position: Option<f64>,
114    pub duration: Option<f64>,
115    pub state: PlayState,
116    pub error: Option<PlayerError>,
117    pub preview_only: bool,
118    pub applied: u64,
119    pub drains_acked: u64,
120    pub drain_timeouts: u64,
121    pub sent_us: u64,
122    pub skipped_us: u64,
123    pub decrypted_us: u64,
124    pub readiness_dips: u64,
125    pub decrypt_fails: u64,
126    pub output_underflows: u64,
127    pub queue: Vec<QueueItem>,
128    pub queue_position: Option<usize>,
129    pub queue_length: usize,
130    pub background_done: u64,
131    #[cfg(feature = "diagnostics")]
132    pub diagnostics: Option<Diagnostics>,
133}
134
135impl Snapshot {
136    pub fn http_statuses(&self, path: &str) -> J {
137        #[cfg(feature = "diagnostics")]
138        if let Some(J::Object(seen)) = self
139            .diagnostics
140            .as_ref()
141            .and_then(|d| d.engine.as_ref())
142            .map(|e| &e["platform"]["xhr"]["endpoints"])
143        {
144            let ours = seen.iter().filter(|(key, _)| key.contains(path));
145            return ours
146                .map(|(key, v)| (key.clone(), v["statuses"].clone()))
147                .collect();
148        }
149        let _ = path;
150        J::Null
151    }
152}
153
154pub(crate) trait Backend {
155    fn tick(&mut self) -> Duration;
156    fn eval_script(&self, name: &str, source: &str) -> Result<()>;
157    fn eval_json(&self, expression: &str) -> Result<J>;
158    fn applied(&self) -> u64;
159    fn current_time(&self) -> Option<f64>;
160    fn output_failed(&self) -> bool;
161    fn output_counts(&self) -> crate::runtime::OutputCounts;
162    #[cfg(feature = "diagnostics")]
163    fn status(&self) -> J;
164}
165
166impl Backend for Engine {
167    fn tick(&mut self) -> Duration {
168        Engine::tick(self)
169    }
170    fn eval_script(&self, name: &str, source: &str) -> Result<()> {
171        Engine::eval_script(self, name, source)
172    }
173    fn eval_json(&self, expression: &str) -> Result<J> {
174        Engine::eval_json(self, expression)
175    }
176    fn applied(&self) -> u64 {
177        Engine::global_number(self, "__player", "applied").map_or(0, |n| n as u64)
178    }
179    fn current_time(&self) -> Option<f64> {
180        Engine::current_time(self)
181    }
182    fn output_failed(&self) -> bool {
183        Engine::output_failed(self)
184    }
185    fn output_counts(&self) -> crate::runtime::OutputCounts {
186        Engine::output_counts(self)
187    }
188    #[cfg(feature = "diagnostics")]
189    fn status(&self) -> J {
190        Engine::status(self)
191    }
192}
193
194pub struct Player;
195
196impl Player {
197    pub fn spawn(
198        mut cfg: PlayerConfig,
199        tokens: Tokens,
200        waker: impl Fn() + Send + 'static,
201    ) -> Result<(Sender<Command>, Receiver<Snapshot>)> {
202        cfg.engine
203            .redact
204            .extend([tokens.developer.clone(), tokens.user.clone()]);
205        let secrets = cfg.engine.redact.clone();
206        let every = cfg.snapshot_interval;
207        let setup = json!({"developerToken": tokens.developer, "userToken": tokens.user, "musickitUrl": MUSICKIT_URL, "build": env!("CARGO_PKG_VERSION"), "commandWaitMs": COMMAND_WAIT_MS, "queueShown": QUEUE_SHOWN});
208        spawn_with(
209            move || {
210                let engine = Engine::new(cfg.engine)?;
211                #[cfg(feature = "diagnostics")]
212                engine.enable_diagnostics()?;
213                #[cfg(feature = "sabotage")]
214                if cfg.truncate_musickit {
215                    engine.sabotage_truncate_scripts();
216                }
217                Ok(engine)
218            },
219            format!("globalThis.__bramley_player = {setup};"),
220            secrets,
221            every,
222            waker,
223        )
224    }
225}
226
227struct Exit<W: Fn()> {
228    snapshots: Sender<Snapshot>,
229    waker: W,
230    applied: u64,
231}
232
233impl<W: Fn()> Exit<W> {
234    fn send(&self, snap: Snapshot) -> bool {
235        let sent = self.snapshots.send(snap).is_ok();
236        (self.waker)();
237        sent
238    }
239}
240
241impl<W: Fn()> Drop for Exit<W> {
242    fn drop(&mut self) {
243        if std::thread::panicking() {
244            self.send(Snapshot {
245                error: Some(PlayerError::LoadFailed("the player thread panicked".into())),
246                applied: self.applied,
247                ..Snapshot::default()
248            });
249        }
250    }
251}
252
253pub(crate) fn spawn_with<B: Backend>(
254    make: impl FnOnce() -> Result<B> + Send + 'static,
255    setup: String,
256    secrets: Vec<String>,
257    every: Duration,
258    waker: impl Fn() + Send + 'static,
259) -> Result<(Sender<Command>, Receiver<Snapshot>)> {
260    let (command_tx, commands) = channel();
261    let (snapshots, snapshot_rx) = channel();
262    std::thread::Builder::new()
263        .name("bramley-player".into())
264        .stack_size(PLAYER_STACK)
265        .spawn(move || {
266            let mut exit = Exit {
267                snapshots,
268                waker,
269                applied: 0,
270            };
271            let started = make().and_then(|b| {
272                b.eval_script("config.js", &setup)?;
273                b.eval_script("player.js", PLAYER)?;
274                Ok(b)
275            });
276            match started {
277                Ok(b) => run(b, &commands, &mut exit, every, &secrets),
278                Err(e) => {
279                    exit.send(Snapshot {
280                        error: Some(PlayerError::LoadFailed(redact(&format!("{e:#}"), &secrets))),
281                        ..Snapshot::default()
282                    });
283                }
284            }
285        })?;
286    Ok((command_tx, snapshot_rx))
287}
288
289fn run<B: Backend, W: Fn()>(
290    mut b: B,
291    commands: &Receiver<Command>,
292    exit: &mut Exit<W>,
293    every: Duration,
294    secrets: &[String],
295) {
296    let mut last: Option<Snapshot> = None;
297    let mut due = Instant::now();
298    let mut sent_at = Instant::now();
299    #[cfg(feature = "diagnostics")]
300    let mut engine_due = Instant::now();
301    loop {
302        let wait = b.tick().min(due.saturating_duration_since(Instant::now()));
303        match commands.recv_timeout(wait) {
304            Ok(Command::Shutdown) | Err(RecvTimeoutError::Disconnected) => return,
305            Ok(command) => {
306                if let Err(e) = b.eval_script("command", &command_js(&command)) {
307                    eprintln!("player command failed: {}", redact(&e.to_string(), secrets));
308                }
309                due = Instant::now();
310                continue;
311            }
312            Err(RecvTimeoutError::Timeout) => {}
313        }
314        if b.applied() > exit.applied {
315            due = Instant::now();
316        }
317        if Instant::now() < due {
318            continue;
319        }
320        due = Instant::now() + every;
321        let mut snap = snapshot(&b, secrets);
322        snap.applied = snap.applied.max(exit.applied);
323        exit.applied = snap.applied;
324        #[cfg(feature = "diagnostics")]
325        if let Some(d) = snap.diagnostics.as_mut()
326            && Instant::now() >= engine_due
327        {
328            engine_due = Instant::now() + ENGINE_STATUS_EVERY;
329            d.engine = Some(b.status());
330        }
331        if last.as_ref() != Some(&snap) || sent_at.elapsed() >= HEARTBEAT {
332            if !exit.send(snap.clone()) {
333                return;
334            }
335            sent_at = Instant::now();
336            last = Some(snap);
337        }
338    }
339}
340
341pub const OUTPUT_FAILED: &str = "the audio output could not be opened";
342
343pub(crate) fn command_js(command: &Command) -> String {
344    let (method, arg) = match command {
345        Command::SetQueue { queue, start } => {
346            let mut options = queue_options(queue);
347            options["startWith"] = json!(start);
348            ("setQueue", options)
349        }
350        Command::Play => ("play", J::Null),
351        Command::Pause => ("pause", J::Null),
352        Command::Seek(t) => ("seek", json!(t)),
353        Command::Next => ("next", J::Null),
354        Command::Previous => ("previous", J::Null),
355        Command::Volume(v) => ("volume", json!(v)),
356        Command::PlayLater(queue) => ("playLater", queue_options(queue)),
357        Command::Shuffle(on) => ("shuffle", json!(on)),
358        Command::TopUp(queue) => ("topUp", queue_options(queue)),
359        Command::PlayNext(queue) => ("playNext", queue_options(queue)),
360        Command::JumpTo { index, id } => ("jumpTo", json!([index, id])),
361        Command::Shutdown => return String::new(),
362    };
363    format!("__player.{method}({arg});")
364}
365
366fn queue_options(queue: &Queue) -> J {
367    match queue {
368        Queue::Songs(ids) => json!({"songs": ids}),
369        Queue::Album(id) => json!({"album": id}),
370        Queue::Playlist(id) => json!({"playlist": id}),
371    }
372}
373
374fn snapshot<B: Backend>(b: &B, secrets: &[String]) -> Snapshot {
375    let js = match b.eval_json("__player.state()") {
376        Ok(js) => serde_json::from_str(&redact(&js.to_string(), secrets)).unwrap_or(J::Null),
377        Err(e) => {
378            return Snapshot {
379                error: Some(PlayerError::LoadFailed(redact(&e.to_string(), secrets))),
380                ..Snapshot::default()
381            };
382        }
383    };
384    let text = |v: &J| v.as_str().map(str::to_owned);
385    let seconds = |v: &J| v.as_f64().map(|ms| ms / 1000.0);
386    let queue = &js["queue"];
387    let shown = queue["items"].as_array().map_or(&[][..], Vec::as_slice);
388    let shown = shown.iter().take(QUEUE_SHOWN).map(|item| QueueItem {
389        id: text(&item["id"]).unwrap_or_default(),
390        title: text(&item["title"]),
391        duration: seconds(&item["durationInMillis"]),
392    });
393    let error = if let Some(reason) = text(&js["signInNeeded"]) {
394        Some(PlayerError::SignInNeeded(reason))
395    } else if let Some(phase @ ("musickit-load-failed" | "configure-failed")) = js["phase"].as_str()
396    {
397        Some(PlayerError::LoadFailed(phase.into()))
398    } else if b.output_failed() {
399        Some(PlayerError::Playback(OUTPUT_FAILED.into()))
400    } else {
401        text(&js["playbackError"]).map(PlayerError::Playback)
402    };
403    let [
404        drains_acked,
405        drain_timeouts,
406        sent_us,
407        skipped_us,
408        decrypted_us,
409        readiness_dips,
410        decrypt_fails,
411        output_underflows,
412    ] = b.output_counts();
413    Snapshot {
414        item_id: text(&js["item"]["id"]),
415        title: text(&js["item"]["title"]),
416        artist: text(&js["item"]["artistName"]),
417        position: b.current_time(),
418        duration: seconds(&js["item"]["durationInMillis"]),
419        state: match js["playbackState"].as_u64() {
420            Some(1 | 6 | 8 | 9) => PlayState::Loading,
421            Some(2) => PlayState::Playing,
422            Some(3) => PlayState::Paused,
423            _ => PlayState::Stopped,
424        },
425        error,
426        preview_only: js["previewOnly"] == true,
427        applied: js["applied"].as_u64().unwrap_or(0),
428        drains_acked,
429        drain_timeouts,
430        sent_us,
431        skipped_us,
432        decrypted_us,
433        readiness_dips,
434        decrypt_fails,
435        output_underflows,
436        queue: shown.collect(),
437        queue_position: queue["position"].as_u64().map(|at| at as usize),
438        queue_length: queue["length"].as_u64().unwrap_or(0) as usize,
439        background_done: js["background"].as_u64().unwrap_or(0),
440        #[cfg(feature = "diagnostics")]
441        diagnostics: Some(Diagnostics {
442            js: serde_json::from_str(&without_queue(&js)).unwrap_or(J::Null),
443            engine: None,
444        }),
445    }
446}
447
448#[cfg(feature = "diagnostics")]
449fn without_queue(js: &J) -> String {
450    let mut js = js.clone();
451    if let Some(state) = js.as_object_mut() {
452        state.remove("queue");
453    }
454    js.to_string()
455}
bramley_engine/player.rs

bramley_engine/
player.rs
