crabidy/cbd-tui/src/rpc.rs

use crabidy_core::proto::crabidy::{
    crabidy_service_client::CrabidyServiceClient, AppendRequest, CaptureLibraryNodeRequest,
    ChangeVolumeRequest, ClearQueueRequest, CreateLibraryNodeRequest, DeleteLibraryNodeRequest,
    GetLibraryNodeRequest, GetUpdateStreamRequest, GetUpdateStreamResponse, InitRequest,
    InitResponse, InsertRequest, LibraryNode, NextRequest, PrevRequest, QueueRequest,
    RemoveRequest, RenameLibraryNodeRequest, ReplaceRequest, RestartTrackRequest, SaveQueueRequest,
    SetCurrentRequest, ToggleMuteRequest, TogglePlayRequest, ToggleRepeatRequest,
    ToggleShuffleRequest,
};

use std::{collections::HashMap, error::Error, fmt, time::Duration};

use base64::Engine;
use tonic::{
    metadata::MetadataValue,
    service::{interceptor::InterceptedService, Interceptor},
    transport::{Channel, Endpoint},
    Request, Status, Streaming,
};

// FIXME: use anyhow + thiserror
#[derive(Debug)]
enum RpcClientError {
    NotFound,
}

impl fmt::Display for RpcClientError {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        match self {
            RpcClientError::NotFound => write!(f, "Requested item not found"),
        }
    }
}

impl Error for RpcClientError {}

/// Attaches the configured `authorization: Basic …` header to every
/// outgoing request (architecture/roles-auth.md). Without configured
/// credentials it attaches nothing, keeping the zero-config local
/// setup working against an open server. The header value is a secret
/// and never logged.
#[derive(Clone)]
pub struct AuthInterceptor {
    header: Option<MetadataValue<tonic::metadata::Ascii>>,
}

impl AuthInterceptor {
    /// `user` empty means "no credentials".
    fn new(user: &str, password: &str) -> Result<Self, Box<dyn Error>> {
        if user.is_empty() {
            return Ok(Self { header: None });
        }
        let encoded =
            base64::engine::general_purpose::STANDARD.encode(format!("{user}:{password}"));
        let header = format!("Basic {encoded}")
            .parse()
            // The value is base64: this cannot fail on credential
            // contents, only on programmer error.
            .map_err(|_| "cannot encode credentials header")?;
        Ok(Self {
            header: Some(header),
        })
    }
}

impl Interceptor for AuthInterceptor {
    fn call(&mut self, mut request: Request<()>) -> Result<Request<()>, Status> {
        if let Some(header) = &self.header {
            request
                .metadata_mut()
                .insert("authorization", header.clone());
        }
        Ok(request)
    }
}

/// The service client with the auth interceptor baked in.
type Client = CrabidyServiceClient<InterceptedService<Channel, AuthInterceptor>>;

pub struct RpcClient {
    library_node_cache: HashMap<String, LibraryNode>,
    client: Client,
    pub update_stream: Streaming<GetUpdateStreamResponse>,
}

/// Whether a library listing may be served from the session cache.
///
/// The server-side folder providers mutate behind the client's back —
/// saves and captures appear under `/crabidy` (`w`/`W`), files change on
/// disk under `/fs` — so a cached listing turns freshly captured content
/// invisible until a restart. Their listings are cheap local directory
/// walks on the server; always refetch them. Remote provider nodes (tidal,
/// youtube) keep the cache that makes back-navigation instant.
fn is_cacheable(path: &str) -> bool {
    const MUTABLE_ROOTS: [&str; 2] = ["/crabidy", "/fs"];
    !MUTABLE_ROOTS.iter().any(|root| {
        path == *root || (path.starts_with(root) && path.as_bytes().get(root.len()) == Some(&b'/'))
    })
}

impl RpcClient {
    pub async fn connect(
        server: &'static crate::config::ServerConfig,
    ) -> Result<RpcClient, Box<dyn Error>> {
        let endpoint = Endpoint::from_static(&server.address).connect_lazy();
        let interceptor = AuthInterceptor::new(&server.user, &server.password)?;
        let mut client = CrabidyServiceClient::with_interceptor(endpoint, interceptor);

        let update_stream = Self::get_update_stream(&mut client).await;
        let library_node_cache: HashMap<String, LibraryNode> = HashMap::new();

        Ok(RpcClient {
            client,
            library_node_cache,
            update_stream,
        })
    }

    async fn get_update_stream(client: &mut Client) -> Streaming<GetUpdateStreamResponse> {
        loop {
            let get_update_stream_request = Request::new(GetUpdateStreamRequest {});
            if let Ok(resp) = client.get_update_stream(get_update_stream_request).await {
                return resp.into_inner();
            } else {
                tokio::time::sleep(Duration::from_secs(2)).await;
            }
        }
    }

    pub async fn reconnect_update_stream(&mut self) {
        self.update_stream = Self::get_update_stream(&mut self.client).await;
    }

