clusterfuck
/
crabidy
4
1
Fork 0
Code Issues 15 Pull Requests Actions Packages Projects 1 Releases 5 Wiki Activity

Fix intermittent playback stops and harden the queue 

Root causes found and fixed:

- QueueManager could panic and kill the playback task permanently:
  is_last_track() underflowed on an empty queue, remove_tracks accepted
  pos == len (Vec::remove panic) and corrupted positions when removing
  multiple tracks (indices shifted mid-loop), shuffle_behind indexed
  out of range on an empty play order, insert_tracks shifted play-order
  entries by the queue length instead of the inserted count and then
  assert!()ed on the resulting inconsistency, and clear() left
  play_order stale. All mutation methods are now guarded, multi-remove
  works highest-position-first, and an inconsistent play order is
  rebuilt instead of panicking. Regression tests cover these cases.

- The tidal access token was only obtained at startup and never
  refreshed, so long-running sessions ended with every track fetch
  failing (playback just stopped at the next track boundary). Login
  state now lives behind a lock; tokens are refreshed proactively
  before expiry (5 min margin) and once reactively on a 401, and all
  API responses are status-checked (new ClientError::ApiError) instead
  of being fed to the JSON decoder blind. The http client also got a
  30s timeout so a hung connection cannot wedge the provider loop.

- (from the rodio rewrite, same bug class) end of stream used to be
  detected by string-comparing an io::Error message; any other decode
  or network error ended the stream silently without an EndOfStream
  message, so playback never advanced. EOS is now a guaranteed
  callback with a generation counter.

Plus workspace-wide clippy cleanup (zero warnings), cargo-machete
cleanup, and fmt.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
AI User 2026-07-19 22:00:22 +02:00
parent 56bc0b0d04
commit d504ebc85f
16 changed files with 394 additions and 278 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
Cargo.lock generated
View File

