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}