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

bramley_engine/
player.rs
