clusterfuck
/
crabidy
4
1
Fork 0
Code Issues 15 Pull Requests Actions Packages Projects 1 Releases 5 Wiki Activity

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 <noreply@anthropic.com>
This commit is contained in:
Test User 2026-07-21 14:15:47 +02:00
parent 8032bec8e1
commit 58f7f9c66b
15 changed files with 1063 additions and 626 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

14
Cargo.lock generated
View File

@ -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"

3
Cargo.toml
View File

@ -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" }

99
architecture/cbd-bundle.md Normal file
View File

@ -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.

305
cbd-tui/src/lib.rs Normal file
View File

@ -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<dyn Error>> {
let (ui_tx, rx): (Sender<MessageFromUi>, Receiver<MessageFromUi>) = flume::unbounded();
let (tx, ui_rx): (Sender<MessageToUi>, Receiver<MessageToUi>) = 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<MessageToUi>, Receiver<MessageFromUi>),
) -> Result<(), Box<dyn Error>> {
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<MessageFromUi>,
tx: &Sender<MessageToUi>,
) -> Result<(), Box<dyn Error>> {
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<MessageFromUi>, rx: Receiver<MessageToUi>) {
// 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();
}

300
cbd-tui/src/main.rs
View File

@ -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<Config> = OnceLock::new();
@ -65,271 +45,5 @@ fn init_tracing() -> Option<tracing_appender::non_blocking::WorkerGuard> {
async fn main() -> Result<(), Box<dyn std::error::Error>> {
let _log_guard = init_tracing();
let config = CONFIG.get_or_init(|| crabidy_core::init_config("cbd-tui.toml"));
let (ui_tx, rx): (Sender<MessageFromUi>, Receiver<MessageFromUi>) = flume::unbounded();
let (tx, ui_rx): (Sender<MessageToUi>, Receiver<MessageToUi>) = 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<MessageToUi>, Receiver<MessageFromUi>),
) -> Result<(), Box<dyn Error>> {
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<MessageFromUi>,
tx: &Sender<MessageToUi>,
) -> Result<(), Box<dyn Error>> {
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<MessageFromUi>, rx: Receiver<MessageToUi>) {
// 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
}

14
cbd/Cargo.toml Normal file
View File

@ -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

183
cbd/src/main.rs Normal file
View File

@ -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<Config> = 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<dyn Error>> {
// 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<Result<(), Box<dyn Error + Send + Sync>>>,
attempts: u32,
delay: Duration,
) -> Result<(), Box<dyn Error>> {
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<tracing_appender::non_blocking::WorkerGuard> {
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<Result<(), Box<dyn Error + Send + Sync>>> {
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<Result<(), Box<dyn Error + Send + Sync>>> =
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();
}
}

330
crabidy-server/src/lib.rs
View File

@ -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<dyn std::error::Error + Send + Sync>> {
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<PlayerMessage>, tx: flume::Sender<PlaybackMessage>) {
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::<Vec<usize>>());
}
}
/// 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<Result<LibraryNode, ProviderError>>,
},
GetTrackUrls {
path: String,
result_tx: flume::Sender<Result<Vec<String>, 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<Vec<Track>>,
},
/// 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<Result<LibraryNode, ProviderError>>,
},
/// 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<Result<LibraryNode, ProviderError>>,
},
/// Deletes a deletable node (see `ProviderClient::delete_lib_node`);
/// replies with the refreshed parent node.
DeleteLibraryNode {
path: String,
result_tx: flume::Sender<Result<LibraryNode, ProviderError>>,
},
/// 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<Result<(), crate::capture::CaptureError>>,
},
}
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<InitResponse>,
},
Replace {
paths: Vec<String>,
},
Queue {
paths: Vec<String>,
},
Append {
paths: Vec<String>,
},
Remove {
positions: Vec<u32>,
},
Insert {
position: u32,
paths: Vec<String>,
},
/// 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<Track>,
},
/// 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<Result<(), crate::queue_store::SaveQueueError>>,
},
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",
}
}
}

327
crabidy-server/src/main.rs
View File

@ -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<dyn std::error::Error>> {
async fn main() -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
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<PlayerMessage>, tx: flume::Sender<PlaybackMessage>) {
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<Result<LibraryNode, ProviderError>>,
},
GetTrackUrls {
path: String,
result_tx: flume::Sender<Result<Vec<String>, 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<Vec<Track>>,
},
/// 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<Result<LibraryNode, ProviderError>>,
},
/// 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<Result<LibraryNode, ProviderError>>,
},
/// Deletes a deletable node (see `ProviderClient::delete_lib_node`);
/// replies with the refreshed parent node.
DeleteLibraryNode {
path: String,
result_tx: flume::Sender<Result<LibraryNode, ProviderError>>,
},
/// 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<Result<(), crabidy_server::capture::CaptureError>>,
},
}
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<InitResponse>,
},
Replace {
paths: Vec<String>,
},
Queue {
paths: Vec<String>,
},
Append {
paths: Vec<String>,
},
Remove {
positions: Vec<u32>,
},
Insert {
position: u32,
paths: Vec<String>,
},
/// 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<Track>,
},
/// 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<Result<(), crabidy_server::queue_store::SaveQueueError>>,
},
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",
}
}
}

4
crabidy-server/src/playback.rs
View File

@ -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};

16
crabidy-server/src/provider.rs
View File

@ -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) => {

4
crabidy-server/src/rpc.rs
View File

@ -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};

28
plan/cbd-bundle.md Normal file
View File

@ -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`.

26
plan/summary.md
View File

@ -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`

36
quality/cbd-bundle.md Normal file
View File

@ -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.