    pub async fn init(&mut self) -> Result<InitResponse, Box<dyn Error>> {
        let init_request = Request::new(InitRequest {});
        let response = self.client.init(init_request).await?;
        Ok(response.into_inner())
    }

    pub async fn get_library_node(
        &mut self,
        path: &str,
    ) -> Result<Option<&LibraryNode>, Box<dyn Error>> {
        if is_cacheable(path) && self.library_node_cache.contains_key(path) {
            return Ok(self.library_node_cache.get(path));
        }
        let get_library_node_request = Request::new(GetLibraryNodeRequest {
            path: path.to_string(),
        });
        let response = self
            .client
            .get_library_node(get_library_node_request)
            .await?;
        if let Some(library_node) = response.into_inner().node {
            // Non-cacheable nodes are stored too (the return borrows from
            // the map) — they are just always refetched above.
            self.library_node_cache
                .insert(path.to_string(), library_node);
            return Ok(self.library_node_cache.get(path));
        }
        Err(Box::new(RpcClientError::NotFound))
    }

    /// Creates a child node under a creatable parent and returns it.
    ///
    /// Cache contract: the created node is inserted into
    /// `library_node_cache`, and the *parent's* cache entry is evicted —
    /// its child listing just changed and would otherwise be served stale
    /// when the user ascends back to it.
    pub async fn create_library_node(
        &mut self,
        parent_path: &str,
        title: &str,
    ) -> Result<&LibraryNode, Box<dyn Error>> {
        let request = Request::new(CreateLibraryNodeRequest {
            parent_path: parent_path.to_string(),
            title: title.to_string(),
        });
        let response = self.client.create_library_node(request).await?;
        let Some(node) = response.into_inner().node else {
            return Err(Box::new(RpcClientError::NotFound));
        };
        // The parent's child listing just changed; a cached copy would hide
        // the new node when the user ascends back to it.
        self.library_node_cache.remove(parent_path);
        let path = node.path.clone();
        self.library_node_cache.insert(path.clone(), node);
        Ok(&self.library_node_cache[&path])
    }

    /// Renames an editable node and returns it at its (changed) path.
    ///
    /// Cache contract: the old path's entry and the parent's entry are
    /// evicted (the parent's child listing changed; the old path is dead),
    /// and the renamed node is inserted under its new path.
    pub async fn rename_library_node(
        &mut self,
        path: &str,
        new_title: &str,
    ) -> Result<&LibraryNode, Box<dyn Error>> {
        let request = Request::new(RenameLibraryNodeRequest {
            path: path.to_string(),
            new_title: new_title.to_string(),
        });
        let response = self.client.rename_library_node(request).await?;
        let Some(node) = response.into_inner().node else {
            return Err(Box::new(RpcClientError::NotFound));
        };
        // The old path is dead and the parent's child listing changed;
        // cached copies would resurrect the old term.
        self.library_node_cache.remove(path);
        if let Some(parent) = crabidy_core::parent_path(path) {
            self.library_node_cache.remove(parent);
        }
        let new_path = node.path.clone();
        self.library_node_cache.insert(new_path.clone(), node);
        Ok(&self.library_node_cache[&new_path])
    }

    /// Deletes a node and returns the refreshed parent listing.
    ///
    /// Cache contract: the deleted path's entry and the parent's stale entry
    /// are evicted, and the returned parent node is inserted fresh.
    pub async fn delete_library_node(
        &mut self,
        path: &str,
    ) -> Result<&LibraryNode, Box<dyn Error>> {
        let request = Request::new(DeleteLibraryNodeRequest {
            path: path.to_string(),
        });
        let response = self.client.delete_library_node(request).await?;
        let Some(parent) = response.into_inner().parent else {
            return Err(Box::new(RpcClientError::NotFound));
        };
        // Drop the deleted node and the stale parent listing; the response
        // carries the fresh parent to cache instead.
        self.library_node_cache.remove(path);
        let parent_path = parent.path.clone();
        self.library_node_cache.insert(parent_path.clone(), parent);
        Ok(&self.library_node_cache[&parent_path])
    }

    pub async fn append_tracks(&mut self, paths: Vec<String>) -> Result<(), Box<dyn Error>> {
        let append_request = Request::new(AppendRequest { paths });
        self.client.append(append_request).await?;
        Ok(())
    }

    pub async fn queue_tracks(&mut self, paths: Vec<String>) -> Result<(), Box<dyn Error>> {
        let queue_request = Request::new(QueueRequest { paths });
        self.client.queue(queue_request).await?;
        Ok(())
    }

    pub async fn insert_tracks(
        &mut self,
        paths: Vec<String>,
        pos: usize,
    ) -> Result<(), Box<dyn Error>> {
        let insert_request = Request::new(InsertRequest {
            paths,
            position: pos as u32,
        });
        self.client.insert(insert_request).await?;
        Ok(())
    }

