crabidy/crabidy-server/src/provider.rs

use crate::crabidy_store::{CrabidyStore, SaveMode, CRABIDY_PROVIDER_ROOT, CURRENT_NAME};
use crate::orphans::{OrphansProvider, ORPHANS_PROVIDER_ROOT};
use crate::settings::ProviderToggles;
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<ProviderMessage>,
    provider_rx: flume::Receiver<ProviderMessage>,
    /// The Tidal provider; `None` when disabled in `crabidy-server.toml` (or,
    /// unlike the other providers, when its config fails to load — that stays
    /// fatal at startup, see [`ProviderOrchestrator::build`]).
    tidal_client: Option<Arc<tidaldy::Client>>,
    /// `None` when the filesystem provider failed to initialize — the
    /// server runs without `/fs` instead of dying (architecture D5).
    fs_client: Option<Arc<fsdy::Client>>,
    /// Second `fsdy` instance mounting the `/crabidy` toml tree
    /// (architecture/crabidy-store.md D1) — saved queues, bookmarks, and
    /// captures collapsed into one provider. `None` without a state
    /// directory — the server then runs without `/crabidy`.
    crabidy_client: Option<Arc<fsdy::Client>>,
    /// The single writer behind `/crabidy`: the content store and the toml
    /// tree. `None` disables saving/capturing (and the `/crabidy` mount goes
    /// with it).
    crabidy_store: Option<Arc<CrabidyStore>>,
    /// The `/orphans` provider: the store's garbage-collection view
    /// (architecture/orphans.md). `None` whenever `crabidy_store` is — it is a
    /// view over the store.
    orphans_client: Option<Arc<OrphansProvider>>,
    /// The YouTube provider (yt-dlp backed); `None` when the binary
    /// probe failed at init (architecture/youtube-provider.md D2).
    youtube_client: Option<Arc<ytdy::Client>>,
}

/// Whether a path belongs to the filesystem provider.
fn fs_owns(path: &str) -> bool {
    path == fsdy::PROVIDER_ROOT || path.starts_with("/fs/")
}

/// Whether a path belongs to the `/crabidy` provider instance (saved
/// queues, bookmarks, and captures).
fn crabidy_owns(path: &str) -> bool {
    path == CRABIDY_PROVIDER_ROOT || path.starts_with("/crabidy/")
}

/// Whether a path belongs to the YouTube provider.
fn youtube_owns(path: &str) -> bool {
    path == ytdy::PROVIDER_ROOT || path.starts_with("/youtube/")
}

/// Whether a path belongs to the `/orphans` provider.
fn orphans_owns(path: &str) -> bool {
    path == ORPHANS_PROVIDER_ROOT || path.starts_with("/orphans/")
}

impl ProviderOrchestrator {
    /// The tidal client, or `MalformedPath` (with a warning) when the
    /// provider is disabled — a `/tidal` path then has no owner.
    fn tidal_provider(&self) -> Result<&tidaldy::Client, ProviderError> {
        self.tidal_client.as_deref().ok_or_else(|| {
            warn!("tidal provider is disabled");
            ProviderError::MalformedPath
        })
    }

