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

bramley_engine/
player.rs
