use crate::{ProviderCommand, ProviderMessage}; use async_trait::async_trait; use crabidy_core::{ proto::crabidy::{LibraryNode, LibraryNodeChild, Track}, ProviderClient, ProviderError, }; use std::{fs, path::PathBuf, sync::Arc}; use tracing::{debug, debug_span, error, instrument, warn, Instrument}; #[derive(Debug)] pub struct ProviderOrchestrator { pub provider_tx: flume::Sender, provider_rx: flume::Receiver, tidal_client: Arc, /// `None` when the filesystem provider failed to initialize — the /// server runs without `/fs` instead of dying (architecture D5). fs_client: Option>, } /// Whether a path belongs to the filesystem provider. fn fs_owns(path: &str) -> bool { path == fsdy::PROVIDER_ROOT || path.starts_with("/fs/") } impl ProviderOrchestrator { /// The fs client, or `MalformedPath` (with a warning) when the /// provider is disabled — a `/fs` path then has no owner. fn fs_provider(&self) -> Result<&fsdy::Client, ProviderError> { self.fs_client.as_deref().ok_or_else(|| { warn!("filesystem provider is disabled"); ProviderError::MalformedPath }) } pub fn run(self) { tokio::spawn(async move { // Behind an Arc so long-running resolves can be spawned onto // their own tasks while the loop keeps serving commands. let this = Arc::new(self); while let Ok(ProviderMessage { span, command }) = this.provider_rx.recv_async().await { let handler_span = debug_span!(parent: &span, "provider_command", command = command.name()); Arc::clone(&this) .handle_command(command) .instrument(handler_span) .await; } warn!("provider message channel closed, loop exiting"); }); } async fn handle_command(self: Arc, command: ProviderCommand) { match command { ProviderCommand::GetLibraryNode { path, result_tx } => { let result = self.get_lib_node(&path).await; if let Err(err) = result_tx.send_async(result).await { error!("failed to send get_library_node result: {err}"); } } ProviderCommand::GetTrackUrls { path, result_tx } => { let result = self.get_urls_for_track(&path).await; if let Err(err) = result_tx.send_async(result).await { error!("failed to send get_track_urls result: {err}"); } } ProviderCommand::ResolveTracks { path, chunk_tx } => { // Spawned: a large resolve must not block this loop, or the // playback side deadlocks waiting for `GetTrackUrls` while // chunks back up. Dropping `chunk_tx` at the end of the // task is the completion signal; there is no reply channel. let this = Arc::clone(&self); tokio::spawn( async move { if let Err(err) = this.resolve_tracks_into(&path, chunk_tx).await { warn!(path, "resolve produced no tracks: {err}"); } } .in_current_span(), ); } ProviderCommand::CreateLibraryNode { parent_path, title, result_tx, } => { let result = self.create_lib_node(&parent_path, &title).await; if let Err(err) = result_tx.send_async(result).await { error!("failed to send create_library_node result: {err}"); } } ProviderCommand::RenameLibraryNode { path, new_title, result_tx, } => { let result = self.rename_lib_node(&path, &new_title).await; if let Err(err) = result_tx.send_async(result).await { error!("failed to send rename_library_node result: {err}"); } } ProviderCommand::DeleteLibraryNode { path, result_tx } => { let result = self.delete_lib_node(&path).await; if let Err(err) = result_tx.send_async(result).await { error!("failed to send delete_library_node result: {err}"); } } } } } #[async_trait] impl ProviderClient for ProviderOrchestrator { #[instrument(skip(_s))] async fn init(_s: &str) -> Result { let config_dir = dirs::config_dir() .map(|d| d.join("crabidy")) .unwrap_or(PathBuf::from("/tmp")); let dir_exists = tokio::fs::try_exists(&config_dir) .await .map_err(|e| ProviderError::Config(e.to_string()))?; if !dir_exists { tokio::fs::create_dir(&config_dir) .await .map_err(|e| ProviderError::Config(e.to_string()))?; } let config_file = config_dir.join("tidaly.toml"); debug!(config_file = %config_file.display(), "loading tidal config"); let raw_toml_settings = fs::read_to_string(&config_file).unwrap_or_default(); let tidal_client = Arc::new(tidaldy::Client::init(&raw_toml_settings).await.map_err( |err| { error!("failed to init tidal client: {err}"); err }, )?); let new_toml_config = tidal_client.settings(); if let Err(err) = tokio::fs::write(&config_file, new_toml_config).await { error!("failed to write tidal config file: {err}"); }; // The filesystem provider is optional: a broken local config only // costs the `/fs` subtree, never the server. let fs_config_file = config_dir.join("fsdy.toml"); debug!(config_file = %fs_config_file.display(), "loading fs config"); let raw_fs_settings = fs::read_to_string(&fs_config_file).unwrap_or_default(); let fs_client = match fsdy::Client::init(&raw_fs_settings).await { Ok(client) => { if let Err(err) = tokio::fs::write(&fs_config_file, client.settings()).await { error!("failed to write fsdy config file: {err}"); } Some(Arc::new(client)) } Err(err) => { warn!("filesystem provider disabled: {err}"); None } }; let (provider_tx, provider_rx) = flume::bounded(100); Ok(Self { provider_rx, provider_tx, tidal_client, fs_client, }) } fn settings(&self) -> String { String::new() } /// Routes to the provider that owns the path. fn is_track_path(&self, path: &str) -> bool { if path == "/tidal" || path.starts_with("/tidal/") { return self.tidal_client.is_track_path(path); } if fs_owns(path) { return self .fs_client .as_ref() .is_some_and(|fs| fs.is_track_path(path)); } false } #[instrument(skip(self))] async fn get_urls_for_track(&self, track_path: &str) -> Result, ProviderError> { if track_path.starts_with("/tidal/") { return self.tidal_client.get_urls_for_track(track_path).await; } if fs_owns(track_path) { return self.fs_provider()?.get_urls_for_track(track_path).await; } warn!(path = track_path, "no provider owns this track path"); Err(ProviderError::MalformedPath) } #[instrument(skip(self))] async fn get_metadata_for_track(&self, track_path: &str) -> Result { if track_path.starts_with("/tidal/") { return self.tidal_client.get_metadata_for_track(track_path).await; } if fs_owns(track_path) { return self.fs_provider()?.get_metadata_for_track(track_path).await; } warn!(path = track_path, "no provider owns this track path"); Err(ProviderError::MalformedPath) } fn get_lib_root(&self) -> LibraryNode { let mut root_node = LibraryNode::new(); let child = LibraryNodeChild::new(tidaldy::PROVIDER_ROOT.to_owned(), "tidal".to_owned(), false); root_node.children.push(child); if self.fs_client.is_some() { let child = LibraryNodeChild::new(fsdy::PROVIDER_ROOT.to_owned(), "fs".to_owned(), false); root_node.children.push(child); } root_node } #[instrument(skip(self))] async fn get_lib_node(&self, path: &str) -> Result { if path == crabidy_core::ROOT_PATH { debug!("serving global library root"); return Ok(self.get_lib_root()); } if path == tidaldy::PROVIDER_ROOT || path.starts_with("/tidal/") { return self.tidal_client.get_lib_node(path).await; } if fs_owns(path) { return self.fs_provider()?.get_lib_node(path).await; } warn!(path, "no provider owns this path"); Err(ProviderError::MalformedPath) } /// Routes to the provider that owns the parent path. The synthetic root /// itself is not creatable. #[instrument(skip(self))] async fn create_lib_node( &self, parent_path: &str, title: &str, ) -> Result { if parent_path == tidaldy::PROVIDER_ROOT || parent_path.starts_with("/tidal/") { return self.tidal_client.create_lib_node(parent_path, title).await; } if fs_owns(parent_path) { return self .fs_provider()? .create_lib_node(parent_path, title) .await; } warn!(parent_path, "no provider supports creating nodes here"); Err(ProviderError::NotSupported) } /// Routes to the provider that owns the path. The synthetic root's own /// children are fixed and never editable. #[instrument(skip(self))] async fn rename_lib_node( &self, path: &str, new_title: &str, ) -> Result { if path == tidaldy::PROVIDER_ROOT || path.starts_with("/tidal/") { return self.tidal_client.rename_lib_node(path, new_title).await; } if fs_owns(path) { return self.fs_provider()?.rename_lib_node(path, new_title).await; } warn!(path, "no provider supports renaming this node"); Err(ProviderError::NotSupported) } /// Routes to the provider that owns the path. The synthetic root is not /// queueable, so only provider-owned paths can resolve. #[instrument(skip(self, chunk_tx))] async fn resolve_tracks_into( &self, path: &str, chunk_tx: flume::Sender>, ) -> Result<(), ProviderError> { if path == tidaldy::PROVIDER_ROOT || path.starts_with("/tidal/") { return self.tidal_client.resolve_tracks_into(path, chunk_tx).await; } if fs_owns(path) { return self .fs_provider()? .resolve_tracks_into(path, chunk_tx) .await; } warn!(path, "no provider owns this path"); Err(ProviderError::MalformedPath) } /// Routes to the provider that owns the path. The synthetic root's own /// children are fixed and never deletable. #[instrument(skip(self))] async fn delete_lib_node(&self, path: &str) -> Result { if path == tidaldy::PROVIDER_ROOT || path.starts_with("/tidal/") { return self.tidal_client.delete_lib_node(path).await; } if fs_owns(path) { return self.fs_provider()?.delete_lib_node(path).await; } warn!(path, "no provider supports deleting this node"); Err(ProviderError::NotSupported) } }