    /// 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
        })
    }

    /// The `/crabidy` client, or `MalformedPath` (with a warning) when the
    /// instance is disabled — a `/crabidy` path then has no owner.
    fn crabidy_provider(&self) -> Result<&fsdy::Client, ProviderError> {
        self.crabidy_client.as_deref().ok_or_else(|| {
            warn!("crabidy library is disabled");
            ProviderError::MalformedPath
        })
    }

    /// The `/crabidy` store (writer), for saves and the startup restore.
    pub fn crabidy_store(&self) -> Option<Arc<CrabidyStore>> {
        self.crabidy_store.clone()
    }

    /// The `/orphans` client, or `MalformedPath` (with a warning) when the
    /// store — and so the view — is disabled.
    fn orphans_provider(&self) -> Result<&OrphansProvider, ProviderError> {
        self.orphans_client.as_deref().ok_or_else(|| {
            warn!("orphans provider is disabled");
            ProviderError::MalformedPath
        })
    }

    /// The YouTube client, or `MalformedPath` (with a warning) when the
    /// provider is disabled — a `/youtube` path then has no owner.
    fn youtube_provider(&self) -> Result<&ytdy::Client, ProviderError> {
        self.youtube_client.as_deref().ok_or_else(|| {
            warn!("youtube 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<Self>, 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}");
                }
            }
            ProviderCommand::CaptureLibraryNode {
                path,
                name,
                download,
                progress_tx,
                result_tx,
            } => {
                // Spawned: capturing a large artist walks many provider
                // nodes (and, for downloads, streams audio) and must not
                // block this loop (the walk itself calls back into
                // `get_lib_node` via `this`). Accept-then-stream
                // (architecture/incremental-captures.md D4): validation
                // answers the RPC, the walk reports through progress
                // events, ending in exactly one terminal event.
                let this = Arc::clone(&self);
                tokio::spawn(
                    async move {
                        let mode = if download {
                            SaveMode::Capture
                        } else {
                            SaveMode::Link
                        };
                        let accepted = match &this.crabidy_store {
                            Some(store) => store.validate(&*this, &path, &name, mode).await,
                            None => Err(crate::capture::CaptureError::Disabled),
                        };
                        if let Err(err) = accepted {
                            warn!(path, name, download, "capture rejected: {err}");
                            if let Err(err) = result_tx.send_async(Err(err)).await {
                                error!("failed to send capture_library_node result: {err}");
                            }
                            return;
                        }
                        if let Err(err) = result_tx.send_async(Ok(())).await {
                            error!("failed to send capture_library_node result: {err}");
                        }
                        let progress = crate::capture::Progress::new(&name, download, progress_tx);
                        let result = match &this.crabidy_store {
                            Some(store) => store.save(&*this, &path, &name, mode, &progress).await,
                            None => Err(crate::capture::CaptureError::Disabled),
                        };
                        if let Err(err) = &result {
                            warn!(path, name, download, "cannot capture subtree: {err}");
                        }
                        progress
                            .finish(result.err().map(|err| err.to_string()))
                            .await;
                    }
                    .in_current_span(),
                );
            }
        }
    }
}

impl ProviderOrchestrator {
    /// Builds the orchestrator, mounting only the providers `enabled` turns on
    /// (from `crabidy-server.toml`'s `providers` list). Tidal, when enabled,
    /// still aborts startup on a broken config — its historical behavior — but
    /// a disabled tidal skips it entirely; the other providers are non-fatal
    /// and just drop their subtree. `/orphans` additionally needs the store,
    /// so it only mounts when `crabidy` does.
    #[instrument]
    pub async fn build(enabled: ProviderToggles) -> Result<Self, ProviderError> {
        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()))?;
        }
        // Tidal: skipped when disabled; when enabled a broken config is still
        // fatal (unlike the local providers), preserving prior behavior.
        let tidal_client = if enabled.tidal {
            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 client = tidaldy::Client::init(&raw_toml_settings)
                .await
                .map_err(|err| {
                    error!("failed to init tidal client: {err}");
                    err
                })?;
            if let Err(err) = tokio::fs::write(&config_file, client.settings()).await {
                error!("failed to write tidal config file: {err}");
            }
            Some(Arc::new(client))
        } else {
            None
        };
        // The filesystem provider is optional: a broken local config only
        // costs the `/fs` subtree, never the server.
        let fs_client = if enabled.fs {
            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();
            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
                }
            }
        } else {
            None
        };
        // The single `/crabidy` provider: one content store + toml tree
        // (architecture/crabidy-store.md). The store owns both roots (state
        // tree + data store) and is the sole writer; the mounted `fsdy`
        // client reads the tree and resolves store playables. Non-fatal like
        // the other local providers — no state/data dir just drops `/crabidy`.
        let crabidy_store = if enabled.crabidy {
            match (
                CrabidyStore::default_tree_root(),
                CrabidyStore::default_store_root(),
            ) {
                (Some(tree), Some(store)) => match CrabidyStore::open(tree, store).await {
                    Ok(store) => Some(Arc::new(store)),
                    Err(err) => {
                        warn!("crabidy library disabled: {err}");
                        None
                    }
                },
                _ => {
                    warn!("crabidy library disabled: no state/data directory");
                    None
                }
            }
        } else {
            None
        };
        // Saved queues/bookmarks are renamable and deletable; the
        // auto-persisted `current` stays untouchable. The whole tree is
        // downloadable (`W` captures a save) and deletable (deleting a toml
        // never touches the shared store, D7). Store playables resolve
        // against the data store root.
        let crabidy_client = crabidy_store.as_ref().and_then(|store| {
            match fsdy::Client::new(CRABIDY_PROVIDER_ROOT, store.tree_dir().to_path_buf()) {
                Ok(client) => Some(Arc::new(
                    client
                        .with_editable_top_level(&[CURRENT_NAME])
                        .with_downloadable_nodes()
                        .with_deletable_tree()
                        .with_store_root(store.store_dir().to_path_buf()),
                )),
                Err(err) => {
                    warn!("crabidy library disabled: {err}");
                    None
                }
            }
        });
        // `/orphans`: a view over the store, so it needs both its own toggle
        // and a live store. Its reference roots are the disk roots of the
        // mounted file providers — the `/crabidy` toml tree and (when enabled)
        // the `/fs` root — the only places a `Playable::Store` link can live
        // (architecture/orphans.md).
        let orphans_client = if enabled.orphans {
            crabidy_store.as_ref().map(|store| {
                let mut ref_roots = vec![store.tree_dir().to_path_buf()];
                if let Some(fs) = &fs_client {
                    ref_roots.push(fs.disk_root().to_path_buf());
                }
                Arc::new(OrphansProvider::new(Arc::clone(store), ref_roots))
            })
        } else {
            None
        };
        // YouTube: non-fatal like the local providers — a missing or
        // broken yt-dlp binary only costs the `/youtube` subtree.
        let youtube_client = if enabled.youtube {
            let yt_config_file = config_dir.join("ytdy.toml");
            debug!(config_file = %yt_config_file.display(), "loading youtube config");
            let raw_yt_settings = fs::read_to_string(&yt_config_file).unwrap_or_default();
            match ytdy::Client::init(&raw_yt_settings).await {
                Ok(client) => {
                    if let Err(err) = tokio::fs::write(&yt_config_file, client.settings()).await {
                        error!("failed to write ytdy config file: {err}");
                    }
                    Some(Arc::new(client))
                }
                Err(err) => {
                    warn!("youtube provider disabled: {err}");
                    None
                }
            }
        } else {
            None
        };
        let (provider_tx, provider_rx) = flume::bounded(100);
        Ok(Self {
            provider_rx,
            provider_tx,
            tidal_client,
            fs_client,
            crabidy_client,
            crabidy_store,
            orphans_client,
            youtube_client,
        })
    }
}