@ -721,7 +721,6 @@ name = "crabidy-core"
version = "0.1.0"
dependencies = [
"async-trait",
"clap",
"clap-serde-derive",
"dirs",
"prost",
@ -743,7 +742,6 @@ dependencies = [
"dirs",
"flume",
"futures",
"log",
"rand 0.10.2",
"tidaldy",
"tokio",

1
Cargo.toml
View File

@ -23,7 +23,6 @@ crossterm = "0.29"
dirs = "6"
flume = "0.12"
futures = "0.3"
log = "0.4"
notify-rust = "4"
prost = "0.14"
rand = "0.10"

12
audio-player/examples/basic.rs
View File

@ -1,5 +1,3 @@
use std::{thread, time::Duration};
use audio_player::{Player, PlayerMessage};
#[tokio::main]
@ -13,7 +11,10 @@ async fn main() {
loop {
match player.messages.recv_async().await {
Ok(PlayerMessage::Elapsed { duration, elapsed }) => {
Ok(PlayerMessage::Elapsed {
duration: _,
elapsed,
}) => {
println!("ELAPSED: {:?}", elapsed);
}
Ok(PlayerMessage::EndOfStream) => {
@ -30,7 +31,10 @@ async fn main() {
loop {
match player.messages.recv_async().await {
Ok(PlayerMessage::Elapsed { duration, elapsed }) => {
Ok(PlayerMessage::Elapsed {
duration: _,
elapsed,
}) => {
println!("ELAPSED: {:?}", elapsed);
}
Ok(PlayerMessage::EndOfStream) => {

9
cbd-tui/src/app/library.rs
View File

@ -54,14 +54,15 @@ impl Library {
}
pub fn ascend(&mut self) {
if let Some(parent) = self.parent.as_ref() {
self.tx.send(MessageFromUi::GetLibraryNode(parent.clone()));
let _ = self.tx.send(MessageFromUi::GetLibraryNode(parent.clone()));
}
}
pub fn dive(&mut self) {
if let Some(idx) = self.list_state.selected() {
let item = &self.list[idx];
if let UiItemKind::Node = item.kind {
self.tx
let _ = self
.tx
.send(MessageFromUi::GetLibraryNode(item.path.clone()));
}
}
@ -103,7 +104,7 @@ impl Library {
}
pub fn toggle_mark(&mut self) {
if let Some(idx) = self.list_state.selected() {
let mut item = &mut self.list[idx];
let item = &mut self.list[idx];
if !item.is_queable {
return;
}
@ -176,7 +177,7 @@ impl Library {
} else {
Style::default()
};
return ListItem::new(Span::from(text)).style(style);
ListItem::new(Span::from(text)).style(style)
})
.collect();

6
cbd-tui/src/app/list.rs
View File

@ -1,5 +1,3 @@
pub use ratatui::widgets::ListState;
// FIXME: Move marking stuff here, to be able to use it in queue as well
pub trait StatefulList {
fn get_size(&self) -> usize;
@ -72,10 +70,6 @@ pub trait StatefulList {
}
}
fn is_selected(&self) -> bool {
self.selected().is_some()
}
fn is_empty(&self) -> bool {
self.get_size() == 0
}

2
cbd-tui/src/app/mod.rs
View File

@ -105,7 +105,7 @@ impl App {
}
pub fn render(&mut self, f: &mut Frame) {
let full_screen = f.area();
let _full_screen = f.area();
let library_focused = matches!(self.focus, UiFocus::Library);
let queue_focused = matches!(self.focus, UiFocus::Queue);

6
cbd-tui/src/app/now_playing.rs
View File

@ -68,7 +68,7 @@ impl NowPlaying {
self.track = active;
}
pub fn update_modifiers(&mut self, mods: &QueueModifiers) {
self.modifiers = mods.clone();
self.modifiers = *mods;
}
pub fn render(&self, f: &mut Frame, area: Rect) {
@ -129,7 +129,7 @@ impl NowPlaying {
f.render_widget(media_info_p, now_playing_layout[0]);
if let (Some(position), Some(duration), Some(track)) =
if let (Some(position), Some(duration), Some(_track)) =
(self.position, self.duration, &self.track)
{
let pos = position.as_secs();
@ -151,7 +151,7 @@ impl NowPlaying {
let progress = LineGauge::default()
.label("")
.block(Block::default().borders(Borders::NONE))
.gauge_style(Style::default().fg(COLOR_SECONDARY).bg(Color::Black))
.filled_style(Style::default().fg(COLOR_SECONDARY).bg(Color::Black))
.ratio(ratio);
f.render_widget(progress, elapsed_layout[0]);

11
cbd-tui/src/app/queue.rs
View File

@ -30,14 +30,14 @@ impl Queue {
}
}
pub fn play_next(&self) {
self.tx.send(MessageFromUi::NextTrack);
let _ = self.tx.send(MessageFromUi::NextTrack);
}
pub fn play_prev(&self) {
self.tx.send(MessageFromUi::PrevTrack);
let _ = self.tx.send(MessageFromUi::PrevTrack);
}
pub fn play_selected(&self) {
if let Some(pos) = self.selected() {
self.tx.send(MessageFromUi::SetCurrentTrack(pos));
let _ = self.tx.send(MessageFromUi::SetCurrentTrack(pos));
}
}
pub fn select_current(&mut self) {
@ -46,7 +46,7 @@ impl Queue {
pub fn remove_track(&mut self) {
if let Some(pos) = self.selected() {
// FIXME: mark multiple tracks on queue and remove them
self.tx.send(MessageFromUi::RemoveTracks(vec![pos]));
let _ = self.tx.send(MessageFromUi::RemoveTracks(vec![pos]));
}
}
pub fn update_position(&mut self, pos: usize) {
@ -57,8 +57,7 @@ impl Queue {
self.list = queue
.tracks
.iter()
.enumerate()
.map(|(i, t)| UiItem {
.map(|t| UiItem {
path: t.path.clone(),
title: format!("{} - {}", t.artist, t.title),
kind: UiItemKind::Track,

31
cbd-tui/src/main.rs
View File

@ -82,7 +82,7 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
Ok(())
}
async fn orchestrate<'a>(
async fn orchestrate(
config: &'static Config,
(tx, rx): (Sender<MessageToUi>, Receiver<MessageFromUi>),
) -> Result<(), Box<dyn Error>> {
@ -114,7 +114,7 @@ async fn poll(
match msg {
MessageFromUi::GetLibraryNode(path) => {
if let Some(node) = rpc_client.get_library_node(&path).await? {
tx.send(MessageToUi::ReplaceLibraryNode(node.clone()));
let _ = tx.send(MessageToUi::ReplaceLibraryNode(node.clone()));
}
},
MessageFromUi::AppendTracks(uuids) => {
@ -211,7 +211,7 @@ fn run_ui(tx: Sender<MessageFromUi>, rx: Receiver<MessageToUi>) {
app.now_playing.update_track(track.track);
app.queue.update_position(track.queue_position as usize);
}
if let Some(ps) = PlayState::try_from(init_data.play_state).ok() {
if let Ok(ps) = PlayState::try_from(init_data.play_state) {
app.now_playing.update_play_state(ps);
}
if let Some(mods) = init_data.mods {
@ -228,7 +228,7 @@ fn run_ui(tx: Sender<MessageFromUi>, rx: Receiver<MessageToUi>) {
}
StreamUpdate::Position(pos) => app.now_playing.update_position(pos),
StreamUpdate::PlayState(play_state) => {
if let Some(ps) = PlayState::try_from(play_state).ok() {
if let Ok(ps) = PlayState::try_from(play_state) {
app.now_playing.update_play_state(ps);
}
}
@ -241,7 +241,10 @@ fn run_ui(tx: Sender<MessageFromUi>, rx: Receiver<MessageToUi>) {
}
}
terminal.draw(|f| app.render(f));
if let Err(err) = terminal.draw(|f| app.render(f)) {
error!("failed to draw frame: {err}");
break;
}
let timeout = tick_rate
.checked_sub(last_tick.elapsed())
@ -256,25 +259,25 @@ fn run_ui(tx: Sender<MessageFromUi>, rx: Receiver<MessageToUi>) {
}
(_, KeyModifiers::NONE, KeyCode::Tab) => app.cycle_active(),
(_, KeyModifiers::NONE, KeyCode::Char(' ')) => {
tx.send(MessageFromUi::TogglePlay);
let _ = tx.send(MessageFromUi::TogglePlay);
}
(_, KeyModifiers::NONE, KeyCode::Char('r')) => {
tx.send(MessageFromUi::RestartTrack);
let _ = tx.send(MessageFromUi::RestartTrack);
}
(_, KeyModifiers::SHIFT, KeyCode::Char('J')) => {
tx.send(MessageFromUi::ChangeVolume(-0.1));
let _ = tx.send(MessageFromUi::ChangeVolume(-0.1));
}
(_, KeyModifiers::SHIFT, KeyCode::Char('K')) => {
tx.send(MessageFromUi::ChangeVolume(0.1));
let _ = tx.send(MessageFromUi::ChangeVolume(0.1));
}
(_, KeyModifiers::NONE, KeyCode::Char('m')) => {
tx.send(MessageFromUi::ToggleMute);
let _ = tx.send(MessageFromUi::ToggleMute);
}
(_, KeyModifiers::NONE, KeyCode::Char('z')) => {
tx.send(MessageFromUi::ToggleShuffle);
let _ = tx.send(MessageFromUi::ToggleShuffle);
}
(_, KeyModifiers::NONE, KeyCode::Char('x')) => {
tx.send(MessageFromUi::ToggleRepeat);
let _ = tx.send(MessageFromUi::ToggleRepeat);
}
(_, KeyModifiers::CONTROL, KeyCode::Char('n')) => {
app.queue.play_next();
@ -351,10 +354,10 @@ fn run_ui(tx: Sender<MessageFromUi>, rx: Receiver<MessageToUi>) {
app.queue.remove_track();
}
(UiFocus::Queue, KeyModifiers::NONE, KeyCode::Char('c')) => {
tx.send(MessageFromUi::ClearQueue(true));
let _ = tx.send(MessageFromUi::ClearQueue(true));
}
(UiFocus::Queue, KeyModifiers::SHIFT, KeyCode::Char('C')) => {
tx.send(MessageFromUi::ClearQueue(false));
let _ = tx.send(MessageFromUi::ClearQueue(false));
}
_ => {}
}

6
crabidy-core/Cargo.toml
View File

@ -5,7 +5,6 @@ edition.workspace = true
[dependencies]
async-trait.workspace = true
clap.workspace = true
clap-serde-derive.workspace = true
dirs.workspace = true
prost.workspace = true
@ -16,3 +15,8 @@ tonic-prost.workspace = true
[build-dependencies]
tonic-prost-build.workspace = true
# prost and tonic-prost are used by the code generated from the proto files,
# which cargo-machete cannot see.
[package.metadata.cargo-machete]
ignored = ["prost", "tonic-prost"]

62
crabidy-core/src/lib.rs
View File

@ -101,37 +101,6 @@ impl LibraryNodeChild {
}
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn parent_path_walks_up_to_root() {
assert_eq!(
parent_path("/tidal/playlists/abc"),
Some("/tidal/playlists")
);
assert_eq!(parent_path("/tidal/playlists"), Some("/tidal"));
assert_eq!(parent_path("/tidal"), Some("/"));
assert_eq!(parent_path("/"), None);
}
#[test]
fn join_path_appends_segments() {
assert_eq!(join_path("/", "tidal"), "/tidal");
assert_eq!(join_path("/tidal", "playlists"), "/tidal/playlists");
}
#[test]
fn path_segments_splits() {
assert_eq!(path_segments("/"), Vec::<&str>::new());
assert_eq!(
path_segments("/tidal/artists/1/2"),
vec!["tidal", "artists", "1", "2"]
);
}
}
pub enum QueueError {
NotQueable,
}
@ -164,3 +133,34 @@ where
}
T::default().merge_clap()
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn parent_path_walks_up_to_root() {
assert_eq!(
parent_path("/tidal/playlists/abc"),
Some("/tidal/playlists")
);
assert_eq!(parent_path("/tidal/playlists"), Some("/tidal"));
assert_eq!(parent_path("/tidal"), Some("/"));
assert_eq!(parent_path("/"), None);
}
#[test]
fn join_path_appends_segments() {
assert_eq!(join_path("/", "tidal"), "/tidal");
assert_eq!(join_path("/tidal", "playlists"), "/tidal/playlists");
}
#[test]
fn path_segments_splits() {
assert_eq!(path_segments("/"), Vec::<&str>::new());
assert_eq!(
path_segments("/tidal/artists/1/2"),
vec!["tidal", "artists", "1", "2"]
);
}
}

1
crabidy-server/Cargo.toml
View File

@ -15,7 +15,6 @@ crabidy-core.workspace = true
dirs.workspace = true
flume.workspace = true
futures.workspace = true
log.workspace = true
rand.workspace = true
tidaldy.workspace = true
tokio = { workspace = true, features = ["full"] }

187
crabidy-server/src/lib.rs
View File

@ -28,6 +28,12 @@ impl From<QueueManager> for Queue {
}
}
impl Default for QueueManager {
fn default() -> Self {
Self::new()
}
}
impl QueueManager {
pub fn new() -> Self {
Self {
@ -48,7 +54,7 @@ impl QueueManager {
}
pub fn is_last_track(&self) -> bool {
self.current_position() == self.tracks.len() - 1
!self.tracks.is_empty() && self.current_position() == self.tracks.len() - 1
}
pub fn shuffle_on(&mut self) {
@ -69,11 +75,15 @@ impl QueueManager {
}
pub fn shuffle_before(&mut self, pos: usize) {
self.play_order[..pos].shuffle(&mut rng());
if let Some(slice) = self.play_order.get_mut(..pos) {
slice.shuffle(&mut rng());
}
}
pub fn shuffle_behind(&mut self, pos: usize) {
self.play_order[pos + 1..].shuffle(&mut rng());
if let Some(slice) = self.play_order.get_mut(pos + 1..) {
slice.shuffle(&mut rng());
}
}
pub fn current_track(&self) -> Option<Track> {
@ -114,7 +124,7 @@ impl QueueManager {
pub fn prev_track(&mut self) -> Option<Track> {
if 0 < self.current_offset {
self.current_offset -= 1;
Some(self.tracks[self.current_position()].clone())
self.current_track()
} else {
None
}
@ -178,28 +188,37 @@ impl QueueManager {
pub fn remove_tracks(&mut self, positions: &[u32]) -> Option<Track> {
let mut play_next = false;
// Remove highest positions first so earlier removals don't shift the
// positions that are still to be removed.
let mut positions: Vec<usize> = positions.iter().map(|p| *p as usize).collect();
positions.sort_unstable_by(|a, b| b.cmp(a));
positions.dedup();
for pos in positions {
if (self.tracks.len() as u32) < *pos {
return None;
};
if *pos == self.current_position() as u32 {
if pos >= self.tracks.len() {
debug!(pos, len = self.tracks.len(), "ignoring out-of-range remove");
continue;
}
if pos == self.current_position() {
play_next = true;
}
let Some(offset) = self.play_order.iter().position(|&i| i == *pos as usize) else {
error!("invalid current position");
error!("queue: {:#?}", self);
let Some(offset) = self.play_order.iter().position(|&i| i == pos) else {
error!(pos, "track position missing from play order, rebuilding");
self.rebuild_play_order();
return None;
};
if offset < self.current_offset {
self.current_offset -= 1;
}
self.tracks.remove(*pos as usize);
self.tracks.remove(pos);
self.play_order.remove(offset);
self.play_order
.iter_mut()
.filter(|i| (*pos as usize) < **i)
.filter(|i| pos < **i)
.for_each(|i| *i -= 1);
}
if self.current_offset >= self.play_order.len() {
self.current_offset = 0;
}
if play_next {
self.current_track()
} else {
@ -212,38 +231,36 @@ impl QueueManager {
if len == 0 {
return self.replace_with_tracks(tracks);
}
let order_additions: Vec<usize> = (len..len + tracks.len()).collect();
let inserted = tracks.len();
let position = (position as usize).min(len - 1);
let order_additions: Vec<usize> = (len..len + inserted).collect();
self.play_order.extend(order_additions);
let tail: Vec<Track> = self
.tracks
.splice((position as usize + 1).., tracks.to_vec())
.splice(position + 1.., tracks.to_vec())
.collect();
self.tracks.extend(tail);
let mut changed: Vec<usize> = Vec::new();
// in shuffle mode, it might be that we played already postions which are behind
// the insertion point and which postions are shifted by the lenght of the inserted
// track
// In shuffle mode we may already have played positions that are
// behind the insertion point; those shift by the number of inserted
// tracks.
for i in self
.play_order
.iter_mut()
.take(self.current_offset)
.filter(|i| (position as usize) < **i)
.filter(|i| position < **i)
{
*i += len;
*i += inserted;
changed.push(*i);
}
if !self.shuffle {
// if we don't shuffle, there should be no positions alredy played behind the
// current track
assert!(changed.is_empty());
}
// the newly inserted indices need to replaced with the ones that we already handled
// The freshly appended order entries need to swap with the shifted
// ones so every index stays unique.
self.play_order
.iter_mut()
.skip(self.current_offset)
.for_each(|i| {
if changed.contains(i) {
*i -= len;
*i -= inserted;
}
});
@ -262,28 +279,126 @@ impl QueueManager {
let current_track = self.current_track();
self.current_offset = 0;
self.tracks.clear();
self.play_order.clear();
if exclude_current {
if let Some(track) = current_track {
self.tracks.push(track);
self.play_order.push(0);
}
}
!exclude_current
}
/// Restores play_order to a consistent state after an inconsistency was
/// detected. Loses shuffle history but keeps the queue playable.
fn rebuild_play_order(&mut self) {
self.play_order = (0..self.tracks.len()).collect();
self.current_offset = 0;
}
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn random_delete_before() {}
#[test]
fn random_delete_track() {}
#[test]
fn random_delete_after() {}
#[test]
fn random_select_track() {}
fn track(id: usize) -> Track {
Track {
path: format!("/tidal/playlists/p/{id}"),
artist: "artist".to_string(),
title: format!("track {id}"),
duration: None,
album: None,
}
}
fn queue_with(n: usize) -> QueueManager {
let mut q = QueueManager::new();
let tracks: Vec<Track> = (0..n).map(track).collect();
q.replace_with_tracks(&tracks);
q
}
#[test]
fn empty_queue_operations_do_not_panic() {
let mut q = QueueManager::new();
assert!(!q.is_last_track());
assert!(q.current_track().is_none());
assert!(q.next_track().is_none());
assert!(q.prev_track().is_none());
assert!(q.remove_tracks(&[0]).is_none());
q.shuffle_on();
q.shuffle_off();
q.clear(true);
q.clear(false);
}
#[test]
fn remove_out_of_range_is_ignored() {
let mut q = queue_with(2);
assert!(q.remove_tracks(&[5]).is_none());
assert_eq!(q.tracks.len(), 2);
// pos == len used to panic via Vec::remove
assert!(q.remove_tracks(&[2]).is_none());
assert_eq!(q.tracks.len(), 2);
}
#[test]
fn remove_multiple_positions() {
let mut q = queue_with(4);
q.remove_tracks(&[1, 3]);
assert_eq!(q.tracks.len(), 2);
assert_eq!(q.play_order.len(), 2);
assert_eq!(q.current_track().unwrap().title, "track 0");
}
#[test]
fn remove_current_returns_successor() {
let mut q = queue_with(3);
let next = q.remove_tracks(&[0]);
assert_eq!(next.unwrap().title, "track 1");
assert_eq!(q.tracks.len(), 2);
}
#[test]
fn clear_keeps_play_order_consistent() {
let mut q = queue_with(3);
q.next_track();
q.clear(true);
assert_eq!(q.tracks.len(), 1);
assert_eq!(q.play_order.len(), 1);
assert!(q.current_track().is_some());
assert!(q.next_track().is_none());
}
#[test]
fn next_track_advances_and_repeats() {
let mut q = queue_with(2);
assert_eq!(q.next_track().unwrap().title, "track 1");
assert!(q.next_track().is_none());
q.repeat = true;
assert_eq!(q.next_track().unwrap().title, "track 0");
}
#[test]
fn insert_past_end_appends() {
let mut q = queue_with(2);
q.insert_tracks(99, &[track(2)]);
assert_eq!(q.tracks.len(), 3);
assert_eq!(q.play_order.len(), 3);
assert_eq!(q.tracks.last().unwrap().title, "track 2");
}
#[test]
fn shuffle_insert_keeps_order_unique() {
let mut q = queue_with(5);
q.shuffle_on();
q.next_track();
q.next_track();
q.queue_tracks(&[track(5), track(6)]);
let mut order = q.play_order.clone();
order.sort_unstable();
assert_eq!(order, (0..7).collect::<Vec<usize>>());
}
}

2
tidaldy/src/config.rs
View File

@ -2,7 +2,7 @@ use serde::{Deserialize, Serialize};
use std::iter::zip;
use thiserror::Error;
#[derive(Debug, Serialize, Deserialize)]
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct Settings {
pub base_url: String,
pub hifi_url: String,

323
tidaldy/src/lib.rs
View File

@ -13,8 +13,14 @@ pub use models::*;
pub struct Client {
http_client: HttpClient,
settings: config::Settings,
/// Login state changes at runtime when tokens are refreshed, while the
/// client is shared immutably, hence the lock. Never held across awaits.
login: std::sync::RwLock<config::LoginConfig>,
}
/// Refresh the access token this long before it actually expires.
const TOKEN_REFRESH_MARGIN_SECS: u64 = 300;
#[async_trait]
impl crabidy_core::ProviderClient for Client {
#[instrument(skip(raw_toml_settings))]
@ -27,17 +33,19 @@ impl crabidy_core::ProviderClient for Client {
};
let mut client = Self::new(settings)?;
if let Ok(_) = client.login_config().await {
if client.login_config().await.is_ok() {
return Ok(client);
}
if let Ok(_) = client.login_web().await {
if client.login_web().await.is_ok() {
return Ok(client);
}
Err(crabidy_core::ProviderError::CouldNotLogin)
}
#[instrument(skip(self))]
fn settings(&self) -> String {
toml::to_string_pretty(&self.settings).unwrap_or_default()
let mut settings = self.settings.clone();
settings.login = self.login_snapshot();
toml::to_string_pretty(&settings).unwrap_or_default()
}
fn is_track_path(&self, path: &str) -> bool {
matches!(
@ -113,7 +121,7 @@ impl crabidy_core::ProviderClient for Client {
&self,
path: &str,
) -> Result<crabidy_core::proto::crabidy::LibraryNode, crabidy_core::ProviderError> {
let Some(user_id) = self.settings.login.user_id.clone() else {
let Some(user_id) = self.get_user_id() else {
return Err(crabidy_core::ProviderError::UnknownUser);
};
let parsed = parse_path(path)?;
@ -293,17 +301,94 @@ impl Client {
pub fn new(settings: config::Settings) -> Result<Self, ClientError> {
let http_client = HttpClient::builder()
.user_agent("Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.124 Safari/537.36 Edg/91.0.864.59")
.timeout(std::time::Duration::from_secs(30))
.build()?;
let login = std::sync::RwLock::new(settings.login.clone());
Ok(Self {
http_client,
settings,
login,
})
}
/// A consistent copy of the current login state.
fn login_snapshot(&self) -> config::LoginConfig {
match self.login.read() {
Ok(login) => login.clone(),
Err(poisoned) => poisoned.into_inner().clone(),
}
}
fn store_refresh(&self, refresh: RefreshResponse) {
let now = chrono::Utc::now().timestamp() as u64;
let mut login = match self.login.write() {
Ok(login) => login,
Err(poisoned) => poisoned.into_inner(),
};
login.expires_after = Some(now + refresh.expires_in);
login.access_token = Some(refresh.access_token);
if let Some(refresh_token) = refresh.refresh_token {
login.refresh_token = Some(refresh_token);
}
}
/// Refreshes the access token unconditionally and stores the result.
async fn force_refresh_token(&self) -> Result<(), ClientError> {
let refresh = self.refresh_access_token().await?;
self.store_refresh(refresh);
info!("access token refreshed");
Ok(())
}
/// Refreshes the access token if it is expired or about to expire.
/// Without this, long-running sessions ended up with an expired token
/// and every track fetch failed, silently stopping playback.
async fn ensure_fresh_token(&self) -> Result<(), ClientError> {
let login = self.login_snapshot();
let Some(expires_after) = login.expires_after else {
return Ok(());
};
let now = chrono::Utc::now().timestamp() as u64;
if now + TOKEN_REFRESH_MARGIN_SECS < expires_after {
return Ok(());
}
info!("access token expired or expiring soon");
self.force_refresh_token().await
}
/// Performs an authenticated GET against the hifi API.
async fn authed_get(
&self,
uri: &str,
query: Option<&[(&str, String)]>,
) -> Result<reqwest::Response, ClientError> {
let login = self.login_snapshot();
let Some(access_token) = login.access_token else {
return Err(ClientError::AuthError("No access token found".to_string()));
};
let Some(country_code) = login.country_code else {
return Err(ClientError::AuthError("No country code found".to_string()));
};
let mut params: Vec<(&str, String)> = vec![("countryCode", country_code)];
if let Some(query) = query {
params.extend(query.iter().cloned());
}
self.http_client
.get(format!("{}/{}", self.settings.hifi_url, uri))
.bearer_auth(access_token)
.query(&params)
.send()
.await
.map_err(|e| {
warn!(uri, "tidal api request failed: {e}");
ClientError::from(e)
})
}
#[instrument(skip(self))]
pub fn get_user_id(&self) -> Option<String> {
self.settings.login.user_id.clone()
self.login_snapshot().user_id
}
#[instrument(skip(self))]
@ -312,37 +397,24 @@ impl Client {
uri: &str,
query: Option<&[(&str, String)]>,
) -> Result<T, ClientError> {
debug!("make_request {}", uri);
let Some(ref access_token) = self.settings.login.access_token.clone() else {
return Err(ClientError::AuthError("No access token found".to_string()));
};
let Some(country_code) = self.settings.login.country_code.clone() else {
return Err(ClientError::AuthError("No country code found".to_string()));
};
let country_param = ("countryCode", country_code);
let mut params: Vec<&(&str, String)> = vec![&country_param];
if let Some(query) = query {
params.extend(query);
trace!(uri, "make_request");
self.ensure_fresh_token().await?;
let mut response = self.authed_get(uri, query).await?;
if response.status() == reqwest::StatusCode::UNAUTHORIZED {
// The token may have been revoked or the clock may be off:
// refresh once and retry (GETs are idempotent).
info!(uri, "got 401, refreshing access token and retrying once");
self.force_refresh_token().await?;
response = self.authed_get(uri, query).await?;
}
let response: T = self
.http_client
.get(format!("{}/{}", self.settings.hifi_url, uri))
.bearer_auth(access_token)
.query(&params)
.send()
.await
.map_err(|e| {
error!("{:?}", e);
e
})?
.json()
.await
.map_err(|e| {
error!("{:?}", e);
e
})?;
Ok(response)
if !response.status().is_success() {
warn!(uri, status = %response.status(), "tidal api request failed");
return Err(ClientError::ApiError(response.status().as_u16()));
}
response.json().await.map_err(|e| {
error!(uri, "failed to decode tidal api response: {e}");
ClientError::from(e)
})
}
#[instrument(skip(self))]
@ -351,68 +423,25 @@ impl Client {
uri: &str,
query: Option<&[(&str, String)]>,
) -> Result<Vec<T>, ClientError> {
debug!("make_paginated_request {}", uri);
let Some(ref access_token) = self.settings.login.access_token.clone() else {
return Err(ClientError::AuthError("No access token found".to_string()));
};
let Some(country_code) = self.settings.login.country_code.clone() else {
return Err(ClientError::AuthError("No country code found".to_string()));
};
let country_param = ("countryCode", country_code);
let limit = 50;
let mut offset = 0;
let limit_param = ("limit", limit.to_string());
let mut params: Vec<&(&str, String)> = vec![&country_param, &limit_param];
trace!(uri, "make_paginated_request");
let limit: usize = 50;
let mut offset: usize = 0;
let mut items = Vec::new();
loop {
let mut params: Vec<(&str, String)> =
vec![("limit", limit.to_string()), ("offset", offset.to_string())];
if let Some(query) = query {
params.extend(query);
params.extend(query.iter().cloned());
}
let mut response: Page<T> = self
.http_client
.get(format!("{}/{}", self.settings.hifi_url, uri))
.bearer_auth(access_token)
.query(&params)
.send()
.await
.map_err(|e| {
error!("{:?}", e);
e
})?
.json()
.await
.map_err(|e| {
error!("{:?}", e);
e
})?;
let mut items = Vec::with_capacity(response.total_number_of_items);
items.extend(response.items);
while response.offset + limit < response.total_number_of_items {
offset += limit;
let offset_param = ("offset", offset.to_string());
let mut params: Vec<&(&str, String)> =
vec![&country_param, &limit_param, &offset_param];
if let Some(query) = query {
params.extend(query);
let page: Page<T> = self.make_request(uri, Some(&params)).await?;
let fetched = page.items.len();
items.extend(page.items);
offset += fetched;
if fetched == 0 || offset >= page.total_number_of_items {
break;
}
response = self
.http_client
.get(format!("{}/{}", self.settings.hifi_url, uri))
.bearer_auth(access_token)
.query(&params)
.send()
.await
.map_err(|e| {
error!("{:?}", e);
e
})?
.json()
.await
.map_err(|e| {
error!("{:?}", e);
e
})?;
items.extend(response.items);
}
debug!(uri, count = items.len(), "fetched paginated collection");
Ok(items)
}
@ -422,35 +451,8 @@ impl Client {
uri: &str,
query: Option<&[(&str, String)]>,
) -> Result<(), ClientError> {
let Some(ref access_token) = self.settings.login.access_token.clone() else {
return Err(ClientError::AuthError("No access token found".to_string()));
};
let Some(country_code) = self.settings.login.country_code.clone() else {
return Err(ClientError::AuthError("No country code found".to_string()));
};
let country_param = ("countryCode", country_code);
let mut params: Vec<&(&str, String)> = vec![&country_param];
if let Some(query) = query {
params.extend(query);
}
let response = self
.http_client
.get(format!("{}/{}", self.settings.hifi_url, uri))
.bearer_auth(access_token)
.query(&params)
.send()
.await
.map_err(|e| {
error!("{:?}", e);
e
})?
.text()
.await
.map_err(|e| {
error!("{:?}", e);
e
})?;
self.ensure_fresh_token().await?;
let response = self.authed_get(uri, query).await?.text().await?;
debug!(?response, "explorer response");
Ok(())
}
@ -458,7 +460,7 @@ impl Client {
#[instrument(skip(self))]
pub async fn search(&self, query: &str) -> Result<(), ClientError> {
let query = vec![("query", query.to_string())];
self.make_explorer_request(&format!("search/artists"), Some(&query))
self.make_explorer_request("search/artists", Some(&query))
.await?;
Ok(())
}
@ -468,37 +470,32 @@ impl Client {
&self,
playlist_uuid: &str,
) -> Result<Vec<Track>, ClientError> {
Ok(self
.make_paginated_request(&format!("playlists/{}/tracks", playlist_uuid), None)
.await?)
self.make_paginated_request(&format!("playlists/{}/tracks", playlist_uuid), None)
.await
}
#[instrument(skip(self))]
pub async fn get_playlist(&self, playlist_uuid: &str) -> Result<Playlist, ClientError> {
Ok(self
.make_request(&format!("playlists/{}", playlist_uuid), None)
.await?)
self.make_request(&format!("playlists/{}", playlist_uuid), None)
.await
}
#[instrument(skip(self))]
pub async fn get_artist(&self, artist_uuid: &str) -> Result<Artist, ClientError> {
Ok(self
.make_request(&format!("artists/{}", artist_uuid), None)
.await?)
self.make_request(&format!("artists/{}", artist_uuid), None)
.await
}
#[instrument(skip(self))]
pub async fn get_artist_albums(&self, artist_uuid: &str) -> Result<Vec<Album>, ClientError> {
Ok(self
.make_paginated_request(&format!("artists/{}/albums", artist_uuid), None)
.await?)
self.make_paginated_request(&format!("artists/{}/albums", artist_uuid), None)
.await
}
#[instrument(skip(self))]
pub async fn get_users_playlists(&self, user_id: u64) -> Result<Vec<Playlist>, ClientError> {
Ok(self
.make_paginated_request(&format!("users/{}/playlists", user_id), None)
.await?)
self.make_paginated_request(&format!("users/{}/playlists", user_id), None)
.await
}
#[instrument(skip(self))]
@ -506,12 +503,11 @@ impl Client {
&self,
user_id: &str,
) -> Result<Vec<PlaylistAndFavorite>, ClientError> {
Ok(self
.make_paginated_request(
self.make_paginated_request(
&format!("users/{}/playlistsAndFavoritePlaylists", user_id),
None,
)
.await?)
.await
}
#[instrument(skip(self))]
@ -527,13 +523,12 @@ impl Client {
#[instrument(skip(self))]
pub async fn get_users_artists(&self, user_id: &str) -> Result<Vec<ArtistItem>, ClientError> {
Ok(self
.make_paginated_request(
self.make_paginated_request(
&format!("users/{}/favorites/artists", user_id),
None,
// Some(&query),
)
.await?)
.await
}
#[instrument(skip(self))]
@ -599,12 +594,18 @@ impl Client {
let timestamp = chrono::Utc::now().timestamp() as u64;
let login_results = login?;
self.settings.login.device_code = Some(code_response.device_code);
self.settings.login.access_token = Some(login_results.access_token);
self.settings.login.refresh_token = login_results.refresh_token;
self.settings.login.expires_after = Some(login_results.expires_in + timestamp);
self.settings.login.user_id = Some(login_results.user.user_id.to_string());
self.settings.login.country_code = Some(login_results.user.country_code);
{
let mut login = match self.login.write() {
Ok(login) => login,
Err(poisoned) => poisoned.into_inner(),
};
login.device_code = Some(code_response.device_code);
login.access_token = Some(login_results.access_token);
login.refresh_token = login_results.refresh_token;
login.expires_after = Some(login_results.expires_in + timestamp);
login.user_id = Some(login_results.user.user_id.to_string());
login.country_code = Some(login_results.user.country_code);
}
info!("device login succeeded");
return Ok(());
}
@ -614,10 +615,11 @@ impl Client {
#[instrument(skip(self))]
pub async fn login_config(&mut self) -> Result<(), ClientError> {
let Some(access_token) = self.settings.login.access_token.clone() else {
let login = self.login_snapshot();
let Some(access_token) = login.access_token else {
return Err(ClientError::AuthError("No access token found".to_string()));
};
//return if our session is still valid
// Return if our session is still valid.
if self
.http_client
.get(format!("{}/sessions", self.settings.base_url))
@ -625,27 +627,23 @@ impl Client {
.send()
.await
.map_err(|e| {
error!("{:?}", e);
warn!("session check failed: {e}");
e
})?
.status()
.is_success()
{
debug!("existing session still valid");
return Ok(());
}
//otherwise refresh our token
let refresh = self.refresh_access_token().await?;
let now = chrono::Utc::now().timestamp() as u64;
self.settings.login.expires_after = Some(refresh.expires_in + now);
self.settings.login.access_token = Some(refresh.access_token);
Ok(())
// Otherwise refresh our token.
self.force_refresh_token().await
}
#[instrument(skip(self))]
pub async fn refresh_access_token(&self) -> Result<RefreshResponse, ClientError> {
let Some(refresh_token) = self.settings.login.refresh_token.clone() else {
let Some(refresh_token) = self.login_snapshot().refresh_token else {
return Err(ClientError::AuthError("No refresh token found".to_string()));
};
let data = DeviceAuthRequest {
@ -693,7 +691,7 @@ impl Client {
.http_client
.post(format!(
"{}/device_authorization",
&self.settings.oauth.base_url
self.settings.oauth.base_url
))
.header("Content-Type", "application/x-www-form-urlencoded")
.body(payload)
@ -779,6 +777,5 @@ mod tests {
println!("{:?}", result);
let result = client.get_album("244167550").await.unwrap();
println!("{:?}", result);
assert!(false);
}
}

3
tidaldy/src/models.rs
View File

@ -51,6 +51,8 @@ pub enum ClientError {
SerdeUrlError(#[from] serde_urlencoded::ser::Error),
#[error("authentication failed")]
AuthError(String),
#[error("tidal api returned status {0}")]
ApiError(u16),
#[error("base64 decoding failed")]
Base64DecodeError(#[from] base64::DecodeError),
#[error("utf8 decoding failed")]
@ -65,6 +67,7 @@ impl From<ClientError> for crabidy_core::ProviderError {
ClientError::ConnectionError => Self::FetchError,
ClientError::HttpClientError(_) => Self::FetchError,
ClientError::SerdeUrlError(_) => Self::FetchError,
ClientError::ApiError(_) => Self::FetchError,
_ => Self::Other,
}
}