crabidy/crabidy-server/src/queue_store.rs

//! Persisted queues on disk (see `architecture/queue-persistence.md`).
//!
//! Every queue is a folder under the store directory
//! (`<config>/crabidy/queues/`) holding one order-prefixed
//! `*.cbd-track.toml` **link** file per entry, plus a hidden
//! [`STATE_FILE_NAME`] sidecar. The automatically maintained queue lives in
//! [`CURRENT_QUEUE_NAME`]; every other folder is a named save. The same
//! directory is mounted read-only into the library as `/queues` by a second
//! `fsdy` instance — this module is the only writer.

use std::path::{Path, PathBuf};
use std::sync::Arc;
use std::time::Duration;

use crabidy_core::proto::crabidy::Track;
use serde::{Deserialize, Serialize};
use tracing::{debug, warn};

/// The library mount point of the queues directory (second `fsdy`
/// instance, see `architecture/queue-persistence.md` D1).
pub const QUEUES_PROVIDER_ROOT: &str = "/queues";

/// Reserved folder name of the automatically maintained queue.
pub const CURRENT_QUEUE_NAME: &str = "current";

/// Hidden per-queue sidecar carrying [`QueueState`]. Dot-prefixed, so
/// library listings never show it.
pub const STATE_FILE_NAME: &str = ".queue-state.toml";

/// The queues directory: `queues/` inside the crabidy config directory.
/// `None` when the platform has no config directory.
pub fn queues_dir() -> Option<PathBuf> {
    dirs::config_dir().map(|d| d.join("crabidy").join("queues"))
}

/// Everything the playback loop knows about the queue that is worth
/// persisting. Sent through the persister's `watch` channel (latest wins)
/// and written by [`QueueStore`].
#[derive(Clone, Debug, PartialEq)]
pub struct QueueSnapshot {
    /// Queue entries in track order (not play order — shuffle order is
    /// deliberately not persisted).
    pub tracks: Vec<Track>,
    /// Index of the current track in `tracks`.
    pub current_position: u32,
    pub repeat: bool,
    pub shuffle: bool,
}

/// The on-disk schema of the [`STATE_FILE_NAME`] sidecar.
#[derive(Clone, Copy, Debug, Default, Deserialize, Serialize)]
pub struct QueueState {
    pub current_position: u32,
    pub repeat: bool,
    pub shuffle: bool,
}