#[async_trait]
impl ProviderClient for ProviderOrchestrator {
    /// Builds with every provider enabled. The server calls [`Self::build`]
    /// with the configured toggles instead; this exists only to satisfy the
    /// trait.
    #[instrument(skip(_s))]
    async fn init(_s: &str) -> Result<Self, ProviderError> {
        Self::build(ProviderToggles::all()).await
    }

    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
                .as_ref()
                .is_some_and(|tidal| tidal.is_track_path(path));
        }
        if fs_owns(path) {
            return self
                .fs_client
                .as_ref()
                .is_some_and(|fs| fs.is_track_path(path));
        }
        if crabidy_owns(path) {
            return self
                .crabidy_client
                .as_ref()
                .is_some_and(|crabidy| crabidy.is_track_path(path));
        }
        if youtube_owns(path) {
            return self
                .youtube_client
                .as_ref()
                .is_some_and(|youtube| youtube.is_track_path(path));
        }
        if orphans_owns(path) {
            // Orphans are addressed as nodes; there are no track paths.
            return false;
        }
        false
    }

    #[instrument(skip(self))]
    async fn get_urls_for_track(&self, track_path: &str) -> Result<Vec<String>, ProviderError> {
        if track_path.starts_with("/tidal/") {
            return self.tidal_provider()?.get_urls_for_track(track_path).await;
        }
        if fs_owns(track_path) {
            return self.fs_provider()?.get_urls_for_track(track_path).await;
        }
        if crabidy_owns(track_path) {
            return self
                .crabidy_provider()?
                .get_urls_for_track(track_path)
                .await;
        }
        if youtube_owns(track_path) {
            return self
                .youtube_provider()?
                .get_urls_for_track(track_path)
                .await;
        }
        if orphans_owns(track_path) {
            return self
                .orphans_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<Track, ProviderError> {
        if track_path.starts_with("/tidal/") {
            return self
                .tidal_provider()?
                .get_metadata_for_track(track_path)
                .await;
        }
        if fs_owns(track_path) {
            return self.fs_provider()?.get_metadata_for_track(track_path).await;
        }
        if crabidy_owns(track_path) {
            return self
                .crabidy_provider()?
                .get_metadata_for_track(track_path)
                .await;
        }
        if youtube_owns(track_path) {
            return self
                .youtube_provider()?
                .get_metadata_for_track(track_path)
                .await;
        }
        if orphans_owns(track_path) {
            return self
                .orphans_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();
        if self.tidal_client.is_some() {
            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);
        }
        if self.crabidy_client.is_some() {
            let child = LibraryNodeChild::new(
                CRABIDY_PROVIDER_ROOT.to_owned(),
                "crabidy".to_owned(),
                false,
            );
            root_node.children.push(child);
        }
        if self.youtube_client.is_some() {
            let child =
                LibraryNodeChild::new(ytdy::PROVIDER_ROOT.to_owned(), "youtube".to_owned(), false);
            root_node.children.push(child);
        }
        if self.orphans_client.is_some() {
            let child = LibraryNodeChild::new(
                ORPHANS_PROVIDER_ROOT.to_owned(),
                "orphans".to_owned(),
                false,
            );
            root_node.children.push(child);
        }
        root_node
    }

    #[instrument(skip(self))]
    async fn get_lib_node(&self, path: &str) -> Result<LibraryNode, ProviderError> {
        if path == crabidy_core::ROOT_PATH {
            debug!("serving global library root");
            return Ok(self.get_lib_root());
        }
        let mut node = if path == tidaldy::PROVIDER_ROOT || path.starts_with("/tidal/") {
            self.tidal_provider()?.get_lib_node(path).await?
        } else if fs_owns(path) {
            self.fs_provider()?.get_lib_node(path).await?
        } else if crabidy_owns(path) {
            self.crabidy_provider()?.get_lib_node(path).await?
        } else if youtube_owns(path) {
            self.youtube_provider()?.get_lib_node(path).await?
        } else if orphans_owns(path) {
            self.orphans_provider()?.get_lib_node(path).await?
        } else {
            warn!(path, "no provider owns this path");
            return Err(ProviderError::MalformedPath);
        };
        // Mark tracks already held in the content store (D3/D8) — cheap, and
        // works while browsing any provider, not just `/crabidy`.
        if let Some(store) = &self.crabidy_store {
            store.annotate_captured(&mut node).await;
        }
        Ok(node)
    }

    /// 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<LibraryNode, ProviderError> {
        if parent_path == tidaldy::PROVIDER_ROOT || parent_path.starts_with("/tidal/") {
            return self
                .tidal_provider()?
                .create_lib_node(parent_path, title)
                .await;
        }
        if fs_owns(parent_path) {
            return self
                .fs_provider()?
                .create_lib_node(parent_path, title)
                .await;
        }
        if crabidy_owns(parent_path) {
            return self
                .crabidy_provider()?
                .create_lib_node(parent_path, title)
                .await;
        }
        if youtube_owns(parent_path) {
            return self
                .youtube_provider()?
                .create_lib_node(parent_path, title)
                .await;
        }
        if orphans_owns(parent_path) {
            return self
                .orphans_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<LibraryNode, ProviderError> {
        if path == tidaldy::PROVIDER_ROOT || path.starts_with("/tidal/") {
            return self
                .tidal_provider()?
                .rename_lib_node(path, new_title)
                .await;
        }
        if fs_owns(path) {
            return self.fs_provider()?.rename_lib_node(path, new_title).await;
        }
        if crabidy_owns(path) {
            return self
                .crabidy_provider()?
                .rename_lib_node(path, new_title)
                .await;
        }
        if youtube_owns(path) {
            return self
                .youtube_provider()?
                .rename_lib_node(path, new_title)
                .await;
        }
        if orphans_owns(path) {
            return self
                .orphans_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<Vec<Track>>,
    ) -> Result<(), ProviderError> {
        if path == tidaldy::PROVIDER_ROOT || path.starts_with("/tidal/") {
            return self
                .tidal_provider()?
                .resolve_tracks_into(path, chunk_tx)
                .await;
        }
        if fs_owns(path) {
            return self
                .fs_provider()?
                .resolve_tracks_into(path, chunk_tx)
                .await;
        }
        if crabidy_owns(path) {
            return self
                .crabidy_provider()?
                .resolve_tracks_into(path, chunk_tx)
                .await;
        }
        if youtube_owns(path) {
            return self
                .youtube_provider()?
                .resolve_tracks_into(path, chunk_tx)
                .await;
        }
        if orphans_owns(path) {
            return self
                .orphans_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<LibraryNode, ProviderError> {
        if path == tidaldy::PROVIDER_ROOT || path.starts_with("/tidal/") {
            return self.tidal_provider()?.delete_lib_node(path).await;
        }
        if fs_owns(path) {
            return self.fs_provider()?.delete_lib_node(path).await;
        }
        if crabidy_owns(path) {
            return self.crabidy_provider()?.delete_lib_node(path).await;
        }
        if youtube_owns(path) {
            return self.youtube_provider()?.delete_lib_node(path).await;
        }
        if orphans_owns(path) {
            return self.orphans_provider()?.delete_lib_node(path).await;
        }
        warn!(path, "no provider supports deleting this node");
        Err(ProviderError::NotSupported)
    }
}