947 lines
31 KiB
Rust
947 lines
31 KiB
Rust
pub mod auth;
|
||
pub mod bookmark_store;
|
||
#[cfg(feature = "web-ui")]
|
||
pub mod web;
|
||
|
||
pub mod capture;
|
||
pub mod capture_store;
|
||
pub mod playback;
|
||
pub mod provider;
|
||
pub mod queue_store;
|
||
pub mod rpc;
|
||
pub mod settings;
|
||
|
||
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, 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>> {
|
||
// Auth first: a malformed crabidy-server.toml must abort startup
|
||
// instead of running an intended-to-be-locked server open
|
||
// (architecture/roles-auth.md). A missing file runs open.
|
||
let config_dir = dirs::config_dir()
|
||
.map(|d| d.join("crabidy"))
|
||
.unwrap_or_else(|| std::path::PathBuf::from("/tmp"));
|
||
let server_settings = settings::ServerSettings::load(&config_dir)?;
|
||
let authenticator = Arc::new(auth::Authenticator::new(&server_settings.auth));
|
||
if authenticator.enabled() {
|
||
info!("role authorization enabled");
|
||
}
|
||
|
||
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");
|
||
|
||
let router = build_router(crabidy_service, authenticator);
|
||
|
||
info!(%addr, "grpc server listening");
|
||
let listener = tokio::net::TcpListener::bind(addr).await?;
|
||
axum::serve(listener, router).await?;
|
||
|
||
Ok(())
|
||
}
|
||
|
||
/// Composes the one axum router that serves everything on one port: the
|
||
/// gRPC service (native HTTP/2 for the TUI *and*, with `web-ui`,
|
||
/// gRPC-web for the browser through the tonic-web layer) plus, with
|
||
/// `web-ui`, the embedded web client as the fallback route
|
||
/// (architecture/web-client.md).
|
||
///
|
||
/// The auth layer wraps only the gRPC route — its default-deny is for
|
||
/// RPC methods; the app shell itself is public, like any login page.
|
||
/// Kept separate from [`serve`] so the routing/auth composition is
|
||
/// testable without a live provider backend.
|
||
pub fn build_router(
|
||
crabidy_service: rpc::RpcService,
|
||
authenticator: Arc<auth::Authenticator>,
|
||
) -> axum::Router {
|
||
let builder = tower::ServiceBuilder::new().layer(auth::AuthLayer::new(authenticator));
|
||
#[cfg(feature = "web-ui")]
|
||
let builder = builder.layer(tonic_web::GrpcWebLayer::new());
|
||
let grpc = builder.service(CrabidyServiceServer::new(crabidy_service));
|
||
let router = axum::Router::new().route_service(
|
||
&format!(
|
||
"/{}/{{*method}}",
|
||
<CrabidyServiceServer<rpc::RpcService> as tonic::server::NamedService>::NAME
|
||
),
|
||
grpc,
|
||
);
|
||
#[cfg(feature = "web-ui")]
|
||
let router = router.fallback(web::serve_asset);
|
||
router
|
||
}
|
||
|
||
/// 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)]
|
||
pub enum ResolveKind {
|
||
/// First chunk replaces the whole queue, later chunks append.
|
||
Replace,
|
||
/// Every chunk appends at the end.
|
||
Append,
|
||
/// Chunks insert after the given position, each advancing the cursor so
|
||
/// the resolved collection stays contiguous and in order. `Queue`
|
||
/// (play-after-current) is an `InsertAfter` at the current position.
|
||
InsertAfter(u32),
|
||
}
|
||
|
||
/// The playback loop's bookkeeping for one in-flight resolve operation.
|
||
///
|
||
/// Created when a `Replace`/`Queue`/`Append`/`Insert` command arrives,
|
||
/// dropped when its forwarder reports completion or a `Replace`/`Clear`
|
||
/// cancels it. Chunk application happens exclusively on the playback loop,
|
||
/// which keeps the loop the single writer of queue state.
|
||
#[derive(Debug)]
|
||
pub struct PendingResolve {
|
||
kind: ResolveKind,
|
||
/// Tracks applied so far; an op finishing at zero is worth a warning.
|
||
applied: usize,
|
||
/// Shared with the op's forwarder task: set on cancellation so the
|
||
/// forwarder drops the chunk receiver, which stops the provider fetch.
|
||
cancelled: Arc<AtomicBool>,
|
||
}
|
||
|
||
impl PendingResolve {
|
||
pub fn new(kind: ResolveKind) -> Self {
|
||
Self {
|
||
kind,
|
||
applied: 0,
|
||
cancelled: Arc::new(AtomicBool::new(false)),
|
||
}
|
||
}
|
||
|
||
/// The cancellation flag to hand to this op's forwarder task.
|
||
pub fn cancel_flag(&self) -> Arc<AtomicBool> {
|
||
Arc::clone(&self.cancelled)
|
||
}
|
||
|
||
/// Marks the op cancelled so its forwarder stops feeding chunks.
|
||
pub fn cancel(&self) {
|
||
self.cancelled
|
||
.store(true, std::sync::atomic::Ordering::Relaxed);
|
||
}
|
||
|
||
/// Total tracks applied by this op so far.
|
||
pub fn applied(&self) -> usize {
|
||
self.applied
|
||
}
|
||
|
||
/// Applies one resolved chunk to the queue and advances this op's
|
||
/// cursor. Returns the track that should start playing, if this chunk
|
||
/// made one current (first chunk of a replace, or any chunk landing in
|
||
/// an empty queue) — later chunks of the same op never restart playback.
|
||
pub fn apply_chunk(&mut self, queue: &mut QueueManager, tracks: &[Track]) -> Option<Track> {
|
||
self.applied += tracks.len();
|
||
match self.kind {
|
||
ResolveKind::Replace => {
|
||
// Only the first chunk replaces; the rest of this op
|
||
// extends the fresh queue.
|
||
self.kind = ResolveKind::Append;
|
||
queue.replace_with_tracks(tracks)
|
||
}
|
||
ResolveKind::Append => queue.append_tracks(tracks),
|
||
ResolveKind::InsertAfter(position) => {
|
||
// Advance the cursor so this op's next chunk lands right
|
||
// behind this one, keeping the collection contiguous.
|
||
// `insert_tracks` clamps positions past the end.
|
||
self.kind = ResolveKind::InsertAfter(position + tracks.len() as u32);
|
||
queue.insert_tracks(position, tracks)
|
||
}
|
||
}
|
||
}
|
||
}
|
||
|
||
#[derive(Clone, Debug)]
|
||
pub struct QueueManager {
|
||
created_at: SystemTime,
|
||
current_offset: usize,
|
||
play_order: Vec<usize>,
|
||
tracks: Vec<Track>,
|
||
pub repeat: bool,
|
||
pub shuffle: bool,
|
||
}
|
||
|
||
impl From<QueueManager> for Queue {
|
||
fn from(queue_manager: QueueManager) -> Self {
|
||
Self {
|
||
// A clock step backwards must not panic the playback loop.
|
||
timestamp: queue_manager
|
||
.created_at
|
||
.elapsed()
|
||
.unwrap_or_default()
|
||
.as_secs(),
|
||
current_position: queue_manager.current_position() as u32,
|
||
tracks: queue_manager.tracks,
|
||
// The manager cannot know about in-flight resolves; the
|
||
// playback loop's broadcast path sets this from its pending-op
|
||
// map (see `Playback::broadcast_queue`).
|
||
resolving: false,
|
||
}
|
||
}
|
||
}
|
||
|
||
impl Default for QueueManager {
|
||
fn default() -> Self {
|
||
Self::new()
|
||
}
|
||
}
|
||
|
||
impl QueueManager {
|
||
pub fn new() -> Self {
|
||
Self {
|
||
created_at: SystemTime::now(),
|
||
current_offset: 0,
|
||
play_order: Vec::new(),
|
||
tracks: Vec::new(),
|
||
repeat: false,
|
||
shuffle: false,
|
||
}
|
||
}
|
||
pub fn current_position(&self) -> usize {
|
||
if self.current_offset < self.play_order.len() {
|
||
self.play_order[self.current_offset]
|
||
} else {
|
||
0
|
||
}
|
||
}
|
||
|
||
/// Number of tracks in the queue (playback uses it to bound skip
|
||
/// loops: at most one full pass, even with repeat on).
|
||
pub fn len(&self) -> usize {
|
||
self.tracks.len()
|
||
}
|
||
|
||
pub fn is_empty(&self) -> bool {
|
||
self.tracks.is_empty()
|
||
}
|
||
|
||
pub fn is_last_track(&self) -> bool {
|
||
!self.tracks.is_empty() && self.current_position() == self.tracks.len() - 1
|
||
}
|
||
|
||
pub fn shuffle_on(&mut self) {
|
||
self.shuffle = true;
|
||
self.shuffle_before(self.current_offset);
|
||
self.shuffle_behind(self.current_offset);
|
||
}
|
||
|
||
pub fn shuffle_off(&mut self) {
|
||
self.shuffle = false;
|
||
let pos = self.current_position();
|
||
self.current_offset = pos;
|
||
self.play_order = (0..self.tracks.len()).collect();
|
||
}
|
||
|
||
pub fn shuffle_all(&mut self) {
|
||
self.play_order.shuffle(&mut rng());
|
||
}
|
||
|
||
pub fn shuffle_before(&mut self, pos: usize) {
|
||
if let Some(slice) = self.play_order.get_mut(..pos) {
|
||
slice.shuffle(&mut rng());
|
||
}
|
||
}
|
||
|
||
pub fn shuffle_behind(&mut self, pos: usize) {
|
||
if let Some(slice) = self.play_order.get_mut(pos + 1..) {
|
||
slice.shuffle(&mut rng());
|
||
}
|
||
}
|
||
|
||
pub fn current_track(&self) -> Option<Track> {
|
||
if self.current_position() < self.tracks.len() {
|
||
Some(self.tracks[self.current_position()].clone())
|
||
} else {
|
||
None
|
||
}
|
||
}
|
||
|
||
pub fn next_track(&mut self) -> Option<Track> {
|
||
let len = self.tracks.len();
|
||
if len == 0 {
|
||
return None;
|
||
};
|
||
if self.current_offset < len - 1 {
|
||
self.current_offset += 1;
|
||
let current_pos = self.current_position();
|
||
if current_pos < len {
|
||
Some(self.tracks[current_pos].clone())
|
||
} else {
|
||
None
|
||
}
|
||
} else {
|
||
debug!("no more tracks");
|
||
if self.repeat {
|
||
debug!("repeat");
|
||
self.current_offset = 0;
|
||
if self.shuffle {
|
||
self.shuffle_all();
|
||
}
|
||
return self.current_track();
|
||
}
|
||
None
|
||
}
|
||
}
|
||
|
||
pub fn prev_track(&mut self) -> Option<Track> {
|
||
if 0 < self.current_offset {
|
||
self.current_offset -= 1;
|
||
self.current_track()
|
||
} else {
|
||
None
|
||
}
|
||
}
|
||
|
||
pub fn set_current_position(&mut self, current_position: u32) -> bool {
|
||
if current_position < self.tracks.len() as u32 {
|
||
if self.shuffle {
|
||
self.shuffle_all();
|
||
}
|
||
let Some(current_offset) = self
|
||
.play_order
|
||
.iter()
|
||
.position(|&i| i == current_position as usize)
|
||
else {
|
||
error!("invalid current position");
|
||
error!("queue: {:#?}", self);
|
||
return false;
|
||
};
|
||
if self.shuffle {
|
||
self.play_order.swap(0, current_offset);
|
||
self.current_offset = 0;
|
||
} else {
|
||
self.current_offset = current_offset;
|
||
}
|
||
true
|
||
} else {
|
||
false
|
||
}
|
||
}
|
||
|
||
pub fn replace_with_tracks(&mut self, tracks: &[Track]) -> Option<Track> {
|
||
self.current_offset = 0;
|
||
self.tracks = tracks.to_vec();
|
||
self.play_order = (0..self.tracks.len()).collect();
|
||
if self.shuffle {
|
||
self.shuffle_all();
|
||
}
|
||
if 0 < self.tracks.len() as u32 {
|
||
Some(self.tracks[self.current_position()].clone())
|
||
} else {
|
||
None
|
||
}
|
||
}
|
||
|
||
pub fn append_tracks(&mut self, tracks: &[Track]) -> Option<Track> {
|
||
let len = self.tracks.len();
|
||
let is_empty = len == 0;
|
||
let order_additions: Vec<usize> = (len..len + tracks.len()).collect();
|
||
self.play_order.extend(order_additions);
|
||
self.tracks.extend(tracks.iter().cloned());
|
||
if self.shuffle {
|
||
self.shuffle_behind(self.current_offset);
|
||
}
|
||
if is_empty {
|
||
self.current_track()
|
||
} else {
|
||
None
|
||
}
|
||
}
|
||
|
||
pub fn remove_tracks(&mut self, positions: &[u32]) -> Option<Track> {
|
||
let mut play_next = false;
|
||
// Remove highest positions first so earlier removals don't shift the
|
||
// positions that are still to be removed.
|
||
let mut positions: Vec<usize> = positions.iter().map(|p| *p as usize).collect();
|
||
positions.sort_unstable_by(|a, b| b.cmp(a));
|
||
positions.dedup();
|
||
for pos in positions {
|
||
if pos >= self.tracks.len() {
|
||
debug!(pos, len = self.tracks.len(), "ignoring out-of-range remove");
|
||
continue;
|
||
}
|
||
if pos == self.current_position() {
|
||
play_next = true;
|
||
}
|
||
let Some(offset) = self.play_order.iter().position(|&i| i == pos) else {
|
||
error!(pos, "track position missing from play order, rebuilding");
|
||
self.rebuild_play_order();
|
||
return None;
|
||
};
|
||
if offset < self.current_offset {
|
||
self.current_offset -= 1;
|
||
}
|
||
self.tracks.remove(pos);
|
||
self.play_order.remove(offset);
|
||
self.play_order
|
||
.iter_mut()
|
||
.filter(|i| pos < **i)
|
||
.for_each(|i| *i -= 1);
|
||
}
|
||
if self.current_offset >= self.play_order.len() {
|
||
self.current_offset = 0;
|
||
}
|
||
if play_next {
|
||
self.current_track()
|
||
} else {
|
||
None
|
||
}
|
||
}
|
||
|
||
pub fn insert_tracks(&mut self, position: u32, tracks: &[Track]) -> Option<Track> {
|
||
let len = self.tracks.len();
|
||
if len == 0 {
|
||
return self.replace_with_tracks(tracks);
|
||
}
|
||
let inserted = tracks.len();
|
||
let position = (position as usize).min(len - 1);
|
||
let order_additions: Vec<usize> = (len..len + inserted).collect();
|
||
self.play_order.extend(order_additions);
|
||
let tail: Vec<Track> = self
|
||
.tracks
|
||
.splice(position + 1.., tracks.to_vec())
|
||
.collect();
|
||
self.tracks.extend(tail);
|
||
let mut changed: Vec<usize> = Vec::new();
|
||
// In shuffle mode we may already have played positions that are
|
||
// behind the insertion point; those shift by the number of inserted
|
||
// tracks.
|
||
for i in self
|
||
.play_order
|
||
.iter_mut()
|
||
.take(self.current_offset)
|
||
.filter(|i| position < **i)
|
||
{
|
||
*i += inserted;
|
||
changed.push(*i);
|
||
}
|
||
// The freshly appended order entries need to swap with the shifted
|
||
// ones so every index stays unique.
|
||
self.play_order
|
||
.iter_mut()
|
||
.skip(self.current_offset)
|
||
.for_each(|i| {
|
||
if changed.contains(i) {
|
||
*i -= inserted;
|
||
}
|
||
});
|
||
|
||
if self.shuffle {
|
||
self.shuffle_behind(self.current_offset);
|
||
}
|
||
None
|
||
}
|
||
|
||
pub fn queue_tracks(&mut self, tracks: &[Track]) -> Option<Track> {
|
||
let pos = self.current_position();
|
||
self.insert_tracks(pos as u32, tracks)
|
||
}
|
||
|
||
pub fn clear(&mut self, exclude_current: bool) -> bool {
|
||
let current_track = self.current_track();
|
||
self.current_offset = 0;
|
||
self.tracks.clear();
|
||
self.play_order.clear();
|
||
|
||
if exclude_current {
|
||
if let Some(track) = current_track {
|
||
self.tracks.push(track);
|
||
self.play_order.push(0);
|
||
}
|
||
}
|
||
|
||
!exclude_current
|
||
}
|
||
|
||
/// Restores play_order to a consistent state after an inconsistency was
|
||
/// detected. Loses shuffle history but keeps the queue playable.
|
||
fn rebuild_play_order(&mut self) {
|
||
self.play_order = (0..self.tracks.len()).collect();
|
||
self.current_offset = 0;
|
||
}
|
||
}
|
||
|
||
#[cfg(test)]
|
||
mod tests {
|
||
use super::*;
|
||
|
||
fn track(id: usize) -> Track {
|
||
Track {
|
||
path: format!("/tidal/playlists/p/{id}"),
|
||
artist: "artist".to_string(),
|
||
title: format!("track {id}"),
|
||
duration: None,
|
||
album: None,
|
||
is_skipped: false,
|
||
}
|
||
}
|
||
|
||
fn queue_with(n: usize) -> QueueManager {
|
||
let mut q = QueueManager::new();
|
||
let tracks: Vec<Track> = (0..n).map(track).collect();
|
||
q.replace_with_tracks(&tracks);
|
||
q
|
||
}
|
||
|
||
#[test]
|
||
fn empty_queue_operations_do_not_panic() {
|
||
let mut q = QueueManager::new();
|
||
assert!(!q.is_last_track());
|
||
assert!(q.current_track().is_none());
|
||
assert!(q.next_track().is_none());
|
||
assert!(q.prev_track().is_none());
|
||
assert!(q.remove_tracks(&[0]).is_none());
|
||
q.shuffle_on();
|
||
q.shuffle_off();
|
||
q.clear(true);
|
||
q.clear(false);
|
||
}
|
||
|
||
#[test]
|
||
fn remove_out_of_range_is_ignored() {
|
||
let mut q = queue_with(2);
|
||
assert!(q.remove_tracks(&[5]).is_none());
|
||
assert_eq!(q.tracks.len(), 2);
|
||
// pos == len used to panic via Vec::remove
|
||
assert!(q.remove_tracks(&[2]).is_none());
|
||
assert_eq!(q.tracks.len(), 2);
|
||
}
|
||
|
||
#[test]
|
||
fn remove_multiple_positions() {
|
||
let mut q = queue_with(4);
|
||
q.remove_tracks(&[1, 3]);
|
||
assert_eq!(q.tracks.len(), 2);
|
||
assert_eq!(q.play_order.len(), 2);
|
||
assert_eq!(q.current_track().unwrap().title, "track 0");
|
||
}
|
||
|
||
#[test]
|
||
fn remove_current_returns_successor() {
|
||
let mut q = queue_with(3);
|
||
let next = q.remove_tracks(&[0]);
|
||
assert_eq!(next.unwrap().title, "track 1");
|
||
assert_eq!(q.tracks.len(), 2);
|
||
}
|
||
|
||
#[test]
|
||
fn clear_keeps_play_order_consistent() {
|
||
let mut q = queue_with(3);
|
||
q.next_track();
|
||
q.clear(true);
|
||
assert_eq!(q.tracks.len(), 1);
|
||
assert_eq!(q.play_order.len(), 1);
|
||
assert!(q.current_track().is_some());
|
||
assert!(q.next_track().is_none());
|
||
}
|
||
|
||
#[test]
|
||
fn next_track_advances_and_repeats() {
|
||
let mut q = queue_with(2);
|
||
assert_eq!(q.next_track().unwrap().title, "track 1");
|
||
assert!(q.next_track().is_none());
|
||
q.repeat = true;
|
||
assert_eq!(q.next_track().unwrap().title, "track 0");
|
||
}
|
||
|
||
#[test]
|
||
fn insert_past_end_appends() {
|
||
let mut q = queue_with(2);
|
||
q.insert_tracks(99, &[track(2)]);
|
||
assert_eq!(q.tracks.len(), 3);
|
||
assert_eq!(q.play_order.len(), 3);
|
||
assert_eq!(q.tracks.last().unwrap().title, "track 2");
|
||
}
|
||
|
||
fn titles(q: &QueueManager) -> Vec<String> {
|
||
q.tracks.iter().map(|t| t.title.clone()).collect()
|
||
}
|
||
|
||
#[test]
|
||
fn replace_op_replaces_first_then_appends_and_plays_once() {
|
||
let mut q = queue_with(2);
|
||
let mut op = PendingResolve::new(ResolveKind::Replace);
|
||
let first = op.apply_chunk(&mut q, &[track(10), track(11)]);
|
||
// The first chunk resets the queue and names the track to start.
|
||
assert_eq!(first.unwrap().title, "track 10");
|
||
assert_eq!(titles(&q), vec!["track 10", "track 11"]);
|
||
let second = op.apply_chunk(&mut q, &[track(12)]);
|
||
// Later chunks extend the same replace without restarting playback.
|
||
assert!(second.is_none());
|
||
assert_eq!(titles(&q), vec!["track 10", "track 11", "track 12"]);
|
||
assert_eq!(op.applied(), 3);
|
||
}
|
||
|
||
#[test]
|
||
fn insert_after_op_keeps_chunks_contiguous_and_in_order() {
|
||
let mut q = queue_with(3); // playing track 0
|
||
let mut op = PendingResolve::new(ResolveKind::InsertAfter(0));
|
||
assert!(op.apply_chunk(&mut q, &[track(10), track(11)]).is_none());
|
||
assert!(op.apply_chunk(&mut q, &[track(12)]).is_none());
|
||
// Both chunks sit as one contiguous run right after the current
|
||
// track, in arrival order — not interleaved with the old tail.
|
||
assert_eq!(
|
||
titles(&q),
|
||
vec!["track 0", "track 10", "track 11", "track 12", "track 1", "track 2"]
|
||
);
|
||
}
|
||
|
||
#[test]
|
||
fn insert_after_op_clamps_past_the_end() {
|
||
let mut q = queue_with(1);
|
||
let mut op = PendingResolve::new(ResolveKind::InsertAfter(99));
|
||
assert!(op.apply_chunk(&mut q, &[track(10)]).is_none());
|
||
assert!(op.apply_chunk(&mut q, &[track(11)]).is_none());
|
||
assert_eq!(titles(&q), vec!["track 0", "track 10", "track 11"]);
|
||
}
|
||
|
||
#[test]
|
||
fn append_op_starts_playback_only_into_an_empty_queue() {
|
||
let mut q = QueueManager::new();
|
||
let mut op = PendingResolve::new(ResolveKind::Append);
|
||
let first = op.apply_chunk(&mut q, &[track(10)]);
|
||
// Landing in an empty queue makes the track current: play it.
|
||
assert_eq!(first.unwrap().title, "track 10");
|
||
assert!(op.apply_chunk(&mut q, &[track(11)]).is_none());
|
||
assert_eq!(titles(&q), vec!["track 10", "track 11"]);
|
||
assert_eq!(op.applied(), 2);
|
||
}
|
||
|
||
#[test]
|
||
fn cancel_flag_is_shared_with_the_forwarder() {
|
||
let op = PendingResolve::new(ResolveKind::Append);
|
||
let flag = op.cancel_flag();
|
||
assert!(!flag.load(std::sync::atomic::Ordering::Relaxed));
|
||
op.cancel();
|
||
assert!(flag.load(std::sync::atomic::Ordering::Relaxed));
|
||
}
|
||
|
||
#[test]
|
||
fn shuffle_insert_keeps_order_unique() {
|
||
let mut q = queue_with(5);
|
||
q.shuffle_on();
|
||
q.next_track();
|
||
q.next_track();
|
||
q.queue_tracks(&[track(5), track(6)]);
|
||
let mut order = q.play_order.clone();
|
||
order.sort_unstable();
|
||
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` and
|
||
/// `architecture/incremental-captures.md`). Handled on a spawned
|
||
/// task — a large walk or download must not block the orchestrator
|
||
/// loop. `result_tx` answers once the capture is *accepted*
|
||
/// (validation only); the walk then streams `CaptureProgress` events
|
||
/// on `progress_tx`, ending in exactly one `finished` event (with
|
||
/// `error` set on failure). A rejected capture answers with the error
|
||
/// and sends no progress events.
|
||
CaptureLibraryNode {
|
||
path: String,
|
||
name: String,
|
||
download: bool,
|
||
progress_tx: flume::Sender<crabidy_core::proto::crabidy::CaptureProgress>,
|
||
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",
|
||
}
|
||
}
|
||
}
|