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

bramley_engine/
player.rs
