From 58f7f9c66b99408ecb5edcdb314fde8ed02052cc Mon Sep 17 00:00:00 2001 From: Test User Date: Tue, 21 Jul 2026 14:15:47 +0200 Subject: [PATCH] Bundle server and TUI into a single cbd binary crabidy-server and cbd-tui become libraries with thin mains: crabidy_server::serve(addr) hosts the whole server stack, cbd_tui::run(config) the client loops. The new cbd binary logs both halves to one file, starts the server in-process, waits for the socket (adopting an already-running standalone server on an occupied port), and runs the TUI against it over the unchanged localhost gRPC wire. Co-Authored-By: Claude Fable 5 --- Cargo.lock | 14 ++ Cargo.toml | 3 + architecture/cbd-bundle.md | 99 ++++++++++ cbd-tui/src/lib.rs | 305 ++++++++++++++++++++++++++++++ cbd-tui/src/main.rs | 300 +----------------------------- cbd/Cargo.toml | 14 ++ cbd/src/main.rs | 183 ++++++++++++++++++ crabidy-server/src/lib.rs | 330 ++++++++++++++++++++++++++++++++- crabidy-server/src/main.rs | 327 +------------------------------- crabidy-server/src/playback.rs | 4 +- crabidy-server/src/provider.rs | 16 +- crabidy-server/src/rpc.rs | 4 +- plan/cbd-bundle.md | 28 +++ plan/summary.md | 26 +++ quality/cbd-bundle.md | 36 ++++ 15 files changed, 1063 insertions(+), 626 deletions(-) create mode 100644 architecture/cbd-bundle.md create mode 100644 cbd-tui/src/lib.rs create mode 100644 cbd/Cargo.toml create mode 100644 cbd/src/main.rs create mode 100644 plan/cbd-bundle.md create mode 100644 quality/cbd-bundle.md diff --git a/Cargo.lock b/Cargo.lock index 9a5a755..efff58a 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -448,6 +448,20 @@ dependencies = [ "rustversion", ] +[[package]] +name = "cbd" +version = "0.1.0" +dependencies = [ + "cbd-tui", + "crabidy-core", + "crabidy-server", + "dirs", + "tokio", + "tracing", + "tracing-appender", + "tracing-subscriber", +] + [[package]] name = "cbd-tui" version = "0.1.0" diff --git a/Cargo.toml b/Cargo.toml index 3bbe36b..e6422d7 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -2,6 +2,7 @@ resolver = "2" members = [ "audio-player", + "cbd", "cbd-tui", "crabidy-core", "crabidy-server", @@ -66,7 +67,9 @@ url = "2" # Local crates audio-player = { path = "audio-player" } +cbd-tui = { path = "cbd-tui" } crabidy-core = { path = "crabidy-core" } +crabidy-server = { path = "crabidy-server" } fsdy = { path = "fsdy" } tidaldy = { path = "tidaldy" } ytdy = { path = "ytdy" } diff --git a/architecture/cbd-bundle.md b/architecture/cbd-bundle.md new file mode 100644 index 0000000..7e94858 --- /dev/null +++ b/architecture/cbd-bundle.md @@ -0,0 +1,99 @@ +# cbd: bundled server + client binary + +## Context and problem statement + +`crabidy-server` and `cbd-tui` are separate binaries: the normal setup +runs a long-lived server and attaches TUIs to it. The user wants a +single binary **`cbd`** for the one-machine case: starting it starts the +server and connects the TUI to it. Everything else works exactly the +same — same configs, same gRPC wire, same features. + +## Assumptions (confirmed) + +- `cbd-tui`'s config (`cbd-tui.toml`) already carries the server + address; the server listens on a constant (`0.0.0.0:50051`). +- Both mains are thin shells over module code: the server's `main.rs` + holds the command/message enums and the startup sequence; the TUI's + holds tracing setup and two loops (`orchestrate`, `run_ui`). +- The gRPC boundary stays: the bundled TUI talks to the in-process + server over localhost exactly like a remote one ("works the same + completely"). No in-process transport special-casing. + +## Decisions + +### D1 — Both binaries become libraries with thin mains + +- **crabidy-server**: `playback`, `provider`, `rpc` move from bin + modules to lib modules; the command/message enums and the startup + sequence move into the lib (`serve(addr)` builds orchestrator, queue + store, playback, rpc service and serves tonic on `addr`). `main.rs` + keeps only stderr tracing setup + `serve(LISTEN_ADDR)`. +- **cbd-tui**: gains `src/lib.rs` exposing `run(config)` (the two loops + and their channels); `main.rs` keeps file-based tracing setup + + config init + `run`. +- Behavior-preserving: no logic changes, only module moves and + `crabidy_server::` → `crate::` path rewrites. All existing tests move + along unchanged. + +Not chosen: `cbd` spawning `crabidy-server` as a subprocess — that +needs the second binary installed, which is exactly what "a single +binary" is for. + +### D2 — `cbd` = start (or adopt) the server, then run the TUI + +New tiny binary crate `cbd`: + +1. Tracing goes to the TUI's log file for **both** halves — the + terminal belongs to the TUI, so the server's stderr logging would + corrupt it. +2. Spawn `crabidy_server::serve(LISTEN_ADDR)` on a background task. + If the port is already taken (`AddrInUse` — a standalone server is + running), log and carry on: the TUI simply connects to the existing + server. Any other server error before readiness is fatal. +3. Wait for readiness by polling a TCP connect against the configured + server address (bounded retries with delay, then a clear error). +4. Run the TUI exactly as `cbd-tui` would, with the same + `cbd-tui.toml`. + +Quitting the TUI ends the process — and with it the in-process server. +That is inherent to bundling and fine: the current queue is persisted +continuously, so the next start restores it (the ≤200 ms persist +debounce window is the same loss window as killing the standalone +server). + +### D3 — Out of scope (explicitly) + +- An in-process (channel) transport instead of localhost gRPC. +- Daemonizing: `cbd` never outlives its TUI. Users who want a + persistent server keep running `crabidy-server`. +- CLI subcommands (`cbd server`, `cbd attach`, …) — later if wanted. + +## Structure + +```d2 +direction: right + +cbd: "cbd (one binary)" { + boot: "main: file tracing,\nspawn server, wait, run TUI" + srv: "crabidy-server lib\nserve(addr)" + tui: "cbd-tui lib\nrun(config)" + boot -> srv: "tokio::spawn\n(AddrInUse → adopt)" + boot -> tui: "after TCP readiness" + tui -> srv: "localhost gRPC\n(unchanged wire)" +} + +standalone: "crabidy-server bin\n(unchanged)" +remote: "cbd-tui bin\n(unchanged)" +remote -> standalone: "gRPC (remote setup\nkeeps working)" +``` + +## Risks and open questions + +- **Port constant**: the server listens on `0.0.0.0:50051` and the TUI + config defaults to localhost; if the user points `cbd-tui.toml` at a + remote server, `cbd` still starts a local one (and connects to the + configured, remote one). Accepted: `cbd` is the one-machine tool. +- **Two log producers, one file**: server and TUI layers share the + bundled tracing subscriber; targets distinguish them. +- Open (future): a `--no-server` flag; graceful server shutdown (flush + the persister) on TUI exit. diff --git a/cbd-tui/src/lib.rs b/cbd-tui/src/lib.rs new file mode 100644 index 0000000..7920e04 --- /dev/null +++ b/cbd-tui/src/lib.rs @@ -0,0 +1,305 @@ +//! The cbd-tui client as a library: the server-facing orchestration loop +//! and the terminal UI loop, exposed as [`run`] so both the standalone +//! `cbd-tui` binary and the bundled `cbd` binary can host them +//! (architecture/cbd-bundle.md D1). Tracing setup stays with the +//! binaries — where logs go is a hosting decision. + +pub mod app; +pub mod config; +pub mod rpc; + +use std::{ + error::Error, + io, + time::{Duration, Instant}, +}; + +use crabidy_core::proto::crabidy::{get_update_stream_response::Update as StreamUpdate, PlayState}; + +use crossterm::{ + event::{self, DisableMouseCapture, EnableMouseCapture, Event, KeyEventKind}, + execute, + terminal::{disable_raw_mode, enable_raw_mode, EnterAlternateScreen, LeaveAlternateScreen}, +}; +use flume::{Receiver, Sender}; + +use ratatui::{backend::CrosstermBackend, Terminal}; +use tokio::select; +use tokio_stream::StreamExt; + +use app::{bindings, App, DispatchResult, MessageFromUi, MessageToUi}; +use config::Config; +use rpc::RpcClient; +use tracing::{error, info, warn}; + +/// Runs the client: the rpc orchestration loop on the runtime, the +/// blocking terminal UI on its own thread. Returns when the user quits +/// the UI. +pub async fn run(config: &'static Config) -> Result<(), Box> { + let (ui_tx, rx): (Sender, Receiver) = flume::unbounded(); + let (tx, ui_rx): (Sender, Receiver) = flume::unbounded(); + + // FIXME: unwrap + tokio::spawn(async move { orchestrate(config, (tx, rx)).await.unwrap() }); + + tokio::task::spawn_blocking(|| { + run_ui(ui_tx, ui_rx); + }) + .await?; + + Ok(()) +} + +async fn orchestrate( + config: &'static Config, + (tx, rx): (Sender, Receiver), +) -> Result<(), Box> { + info!(address = config.server.address, "connecting to server"); + let mut rpc_client = rpc::RpcClient::connect(&config.server.address).await?; + + if let Some(root_node) = rpc_client.get_library_node(crabidy_core::ROOT_PATH).await? { + tx.send(MessageToUi::ReplaceLibraryNode(root_node.clone()))?; + } + + let init_data = rpc_client.init().await?; + info!("received initial state from server"); + tx.send_async(MessageToUi::Init(init_data)).await?; + + loop { + if let Err(err) = poll(&mut rpc_client, &rx, &tx).await { + error!("request to server failed: {err}"); + } + } +} + +async fn poll( + rpc_client: &mut RpcClient, + rx: &Receiver, + tx: &Sender, +) -> Result<(), Box> { + select! { + Ok(msg) = &mut rx.recv_async() => { + match msg { + MessageFromUi::GetLibraryNode(path) => { + if let Some(node) = rpc_client.get_library_node(&path).await? { + let _ = tx.send(MessageToUi::ReplaceLibraryNode(node.clone())); + } + }, + MessageFromUi::CreateNode { parent_path, title } => { + // Navigates the library into the created node on + // success; on failure the library stays where it is. + match rpc_client.create_library_node(&parent_path, &title).await { + Ok(node) => { + let _ = tx.send(MessageToUi::ReplaceLibraryNode(node.clone())); + } + Err(err) => { + error!(parent_path, title, "failed to create node: {err}"); + } + } + }, + MessageFromUi::RenameNode { path, new_title } => { + // Navigates the library into the renamed node on success; + // on failure the library stays where it is. + match rpc_client.rename_library_node(&path, &new_title).await { + Ok(node) => { + let _ = tx.send(MessageToUi::ReplaceLibraryNode(node.clone())); + } + Err(err) => { + error!(path, new_title, "failed to rename node: {err}"); + } + } + }, + MessageFromUi::DeleteNode { path } => { + // Shows the refreshed parent listing on success; on + // failure the library stays where it is. + match rpc_client.delete_library_node(&path).await { + Ok(parent) => { + let _ = tx.send(MessageToUi::ReplaceLibraryNode(parent.clone())); + } + Err(err) => { + error!(path, "failed to delete node: {err}"); + } + } + }, + MessageFromUi::AppendTracks(uuids) => { + rpc_client.append_tracks(uuids).await? + } + MessageFromUi::QueueTracks(uuids) => { + rpc_client.queue_tracks(uuids).await? + } + MessageFromUi::InsertTracks(uuids, pos) => { + rpc_client.insert_tracks(uuids, pos).await? + } + MessageFromUi::RemoveTracks(positions) => { + rpc_client.remove_tracks(positions).await? + } + MessageFromUi::ReplaceQueue(uuids) => { + rpc_client.replace_queue(uuids).await? + } + MessageFromUi::NextTrack => { + rpc_client.next_track().await? + } + MessageFromUi::PrevTrack => { + rpc_client.prev_track().await? + } + MessageFromUi::RestartTrack => { + rpc_client.restart_track().await? + } + MessageFromUi::SetCurrentTrack(pos) => { + rpc_client.set_current_track(pos).await? + } + MessageFromUi::TogglePlay => { + rpc_client.toggle_play().await? + } + MessageFromUi::ChangeVolume(delta) => { + rpc_client.change_volume(delta).await? + } + MessageFromUi::ToggleMute => { + rpc_client.toggle_mute().await? + } + MessageFromUi::ToggleShuffle => { + rpc_client.toggle_shuffle().await? + } + MessageFromUi::ToggleRepeat => { + rpc_client.toggle_repeat().await? + } + MessageFromUi::ClearQueue(exclude_current) => { + rpc_client.clear_queue(exclude_current).await? + } + MessageFromUi::SaveQueue(name) => { + // A rejected save (bad name, empty queue) must not tear + // down the poll loop; the server logs the cause. + if let Err(err) = rpc_client.save_queue(name.clone()).await { + error!(name, "failed to save queue: {err}"); + } + } + MessageFromUi::CaptureNode { path, name, download } => { + // A rejected capture (bad name, over-cap subtree, failed + // download) must not tear down the poll loop either. + if let Err(err) = rpc_client + .capture_library_node(path.clone(), name.clone(), download) + .await + { + error!(path, name, download, "failed to capture subtree: {err}"); + } + } + } + } + Some(resp) = rpc_client.update_stream.next() => { + match resp { + Ok(resp) => { + if let Some(update) = resp.update { + tx.send_async(MessageToUi::Update(update)).await?; + } + } + Err(err) => { + warn!("update stream broke, reconnecting: {err}"); + rpc_client.reconnect_update_stream().await; + info!("update stream reconnected"); + } + + } + } + } + + Ok(()) +} + +fn run_ui(tx: Sender, rx: Receiver) { + // setup terminal + enable_raw_mode().unwrap(); + let mut stdout = io::stdout(); + execute!(stdout, EnterAlternateScreen, EnableMouseCapture).unwrap(); + let backend = CrosstermBackend::new(stdout); + let mut terminal = Terminal::new(backend).unwrap(); + + // create app and run it + let mut app = App::new(tx); + let tick_rate = Duration::from_millis(100); + let mut last_tick = Instant::now(); + + loop { + for message in rx.try_iter() { + match message { + MessageToUi::ReplaceLibraryNode(node) => { + app.library.update(node); + } + MessageToUi::Init(init_data) => { + if let Some(queue) = init_data.queue { + app.queue.update_queue(queue); + } + if let Some(track) = init_data.queue_track { + app.now_playing.update_track(track.track); + app.queue.update_position(track.queue_position as usize); + } + if let Ok(ps) = PlayState::try_from(init_data.play_state) { + app.now_playing.update_play_state(ps); + } + if let Some(mods) = init_data.mods { + app.now_playing.update_modifiers(&mods); + } + } + MessageToUi::Update(update) => match update { + StreamUpdate::Queue(queue) => { + app.queue.update_queue(queue); + } + StreamUpdate::QueueTrack(track) => { + app.now_playing.update_track(track.track); + app.queue.update_position(track.queue_position as usize); + } + StreamUpdate::Position(pos) => app.now_playing.update_position(pos), + StreamUpdate::PlayState(play_state) => { + if let Ok(ps) = PlayState::try_from(play_state) { + app.now_playing.update_play_state(ps); + } + } + StreamUpdate::Mods(mods) => { + app.now_playing.update_modifiers(&mods); + } + StreamUpdate::Mute(_) => { /* FIXME: implement */ } + StreamUpdate::Volume(_) => { /* FIXME: implement */ } + }, + } + } + + if let Err(err) = terminal.draw(|f| app.render(f)) { + error!("failed to draw frame: {err}"); + break; + } + + let timeout = tick_rate + .checked_sub(last_tick.elapsed()) + .unwrap_or_else(|| Duration::from_secs(0)); + + if event::poll(timeout).unwrap() { + if let Event::Key(key) = event::read().unwrap() { + if key.kind == KeyEventKind::Press { + // The input overlay is strictly modal: while it is open, + // keys edit the buffer and the bindings table (including + // quit) is unreachable. + if app.input.is_some() { + app.handle_input_key(key); + } else if let Some(action) = bindings::lookup(app.focus, app.show_help, key) { + if app.dispatch(action) == DispatchResult::Quit { + break; + } + } + } + } + } + + if last_tick.elapsed() >= tick_rate { + last_tick = Instant::now(); + } + } + + // restore terminal + disable_raw_mode().unwrap(); + execute!( + terminal.backend_mut(), + LeaveAlternateScreen, + DisableMouseCapture + ) + .unwrap(); + terminal.show_cursor().unwrap(); +} diff --git a/cbd-tui/src/main.rs b/cbd-tui/src/main.rs index 066c11a..f2e600e 100644 --- a/cbd-tui/src/main.rs +++ b/cbd-tui/src/main.rs @@ -1,31 +1,11 @@ -mod app; -mod config; -mod rpc; +//! The standalone TUI binary: file-based tracing (the terminal belongs +//! to the UI), config init, and [`cbd_tui::run`]. All client logic lives +//! in the library so the bundled `cbd` binary can host it too +//! (architecture/cbd-bundle.md D1). -use std::{ - error::Error, - io, - sync::OnceLock, - time::{Duration, Instant}, -}; +use std::sync::OnceLock; -use crabidy_core::proto::crabidy::{get_update_stream_response::Update as StreamUpdate, PlayState}; - -use crossterm::{ - event::{self, DisableMouseCapture, EnableMouseCapture, Event, KeyEventKind}, - execute, - terminal::{disable_raw_mode, enable_raw_mode, EnterAlternateScreen, LeaveAlternateScreen}, -}; -use flume::{Receiver, Sender}; - -use ratatui::{backend::CrosstermBackend, Terminal}; -use tokio::select; -use tokio_stream::StreamExt; - -use app::{bindings, App, DispatchResult, MessageFromUi, MessageToUi}; -use config::Config; -use rpc::RpcClient; -use tracing::{error, info, warn}; +use cbd_tui::config::Config; static CONFIG: OnceLock = OnceLock::new(); @@ -65,271 +45,5 @@ fn init_tracing() -> Option { async fn main() -> Result<(), Box> { let _log_guard = init_tracing(); let config = CONFIG.get_or_init(|| crabidy_core::init_config("cbd-tui.toml")); - - let (ui_tx, rx): (Sender, Receiver) = flume::unbounded(); - let (tx, ui_rx): (Sender, Receiver) = flume::unbounded(); - - // FIXME: unwrap - tokio::spawn(async move { orchestrate(config, (tx, rx)).await.unwrap() }); - - tokio::task::spawn_blocking(|| { - run_ui(ui_tx, ui_rx); - }) - .await?; - - Ok(()) -} - -async fn orchestrate( - config: &'static Config, - (tx, rx): (Sender, Receiver), -) -> Result<(), Box> { - info!(address = config.server.address, "connecting to server"); - let mut rpc_client = rpc::RpcClient::connect(&config.server.address).await?; - - if let Some(root_node) = rpc_client.get_library_node(crabidy_core::ROOT_PATH).await? { - tx.send(MessageToUi::ReplaceLibraryNode(root_node.clone()))?; - } - - let init_data = rpc_client.init().await?; - info!("received initial state from server"); - tx.send_async(MessageToUi::Init(init_data)).await?; - - loop { - if let Err(err) = poll(&mut rpc_client, &rx, &tx).await { - error!("request to server failed: {err}"); - } - } -} - -async fn poll( - rpc_client: &mut RpcClient, - rx: &Receiver, - tx: &Sender, -) -> Result<(), Box> { - select! { - Ok(msg) = &mut rx.recv_async() => { - match msg { - MessageFromUi::GetLibraryNode(path) => { - if let Some(node) = rpc_client.get_library_node(&path).await? { - let _ = tx.send(MessageToUi::ReplaceLibraryNode(node.clone())); - } - }, - MessageFromUi::CreateNode { parent_path, title } => { - // Navigates the library into the created node on - // success; on failure the library stays where it is. - match rpc_client.create_library_node(&parent_path, &title).await { - Ok(node) => { - let _ = tx.send(MessageToUi::ReplaceLibraryNode(node.clone())); - } - Err(err) => { - error!(parent_path, title, "failed to create node: {err}"); - } - } - }, - MessageFromUi::RenameNode { path, new_title } => { - // Navigates the library into the renamed node on success; - // on failure the library stays where it is. - match rpc_client.rename_library_node(&path, &new_title).await { - Ok(node) => { - let _ = tx.send(MessageToUi::ReplaceLibraryNode(node.clone())); - } - Err(err) => { - error!(path, new_title, "failed to rename node: {err}"); - } - } - }, - MessageFromUi::DeleteNode { path } => { - // Shows the refreshed parent listing on success; on - // failure the library stays where it is. - match rpc_client.delete_library_node(&path).await { - Ok(parent) => { - let _ = tx.send(MessageToUi::ReplaceLibraryNode(parent.clone())); - } - Err(err) => { - error!(path, "failed to delete node: {err}"); - } - } - }, - MessageFromUi::AppendTracks(uuids) => { - rpc_client.append_tracks(uuids).await? - } - MessageFromUi::QueueTracks(uuids) => { - rpc_client.queue_tracks(uuids).await? - } - MessageFromUi::InsertTracks(uuids, pos) => { - rpc_client.insert_tracks(uuids, pos).await? - } - MessageFromUi::RemoveTracks(positions) => { - rpc_client.remove_tracks(positions).await? - } - MessageFromUi::ReplaceQueue(uuids) => { - rpc_client.replace_queue(uuids).await? - } - MessageFromUi::NextTrack => { - rpc_client.next_track().await? - } - MessageFromUi::PrevTrack => { - rpc_client.prev_track().await? - } - MessageFromUi::RestartTrack => { - rpc_client.restart_track().await? - } - MessageFromUi::SetCurrentTrack(pos) => { - rpc_client.set_current_track(pos).await? - } - MessageFromUi::TogglePlay => { - rpc_client.toggle_play().await? - } - MessageFromUi::ChangeVolume(delta) => { - rpc_client.change_volume(delta).await? - } - MessageFromUi::ToggleMute => { - rpc_client.toggle_mute().await? - } - MessageFromUi::ToggleShuffle => { - rpc_client.toggle_shuffle().await? - } - MessageFromUi::ToggleRepeat => { - rpc_client.toggle_repeat().await? - } - MessageFromUi::ClearQueue(exclude_current) => { - rpc_client.clear_queue(exclude_current).await? - } - MessageFromUi::SaveQueue(name) => { - // A rejected save (bad name, empty queue) must not tear - // down the poll loop; the server logs the cause. - if let Err(err) = rpc_client.save_queue(name.clone()).await { - error!(name, "failed to save queue: {err}"); - } - } - MessageFromUi::CaptureNode { path, name, download } => { - // A rejected capture (bad name, over-cap subtree, failed - // download) must not tear down the poll loop either. - if let Err(err) = rpc_client - .capture_library_node(path.clone(), name.clone(), download) - .await - { - error!(path, name, download, "failed to capture subtree: {err}"); - } - } - } - } - Some(resp) = rpc_client.update_stream.next() => { - match resp { - Ok(resp) => { - if let Some(update) = resp.update { - tx.send_async(MessageToUi::Update(update)).await?; - } - } - Err(err) => { - warn!("update stream broke, reconnecting: {err}"); - rpc_client.reconnect_update_stream().await; - info!("update stream reconnected"); - } - - } - } - } - - Ok(()) -} - -fn run_ui(tx: Sender, rx: Receiver) { - // setup terminal - enable_raw_mode().unwrap(); - let mut stdout = io::stdout(); - execute!(stdout, EnterAlternateScreen, EnableMouseCapture).unwrap(); - let backend = CrosstermBackend::new(stdout); - let mut terminal = Terminal::new(backend).unwrap(); - - // create app and run it - let mut app = App::new(tx); - let tick_rate = Duration::from_millis(100); - let mut last_tick = Instant::now(); - - loop { - for message in rx.try_iter() { - match message { - MessageToUi::ReplaceLibraryNode(node) => { - app.library.update(node); - } - MessageToUi::Init(init_data) => { - if let Some(queue) = init_data.queue { - app.queue.update_queue(queue); - } - if let Some(track) = init_data.queue_track { - app.now_playing.update_track(track.track); - app.queue.update_position(track.queue_position as usize); - } - if let Ok(ps) = PlayState::try_from(init_data.play_state) { - app.now_playing.update_play_state(ps); - } - if let Some(mods) = init_data.mods { - app.now_playing.update_modifiers(&mods); - } - } - MessageToUi::Update(update) => match update { - StreamUpdate::Queue(queue) => { - app.queue.update_queue(queue); - } - StreamUpdate::QueueTrack(track) => { - app.now_playing.update_track(track.track); - app.queue.update_position(track.queue_position as usize); - } - StreamUpdate::Position(pos) => app.now_playing.update_position(pos), - StreamUpdate::PlayState(play_state) => { - if let Ok(ps) = PlayState::try_from(play_state) { - app.now_playing.update_play_state(ps); - } - } - StreamUpdate::Mods(mods) => { - app.now_playing.update_modifiers(&mods); - } - StreamUpdate::Mute(_) => { /* FIXME: implement */ } - StreamUpdate::Volume(_) => { /* FIXME: implement */ } - }, - } - } - - if let Err(err) = terminal.draw(|f| app.render(f)) { - error!("failed to draw frame: {err}"); - break; - } - - let timeout = tick_rate - .checked_sub(last_tick.elapsed()) - .unwrap_or_else(|| Duration::from_secs(0)); - - if event::poll(timeout).unwrap() { - if let Event::Key(key) = event::read().unwrap() { - if key.kind == KeyEventKind::Press { - // The input overlay is strictly modal: while it is open, - // keys edit the buffer and the bindings table (including - // quit) is unreachable. - if app.input.is_some() { - app.handle_input_key(key); - } else if let Some(action) = bindings::lookup(app.focus, app.show_help, key) { - if app.dispatch(action) == DispatchResult::Quit { - break; - } - } - } - } - } - - if last_tick.elapsed() >= tick_rate { - last_tick = Instant::now(); - } - } - - // restore terminal - disable_raw_mode().unwrap(); - execute!( - terminal.backend_mut(), - LeaveAlternateScreen, - DisableMouseCapture - ) - .unwrap(); - terminal.show_cursor().unwrap(); + cbd_tui::run(config).await } diff --git a/cbd/Cargo.toml b/cbd/Cargo.toml new file mode 100644 index 0000000..62b000f --- /dev/null +++ b/cbd/Cargo.toml @@ -0,0 +1,14 @@ +[package] +name = "cbd" +version.workspace = true +edition.workspace = true + +[dependencies] +cbd-tui.workspace = true +crabidy-core.workspace = true +crabidy-server.workspace = true +dirs.workspace = true +tokio = { workspace = true, features = ["full"] } +tracing.workspace = true +tracing-appender.workspace = true +tracing-subscriber.workspace = true diff --git a/cbd/src/main.rs b/cbd/src/main.rs new file mode 100644 index 0000000..4d561fb --- /dev/null +++ b/cbd/src/main.rs @@ -0,0 +1,183 @@ +//! `cbd`: server and TUI bundled into one binary +//! (see `architecture/cbd-bundle.md`). +//! +//! Starting `cbd` starts the crabidy server in-process, waits for it to +//! accept connections, and runs the TUI against it — the same configs, +//! the same localhost gRPC wire as the standalone pair. If a server is +//! already listening (a standalone `crabidy-server`), `cbd` adopts it +//! instead of failing. Quitting the TUI ends the process, and with it +//! the in-process server; the current queue is persisted continuously, +//! so the next start restores it. + +use std::error::Error; +use std::sync::OnceLock; +use std::time::Duration; + +use cbd_tui::config::Config; +use tracing::{info, warn}; + +static CONFIG: OnceLock = OnceLock::new(); + +/// How long to wait for the server socket before giving up. Generous: +/// the first server start may run a provider login flow. +const READINESS_ATTEMPTS: u32 = 120; +const READINESS_DELAY: Duration = Duration::from_millis(500); + +#[tokio::main] +async fn main() -> Result<(), Box> { + // Both halves share one file-based subscriber: the terminal belongs + // to the TUI, so the server's usual stderr logging would corrupt it. + let _log_guard = init_tracing(); + let config = CONFIG.get_or_init(|| crabidy_core::init_config("cbd-tui.toml")); + + let addr: std::net::SocketAddr = crabidy_server::LISTEN_ADDR.parse()?; + let mut server = tokio::spawn(crabidy_server::serve(addr)); + wait_for_server( + &config.server.address, + &mut server, + READINESS_ATTEMPTS, + READINESS_DELAY, + ) + .await?; + + cbd_tui::run(config).await +} + +/// Waits until something accepts TCP connections on the TUI's configured +/// server address (scheme stripped): the in-process server coming up, or +/// an already-running standalone one (in which case our `serve` fails +/// with the port taken and is deliberately ignored). Fails when the +/// in-process server dies while nothing is listening, or after +/// `attempts` polls. +async fn wait_for_server( + address: &str, + server: &mut tokio::task::JoinHandle>>, + attempts: u32, + delay: Duration, +) -> Result<(), Box> { + let host_port = address + .trim_start_matches("http://") + .trim_start_matches("https://") + .trim_end_matches('/'); + for _ in 0..attempts { + if tokio::net::TcpStream::connect(host_port).await.is_ok() { + if server.is_finished() { + warn!("a server is already listening; connecting to it instead"); + } else { + info!(address, "server is ready"); + } + return Ok(()); + } + if server.is_finished() { + // Nothing listening and our server is gone: a real failure + // (provider init, bad address), not an occupied port. + return match server.await { + Ok(Ok(())) => Err("the server exited before becoming ready".into()), + Ok(Err(err)) => Err(err.to_string().into()), + Err(err) => Err(err.to_string().into()), + }; + } + tokio::time::sleep(delay).await; + } + Err(format!("no server reachable at {host_port} after {attempts} attempts").into()) +} + +/// Logs to a file (`crabidy/cbd.log` in the state dir), like `cbd-tui` — +/// but with the server crates' filter, since they run in-process here. +fn init_tracing() -> Option { + use tracing_subscriber::{prelude::*, EnvFilter}; + + let log_dir = dirs::state_dir() + .or_else(dirs::cache_dir) + .unwrap_or_else(std::env::temp_dir) + .join("crabidy"); + if let Err(err) = std::fs::create_dir_all(&log_dir) { + eprintln!( + "could not create log directory {}: {err}", + log_dir.display() + ); + return None; + } + let file_appender = tracing_appender::rolling::daily(&log_dir, "cbd.log"); + let (non_blocking, guard) = tracing_appender::non_blocking(file_appender); + let env_filter = EnvFilter::try_from_default_env().unwrap_or_else(|_| { + EnvFilter::new( + "info,cbd=debug,cbd_tui=debug,crabidy_server=debug,crabidy_core=debug,tidaldy=debug,ytdy=debug,audio_player=debug", + ) + }); + tracing_subscriber::registry() + .with(env_filter) + .with( + tracing_subscriber::fmt::layer() + .with_writer(non_blocking) + .with_ansi(false) + .with_target(true), + ) + .init(); + Some(guard) +} + +#[cfg(test)] +mod tests { + use super::*; + + /// A server task that never finishes, standing in for a healthy + /// in-process server still starting up. + fn pending_server() -> tokio::task::JoinHandle>> { + tokio::spawn(async { + std::future::pending::<()>().await; + Ok(()) + }) + } + + #[tokio::test] + async fn readiness_polls_until_the_socket_accepts() { + let listener = tokio::net::TcpListener::bind("127.0.0.1:0") + .await + .expect("bind"); + let addr = listener.local_addr().expect("addr"); + // The scheme prefix must be stripped like the TUI config's URL. + let address = format!("http://{addr}/"); + let mut server = pending_server(); + wait_for_server(&address, &mut server, 10, Duration::from_millis(10)) + .await + .expect("socket accepts"); + server.abort(); + } + + #[tokio::test] + async fn readiness_gives_up_and_reports_a_dead_server() { + // Nothing listens on this address (bound, then dropped). + let listener = tokio::net::TcpListener::bind("127.0.0.1:0") + .await + .expect("bind"); + let addr = listener.local_addr().expect("addr"); + drop(listener); + + // A dead server task with nothing listening is a real failure. + let mut dead: tokio::task::JoinHandle>> = + tokio::spawn(async { Err("provider init failed".into()) }); + let err = wait_for_server( + &format!("http://{addr}"), + &mut dead, + 10, + Duration::from_millis(10), + ) + .await + .expect_err("dead server surfaces"); + assert!(err.to_string().contains("provider init failed")); + + // A healthy-but-slow server just runs out of attempts. + let mut server = pending_server(); + let err = wait_for_server( + &format!("http://{addr}"), + &mut server, + 3, + Duration::from_millis(10), + ) + .await + .expect_err("gives up eventually"); + assert!(err.to_string().contains("after 3 attempts")); + server.abort(); + } +} diff --git a/crabidy-server/src/lib.rs b/crabidy-server/src/lib.rs index 7d9cd9f..de500e2 100644 --- a/crabidy-server/src/lib.rs +++ b/crabidy-server/src/lib.rs @@ -1,13 +1,132 @@ pub mod bookmark_store; pub mod capture; pub mod capture_store; +pub mod playback; +pub mod provider; pub mod queue_store; +pub mod rpc; -use crabidy_core::proto::crabidy::{Queue, Track}; +use audio_player::PlayerMessage; +use crabidy_core::proto::crabidy::{ + crabidy_service_server::CrabidyServiceServer, InitResponse, LibraryNode, PlayState, Queue, + Track, +}; +use crabidy_core::{ProviderClient, ProviderError}; use rand::{rng, seq::SliceRandom}; use std::sync::{atomic::AtomicBool, Arc}; use std::time::SystemTime; -use tracing::{debug, error}; +use tracing::{debug, error, info, instrument, warn, Span}; + +/// The gRPC listen address of the server. +pub const LISTEN_ADDR: &str = "0.0.0.0:50051"; + +/// Builds and runs the whole server stack on `addr`: provider +/// orchestrator, queue persistence, playback loop, player message +/// forwarder, and the tonic gRPC service. Runs until the server is shut +/// down or fails. +/// +/// Extracted from the `crabidy-server` binary so the bundled `cbd` +/// binary can host the same server in-process +/// (architecture/cbd-bundle.md D1). Errors are returned, never +/// panicked: a failed provider init or an occupied port is the +/// caller's decision. +pub async fn serve( + addr: std::net::SocketAddr, +) -> Result<(), Box> { + let (update_tx, _) = tokio::sync::broadcast::channel(2048); + let orchestrator = provider::ProviderOrchestrator::init("") + .await + .map_err(|err| { + error!("failed to init provider orchestrator: {err}"); + err + })?; + + // Queue persistence is optional: without a usable queues directory the + // server runs with an in-memory queue only. + let queue_store = match queue_store::queues_dir() { + Some(dir) => match queue_store::QueueStore::open(dir).await { + Ok(store) => Some(Arc::new(store)), + Err(err) => { + warn!("queue persistence disabled: {err}"); + None + } + }, + None => { + warn!("queue persistence disabled: no config directory"); + None + } + }; + + let playback = playback::Playback::new( + update_tx.clone(), + orchestrator.provider_tx.clone(), + queue_store, + ); + // Reload the persisted current queue before anything can observe or + // mutate state; never starts playback. + playback.restore_current().await; + + let playback_tx = playback.playback_tx.clone(); + let player_msg = playback.player.messages.clone(); + + std::thread::spawn(|| { + poll_play_bus(player_msg, playback_tx); + }); + info!("player message forwarder started"); + + let crabidy_service = rpc::RpcService::new( + update_tx, + playback.playback_tx.clone(), + orchestrator.provider_tx.clone(), + ); + orchestrator.run(); + info!("provider orchestrator started"); + playback.run(); + info!("playback started"); + + info!(%addr, "grpc server listening"); + tonic::transport::Server::builder() + .add_service(CrabidyServiceServer::new(crabidy_service)) + .serve(addr) + .await?; + + Ok(()) +} + +/// Forwards player engine events into the playback message loop. +#[instrument(skip(rx, tx))] +fn poll_play_bus(rx: flume::Receiver, tx: flume::Sender) { + for msg in rx.iter() { + let command = match msg { + PlayerMessage::EndOfStream => { + debug!("player reported end of stream"); + PlaybackCommand::Next + } + PlayerMessage::Stopped => PlaybackCommand::StateChanged { + state: PlayState::Stopped, + }, + PlayerMessage::Paused => PlaybackCommand::StateChanged { + state: PlayState::Paused, + }, + PlayerMessage::Playing => PlaybackCommand::StateChanged { + state: PlayState::Playing, + }, + PlayerMessage::Elapsed { duration, elapsed } => PlaybackCommand::PositionChanged { + duration: duration.as_millis() as u32, + position: elapsed.as_millis() as u32, + }, + PlayerMessage::Duration { duration } => PlaybackCommand::PositionChanged { + duration: duration.as_millis() as u32, + position: 0, + }, + }; + if let Err(err) = tx.send(PlaybackMessage::new(command)) { + error!("failed to forward player message: {err}"); + return; + } + } + warn!("player message channel closed"); +} /// How a pending queue operation places its resolved chunks. #[derive(Clone, Copy, Debug, PartialEq, Eq)] @@ -554,3 +673,210 @@ mod tests { assert_eq!(order, (0..7).collect::>()); } } +/// A command for the provider orchestrator, tagged with the tracing span that +/// was current when it was sent so the handler can attribute its events to +/// the originating request. +#[derive(Debug)] +pub struct ProviderMessage { + pub span: Span, + pub command: ProviderCommand, +} + +impl ProviderMessage { + pub fn new(command: ProviderCommand) -> Self { + Self { + span: Span::current(), + command, + } + } +} + +#[derive(Debug)] +pub enum ProviderCommand { + GetLibraryNode { + path: String, + result_tx: flume::Sender>, + }, + GetTrackUrls { + path: String, + result_tx: flume::Sender, ProviderError>>, + }, + /// Resolves a path into playable tracks: a track path yields that single + /// track, a node path yields all tracks reachable below it. Streamed: + /// zero or more in-order chunks arrive on `chunk_tx`, the sender is + /// dropped when resolution finishes, and dropping the receiver cancels + /// it (see `ProviderClient::resolve_tracks_into`). The orchestrator + /// handles this command on a spawned task so its loop stays free for + /// other commands (notably `GetTrackUrls` for the first chunk's track). + ResolveTracks { + path: String, + chunk_tx: flume::Sender>, + }, + /// Creates a child under a creatable node (see + /// `ProviderClient::create_lib_node`); replies with the created node. + CreateLibraryNode { + parent_path: String, + title: String, + result_tx: flume::Sender>, + }, + /// Renames an editable node (see `ProviderClient::rename_lib_node`); + /// replies with the renamed node at its new path. + RenameLibraryNode { + path: String, + new_title: String, + result_tx: flume::Sender>, + }, + /// Deletes a deletable node (see `ProviderClient::delete_lib_node`); + /// replies with the refreshed parent node. + DeleteLibraryNode { + path: String, + result_tx: flume::Sender>, + }, + /// Captures the queueable subtree at `path` as the bookmark `name` + /// (see `architecture/bookmarks.md` D1–D3) — or, with `download`, as + /// the download capture `name` under `/captures`, fetching every + /// track's audio (see `architecture/captures.md`). Handled on a + /// spawned task — a large walk or download must not block the + /// orchestrator loop. + CaptureLibraryNode { + path: String, + name: String, + download: bool, + result_tx: flume::Sender>, + }, +} + +impl ProviderCommand { + pub fn name(&self) -> &'static str { + match self { + Self::GetLibraryNode { .. } => "get_library_node", + Self::GetTrackUrls { .. } => "get_track_urls", + Self::ResolveTracks { .. } => "resolve_tracks", + Self::CreateLibraryNode { .. } => "create_library_node", + Self::RenameLibraryNode { .. } => "rename_library_node", + Self::DeleteLibraryNode { .. } => "delete_library_node", + Self::CaptureLibraryNode { .. } => "capture_library_node", + } + } +} + +/// A command for the playback loop, tagged like [`ProviderMessage`]. +#[derive(Debug)] +pub struct PlaybackMessage { + pub span: Span, + pub command: PlaybackCommand, +} + +impl PlaybackMessage { + pub fn new(command: PlaybackCommand) -> Self { + Self { + span: Span::current(), + command, + } + } +} + +#[derive(Debug)] +pub enum PlaybackCommand { + Init { + result_tx: flume::Sender, + }, + Replace { + paths: Vec, + }, + Queue { + paths: Vec, + }, + Append { + paths: Vec, + }, + Remove { + positions: Vec, + }, + Insert { + position: u32, + paths: Vec, + }, + /// Internal: a resolved chunk of tracks for the pending queue operation + /// `op_id`, sent by that operation's forwarder task. Chunks for an + /// unknown (finished or cancelled) op are dropped silently. + ApplyResolvedChunk { + op_id: u64, + tracks: Vec, + }, + /// Internal: the forwarder task for `op_id` has seen the provider drop + /// its chunk sender — the operation is complete and the `resolving` + /// flag clears once no pending operations remain. + ResolveFinished { + op_id: u64, + }, + Clear { + exclude_current: bool, + }, + SetCurrent { + position: u32, + }, + /// Saves the current queue under a name (see + /// `architecture/queue-persistence.md` D6). Handled on the loop so the + /// snapshot is consistent; the disk write happens on a spawned task and + /// reports through `result_tx`. + SaveQueue { + name: String, + result_tx: flume::Sender>, + }, + ToggleShuffle, + ToggleRepeat, + TogglePlay, + Stop, + ChangeVolume { + delta: f32, + }, + ToggleMute, + Next, + Prev, + RestartTrack, + StateChanged { + state: PlayState, + }, + VolumeChanged { + volume: f32, + }, + MuteChanged { + muted: bool, + }, + PositionChanged { + duration: u32, + position: u32, + }, +} + +impl PlaybackCommand { + pub fn name(&self) -> &'static str { + match self { + Self::Init { .. } => "init", + Self::Replace { .. } => "replace", + Self::Queue { .. } => "queue", + Self::Append { .. } => "append", + Self::Remove { .. } => "remove", + Self::Insert { .. } => "insert", + Self::ApplyResolvedChunk { .. } => "apply_resolved_chunk", + Self::ResolveFinished { .. } => "resolve_finished", + Self::Clear { .. } => "clear", + Self::SetCurrent { .. } => "set_current", + Self::SaveQueue { .. } => "save_queue", + Self::ToggleShuffle => "toggle_shuffle", + Self::ToggleRepeat => "toggle_repeat", + Self::TogglePlay => "toggle_play", + Self::Stop => "stop", + Self::ChangeVolume { .. } => "change_volume", + Self::ToggleMute => "toggle_mute", + Self::Next => "next", + Self::Prev => "prev", + Self::RestartTrack => "restart_track", + Self::StateChanged { .. } => "state_changed", + Self::VolumeChanged { .. } => "volume_changed", + Self::MuteChanged { .. } => "mute_changed", + Self::PositionChanged { .. } => "position_changed", + } + } +} diff --git a/crabidy-server/src/main.rs b/crabidy-server/src/main.rs index 082895d..821d051 100644 --- a/crabidy-server/src/main.rs +++ b/crabidy-server/src/main.rs @@ -1,82 +1,14 @@ -use audio_player::PlayerMessage; -use crabidy_core::proto::crabidy::{ - crabidy_service_server::CrabidyServiceServer, InitResponse, LibraryNode, PlayState, Track, -}; -use crabidy_core::{ProviderClient, ProviderError}; -use tracing::{debug, error, info, instrument, warn, Span}; +//! The standalone server binary: stderr tracing plus +//! [`crabidy_server::serve`] on the fixed listen address. The whole stack +//! lives in the library so the bundled `cbd` binary can host it too +//! (architecture/cbd-bundle.md D1). + use tracing_subscriber::{prelude::*, EnvFilter}; -mod playback; -use playback::Playback; -mod provider; -use provider::ProviderOrchestrator; -mod rpc; -use rpc::RpcService; - -use tonic::{transport::Server, Result}; - -const LISTEN_ADDR: &str = "0.0.0.0:50051"; - #[tokio::main] -async fn main() -> Result<(), Box> { +async fn main() -> Result<(), Box> { let _log_guard = init_tracing(); - - let (update_tx, _) = tokio::sync::broadcast::channel(2048); - let orchestrator = ProviderOrchestrator::init("").await.map_err(|err| { - error!("failed to init provider orchestrator: {err}"); - err - })?; - - // Queue persistence is optional: without a usable queues directory the - // server runs with an in-memory queue only. - let queue_store = match crabidy_server::queue_store::queues_dir() { - Some(dir) => match crabidy_server::queue_store::QueueStore::open(dir).await { - Ok(store) => Some(std::sync::Arc::new(store)), - Err(err) => { - warn!("queue persistence disabled: {err}"); - None - } - }, - None => { - warn!("queue persistence disabled: no config directory"); - None - } - }; - - let playback = Playback::new( - update_tx.clone(), - orchestrator.provider_tx.clone(), - queue_store, - ); - // Reload the persisted current queue before anything can observe or - // mutate state; never starts playback. - playback.restore_current().await; - - let playback_tx = playback.playback_tx.clone(); - let player_msg = playback.player.messages.clone(); - - std::thread::spawn(|| { - poll_play_bus(player_msg, playback_tx); - }); - info!("player message forwarder started"); - - let crabidy_service = RpcService::new( - update_tx, - playback.playback_tx.clone(), - orchestrator.provider_tx.clone(), - ); - orchestrator.run(); - info!("provider orchestrator started"); - playback.run(); - info!("playback started"); - - let addr = LISTEN_ADDR.parse()?; - info!(%addr, "grpc server listening"); - Server::builder() - .add_service(CrabidyServiceServer::new(crabidy_service)) - .serve(addr) - .await?; - + crabidy_server::serve(crabidy_server::LISTEN_ADDR.parse()?).await?; Ok(()) } @@ -90,7 +22,7 @@ fn init_tracing() -> tracing_appender::non_blocking::WorkerGuard { let env_filter = EnvFilter::try_from_default_env().unwrap_or_else(|_| { EnvFilter::new( - "info,crabidy_server=debug,crabidy_core=debug,tidaldy=debug,audio_player=debug", + "info,crabidy_server=debug,crabidy_core=debug,tidaldy=debug,ytdy=debug,audio_player=debug", ) }); @@ -109,246 +41,3 @@ fn init_tracing() -> tracing_appender::non_blocking::WorkerGuard { guard } - -/// Forwards player engine events into the playback message loop. -#[instrument(skip(rx, tx))] -fn poll_play_bus(rx: flume::Receiver, tx: flume::Sender) { - for msg in rx.iter() { - let command = match msg { - PlayerMessage::EndOfStream => { - debug!("player reported end of stream"); - PlaybackCommand::Next - } - PlayerMessage::Stopped => PlaybackCommand::StateChanged { - state: PlayState::Stopped, - }, - PlayerMessage::Paused => PlaybackCommand::StateChanged { - state: PlayState::Paused, - }, - PlayerMessage::Playing => PlaybackCommand::StateChanged { - state: PlayState::Playing, - }, - PlayerMessage::Elapsed { duration, elapsed } => PlaybackCommand::PositionChanged { - duration: duration.as_millis() as u32, - position: elapsed.as_millis() as u32, - }, - PlayerMessage::Duration { duration } => PlaybackCommand::PositionChanged { - duration: duration.as_millis() as u32, - position: 0, - }, - }; - if let Err(err) = tx.send(PlaybackMessage::new(command)) { - error!("failed to forward player message: {err}"); - return; - } - } - warn!("player message channel closed"); -} - -/// A command for the provider orchestrator, tagged with the tracing span that -/// was current when it was sent so the handler can attribute its events to -/// the originating request. -#[derive(Debug)] -pub struct ProviderMessage { - pub span: Span, - pub command: ProviderCommand, -} - -impl ProviderMessage { - pub fn new(command: ProviderCommand) -> Self { - Self { - span: Span::current(), - command, - } - } -} - -#[derive(Debug)] -pub enum ProviderCommand { - GetLibraryNode { - path: String, - result_tx: flume::Sender>, - }, - GetTrackUrls { - path: String, - result_tx: flume::Sender, ProviderError>>, - }, - /// Resolves a path into playable tracks: a track path yields that single - /// track, a node path yields all tracks reachable below it. Streamed: - /// zero or more in-order chunks arrive on `chunk_tx`, the sender is - /// dropped when resolution finishes, and dropping the receiver cancels - /// it (see `ProviderClient::resolve_tracks_into`). The orchestrator - /// handles this command on a spawned task so its loop stays free for - /// other commands (notably `GetTrackUrls` for the first chunk's track). - ResolveTracks { - path: String, - chunk_tx: flume::Sender>, - }, - /// Creates a child under a creatable node (see - /// `ProviderClient::create_lib_node`); replies with the created node. - CreateLibraryNode { - parent_path: String, - title: String, - result_tx: flume::Sender>, - }, - /// Renames an editable node (see `ProviderClient::rename_lib_node`); - /// replies with the renamed node at its new path. - RenameLibraryNode { - path: String, - new_title: String, - result_tx: flume::Sender>, - }, - /// Deletes a deletable node (see `ProviderClient::delete_lib_node`); - /// replies with the refreshed parent node. - DeleteLibraryNode { - path: String, - result_tx: flume::Sender>, - }, - /// Captures the queueable subtree at `path` as the bookmark `name` - /// (see `architecture/bookmarks.md` D1–D3) — or, with `download`, as - /// the download capture `name` under `/captures`, fetching every - /// track's audio (see `architecture/captures.md`). Handled on a - /// spawned task — a large walk or download must not block the - /// orchestrator loop. - CaptureLibraryNode { - path: String, - name: String, - download: bool, - result_tx: flume::Sender>, - }, -} - -impl ProviderCommand { - pub fn name(&self) -> &'static str { - match self { - Self::GetLibraryNode { .. } => "get_library_node", - Self::GetTrackUrls { .. } => "get_track_urls", - Self::ResolveTracks { .. } => "resolve_tracks", - Self::CreateLibraryNode { .. } => "create_library_node", - Self::RenameLibraryNode { .. } => "rename_library_node", - Self::DeleteLibraryNode { .. } => "delete_library_node", - Self::CaptureLibraryNode { .. } => "capture_library_node", - } - } -} - -/// A command for the playback loop, tagged like [`ProviderMessage`]. -#[derive(Debug)] -pub struct PlaybackMessage { - pub span: Span, - pub command: PlaybackCommand, -} - -impl PlaybackMessage { - pub fn new(command: PlaybackCommand) -> Self { - Self { - span: Span::current(), - command, - } - } -} - -#[derive(Debug)] -pub enum PlaybackCommand { - Init { - result_tx: flume::Sender, - }, - Replace { - paths: Vec, - }, - Queue { - paths: Vec, - }, - Append { - paths: Vec, - }, - Remove { - positions: Vec, - }, - Insert { - position: u32, - paths: Vec, - }, - /// Internal: a resolved chunk of tracks for the pending queue operation - /// `op_id`, sent by that operation's forwarder task. Chunks for an - /// unknown (finished or cancelled) op are dropped silently. - ApplyResolvedChunk { - op_id: u64, - tracks: Vec, - }, - /// Internal: the forwarder task for `op_id` has seen the provider drop - /// its chunk sender — the operation is complete and the `resolving` - /// flag clears once no pending operations remain. - ResolveFinished { - op_id: u64, - }, - Clear { - exclude_current: bool, - }, - SetCurrent { - position: u32, - }, - /// Saves the current queue under a name (see - /// `architecture/queue-persistence.md` D6). Handled on the loop so the - /// snapshot is consistent; the disk write happens on a spawned task and - /// reports through `result_tx`. - SaveQueue { - name: String, - result_tx: flume::Sender>, - }, - ToggleShuffle, - ToggleRepeat, - TogglePlay, - Stop, - ChangeVolume { - delta: f32, - }, - ToggleMute, - Next, - Prev, - RestartTrack, - StateChanged { - state: PlayState, - }, - VolumeChanged { - volume: f32, - }, - MuteChanged { - muted: bool, - }, - PositionChanged { - duration: u32, - position: u32, - }, -} - -impl PlaybackCommand { - pub fn name(&self) -> &'static str { - match self { - Self::Init { .. } => "init", - Self::Replace { .. } => "replace", - Self::Queue { .. } => "queue", - Self::Append { .. } => "append", - Self::Remove { .. } => "remove", - Self::Insert { .. } => "insert", - Self::ApplyResolvedChunk { .. } => "apply_resolved_chunk", - Self::ResolveFinished { .. } => "resolve_finished", - Self::Clear { .. } => "clear", - Self::SetCurrent { .. } => "set_current", - Self::SaveQueue { .. } => "save_queue", - Self::ToggleShuffle => "toggle_shuffle", - Self::ToggleRepeat => "toggle_repeat", - Self::TogglePlay => "toggle_play", - Self::Stop => "stop", - Self::ChangeVolume { .. } => "change_volume", - Self::ToggleMute => "toggle_mute", - Self::Next => "next", - Self::Prev => "prev", - Self::RestartTrack => "restart_track", - Self::StateChanged { .. } => "state_changed", - Self::VolumeChanged { .. } => "volume_changed", - Self::MuteChanged { .. } => "mute_changed", - Self::PositionChanged { .. } => "position_changed", - } - } -} diff --git a/crabidy-server/src/playback.rs b/crabidy-server/src/playback.rs index 6eab118..b3a2b6d 100644 --- a/crabidy-server/src/playback.rs +++ b/crabidy-server/src/playback.rs @@ -1,3 +1,5 @@ +use crate::queue_store::{self, QueueSnapshot, QueueStore, SaveQueueError}; +use crate::{PendingResolve, QueueManager, ResolveKind}; use crate::{PlaybackCommand, PlaybackMessage, ProviderCommand, ProviderMessage}; use audio_player::Player; use crabidy_core::proto::crabidy::QueueModifiers; @@ -6,8 +8,6 @@ use crabidy_core::proto::crabidy::{ Queue as ProtoQueue, QueueTrack, Track, TrackPosition, }; use crabidy_core::ProviderError; -use crabidy_server::queue_store::{self, QueueSnapshot, QueueStore, SaveQueueError}; -use crabidy_server::{PendingResolve, QueueManager, ResolveKind}; use std::collections::HashMap; use std::sync::atomic::{AtomicU64, Ordering}; use std::sync::{Arc, Mutex}; diff --git a/crabidy-server/src/provider.rs b/crabidy-server/src/provider.rs index 695bb90..6d53b92 100644 --- a/crabidy-server/src/provider.rs +++ b/crabidy-server/src/provider.rs @@ -1,12 +1,12 @@ +use crate::bookmark_store::{BookmarkStore, BOOKMARKS_PROVIDER_ROOT}; +use crate::capture_store::{CaptureStore, CAPTURES_PROVIDER_ROOT}; +use crate::queue_store::{CURRENT_QUEUE_NAME, QUEUES_PROVIDER_ROOT}; use crate::{ProviderCommand, ProviderMessage}; use async_trait::async_trait; use crabidy_core::{ proto::crabidy::{LibraryNode, LibraryNodeChild, Track}, ProviderClient, ProviderError, }; -use crabidy_server::bookmark_store::{BookmarkStore, BOOKMARKS_PROVIDER_ROOT}; -use crabidy_server::capture_store::{CaptureStore, CAPTURES_PROVIDER_ROOT}; -use crabidy_server::queue_store::{CURRENT_QUEUE_NAME, QUEUES_PROVIDER_ROOT}; use std::{fs, path::PathBuf, sync::Arc}; use tracing::{debug, debug_span, error, instrument, warn, Instrument}; @@ -200,12 +200,12 @@ impl ProviderOrchestrator { let result = if download { match &this.capture_store { Some(store) => store.capture(&*this, &path, &name).await, - None => Err(crabidy_server::capture::CaptureError::Disabled), + None => Err(crate::capture::CaptureError::Disabled), } } else { match &this.bookmark_store { Some(store) => store.capture(&*this, &path, &name).await, - None => Err(crabidy_server::capture::CaptureError::Disabled), + None => Err(crate::capture::CaptureError::Disabled), } }; if let Err(err) = &result { @@ -271,7 +271,7 @@ impl ProviderClient for ProviderOrchestrator { // into; a folder that does not exist yet lists as empty-on-arrival // (created by QueueStore::open in main). Saved queues are renamable // and deletable; the auto-persisted `current` stays untouchable. - let queues_client = match crabidy_server::queue_store::queues_dir() { + let queues_client = match crate::queue_store::queues_dir() { Some(dir) => match fsdy::Client::new(QUEUES_PROVIDER_ROOT, dir) { Ok(client) => Some(Arc::new( client @@ -291,7 +291,7 @@ impl ProviderClient for ProviderOrchestrator { // Bookmarks: the orchestrator owns the store (it is the capture // writer) and mounts the same folder read-only. Non-fatal like the // other local providers. - let bookmark_store = match crabidy_server::bookmark_store::bookmarks_dir() { + let bookmark_store = match crate::bookmark_store::bookmarks_dir() { Some(dir) => match BookmarkStore::open(dir).await { Ok(store) => Some(Arc::new(store)), Err(err) => { @@ -319,7 +319,7 @@ impl ProviderClient for ProviderOrchestrator { }); // Captures: like bookmarks, but the store downloads every track's // audio next to its toml (architecture/captures.md D1). Non-fatal. - let capture_store = match crabidy_server::capture_store::captures_dir() { + let capture_store = match crate::capture_store::captures_dir() { Some(dir) => match CaptureStore::open(dir).await { Ok(store) => Some(Arc::new(store)), Err(err) => { diff --git a/crabidy-server/src/rpc.rs b/crabidy-server/src/rpc.rs index 6d3ed7e..8e2f2af 100644 --- a/crabidy-server/src/rpc.rs +++ b/crabidy-server/src/rpc.rs @@ -1,3 +1,5 @@ +use crate::bookmark_store::CaptureError; +use crate::queue_store::SaveQueueError; use crate::{PlaybackCommand, PlaybackMessage, ProviderCommand, ProviderMessage}; use crabidy_core::proto::crabidy::{ crabidy_service_server::CrabidyService, get_update_stream_response::Update as StreamUpdate, @@ -15,8 +17,6 @@ use crabidy_core::proto::crabidy::{ ToggleShuffleRequest, ToggleShuffleResponse, }; use crabidy_core::ProviderError; -use crabidy_server::bookmark_store::CaptureError; -use crabidy_server::queue_store::SaveQueueError; use std::pin::Pin; use tokio_stream::StreamExt; use tonic::{Request, Response, Status}; diff --git a/plan/cbd-bundle.md b/plan/cbd-bundle.md new file mode 100644 index 0000000..63e1b2c --- /dev/null +++ b/plan/cbd-bundle.md @@ -0,0 +1,28 @@ +# Plan: cbd-bundle + +Ordered tasks; each names its verification (tests in `cbd/src/main.rs` +and/or gates in `quality/cbd-bundle.md`). + +- [x] **T1 — Server library extraction.** Move `playback`/`provider`/ + `rpc` and the command/message enums into the `crabidy-server` lib; + add `serve(addr)` + `LISTEN_ADDR`; thin `main.rs`. Verifies: the + moved server suite passes unchanged; gates "Refactor". +- [x] **T2 — TUI library extraction.** `cbd-tui/src/lib.rs` with + `run(config)` (orchestrate + run_ui + channels); thin `main.rs` keeps + file tracing + config. Verifies: the moved TUI suite passes + unchanged. +- [x] **T3 — cbd crate.** Workspace member; main = shared file tracing, + `cbd-tui.toml` config, spawn `serve(LISTEN_ADDR)`, `wait_for_server` + (scheme-stripped TCP poll, adopt-on-occupied-port, dead-server + error, bounded attempts), then `cbd_tui::run`. Verifies: + `readiness_polls_until_the_socket_accepts`, + `readiness_gives_up_and_reports_a_dead_server`; gates "Bundled + behavior". +- [x] **T4 — Full verification.** Workspace suite green; clippy/fmt/ + taplo/markdownlint clean. +- [x] **T5 — Live smoke test.** Boot the extracted `serve()` on a free + port through `wait_for_server` (temporary ignored probe, removed + after passing): the real stack — tidal login, providers, playback, + queue restore — accepted a TCP connection in ~1 s. +- [x] **T6 — Docs.** `plan/summary.md` section incl. deviations; + reconcile `architecture/cbd-bundle.md`. diff --git a/plan/summary.md b/plan/summary.md index f97ee35..7ddc081 100644 --- a/plan/summary.md +++ b/plan/summary.md @@ -1,5 +1,31 @@ # Implementation summaries +## cbd-bundle (2026-07-21) + +Built per `plan/cbd-bundle.md`: a new **`cbd`** binary bundles server +and TUI. Both former binaries became libraries with thin mains — +`crabidy_server::serve(addr)` is the extracted server startup +(orchestrator, queue store, playback, player forwarder, tonic), and +`cbd_tui::run(config)` the extracted client loops; the standalone +binaries behave exactly as before. `cbd` sets up one file-based tracing +subscriber for both halves (the terminal belongs to the TUI), spawns +`serve` on the fixed listen address, polls a TCP connect against the +TUI's configured server address until ready (bounded, generous — first +runs may sit in a provider login), then runs the TUI. An +already-running standalone server just gets adopted (the in-process +bind fails on the occupied port and is deliberately ignored once the +socket is reachable); a server that dies before readiness surfaces its +real error. Quitting the TUI ends the process and the in-process +server — the continuously persisted current queue makes that safe. + +Deviations: none of substance — the refactor moved code verbatim +(`crabidy_server::` → `crate::` path rewrites aside). The live probe +booted the extracted stack on a free port through the same readiness +poll `cbd` uses: real tidal login, all providers, playback, queue +restore, TCP accept in ~1 s (probe removed after passing). 154 +workspace tests green (2 new in `cbd`); every gate in +`quality/cbd-bundle.md` checked. + ## captures follow-up: W on queues and bookmarks (2026-07-21) Small fix on top of the captures feature: `/queues` and `/bookmarks` diff --git a/quality/cbd-bundle.md b/quality/cbd-bundle.md new file mode 100644 index 0000000..5d451e1 --- /dev/null +++ b/quality/cbd-bundle.md @@ -0,0 +1,36 @@ +# Quality gates: cbd-bundle + +Criteria beyond the automatic tests (`cbd/src/main.rs` readiness tests +plus the moved suites). Each gate is pass/fail by reading the code. + +## Refactor (behavior-preserving) + +- [x] The server's modules (`playback`, `provider`, `rpc`) and command/ + message types moved to the library **unchanged**; `serve(addr)` is the + old `main` body verbatim (minus tracing); both thin mains only set up + tracing and delegate. +- [x] Every pre-existing test moved along and passes unchanged; no + test was rewritten to accommodate the refactor. +- [x] `cbd-tui`'s library exposes `run(config)`; tracing setup stays in + the binaries (where logs go is a hosting decision). + +## Bundled behavior + +- [x] `cbd` uses the same configs as the pair (`cbd-tui.toml` and the + server-side configs) — nothing bundled-specific to configure. +- [x] Both halves log to one file; the terminal is never written to + outside the TUI. +- [x] An already-running standalone server is adopted (occupied port + + reachable socket → connect), never treated as an error. +- [x] A server that dies before readiness surfaces its actual error; + a slow one gets a bounded, generous wait (first-run login flows). +- [x] The readiness poll strips the config URL's scheme and never + parses beyond host:port. + +## Hygiene + +- [x] New public items are documented; docs state error/edge behavior. +- [x] `clippy -D warnings`, `fmt`, `taplo`, `markdownlint` clean; all + tests green. +- [x] `architecture/cbd-bundle.md` reconciled where the implementation + diverged.