Skip to main content

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