bramley_engine/
player.rs

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

bramley_engine/
player.rs