    pub async fn remove_tracks(&mut self, positions: Vec<usize>) -> Result<(), Box<dyn Error>> {
        let remove_request = Request::new(RemoveRequest {
            positions: positions.iter().map(|p| *p as u32).collect(),
        });
        self.client.remove(remove_request).await?;
        Ok(())
    }

    pub async fn clear_queue(&mut self, exclude_current: bool) -> Result<(), Box<dyn Error>> {
        let clear_queue_request = Request::new(ClearQueueRequest { exclude_current });
        self.client.clear_queue(clear_queue_request).await?;
        Ok(())
    }

    pub async fn save_queue(&mut self, name: String) -> Result<(), Box<dyn Error>> {
        let save_queue_request = Request::new(SaveQueueRequest { name });
        self.client.save_queue(save_queue_request).await?;
        Ok(())
    }

    pub async fn capture_library_node(
        &mut self,
        path: String,
        name: String,
        download: bool,
    ) -> Result<(), Box<dyn Error>> {
        let capture_request = Request::new(CaptureLibraryNodeRequest {
            path,
            name,
            download,
        });
        self.client.capture_library_node(capture_request).await?;
        Ok(())
    }

    pub async fn replace_queue(&mut self, paths: Vec<String>) -> Result<(), Box<dyn Error>> {
        let replace_request = Request::new(ReplaceRequest { paths });
        self.client.replace(replace_request).await?;
        Ok(())
    }

    pub async fn next_track(&mut self) -> Result<(), Box<dyn Error>> {
        let next_request = Request::new(NextRequest {});
        self.client.next(next_request).await?;
        Ok(())
    }

    pub async fn prev_track(&mut self) -> Result<(), Box<dyn Error>> {
        let prev_request = Request::new(PrevRequest {});
        self.client.prev(prev_request).await?;
        Ok(())
    }

    pub async fn restart_track(&mut self) -> Result<(), Box<dyn Error>> {
        let restart_track_request = Request::new(RestartTrackRequest {});
        self.client.restart_track(restart_track_request).await?;
        Ok(())
    }

    pub async fn set_current_track(&mut self, pos: usize) -> Result<(), Box<dyn Error>> {
        let set_current_request = Request::new(SetCurrentRequest {
            position: pos as u32,
        });
        self.client.set_current(set_current_request).await?;
        Ok(())
    }

    pub async fn toggle_play(&mut self) -> Result<(), Box<dyn Error>> {
        let toggle_play_request = Request::new(TogglePlayRequest {});
        self.client.toggle_play(toggle_play_request).await?;
        Ok(())
    }

    pub async fn toggle_shuffle(&mut self) -> Result<(), Box<dyn Error>> {
        let toggle_shuffle_request = Request::new(ToggleShuffleRequest {});
        self.client.toggle_shuffle(toggle_shuffle_request).await?;
        Ok(())
    }

    pub async fn toggle_repeat(&mut self) -> Result<(), Box<dyn Error>> {
        let toggle_repeat_request = Request::new(ToggleRepeatRequest {});
        self.client.toggle_repeat(toggle_repeat_request).await?;
        Ok(())
    }

    pub async fn change_volume(&mut self, delta: f32) -> Result<(), Box<dyn Error>> {
        let change_volume_request = Request::new(ChangeVolumeRequest { delta });
        self.client.change_volume(change_volume_request).await?;
        Ok(())
    }

    pub async fn toggle_mute(&mut self) -> Result<(), Box<dyn Error>> {
        let toggle_mute_request = Request::new(ToggleMuteRequest {});
        self.client.toggle_mute(toggle_mute_request).await?;
        Ok(())
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn without_credentials_no_authorization_header_is_sent() {
        let mut interceptor = AuthInterceptor::new("", "ignored").expect("build");
        let request = interceptor.call(Request::new(())).expect("intercept");
        assert!(request.metadata().get("authorization").is_none());
    }

    #[test]
    fn credentials_become_a_basic_authorization_header() {
        let mut interceptor = AuthInterceptor::new("queue-owner", "secret").expect("build");
        let request = interceptor.call(Request::new(())).expect("intercept");
        let header = request
            .metadata()
            .get("authorization")
            .expect("header attached")
            .to_str()
            .expect("ascii");
        // base64("queue-owner:secret")
        assert_eq!(header, "Basic cXVldWUtb3duZXI6c2VjcmV0");
    }

    #[test]
    fn mutable_provider_listings_are_never_cached() {
        // Freshly captured/saved content must show up on the next visit
        // (a cached /crabidy hid new saves until a TUI restart).
        for path in [
            "/crabidy",
            "/crabidy/faves",
            "/crabidy/current",
            "/crabidy/road trip",
            "/fs/music",
        ] {
            assert!(!is_cacheable(path), "{path}");
        }
        // Remote providers keep instant back-navigation…
        for path in ["/", "/tidal", "/tidal/artists/1", "/youtube/search/x"] {
            assert!(is_cacheable(path), "{path}");
        }
        // …and prefix look-alikes are not swept up.
        assert!(is_cacheable("/fsdy"));
        assert!(is_cacheable("/crabidystore"));
    }
}