/// Lots of stuff and especially the auth handling is shamelessly copied from /// https://github.com/MinisculeGirraffe/tdl use reqwest::Client as HttpClient; use serde::de::DeserializeOwned; use tokio::time::{sleep, Duration, Instant}; use tracing::{debug, error, info, instrument, trace, warn}; pub mod config; pub mod models; use async_trait::async_trait; pub use models::*; #[derive(Debug)] 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, /// Search terms created under `/tidal/search`, in creation order, /// deduplicated. In-memory only: terms die with the process, like the /// queue. Same locking discipline as `login`: never held across awaits. search_terms: std::sync::RwLock>, } /// 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))] async fn init(raw_toml_settings: &str) -> Result { let settings: config::Settings = if let Ok(settings) = toml::from_str(raw_toml_settings) { settings } else { warn!("could not parse toml settings, using defaults"); config::Settings::default() }; let mut client = Self::new(settings)?; if client.login_config().await.is_ok() { return Ok(client); } if client.login_web().await.is_ok() { return Ok(client); } Err(crabidy_core::ProviderError::CouldNotLogin) } #[instrument(skip(self))] fn settings(&self) -> String { 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!( parse_path(path), Ok(TidalPath::PlaylistTrack { .. } | TidalPath::AlbumTrack { .. } | TidalPath::SearchTrack { .. }) ) } #[instrument(skip(self))] async fn get_urls_for_track( &self, track_path: &str, ) -> Result, crabidy_core::ProviderError> { let track_id = track_id_from_path(track_path)?; let playback = self.get_track_playback(track_id).await.map_err(|err| { warn!(track = track_id, "failed to fetch playback info: {err}"); crabidy_core::ProviderError::FetchError })?; trace!(?playback, "got playback info"); let manifest = playback.get_manifest().map_err(|err| { warn!(track = track_id, "failed to decode manifest: {err}"); crabidy_core::ProviderError::FetchError })?; debug!( track = track_id, urls = manifest.urls.len(), "resolved stream urls" ); Ok(manifest.urls) } #[instrument(skip(self))] async fn get_metadata_for_track( &self, track_path: &str, ) -> Result { let track_id = track_id_from_path(track_path)?; let track = self.get_track(track_id).await.map_err(|err| { warn!(track = track_id, "failed to fetch track metadata: {err}"); crabidy_core::ProviderError::FetchError })?; let parent = crabidy_core::parent_path(track_path) .unwrap_or(PROVIDER_ROOT) .to_string(); Ok(track.to_proto(&parent)) } #[instrument(skip(self))] fn get_lib_root(&self) -> crabidy_core::proto::crabidy::LibraryNode { crabidy_core::proto::crabidy::LibraryNode { path: PROVIDER_ROOT.to_string(), title: "tidal".to_string(), parent: Some(crabidy_core::ROOT_PATH.to_string()), tracks: Vec::new(), children: vec![ crabidy_core::proto::crabidy::LibraryNodeChild::new( format!("{PROVIDER_ROOT}/playlists"), "playlists".to_string(), false, ), crabidy_core::proto::crabidy::LibraryNodeChild::new( format!("{PROVIDER_ROOT}/artists"), "artists".to_string(), false, ), crabidy_core::proto::crabidy::LibraryNodeChild { is_creatable: true, ..crabidy_core::proto::crabidy::LibraryNodeChild::new( format!("{PROVIDER_ROOT}/search"), "search".to_string(), false, ) }, ], is_queable: false, is_creatable: false, is_downloadable: false, tracks_deletable: false, } } #[instrument(skip(self))] async fn get_lib_node( &self, path: &str, ) -> Result { // The user id is only needed for the favorites listings; search must // work even when it is missing, so the gate lives in those arms. let user_id = self.get_user_id(); let parsed = parse_path(path)?; debug!(?parsed, "resolving library node"); let parent = crabidy_core::parent_path(path) .unwrap_or(crabidy_core::ROOT_PATH) .to_string(); let node = match parsed { TidalPath::Root => self.get_lib_root(), TidalPath::Playlists => { let mut node = crabidy_core::proto::crabidy::LibraryNode { path: path.to_string(), title: "playlists".to_string(), parent: Some(parent), tracks: Vec::new(), children: Vec::new(), is_queable: false, is_creatable: false, is_downloadable: false, tracks_deletable: false, }; let user_id = user_id.ok_or(crabidy_core::ProviderError::UnknownUser)?; for playlist in self .get_users_playlists_and_favorite_playlists(&user_id) .await? { node.children .push(crabidy_core::proto::crabidy::LibraryNodeChild::new( crabidy_core::join_path(path, &playlist.playlist.uuid), playlist.playlist.title, true, )); } node } TidalPath::Playlist(playlist_id) => { let playlist = self.get_playlist(playlist_id).await?; let tracks = self .get_playlist_tracks(playlist_id) .await? .iter() .map(|t| t.to_proto(path)) .collect(); crabidy_core::proto::crabidy::LibraryNode { path: path.to_string(), title: playlist.title, parent: Some(parent), tracks, children: Vec::new(), is_queable: true, is_creatable: false, is_downloadable: false, tracks_deletable: false, } } TidalPath::Artists => { let mut node = crabidy_core::proto::crabidy::LibraryNode { path: path.to_string(), title: "artists".to_string(), parent: Some(parent), tracks: Vec::new(), children: Vec::new(), is_queable: false, is_creatable: false, is_downloadable: false, tracks_deletable: false, }; let user_id = user_id.ok_or(crabidy_core::ProviderError::UnknownUser)?; for artist in self.get_users_artists(&user_id).await? { node.children .push(crabidy_core::proto::crabidy::LibraryNodeChild::new( crabidy_core::join_path(path, &artist.item.id.to_string()), artist.item.name, true, )); } node } TidalPath::Artist(artist_id) => { let artist = self.get_artist(artist_id).await?; let children = self .get_artist_albums(artist_id) .await? .iter() .map(|album| { crabidy_core::proto::crabidy::LibraryNodeChild::new( crabidy_core::join_path(path, &album.id.to_string()), album.title.clone(), true, ) }) .collect(); crabidy_core::proto::crabidy::LibraryNode { path: path.to_string(), title: artist.name, parent: Some(parent), tracks: Vec::new(), children, is_queable: true, is_creatable: false, is_downloadable: false, tracks_deletable: false, } } TidalPath::Album { album, .. } => { let album_data = self.get_album(album).await?; let tracks = self .get_album_tracks(album) .await? .iter() .map(|t| t.to_proto(path)) .collect(); crabidy_core::proto::crabidy::LibraryNode { path: path.to_string(), title: album_data.title, parent: Some(parent), tracks, children: Vec::new(), is_queable: true, is_creatable: false, is_downloadable: false, tracks_deletable: false, } } TidalPath::Search => crabidy_core::proto::crabidy::LibraryNode { path: path.to_string(), title: "search".to_string(), parent: Some(parent), tracks: Vec::new(), children: self .search_terms_snapshot() .iter() .map(|term| { // Term nodes are the modifiable nodes: renamable // (`e`) and deletable (`d`). crabidy_core::proto::crabidy::LibraryNodeChild { is_editable: true, is_deletable: true, ..crabidy_core::proto::crabidy::LibraryNodeChild::new( crabidy_core::join_path(path, &crabidy_core::encode_segment(term)), term.clone(), false, ) } }) .collect(), is_queable: false, is_creatable: true, is_downloadable: false, tracks_deletable: false, }, TidalPath::SearchTerm(encoded) => { let term = crabidy_core::decode_segment(encoded); // Unknown terms (stale client cache, server restart) are // recreated implicitly instead of erroring. self.register_search_term(&term); self.search_term_node(path, &term, parent).await? } TidalPath::PlaylistTrack { .. } | TidalPath::AlbumTrack { .. } | TidalPath::SearchTrack { .. } => { warn!(path, "get_lib_node called with a track path"); return Err(crabidy_core::ProviderError::MalformedPath); } }; // Tidal implements download captures (architecture/captures.md D4): // every queueable subtree may be captured with `W`, and every node // that lists tracks blesses them for download (listed tracks // inherit their node's flag — this covers search-term results, // whose node is not queueable as a whole). let mut node = node; node.is_downloadable = node.is_queable || !node.tracks.is_empty(); for child in &mut node.children { child.is_downloadable = child.is_queable; } Ok(node) } #[instrument(skip(self))] async fn create_lib_node( &self, parent_path: &str, title: &str, ) -> Result { let term = title.trim(); if term.is_empty() { return Err(crabidy_core::ProviderError::InvalidInput); } // Only /tidal/search is creatable. if parse_path(parent_path)? != TidalPath::Search { warn!(parent_path, "node creation not supported here"); return Err(crabidy_core::ProviderError::NotSupported); } self.register_search_term(term); let term_path = crabidy_core::join_path(parent_path, &crabidy_core::encode_segment(term)); self.get_lib_node(&term_path).await } #[instrument(skip(self))] async fn rename_lib_node( &self, path: &str, new_title: &str, ) -> Result { let new_term = new_title.trim(); if new_term.is_empty() { return Err(crabidy_core::ProviderError::InvalidInput); } // Only search terms are editable. let TidalPath::SearchTerm(encoded_old) = parse_path(path)? else { warn!(path, "renaming not supported here"); return Err(crabidy_core::ProviderError::NotSupported); }; let old_term = crabidy_core::decode_segment(encoded_old); // In-place replace keeps the list position; renaming onto an // existing term merges with it. self.rename_search_term(&old_term, new_term); // A SearchTerm path always has the search node as its parent. let parent = crabidy_core::parent_path(path).unwrap_or(PROVIDER_ROOT); let new_path = crabidy_core::join_path(parent, &crabidy_core::encode_segment(new_term)); self.get_lib_node(&new_path).await } #[instrument(skip(self))] async fn delete_lib_node( &self, path: &str, ) -> Result { // Only search terms are deletable. let TidalPath::SearchTerm(encoded) = parse_path(path)? else { warn!(path, "deleting not supported here"); return Err(crabidy_core::ProviderError::NotSupported); }; // Idempotent: removing an already-gone term is a success. self.remove_search_term(&crabidy_core::decode_segment(encoded)); // A SearchTerm path always has the search node as its parent. let parent = crabidy_core::parent_path(path).unwrap_or(PROVIDER_ROOT); self.get_lib_node(parent).await } /// Streams tracks chunk-wise (see the trait docs for the channel /// contract). Overridden so that the paginated collections — playlists /// and albums — emit one chunk per fetched page (50 tracks) instead of /// paginating to exhaustion inside a single node fetch; a large playlist /// starts playing after its first page. All other paths follow the /// generic pre-order walk, in listing order. #[instrument(skip(self, chunk_tx))] async fn resolve_tracks_into( &self, path: &str, chunk_tx: flume::Sender>, ) -> Result<(), crabidy_core::ProviderError> { // Validate before any I/O; foreign paths never touch the network. if let TidalPath::PlaylistTrack { .. } | TidalPath::AlbumTrack { .. } | TidalPath::SearchTrack { .. } = parse_path(path)? { match self.get_metadata_for_track(path).await { Ok(track) => { let _ = chunk_tx.send_async(vec![track]).await; } Err(err) => warn!(path, "failed to resolve track: {err}"), } return Ok(()); } // Depth-first pre-order (children pushed reversed, worklist pops // from the back) so tracks arrive in listing order. let mut nodes_to_go = vec![path.to_string()]; let mut at_root = true; while let Some(node_path) = nodes_to_go.pop() { let root = std::mem::take(&mut at_root); // The track-bearing collections stream one chunk per fetched // page instead of paginating to exhaustion first. let paged_uri = match parse_path(&node_path) { Ok(TidalPath::Playlist(playlist_id)) => { Some(format!("playlists/{playlist_id}/tracks")) } Ok(TidalPath::Album { album, .. }) => Some(format!("albums/{album}/tracks")), _ => None, }; if let Some(uri) = paged_uri { match self .stream_track_pages_into(&uri, &node_path, &chunk_tx) .await { Ok(true) => {} // Receiver gone: the consumer cancelled, stop fetching. Ok(false) => return Ok(()), Err(err) if root => return Err(err.into()), Err(err) => warn!(node = node_path, "skipping unreadable node: {err}"), } continue; } let node = match self.get_lib_node(&node_path).await { Ok(node) => node, Err(err) if root => return Err(err), Err(err) => { warn!(node = node_path, "skipping unreadable node: {err}"); continue; } }; if !node.is_queable { continue; } if !node.tracks.is_empty() && chunk_tx.send_async(node.tracks).await.is_err() { // Receiver gone: the consumer cancelled, stop fetching. return Ok(()); } nodes_to_go.extend(node.children.into_iter().rev().map(|c| c.path)); } Ok(()) } } /// The root of this provider in the global library tree. pub const PROVIDER_ROOT: &str = "/tidal"; /// Maximum results fetched per category (tracks/artists/albums) for one /// search term. First page only — search is exploratory, not a collection. pub const SEARCH_RESULT_LIMIT: usize = 20; /// A parsed tidal library path. The position in the tree is fully encoded /// in the path itself. #[derive(Debug, Clone, Copy, PartialEq, Eq)] pub enum TidalPath<'a> { Root, Playlists, Playlist(&'a str), PlaylistTrack { playlist: &'a str, track: &'a str, }, Artists, Artist(&'a str), Album { artist: &'a str, album: &'a str, }, AlbumTrack { artist: &'a str, album: &'a str, track: &'a str, }, /// `/tidal/search` — the creatable node listing created search terms. Search, /// `/tidal/search/` — one created term; `term` is the /// percent-encoded segment, not the raw text. SearchTerm(&'a str), /// `/tidal/search//` — a track search result. Artist and /// album results are not nested here: their children carry canonical /// `/tidal/artists/...` paths (see architecture/search.md). SearchTrack { term: &'a str, track: &'a str, }, } pub fn parse_path(path: &str) -> Result, crabidy_core::ProviderError> { let segments = crabidy_core::path_segments(path); match segments.as_slice() { ["tidal"] => Ok(TidalPath::Root), ["tidal", "playlists"] => Ok(TidalPath::Playlists), ["tidal", "playlists", playlist] => Ok(TidalPath::Playlist(playlist)), ["tidal", "playlists", playlist, track] => Ok(TidalPath::PlaylistTrack { playlist, track }), ["tidal", "artists"] => Ok(TidalPath::Artists), ["tidal", "artists", artist] => Ok(TidalPath::Artist(artist)), ["tidal", "artists", artist, album] => Ok(TidalPath::Album { artist, album }), ["tidal", "artists", artist, album, track] => Ok(TidalPath::AlbumTrack { artist, album, track, }), ["tidal", "search"] => Ok(TidalPath::Search), ["tidal", "search", term] => Ok(TidalPath::SearchTerm(term)), ["tidal", "search", term, track] => Ok(TidalPath::SearchTrack { term, track }), _ => { warn!(path, "malformed tidal path"); Err(crabidy_core::ProviderError::MalformedPath) } } } fn track_id_from_path(path: &str) -> Result<&str, crabidy_core::ProviderError> { match parse_path(path)? { TidalPath::PlaylistTrack { track, .. } | TidalPath::AlbumTrack { track, .. } | TidalPath::SearchTrack { track, .. } => Ok(track), _ => { warn!(path, "expected a track path"); Err(crabidy_core::ProviderError::MalformedPath) } } } impl Client { pub fn new(settings: config::Settings) -> Result { 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, search_terms: std::sync::RwLock::new(Vec::new()), }) } /// 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. /// /// A failed proactive refresh never fails the request: the current /// access token may still be valid (stale `expires_after` metadata from /// an old config, for example), and `make_request`'s 401-retry remains /// the backstop for a token that is actually dead. 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"); match self.force_refresh_token().await { Ok(()) => Ok(()), Err(err @ ClientError::AuthError(_)) => { // The refresh token is permanently unusable. Stop trying // proactively so every request doesn't hammer the endpoint; // a new device login (server restart) is the only cure once // the access token dies. warn!( "stored refresh token was rejected and can never work; \ continuing with the current access token — restart the \ server to re-login once it expires: {err}" ); let mut login = match self.login.write() { Ok(login) => login, Err(poisoned) => poisoned.into_inner(), }; login.expires_after = None; Ok(()) } Err(err) => { warn!("token refresh failed transiently, continuing with the current token: {err}"); Ok(()) } } } /// Performs an authenticated GET against the hifi API. async fn authed_get( &self, uri: &str, query: Option<&[(&str, String)]>, ) -> Result { 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(¶ms) .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 { self.login_snapshot().user_id } #[instrument(skip(self))] pub async fn make_request( &self, uri: &str, query: Option<&[(&str, String)]>, ) -> Result { 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?; } 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))] pub async fn make_paginated_request( &self, uri: &str, query: Option<&[(&str, String)]>, ) -> Result, ClientError> { 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.iter().cloned()); } let page: Page = self.make_request(uri, Some(¶ms)).await?; let fetched = page.items.len(); items.extend(page.items); offset += fetched; if fetched == 0 || offset >= page.total_number_of_items { break; } } debug!(uri, count = items.len(), "fetched paginated collection"); Ok(items) } /// Like [`make_paginated_request`](Self::make_paginated_request) for a /// paginated track collection, but streams each fetched page into /// `chunk_tx` (mapped to proto tracks under `parent_path`) as soon as /// it arrives instead of collecting to exhaustion first. Returns /// `Ok(false)` when the receiver disappeared before the collection was /// exhausted — cancellation, not an error — and `Ok(true)` on normal /// completion. Reuses [`make_request`](Self::make_request), so token /// refresh and retry behave exactly like every other fetch. async fn stream_track_pages_into( &self, uri: &str, parent_path: &str, chunk_tx: &flume::Sender>, ) -> Result { let limit: usize = 50; let mut offset: usize = 0; loop { let params: Vec<(&str, String)> = vec![("limit", limit.to_string()), ("offset", offset.to_string())]; let page: Page = self.make_request(uri, Some(¶ms)).await?; let fetched = page.items.len(); offset += fetched; let exhausted = fetched == 0 || offset >= page.total_number_of_items; let chunk: Vec<_> = page.items.iter().map(|t| t.to_proto(parent_path)).collect(); if !chunk.is_empty() && chunk_tx.send_async(chunk).await.is_err() { debug!(uri, offset, "paginated fetch cancelled by its consumer"); return Ok(false); } if exhausted { return Ok(true); } } } #[instrument(skip(self))] pub async fn make_explorer_request( &self, uri: &str, query: Option<&[(&str, String)]>, ) -> Result<(), ClientError> { self.ensure_fresh_token().await?; let response = self.authed_get(uri, query).await?.text().await?; debug!(?response, "explorer response"); Ok(()) } /// Explorer helper kept for probing the raw `search/*` payloads against /// the live API (the typed methods below assume our models fit them). #[instrument(skip(self))] pub async fn search(&self, query: &str) -> Result<(), ClientError> { let query = vec![("query", query.to_string())]; self.make_explorer_request("search/artists", Some(&query)) .await?; Ok(()) } /// One first-page search request. Search is exploratory: unlike the /// library collections this deliberately does NOT paginate to /// exhaustion. async fn search_page( &self, category: &str, query: &str, ) -> Result, ClientError> { let params = vec![ ("query", query.to_string()), ("limit", SEARCH_RESULT_LIMIT.to_string()), ("offset", "0".to_string()), ]; let page: Page = self .make_request(&format!("search/{category}"), Some(¶ms)) .await?; Ok(page.items) } /// First page of track results for a search term, at most /// [`SEARCH_RESULT_LIMIT`]. #[instrument(skip(self))] pub async fn search_tracks(&self, query: &str) -> Result, ClientError> { self.search_page("tracks", query).await } /// First page of artist results, at most [`SEARCH_RESULT_LIMIT`]. #[instrument(skip(self))] pub async fn search_artists(&self, query: &str) -> Result, ClientError> { self.search_page("artists", query).await } /// First page of album results, at most [`SEARCH_RESULT_LIMIT`]. #[instrument(skip(self))] pub async fn search_albums(&self, query: &str) -> Result, ClientError> { self.search_page("albums", query).await } /// A consistent copy of the created search terms. fn search_terms_snapshot(&self) -> Vec { match self.search_terms.read() { Ok(terms) => terms.clone(), Err(poisoned) => poisoned.into_inner().clone(), } } /// Registers a term idempotently, preserving creation order. The lock is /// released before any await point. fn register_search_term(&self, term: &str) { let mut terms = match self.search_terms.write() { Ok(terms) => terms, Err(poisoned) => poisoned.into_inner(), }; if !terms.iter().any(|t| t == term) { terms.push(term.to_string()); } } /// Renames `old` to `new` in place, keeping its position. If `new` /// already exists elsewhere the entries merge (the `old` slot is /// removed) — the list never holds duplicates. An unknown `old` behaves /// like registration of `new` (stale-client forgiveness). The lock is /// released before any await point. fn rename_search_term(&self, old: &str, new: &str) { let mut terms = match self.search_terms.write() { Ok(terms) => terms, Err(poisoned) => poisoned.into_inner(), }; if terms.iter().any(|t| t == new) { // Merge — but a self-rename must not delete the very term it // is supposed to keep. if old != new { terms.retain(|t| t != old); } } else if let Some(slot) = terms.iter_mut().find(|t| *t == old) { *slot = new.to_string(); } else { terms.push(new.to_string()); } } /// Removes a term. Unknown terms are a no-op — delete is idempotent. /// The lock is released before any await point. fn remove_search_term(&self, term: &str) { let mut terms = match self.search_terms.write() { Ok(terms) => terms, Err(poisoned) => poisoned.into_inner(), }; terms.retain(|t| t != term); } /// Builds the node for one search term: its `tracks` are the track /// results (queueable individually, paths under this node), its children /// are artist and album results pointing at **canonical** /// `/tidal/artists/...` paths. The node itself is not queueable — the /// server's resolve sweep would otherwise pull every artist's full /// discography into the queue (architecture/search.md). /// /// The three categories are fetched concurrently. A failing category /// degrades to empty with a warning; only all three failing is an error. async fn search_term_node( &self, path: &str, term: &str, parent: String, ) -> Result { let (tracks, artists, albums) = tokio::join!( self.search_tracks(term), self.search_artists(term), self.search_albums(term), ); if let (Err(t), Err(a), Err(al)) = (&tracks, &artists, &albums) { warn!(term, "all search categories failed: {t}; {a}; {al}"); return Err(crabidy_core::ProviderError::FetchError); } fn or_empty(term: &str, name: &str, r: Result, ClientError>) -> Vec { match r { Ok(items) => items, Err(err) => { warn!(term, category = name, "search category failed: {err}"); Vec::new() } } } let tracks = or_empty(term, "tracks", tracks); let artists = or_empty(term, "artists", artists); let albums = or_empty(term, "albums", albums); let mut children = Vec::with_capacity(artists.len() + albums.len()); children.extend(artists.iter().map(|a| { crabidy_core::proto::crabidy::LibraryNodeChild::new( format!("{PROVIDER_ROOT}/artists/{}", a.id), format!("Artist: {}", a.name), true, ) })); children.extend(albums.iter().filter_map(|album| { // The canonical album path needs the artist id; results without // one (not observed in practice) are skipped, not guessed at. let artist = album.artist.as_ref()?; Some(crabidy_core::proto::crabidy::LibraryNodeChild::new( format!("{PROVIDER_ROOT}/artists/{}/{}", artist.id, album.id), format!("Album: {}", album.title), true, )) })); Ok(crabidy_core::proto::crabidy::LibraryNode { path: path.to_string(), title: term.to_string(), parent: Some(parent), tracks: tracks.iter().map(|t| t.to_proto(path)).collect(), children, is_queable: false, is_creatable: false, is_downloadable: false, tracks_deletable: false, }) } #[instrument(skip(self))] pub async fn get_playlist_tracks( &self, playlist_uuid: &str, ) -> Result, ClientError> { self.make_paginated_request(&format!("playlists/{}/tracks", playlist_uuid), None) .await } #[instrument(skip(self))] pub async fn get_playlist(&self, playlist_uuid: &str) -> Result { self.make_request(&format!("playlists/{}", playlist_uuid), None) .await } #[instrument(skip(self))] pub async fn get_artist(&self, artist_uuid: &str) -> Result { self.make_request(&format!("artists/{}", artist_uuid), None) .await } #[instrument(skip(self))] pub async fn get_artist_albums(&self, artist_uuid: &str) -> Result, ClientError> { 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, ClientError> { self.make_paginated_request(&format!("users/{}/playlists", user_id), None) .await } #[instrument(skip(self))] pub async fn get_users_playlists_and_favorite_playlists( &self, user_id: &str, ) -> Result, ClientError> { self.make_paginated_request( &format!("users/{}/playlistsAndFavoritePlaylists", user_id), None, ) .await } #[instrument(skip(self))] pub async fn get_users_favorites(&self, user_id: u64) -> Result<(), ClientError> { self.make_explorer_request( &format!("users/{}/favorites", user_id), None, // Some(&query), ) .await?; Ok(()) } #[instrument(skip(self))] pub async fn get_users_artists(&self, user_id: &str) -> Result, ClientError> { self.make_paginated_request( &format!("users/{}/favorites/artists", user_id), None, // Some(&query), ) .await } #[instrument(skip(self))] pub async fn get_user(&self, user_id: u64) -> Result<(), ClientError> { self.make_explorer_request( &format!("users/{}", user_id), None, // Some(&query), ) .await?; Ok(()) } #[instrument(skip(self))] pub async fn get_album(&self, album_id: &str) -> Result { self.make_request(&format!("albums/{}/", album_id), None) .await } #[instrument(skip(self))] pub async fn get_album_tracks(&self, album_id: &str) -> Result, ClientError> { self.make_paginated_request(&format!("albums/{}/tracks", album_id), None) .await } #[instrument(skip(self))] pub async fn get_track_playback(&self, track_id: &str) -> Result { let query = vec![ ("audioquality", "LOSSLESS".to_string()), ("playbackmode", "STREAM".to_string()), ("assetpresentation", "FULL".to_string()), ]; self.make_request( &format!("tracks/{}/playbackinfopostpaywall", track_id), Some(&query), ) .await } #[instrument(skip(self))] pub async fn get_track(&self, track_id: &str) -> Result { self.make_request(&format!("tracks/{}", track_id), None) .await } #[instrument(skip(self))] pub async fn login_web(&mut self) -> Result<(), ClientError> { let code_response = self.get_device_code().await?; let started = Instant::now(); // The verification link must reach the user even without a log // subscriber configured. println!("https://{}", code_response.verification_uri_complete); info!( expires_in = code_response.expires_in, interval = code_response.interval, "waiting for device login at https://{}", code_response.verification_uri_complete ); // Poll no faster than the server asked for, and never busy-loop. let mut interval = code_response.interval.max(1); while started.elapsed().as_secs() <= code_response.expires_in { match self.poll_device_token(&code_response.device_code).await { Ok(Some(login_results)) => { let timestamp = chrono::Utc::now().timestamp() as u64; { 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(()); } Ok(None) => { debug!(interval, "authorization pending"); } Err(PollError::SlowDown) => { // Per RFC 8628 the client must back off by 5 seconds. interval += 5; debug!(interval, "server asked us to slow down"); } Err(PollError::Transport(err)) => { // Transient network problems shouldn't kill the login. warn!("device token poll failed, retrying: {err}"); } Err(PollError::Fatal(err)) => { error!("device login failed: {err}"); return Err(err); } } sleep(Duration::from_secs(interval)).await; } warn!("device login attempt expired before it was authorized"); Err(ClientError::AuthError("device login expired".to_string())) } #[instrument(skip(self))] pub async fn login_config(&mut self) -> Result<(), ClientError> { 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. if self .http_client .get(format!("{}/sessions", self.settings.base_url)) .bearer_auth(access_token) .send() .await .map_err(|e| { warn!("session check failed: {e}"); e })? .status() .is_success() { debug!("existing session still valid"); return Ok(()); } // Otherwise refresh our token. self.force_refresh_token().await } #[instrument(skip(self))] pub async fn refresh_access_token(&self) -> Result { let Some(refresh_token) = self.login_snapshot().refresh_token else { return Err(ClientError::AuthError("No refresh token found".to_string())); }; let data = DeviceAuthRequest { client_id: self.settings.oauth.client_id.clone(), client_secret: Some(self.settings.oauth.client_secret.clone()), refresh_token: Some(refresh_token.to_string()), grant_type: Some("refresh_token".to_string()), ..Default::default() }; let body = serde_urlencoded::to_string(&data)?; // Secret in the body, not Basic auth — see [`Self::poll_device_token`]. let req = self .http_client .post(format!("{}/token", self.settings.oauth.base_url)) .body(body) .header("Content-Type", "application/x-www-form-urlencoded") .send() .await .map_err(|e| { error!("{:?}", e); e })?; let status = req.status(); if status.is_success() { let res = req.json::().await?; Ok(res) } else if status.is_client_error() { // A 4xx means the endpoint understood us and rejected the token // (e.g. it was issued by a client that no longer exists). This // cannot succeed on retry — only a new device login helps. let body = req.text().await.unwrap_or_default(); let snippet: String = body.chars().take(300).collect(); Err(ClientError::AuthError(format!( "token refresh returned {status}: {snippet}" ))) } else { let body = req.text().await.unwrap_or_default(); let snippet: String = body.chars().take(300).collect(); error!("token refresh returned {status}: {snippet}"); Err(ClientError::ApiError(status.as_u16())) } } #[instrument(skip(self))] async fn get_device_code(&self) -> Result { let req = DeviceAuthRequest { client_id: self.settings.oauth.client_id.clone(), scope: Some("r_usr+w_usr+w_sub".to_string()), ..Default::default() }; let payload = serde_urlencoded::to_string(&req)?; let res = self .http_client .post(format!( "{}/device_authorization", self.settings.oauth.base_url )) .header("Content-Type", "application/x-www-form-urlencoded") .body(payload) .send() .await .map_err(|e| { error!("{:?}", e); e })?; if !res.status().is_success() { return Err(ClientError::AuthError(res.status().to_string())); } let code: DeviceAuthResponse = res.json().await?; Ok(code) } /// One poll of the device-flow token endpoint. /// /// `Ok(Some(_))` means the user authorized us, `Ok(None)` means the /// authorization is still pending and the caller should keep polling. #[instrument(skip(self, device_code))] async fn poll_device_token( &self, device_code: &str, ) -> Result, PollError> { let req = DeviceAuthRequest { client_id: self.settings.oauth.client_id.clone(), // The secret must travel in the body: Tidal's edge rejects HTTP // Basic auth on this endpoint with an HTML 403. client_secret: Some(self.settings.oauth.client_secret.clone()), device_code: Some(device_code.to_string()), scope: Some("r_usr+w_usr+w_sub".to_string()), grant_type: Some("urn:ietf:params:oauth:grant-type:device_code".to_string()), ..Default::default() }; let payload = serde_urlencoded::to_string(&req).map_err(|err| PollError::Fatal(err.into()))?; let res = self .http_client .post(format!("{}/token", self.settings.oauth.base_url)) .body(payload) .header("Content-Type", "application/x-www-form-urlencoded") .send() .await .map_err(|err| PollError::Transport(err.into()))?; let status = res.status(); let body = res .text() .await .map_err(|err| PollError::Transport(err.into()))?; if status.is_success() { // NB: don't log the body here, it contains the tokens. return match serde_json::from_str::(&body) { Ok(refresh) => Ok(Some(refresh)), Err(err) => Err(PollError::Fatal(ClientError::AuthError(format!( "could not decode token response: {err}" )))), }; } // OAuth error responses carry the reason in an `error` field // (RFC 8628 §3.5); Tidal additionally sets `sub_status`. let oauth_error = serde_json::from_str::(&body).unwrap_or_default(); match oauth_error.error.as_deref() { Some("authorization_pending") => Ok(None), Some("slow_down") => Err(PollError::SlowDown), _ if status.is_server_error() => { Err(PollError::Transport(ClientError::ApiError(status.as_u16()))) } _ => { // Error bodies carry no secrets, so quoting them is safe // and beats guessing why the flow died. let body_snippet: String = body.chars().take(300).collect(); Err(PollError::Fatal(ClientError::AuthError(format!( "token endpoint returned {status}: {body_snippet}" )))) } } } } /// Outcome classification for one device-flow token poll. #[derive(Debug)] enum PollError { /// The server asked us to poll less often (RFC 8628 `slow_down`). SlowDown, /// A transient failure; keep polling. Transport(ClientError), /// The flow cannot succeed anymore; stop polling. Fatal(ClientError), } /// Lenient shape of an OAuth token-endpoint error body: only `error` is /// used to classify the failure; the full body is logged separately for /// anything not otherwise handled. #[derive(Debug, Default, serde::Deserialize)] struct OauthErrorBody { error: Option, } #[cfg(test)] mod tests { use crabidy_core::ProviderClient; use super::*; async fn setup() -> Client { let raw_toml_settings = std::fs::read_to_string("/home/hans/.config/crabidy/tidaldy.toml").unwrap(); Client::init(&raw_toml_settings).await.unwrap() } /// A client with default settings and no login — enough for the pure /// path/tree methods, no network involved. fn offline_client() -> Client { Client::new(config::Settings::default()).expect("offline client") } #[tokio::test] async fn resolve_rejects_foreign_paths_before_any_network_call() { use crabidy_core::ProviderError; let client = offline_client(); for path in ["/spotify/x", "/nope", ""] { let (chunk_tx, chunk_rx) = flume::bounded(1); let result = client.resolve_tracks_into(path, chunk_tx).await; assert_eq!(result, Err(ProviderError::MalformedPath), "path {path:?}"); assert!(chunk_rx.try_recv().is_err(), "no chunks for {path:?}"); } } /// Dumps the raw search payloads to re-verify the models if the API /// drifts (last verified 2026-07-20, shapes matched `Page` + the /// existing Track/Artist/Album models). #[tokio::test] #[ignore = "requires local tidal config and network"] async fn probe_search_shapes() { let raw = std::fs::read_to_string("/home/hans/.config/crabidy/tidaly.toml").unwrap(); let settings: config::Settings = toml::from_str(&raw).unwrap(); let client = Client::new(settings).unwrap(); client.ensure_fresh_token().await.unwrap(); for cat in ["tracks", "artists", "albums"] { let resp = client .authed_get( &format!("search/{cat}"), Some(&[("query", "beatles".to_string()), ("limit", "2".to_string())]), ) .await .unwrap(); println!("=== {cat} status={}", resp.status()); println!("{}", resp.text().await.unwrap()); } } #[test] fn parse_path_recognizes_search_paths() { assert_eq!(parse_path("/tidal/search"), Ok(TidalPath::Search)); assert_eq!( parse_path("/tidal/search/abba"), Ok(TidalPath::SearchTerm("abba")) ); assert_eq!( parse_path("/tidal/search/abba/12345"), Ok(TidalPath::SearchTrack { term: "abba", track: "12345" }) ); assert!(parse_path("/tidal/search/a/b/c").is_err()); } #[test] fn search_track_paths_are_track_paths() { let client = offline_client(); assert!(client.is_track_path("/tidal/search/abba/12345")); assert!(!client.is_track_path("/tidal/search/abba")); assert!(!client.is_track_path("/tidal/search")); } #[test] fn track_id_is_extracted_from_search_track_paths() { assert_eq!(track_id_from_path("/tidal/search/abba/12345"), Ok("12345")); } #[tokio::test] async fn create_rejects_empty_titles_and_foreign_parents() { use crabidy_core::ProviderError; let client = offline_client(); // Validation happens before any network call, so this works offline. assert_eq!( client.create_lib_node("/tidal/search", " ").await, Err(ProviderError::InvalidInput) ); assert_eq!( client.create_lib_node("/tidal/playlists", "abba").await, Err(ProviderError::NotSupported) ); assert_eq!( client.create_lib_node("/nope", "abba").await, Err(ProviderError::MalformedPath) ); } #[tokio::test] async fn search_node_lists_created_terms() { let client = offline_client(); client.register_search_term("AC/DC"); client.register_search_term("abba"); client.register_search_term("abba"); // idempotent let node = client.get_lib_node("/tidal/search").await.expect("node"); assert!(node.is_creatable); assert!(!node.is_queable); let titles: Vec<_> = node.children.iter().map(|c| c.title.as_str()).collect(); assert_eq!(titles, vec!["AC/DC", "abba"]); assert_eq!(node.children[0].path, "/tidal/search/AC%2FDC"); assert!(node.children.iter().all(|c| !c.is_queable)); // Term nodes are the modifiable nodes: renamable and deletable. assert!(node .children .iter() .all(|c| c.is_editable && c.is_deletable)); } #[tokio::test] async fn rename_rejects_empty_titles_and_foreign_paths() { use crabidy_core::ProviderError; let client = offline_client(); // Validation happens before any network call, so this works offline. assert_eq!( client.rename_lib_node("/tidal/search/abba", " ").await, Err(ProviderError::InvalidInput) ); assert_eq!( client .rename_lib_node("/tidal/playlists/xyz", "queen") .await, Err(ProviderError::NotSupported) ); // The search node itself is creatable, not editable. assert_eq!( client.rename_lib_node("/tidal/search", "queen").await, Err(ProviderError::NotSupported) ); assert_eq!( client.rename_lib_node("/nope", "queen").await, Err(ProviderError::MalformedPath) ); } #[tokio::test] async fn delete_rejects_foreign_paths() { use crabidy_core::ProviderError; let client = offline_client(); assert_eq!( client.delete_lib_node("/tidal/playlists/xyz").await, Err(ProviderError::NotSupported) ); // The search node itself is not deletable, only its terms are. assert_eq!( client.delete_lib_node("/tidal/search").await, Err(ProviderError::NotSupported) ); assert_eq!( client.delete_lib_node("/nope").await, Err(ProviderError::MalformedPath) ); } #[tokio::test] async fn delete_removes_terms_idempotently_and_returns_the_parent() { let client = offline_client(); client.register_search_term("abba"); client.register_search_term("queen"); let parent = client .delete_lib_node("/tidal/search/abba") .await .expect("refreshed parent"); assert_eq!(parent.path, "/tidal/search"); let titles: Vec<_> = parent.children.iter().map(|c| c.title.as_str()).collect(); assert_eq!(titles, vec!["queen"]); // Deleting an already-gone term succeeds — delete is idempotent. let parent = client .delete_lib_node("/tidal/search/abba") .await .expect("idempotent delete"); assert_eq!(parent.children.len(), 1); } #[test] fn rename_replaces_in_place_and_merges_duplicates() { let client = offline_client(); client.register_search_term("abba"); client.register_search_term("queen"); client.register_search_term("kiss"); // In-place: the renamed term keeps its list position. client.rename_search_term("queen", "wham"); assert_eq!(client.search_terms_snapshot(), vec!["abba", "wham", "kiss"]); // Merge: renaming onto an existing term drops the old slot. client.rename_search_term("abba", "kiss"); assert_eq!(client.search_terms_snapshot(), vec!["wham", "kiss"]); // An unknown old term registers the new one (stale-client forgiveness). client.rename_search_term("ghost", "toto"); assert_eq!(client.search_terms_snapshot(), vec!["wham", "kiss", "toto"]); // Renaming a term to itself changes nothing. client.rename_search_term("kiss", "kiss"); assert_eq!(client.search_terms_snapshot(), vec!["wham", "kiss", "toto"]); } #[test] fn lib_root_offers_a_creatable_search_node() { let client = offline_client(); let root = client.get_lib_root(); let search = root .children .iter() .find(|c| c.path == format!("{PROVIDER_ROOT}/search")) .expect("search child in provider root"); assert!(search.is_creatable); assert!(!search.is_queable); } #[tokio::test] #[ignore = "requires a local tidal config and network access"] async fn test() { let client = setup().await; let user = client.settings.login.user_id.clone().unwrap(); let result = client.get_users_artists(&user).await.unwrap(); println!("{:?}", result); let result = client.get_artist("5293333").await.unwrap(); println!("{:?}", result); let result = client.get_album("244167550").await.unwrap(); println!("{:?}", result); } }