/// Errors from validating or writing a persisted queue.
///
/// At the RPC boundary: `InvalidName` → `invalid_argument`, `EmptyQueue` →
/// `failed_precondition`, the rest → `internal`. Messages carry names and
/// paths, never file contents.
#[derive(Debug, thiserror::Error)]
pub enum SaveQueueError {
    #[error("invalid queue name: {0}")]
    InvalidName(&'static str),
    #[error("the queue is empty")]
    EmptyQueue,
    #[error("queue persistence is disabled")]
    Disabled,
    #[error("cannot write queue: {0}")]
    Io(#[from] std::io::Error),
    #[error(transparent)]
    TrackFile(#[from] fsdy::TrackFileError),
    #[error("cannot serialize queue state: {0}")]
    State(#[from] toml::ser::Error),
}

/// Reads and writes persisted queue folders. Cheap to clone behind an
/// `Arc`; all I/O is `tokio::fs`.
#[derive(Debug)]
pub struct QueueStore {
    dir: PathBuf,
}

impl QueueStore {
    /// Opens the store at `dir`, creating the directory (and parents) if
    /// missing.
    pub async fn open(dir: PathBuf) -> Result<Self, std::io::Error> {
        tokio::fs::create_dir_all(&dir).await?;
        Ok(Self { dir })
    }

    /// The store directory (what the `/queues` provider instance mounts).
    pub fn dir(&self) -> &Path {
        &self.dir
    }

    /// Validates a user-supplied queue name, returning the trimmed name.
    ///
    /// Rejected: empty after trimming, containing `/`, `\` or NUL, starting
    /// with a dot (hidden folders are invisible to listings), and the
    /// reserved [`CURRENT_QUEUE_NAME`].
    pub fn validate_name(name: &str) -> Result<&str, SaveQueueError> {
        let name = name.trim();
        if name.is_empty() {
            return Err(SaveQueueError::InvalidName("must not be empty"));
        }
        if name.contains(['/', '\\', '\0']) {
            return Err(SaveQueueError::InvalidName(
                "must not contain a path separator or NUL",
            ));
        }
        if name.starts_with('.') {
            return Err(SaveQueueError::InvalidName(
                "must not start with a dot (hidden folders are invisible)",
            ));
        }
        if name == CURRENT_QUEUE_NAME {
            return Err(SaveQueueError::InvalidName(
                "is reserved for the automatically maintained queue",
            ));
        }
        Ok(name)
    }

    /// Saves `snapshot` as the named queue, overwriting an existing one.
    ///
    /// Validates `name` per [`Self::validate_name`] and rejects an empty
    /// snapshot with [`SaveQueueError::EmptyQueue`]. The folder is written
    /// to a hidden temp sibling first, then swapped into place (remove old,
    /// rename) — a crash can lose the folder, never corrupt it half-written
    /// next to intact files.
    pub async fn save(&self, name: &str, snapshot: &QueueSnapshot) -> Result<(), SaveQueueError> {
        let name = Self::validate_name(name)?;
        if snapshot.tracks.is_empty() {
            return Err(SaveQueueError::EmptyQueue);
        }
        self.write_queue_dir(name, snapshot).await
    }

    /// Persists `snapshot` as the current queue ([`CURRENT_QUEUE_NAME`]).
    ///
    /// Same write path as [`Self::save`] but without name validation and
    /// with an empty snapshot allowed — clearing the queue must persist as
    /// cleared.
    pub async fn persist_current(&self, snapshot: &QueueSnapshot) -> Result<(), SaveQueueError> {
        self.write_queue_dir(CURRENT_QUEUE_NAME, snapshot).await
    }

    /// The shared write path: build the whole folder as a hidden temp
    /// sibling, then swap it into place (remove old, rename). A crash can
    /// lose the folder, never leave it half-written next to intact files
    /// (architecture/queue-persistence.md D3).
    async fn write_queue_dir(
        &self,
        name: &str,
        snapshot: &QueueSnapshot,
    ) -> Result<(), SaveQueueError> {
        let tmp = self.dir.join(format!(".tmp-{name}"));
        // A leftover temp folder from a crashed or racing write is stale.
        if tokio::fs::try_exists(&tmp).await? {
            tokio::fs::remove_dir_all(&tmp).await?;
        }
        tokio::fs::create_dir_all(&tmp).await?;
        for (index, track) in snapshot.tracks.iter().enumerate() {
            let text = fsdy::TrackFile::from_track(track).to_toml()?;
            let file = tmp.join(fsdy::track_file_name(index, &track.title));
            tokio::fs::write(file, text).await?;
        }
        let state = QueueState {
            current_position: snapshot.current_position,
            repeat: snapshot.repeat,
            shuffle: snapshot.shuffle,
        };
        tokio::fs::write(tmp.join(STATE_FILE_NAME), toml::to_string_pretty(&state)?).await?;

        let target = self.dir.join(name);
        if tokio::fs::try_exists(&target).await? {
            tokio::fs::remove_dir_all(&target).await?;
        }
        tokio::fs::rename(&tmp, &target).await?;
        Ok(())
    }

    /// Loads the persisted current queue for the startup restore.
    ///
    /// Reads the folder like a library listing (sorted case-insensitively,
    /// broken/hidden/foreign entries skipped with warnings) plus the
    /// [`QueueState`] sidecar (missing or broken sidecar → default state).
    /// `None` when the folder does not exist — a fresh start. Never fails
    /// the server; every defect is a warning and degrades to less state.
    pub async fn load_current(&self) -> Option<QueueSnapshot> {
        let dir = self.dir.join(CURRENT_QUEUE_NAME);
        let mut read_dir = match tokio::fs::read_dir(&dir).await {
            Ok(read_dir) => read_dir,
            Err(err) if err.kind() == std::io::ErrorKind::NotFound => return None,
            Err(err) => {
                warn!(dir = %dir.display(), "cannot read the persisted queue: {err}");
                return None;
            }
        };

        // Mirror the provider listing: visible regular `*.cbd-track.toml`
        // files, sorted case-insensitively — restore order == listing order.
        let mut names: Vec<String> = Vec::new();
        loop {
            let entry = match read_dir.next_entry().await {
                Ok(Some(entry)) => entry,
                Ok(None) => break,
                Err(err) => {
                    warn!(dir = %dir.display(), "error while reading the persisted queue: {err}");
                    break;
                }
            };
            let is_file = entry
                .file_type()
                .await
                .is_ok_and(|file_type| file_type.is_file());
            let Some(name) = entry.file_name().to_str().map(str::to_string) else {
                warn!(dir = %dir.display(), "skipping queue entry with non-UTF-8 name");
                continue;
            };
            if is_file && !name.starts_with('.') && name.ends_with(fsdy::TRACK_FILE_SUFFIX) {
                names.push(name);
            }
        }
        names.sort_by_key(|name| name.to_lowercase());

        let mut tracks = Vec::new();
        for name in names {
            let file = dir.join(&name);
            let text = match tokio::fs::read_to_string(&file).await {
                Ok(text) => text,
                Err(err) => {
                    warn!(file = %file.display(), "cannot read queue entry: {err}");
                    continue;
                }
            };
            match fsdy::TrackFile::parse(&text) {
                Ok(track_file) => {
                    // The same library path the /queues listing would give
                    // the entry, so non-link playables behave identically.
                    let lib_path = crabidy_core::join_path(
                        &crabidy_core::join_path(QUEUES_PROVIDER_ROOT, CURRENT_QUEUE_NAME),
                        &crabidy_core::encode_segment(&name),
                    );
                    tracks.push(track_file.to_track(&lib_path));
                }
                Err(err) => {
                    warn!(file = %file.display(), "skipping invalid queue entry: {err}");
                }
            }
        }

        let state_file = dir.join(STATE_FILE_NAME);
        let state = match tokio::fs::read_to_string(&state_file).await {
            Ok(text) => toml::from_str(&text).unwrap_or_else(|err| {
                warn!(file = %state_file.display(), "broken queue state, using defaults: {err}");
                QueueState::default()
            }),
            Err(err) => {
                debug!(file = %state_file.display(), "no queue state, using defaults: {err}");
                QueueState::default()
            }
        };
        Some(QueueSnapshot {
            tracks,
            current_position: state.current_position,
            repeat: state.repeat,
            shuffle: state.shuffle,
        })
    }
}

/// Spawns the auto-persist task: awaits snapshot changes on `rx`, debounces
/// briefly to coalesce bursts (resolve chunks), skips writes whose snapshot
/// equals the last one written, and rewrites the current queue folder.
/// Write failures are warnings; the task never affects playback. Exits when
/// the sender side is dropped.
pub fn spawn_persister(
    store: Arc<QueueStore>,
    mut rx: tokio::sync::watch::Receiver<Option<QueueSnapshot>>,
) {
    tokio::spawn(async move {
        let mut last_written: Option<QueueSnapshot> = None;
        while rx.changed().await.is_ok() {
            // Debounce: a resolve streams many chunks in quick succession;
            // the watch channel keeps only the newest snapshot, so waiting
            // briefly coalesces the burst into one write.
            tokio::time::sleep(Duration::from_millis(200)).await;
            let Some(snapshot) = rx.borrow_and_update().clone() else {
                continue;
            };
            // Broadcasts that only toggled the `resolving` flag carry an
            // unchanged snapshot — skip the write.
            if last_written.as_ref() == Some(&snapshot) {
                continue;
            }
            match store.persist_current(&snapshot).await {
                Ok(()) => last_written = Some(snapshot),
                Err(err) => warn!("cannot persist the current queue: {err}"),
            }
        }
        debug!("queue snapshot channel closed, persister exiting");
    });
}

#[cfg(test)]
mod tests {
    use super::*;
    use crabidy_core::proto::crabidy::Album;
    use std::sync::Arc;
    use std::time::Duration;
    use tempfile::TempDir;

    fn track(i: usize) -> Track {
        Track {
            path: format!("/tidal/playlists/p/{i}"),
            artist: "artist".to_string(),
            title: format!("track {i}"),
            duration: Some(60 + i as u32),
            album: Some(Album {
                title: "album".to_string(),
                release_date: None,
            }),
        }
    }

    fn snapshot(n: usize) -> QueueSnapshot {
        QueueSnapshot {
            tracks: (0..n).map(track).collect(),
            current_position: 0,
            repeat: false,
            shuffle: false,
        }
    }

    async fn store() -> (QueueStore, TempDir) {
        let dir = TempDir::new().expect("tempdir");
        let store = QueueStore::open(dir.path().join("queues"))
            .await
            .expect("open creates the directory");
        (store, dir)
    }

    /// Sorted visible file names of a queue folder.
    fn visible_files(dir: &Path) -> Vec<String> {
        let mut names: Vec<String> = std::fs::read_dir(dir)
            .expect("queue folder")
            .map(|e| e.expect("entry").file_name().to_string_lossy().into_owned())
            .filter(|n| !n.starts_with('.'))
            .collect();
        names.sort_by_key(|n| n.to_lowercase());
        names
    }

    #[tokio::test]
    async fn save_writes_ordered_link_files_and_the_sidecar() {
        let (store, _dir) = store().await;
        let mut snap = snapshot(3);
        // A queue may hold fs tracks; persisting must link to them too.
        snap.tracks[1].path = "/fs/mix/song.cbd-track.toml".to_string();
        store.save("road trip", &snap).await.expect("save");

        let queue_dir = store.dir().join("road trip");
        let names = visible_files(&queue_dir);
        assert_eq!(names.len(), 3);
        for (i, name) in names.iter().enumerate() {
            assert!(name.starts_with(&format!("{:04} ", i + 1)), "{name}");
            let text = std::fs::read_to_string(queue_dir.join(name)).expect("read entry");
            let file = fsdy::TrackFile::parse(&text).expect("entry parses");
            // The listing rewrite restores the original track exactly.
            assert_eq!(file.to_track("/queues/irrelevant"), snap.tracks[i]);
        }
        assert!(
            queue_dir.join(STATE_FILE_NAME).exists(),
            "sidecar written (hidden from listings by its dot prefix)"
        );
    }

    #[tokio::test]
    async fn save_validates_names_and_rejects_an_empty_queue() {
        let (store, _dir) = store().await;
        for bad in ["", "   ", "a/b", "a\\b", ".hidden", CURRENT_QUEUE_NAME] {
            assert!(
                matches!(
                    store.save(bad, &snapshot(1)).await,
                    Err(SaveQueueError::InvalidName(_))
                ),
                "name {bad:?} must be rejected"
            );
        }
        assert!(matches!(
            store.save("fine", &snapshot(0)).await,
            Err(SaveQueueError::EmptyQueue)
        ));
        // A valid name is used trimmed.
        store
            .save("  padded  ", &snapshot(1))
            .await
            .expect("trimmed name saves");
        assert!(store.dir().join("padded").is_dir());
    }

    #[tokio::test]
    async fn save_overwrites_an_existing_queue_completely() {
        let (store, _dir) = store().await;
        store.save("mix", &snapshot(3)).await.expect("first save");
        store.save("mix", &snapshot(1)).await.expect("overwrite");
        // No stale entries from the longer first save survive.
        assert_eq!(visible_files(&store.dir().join("mix")).len(), 1);
    }

    #[tokio::test]
    async fn persist_current_and_load_round_trip() {
        let (store, _dir) = store().await;
        let snap = QueueSnapshot {
            current_position: 2,
            repeat: true,
            shuffle: true,
            ..snapshot(4)
        };
        store.persist_current(&snap).await.expect("persist");
        let loaded = store.load_current().await.expect("load");
        assert_eq!(loaded, snap);
    }

    #[tokio::test]
    async fn persist_current_accepts_an_empty_queue() {
        // Clearing the queue must persist as cleared, not keep yesterday's
        // tracks for the next restart.
        let (store, _dir) = store().await;
        store
            .persist_current(&snapshot(2))
            .await
            .expect("non-empty");
        store.persist_current(&snapshot(0)).await.expect("empty");
        let loaded = store.load_current().await.expect("load");
        assert!(loaded.tracks.is_empty());
    }

    #[tokio::test]
    async fn load_current_without_a_folder_is_a_fresh_start() {
        let (store, _dir) = store().await;
        assert!(store.load_current().await.is_none());
    }

    #[tokio::test]
    async fn load_current_skips_broken_entries_and_survives_a_broken_sidecar() {
        let (store, _dir) = store().await;
        store.persist_current(&snapshot(2)).await.expect("persist");
        let current = store.dir().join(CURRENT_QUEUE_NAME);
        std::fs::write(current.join("0000 broken.cbd-track.toml"), "not [ toml")
            .expect("write broken entry");
        std::fs::write(current.join(STATE_FILE_NAME), "also not [ toml")
            .expect("break the sidecar");
        let loaded = store.load_current().await.expect("load");
        // The two good tracks load; the broken entry is skipped and the
        // broken sidecar degrades to default state instead of failing.
        assert_eq!(loaded.tracks.len(), 2);
        assert_eq!(loaded.current_position, 0);
    }

    #[tokio::test]
    async fn persister_writes_the_latest_snapshot() {
        let (store, _dir) = store().await;
        let store = Arc::new(store);
        let (tx, rx) = tokio::sync::watch::channel(None);
        spawn_persister(Arc::clone(&store), rx);
        // A burst: only the newest snapshot matters (latest-wins channel).
        tx.send(Some(snapshot(5))).expect("send");
        tx.send(Some(snapshot(3))).expect("send");
        let deadline = tokio::time::Instant::now() + Duration::from_secs(5);
        loop {
            if let Some(loaded) = store.load_current().await {
                if loaded.tracks.len() == 3 {
                    break;
                }
            }
            assert!(
                tokio::time::Instant::now() < deadline,
                "persister never wrote the latest snapshot"
            );
            tokio::time::sleep(Duration::from_millis(50)).await;
        }
    }
}