Serverjonas/SJNmusic
2
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

add a bunch of qol features

Browse Source
This commit is contained in:
jhkaktus
2026-06-24 20:02:07 +02:00
parent 1b430aab74
commit 5e14a9e308
11 changed files with 1881 additions and 246 deletions
Download Patch File Download Diff File Expand all files Collapse all files
daemon/Cargo.lock
Generated
+40
View File
@@ -863,6 +863,15 @@ dependencies = [
"portable-atomic",
]
[[package]]
name = "ppv-lite86"
version = "0.2.21"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "85eae3c4ed2f50dcfe72643da4befc30deadb458a9b590d720cde2f2b1e97da9"
dependencies = [
"zerocopy",
]
[[package]]
name = "proc-macro-crate"
version = "3.5.0"
@@ -918,6 +927,36 @@ version = "5.3.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "69cdb34c158ceb288df11e18b4bd39de994f6657d83847bdffdbd7f346754b0f"
[[package]]
name = "rand"
version = "0.8.6"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "5ca0ecfa931c29007047d1bc58e623ab12e5590e8c7cc53200d5202b69266d8a"
dependencies = [
"libc",
"rand_chacha",
"rand_core",
]
[[package]]
name = "rand_chacha"
version = "0.3.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "e6c10a63a0fa32252be49d21e7709d4d4baf8d231c2dbce1eaa8141b9b127d88"
dependencies = [
"ppv-lite86",
"rand_core",
]
[[package]]
name = "rand_core"
version = "0.6.4"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "ec0be4795e2f6a28069bec0b5ff3e2ac9bafc99e6a9a7dc3547996c5c816922c"
dependencies = [
"getrandom 0.2.17",
]
[[package]]
name = "redox_users"
version = "0.4.6"
@@ -1079,6 +1118,7 @@ dependencies = [
"env_logger",
"log",
"percent-encoding",
"rand",
"rodio",
"rusqlite",
"serde",
daemon/Cargo.toml
+1
View File
@@ -15,3 +15,4 @@ log = "0.4"
env_logger = "0.11"
toml = "0.8"
percent-encoding = "2"
rand = "0.8"
daemon/src/audio.rs
+324 -29
View File
@@ -1,42 +1,337 @@
use log::debug;
use rodio::{Decoder, OutputStream, Sink};
//! Audio playback lives on a dedicated thread because rodio's `OutputStream`
//! is `!Send` (cpal wraps a non-Send platform stream). We expose a
//! `Send + Sync` `AudioHandle` to callers via:
//! - `mpsc::Sender<AudioCmd>` for commands (Play, Pause, Resume, ...)
//! - `Arc<Mutex<AudioStatus>>` for read snapshots (volume, paused,
//! elapsed/duration, ...)
//!
//! The audio thread is the sole writer of `AudioStatus`; readers (daemon
//! tick, HTTP handlers) just clone or peek.
use std::fs::File;
use std::io::BufReader;
use std::sync::mpsc::{self, Receiver, RecvTimeoutError, Sender};
use std::sync::{Arc, Mutex};
use std::thread;
use std::time::{Duration, Instant};
pub struct AudioEngine {
sink: Option<Sink>,
_stream: Option<OutputStream>,
use log::{debug, info, warn};
use rodio::{Decoder, OutputStream, Sink, Source};
/// Shared, `Send + Sync` view of the engine's state. Updated by the audio
/// thread on every command and on natural end-of-track detection. Readers
/// lock briefly to clone this struct.
#[derive(Default, Clone, Debug)]
pub struct AudioStatus {
pub volume: f32,
pub paused: bool,
/// True iff an active `Sink` exists. Goes false on `Stop`, on natural
/// end-of-track, or before the very first `Play`.
pub playing: bool,
/// Path of the file currently producing sound. `None` while idle and
/// during the gap between two `Play` commands.
pub current_path: Option<String>,
/// Source-reported duration in seconds; `None` when unknown.
pub duration_secs: Option<f64>,
/// `Instant` of the most recent play-start or seek reset.
pub play_started_at: Option<Instant>,
/// Cumulative time spent paused during the current track, in seconds.
pub paused_total_secs: f64,
/// `Instant` of the current pause; `None` when not paused.
pub paused_at: Option<Instant>,
/// `true` between the moment a `Play` command is dequeued and the moment
/// the audio thread finishes setting the sink up (success *or* failure).
/// Lets the daemon tick distinguish "still working" from "decided".
pub audio_pending_play: bool,
/// Set `true` when a `Play` command completed by *failing* to open the
/// output, file, decoder, etc. The daemon reads this to avoid writing a
/// bogus 0-second history row for songs that never actually played.
pub last_play_failed: bool,
}
impl AudioEngine {
pub fn new() -> Self {
Self {
sink: None,
_stream: None,
impl AudioStatus {
/// Time spent playing the current track so far, in seconds. Pauses are
/// excluded via `paused_total_secs`.
pub fn elapsed_secs(&self) -> f64 {
let Some(started) = self.play_started_at else {
return 0.0;
};
if let Some(paused_at) = self.paused_at {
let raw = paused_at.duration_since(started).as_secs_f64();
return (raw - self.paused_total_secs).max(0.0);
}
let raw = started.elapsed().as_secs_f64();
(raw - self.paused_total_secs).max(0.0)
}
}
/// Commands the audio thread processes FIFO.
#[derive(Debug)]
pub enum AudioCmd {
Play(String),
Pause,
Resume,
SetVolume(f32),
Seek(f64),
Stop,
Shutdown,
}
/// Cheap, `Send + Sync` handle HTTP / daemon code use to talk to the audio
/// thread without holding the rodio `OutputStream` directly.
#[derive(Clone)]
pub struct AudioHandle {
tx: Sender<AudioCmd>,
status: Arc<Mutex<AudioStatus>>,
}
impl AudioHandle {
/// Enqueue a command. Returns silently if the audio thread has already
/// shut down — best-effort, idempotently safe because the daemon is
/// shutting down in that case.
pub fn send(&self, cmd: AudioCmd) {
let _ = self.tx.send(cmd);
}
pub fn play(&mut self, path: &str) -> Result<(), String> {
let (_stream, handle) = OutputStream::try_default()
.map_err(|e| format!("could not open audio output: {e}"))?;
let sink = Sink::try_new(&handle).map_err(|e| format!("could not create sink: {e}"))?;
let file = File::open(path).map_err(|e| format!("could not open file: {e}"))?;
let source =
Decoder::new(BufReader::new(file)).map_err(|e| format!("decode error: {e}"))?;
sink.append(source);
debug!("rodio sink appended for {path}");
self.sink = Some(sink);
self._stream = Some(_stream);
Ok(())
/// Take a consistent snapshot of the audio state. Cheap: the inner lock
/// is held only long enough to clone a few primitives.
pub fn snapshot(&self) -> AudioStatus {
self.status
.lock()
.unwrap_or_else(|p| p.into_inner())
.clone()
}
pub fn is_playing(&self) -> bool {
match &self.sink {
Some(s) => !s.empty(),
None => false,
/// Signal the audio thread to exit. Safe to call multiple times.
pub fn shutdown(&self) {
let _ = self.tx.send(AudioCmd::Shutdown);
}
}
/// Spawn the audio thread and return its handle. Caller is responsible for
/// eventually calling `AudioHandle::shutdown` (or simply dropping the handle
/// sender end, which causes the thread to exit on its next `recv_timeout`).
pub fn spawn_audio_thread() -> AudioHandle {
let (tx, rx) = mpsc::channel();
let status = Arc::new(Mutex::new(AudioStatus {
volume: 1.0,
..Default::default()
}));
let status_clone = Arc::clone(&status);
thread::Builder::new()
.name("sjnmusic-audio".into())
.spawn(move || run_audio(rx, status_clone))
.expect("failed to spawn audio thread");
AudioHandle { tx, status }
}
/// Probe-only helper for download workers that aren't allowed to start
/// full playback. Returns `None` if the source doesn't expose duration.
pub fn probe_duration_secs(path: &str) -> Option<f64> {
let file = File::open(path).ok()?;
let source = Decoder::new(BufReader::new(file)).ok()?;
source.total_duration().map(|d| d.as_secs_f64())
}
/// Audio thread body: owns the rodio `OutputStream` and `Sink`. Receives
/// commands from the channel and, when the channel is idle for 200 ms,
/// checks whether the track ended naturally so HTTP/daemon can react.
#[allow(unused_assignments)] // Some commands (Stop / Play) deliberately assign sink = None
// to drop the prior value before optionally reassigning; lint
// doesn't track Drop side effects.
fn run_audio(rx: Receiver<AudioCmd>, status: Arc<Mutex<AudioStatus>>) {
let mut sink: Option<Sink> = None;
let mut _stream: Option<OutputStream> = None;
loop {
// End-of-track detection. rodio's `Sink::empty()` is true iff the
// source is exhausted AND playback is not paused. So while we are
// paused, the check is harmless; once we resume and the buffer
// drains, `empty()` flips to true and we clear `playing`.
if !is_paused(&status) {
let ended = sink.as_ref().map(|s| s.empty()).unwrap_or(false);
if ended && status_snapshot_path(&status).is_some() {
debug!("audio: track ended naturally");
sink = None;
_stream = None;
let mut s = status.lock().unwrap();
s.playing = false;
s.paused = false;
s.current_path = None;
s.play_started_at = None;
s.paused_at = None;
s.paused_total_secs = 0.0;
}
}
match rx.recv_timeout(Duration::from_millis(200)) {
Ok(cmd) => match cmd {
AudioCmd::Play(path) => {
// Drop any prior stream/sink before opening a new one
// so the audio device isn't held across calls.
sink = None;
_stream = None;
{
let mut s = status.lock().unwrap();
s.playing = false;
s.paused = false;
s.current_path = None;
s.play_started_at = None;
s.paused_at = None;
s.paused_total_secs = 0.0;
// Mark "we're working on a Play" so the daemon tick
// knows not to treat a transient `playing=false` as
// "song ended naturally".
s.audio_pending_play = true;
s.last_play_failed = false;
}
let vol = status.lock().unwrap().volume;
// Open device + sink + decoder in one place so success /
// failure book-keeping lives at the bottom of the block.
let result: Result<(), String> = (|| {
let (stream, handle) = OutputStream::try_default()
.map_err(|e| format!("output: {e}"))?;
let sink_obj =
Sink::try_new(&handle).map_err(|e| format!("sink: {e}"))?;
let file = File::open(&path)
.map_err(|e| format!("file: {e}"))?;
let source = Decoder::new(BufReader::new(file))
.map_err(|e| format!("decode: {e}"))?;
let dur = source
.total_duration()
.map(|d| d.as_secs_f64())
.unwrap_or(0.0);
sink_obj.set_volume(vol);
sink_obj.append(source);
sink = Some(sink_obj);
_stream = Some(stream);
let mut s = status.lock().unwrap();
s.current_path = Some(path.clone());
s.duration_secs = Some(dur);
s.playing = true;
s.paused = false;
s.play_started_at = Some(Instant::now());
s.paused_at = None;
s.paused_total_secs = 0.0;
debug!("audio: started {path} (~{dur:.1}s)");
Ok(())
})();
{
let mut s = status.lock().unwrap();
s.audio_pending_play = false;
if let Err(e) = result {
// Don't leave the daemon thinking a song is
// "playing" — flip the explicit failure flag so
// the next tick can clean up state.current
// without writing a bogus history row.
s.last_play_failed = true;
warn!("audio: setup failed for {path}: {e}");
s.playing = false;
s.current_path = None;
s.play_started_at = None;
}
}
}
AudioCmd::Pause => {
if let Some(s) = sink.as_ref() {
let already_paused = {
let st = status.lock().unwrap();
st.paused
};
if !already_paused {
s.pause();
let mut st = status.lock().unwrap();
st.paused = true;
st.paused_at = Some(Instant::now());
debug!("audio: paused");
}
} else {
warn!("audio: pause requested but nothing playing");
}
}
AudioCmd::Resume => {
if let Some(s) = sink.as_ref() {
let mut st = status.lock().unwrap();
if st.paused {
if let Some(p) = st.paused_at.take() {
st.paused_total_secs += p.elapsed().as_secs_f64();
}
drop(st);
s.play();
status.lock().unwrap().paused = false;
debug!("audio: resumed");
}
} else {
warn!("audio: resume requested but nothing playing");
}
}
AudioCmd::SetVolume(v) => {
let v_clamped = v.clamp(0.0, 2.0);
if let Some(s) = sink.as_ref() {
s.set_volume(v_clamped);
}
status.lock().unwrap().volume = v_clamped;
debug!("audio: volume -> {v_clamped}");
}
AudioCmd::Seek(secs) => {
if let Some(s) = sink.as_ref() {
if !secs.is_finite() || secs < 0.0 {
warn!("audio: invalid seek {secs}");
} else {
let d = Duration::from_secs_f64(secs);
match s.try_seek(d) {
Ok(()) => {
let mut st = status.lock().unwrap();
st.play_started_at = Some(Instant::now());
st.paused_total_secs = 0.0;
if st.paused {
st.paused_at = Some(Instant::now());
}
debug!("audio: seeked to {secs:.2}s");
}
Err(e) => warn!("audio: seek failed: {e:?}"),
}
}
} else {
warn!("audio: seek requested but nothing playing");
}
}
AudioCmd::Stop => {
if sink.is_some() {
debug!("audio: stop");
}
sink = None;
_stream = None;
let mut s = status.lock().unwrap();
s.playing = false;
s.paused = false;
s.current_path = None;
s.play_started_at = None;
s.paused_at = None;
s.paused_total_secs = 0.0;
}
AudioCmd::Shutdown => {
info!("audio: shutdown");
sink = None;
_stream = None;
break;
}
},
Err(RecvTimeoutError::Timeout) => {}
Err(RecvTimeoutError::Disconnected) => {
warn!("audio: command channel disconnected");
break;
}
}
}
}
fn is_paused(status: &Arc<Mutex<AudioStatus>>) -> bool {
status.lock().unwrap().paused
}
fn status_snapshot_path(status: &Arc<Mutex<AudioStatus>>) -> Option<String> {
status.lock().unwrap().current_path.clone()
}
daemon/src/config.rs
+49
View File
@@ -36,12 +36,61 @@ impl Default for LoggingConfig {
}
}
/// Library / filesystem configuration.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct LibraryConfig {
/// Directory where downloaded songs live. Defaults to
/// `$HOME/.sjn/music/songs`. May be absolute or relative to the config file.
#[serde(default)]
pub music_dir: Option<String>,
/// Maximum number of songs allowed in the playback queue. 0 = unlimited.
#[serde(default)]
pub max_queue_size: usize,
/// Default repeat mode used on daemon start: one of "off", "one", "all".
#[serde(default)]
pub default_repeat: String,
}
impl Default for LibraryConfig {
fn default() -> Self {
Self {
music_dir: None,
max_queue_size: 0,
default_repeat: "off".to_string(),
}
}
}
/// Controls how fuzzy search behaves and which routes /search and /search/all use.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct SearchConfig {
/// Jaro-Winkler score in [0.0, 1.0] below which fuzzy matches are rejected.
#[serde(default = "default_fuzzy")]
pub fuzzy_threshold: f64,
}
fn default_fuzzy() -> f64 {
0.65
}
impl Default for SearchConfig {
fn default() -> Self {
Self {
fuzzy_threshold: default_fuzzy(),
}
}
}
#[derive(Debug, Clone, Serialize, Deserialize, Default)]
pub struct Config {
#[serde(default)]
pub server: ServerConfig,
#[serde(default)]
pub logging: LoggingConfig,
#[serde(default)]
pub library: LibraryConfig,
#[serde(default)]
pub search: SearchConfig,
}
impl Config {
daemon/src/daemon.rs
+144 -64
View File
@@ -4,18 +4,27 @@ use std::thread;
use log::{debug, error, info, warn};
use crate::{
audio::AudioEngine,
audio::spawn_audio_thread,
config::Config,
db::open_db,
server::start_server,
state::DaemonState,
state::{DaemonState, RepeatMode, Song},
};
pub fn run(cfg: Config) {
info!("Starting sjnmusicd");
let conn = open_db();
let state = Arc::new(Mutex::new(DaemonState::new(conn)));
let audio_handle = Arc::new(spawn_audio_thread());
let default_repeat = RepeatMode::parse(&cfg.library.default_repeat).unwrap_or(RepeatMode::Off);
let state = Arc::new(Mutex::new(DaemonState::new(
conn,
cfg.search.fuzzy_threshold,
cfg.library.max_queue_size,
default_repeat,
audio_handle,
)));
// HTTP server thread.
let server_state = state.clone();
@@ -30,70 +39,141 @@ pub fn run(cfg: Config) {
error!("Failed to spawn HTTP server thread: {e}");
}
// Audio playback loop (runs on the main thread).
let mut audio = AudioEngine::new();
info!("Playback loop ready");
loop {
{
let mut state = state.lock().unwrap();
// Clear "now playing" once the audio engine reports the song has
// ended so /now-playing and /queue reflect reality while the
// next iteration decides what's up next.
if !audio.is_playing() {
if let Some(prev) = state.current.take() {
debug!("Playback finished for {}", prev.name);
}
}
if !state.queue.is_empty() && !audio.is_playing() {
// Snapshot the next song first, attempt audio.play, and only
// remove from the queue (memory + DB) when playback actually
// starts so decode/init failures don't silently lose it.
let next = state.queue[0].clone();
debug!("Starting playback: {} ({})", next.name, next.path);
match audio.play(&next.path) {
Ok(()) => match state.pop_played() {
Ok(Some(popped)) => {
let name = popped.name.clone();
state.current = Some(popped);
debug!("DB queue row removed for {name}");
info!("Now playing: {name}");
}
Ok(None) => {
// Should not happen — we peeked a non-empty queue.
state.current = Some(next.clone());
warn!(
"pop_played returned None after successful audio.play for {}",
next.name
);
}
Err(e) => {
// pop_played is DB-first; on Err the in-memory
// queue has NOT been mutated, so the song is
// already at index 0 — do not re-insert or we
// would duplicate it.
error!(
"Played {} but failed to update DB queue ({e}); \
leaving in-memory queue untouched",
next.name
);
state.current = Some(next);
}
},
Err(e) => {
// The previous "now playing" is no longer accurate.
state.current = None;
error!(
"Failed to play {} ({e}); leaving in queue",
next.path
);
}
}
}
}
tick(&state);
thread::sleep(std::time::Duration::from_millis(200));
}
}
/// One iteration of the playback loop. Pulled out so it can be unit-tested
/// without a sleep or a network server.
fn tick(state: &Arc<Mutex<DaemonState>>) {
let mut state = state.lock().unwrap();
// Single snapshot per tick — cheaper (one lock acquisition) and avoids
// races where the audio thread flips `playing` between two reads.
let snap = state.audio_handle.snapshot();
// 1) End-of-track / failed-play handling. We only decide what happened
// AFTER the audio thread has finished deciding things itself, i.e.
// `audio_pending_play == false`. The two outcomes we differentiate now:
// - `snap.last_play_failed == true` ⇒ the `Play` command errored. We
// drop `state.current` and pop the queue row, but we DON'T write a
// history row (the song was never actually heard).
// - `snap.last_play_failed == false` & `!snap.playing` & some time
// elapsed ⇒ natural end-of-track. Record history with the elapsed
// time and apply repeat-mode rules.
if state.current.is_some() && !snap.audio_pending_play {
if snap.last_play_failed {
let prev = state.current.take();
state.popped_for_current = false;
let _ = state.pop_played(None); // drop queue row, no history
if let Some(p) = prev {
debug!("audio: {} never actually played (setup failed)", p.name);
apply_repeat_after_end(&mut state, &p);
}
} else if !snap.playing {
if let Some(prev) = state.current.take() {
let played_secs = snap.elapsed_secs();
debug!(
"Playback finished for {} (~{played_secs:.1}s)",
prev.name
);
let _ = state.pop_played(Some(played_secs));
apply_repeat_after_end(&mut state, &prev);
state.popped_for_current = false;
}
}
}
// 2) Pop the queue row for the song we just kicked off, but only once
// the audio thread confirmed it actually got the sink running. Until
// then, we risk a double-pop if the play failed.
if state.current.is_some() && snap.playing && !state.popped_for_current {
let _ = state.pop_played(None); // drop queue row only, no history
state.popped_for_current = true;
}
// 3) Start the next song if the queue has one and the audio engine is
// idle (idle = not playing AND not mid-setup).
if !state.queue.is_empty()
&& !snap.playing
&& !snap.audio_pending_play
&& state.current.is_none()
{
let next = state.queue[0].clone();
debug!("Starting playback: {} ({})", next.name, next.path);
state.current = Some(next.clone());
state.popped_for_current = false;
state.audio_handle.send(crate::audio::AudioCmd::Play(next.path.clone()));
}
}
/// Apply repeat-mode rules after a track has ended and `pop_played` has
/// already removed it from the queue.
fn apply_repeat_after_end(state: &mut DaemonState, prev: &Song) {
match state.repeat_mode {
RepeatMode::Off => {
// `cycle_snapshot` is unused for Off; no need to update it.
}
RepeatMode::One => {
state.queue.push(prev.clone());
if let Err(e) = append_queue_row(state, prev.id) {
warn!("failed to re-queue {} for repeat-one: {e}", prev.name);
}
state.cycle_snapshot = state.queue.clone();
}
RepeatMode::All => {
if state.queue.is_empty() && !state.cycle_snapshot.is_empty() {
info!(
"Repeat-all: refilling queue from snapshot ({} songs)",
state.cycle_snapshot.len()
);
let snapshot = state.cycle_snapshot.clone();
if let Err(e) = refill_queue(state, &snapshot) {
warn!("failed to refill queue for repeat-all: {e}");
}
state.queue = snapshot;
} else {
state.cycle_snapshot = state.queue.clone();
}
}
}
}
fn append_queue_row(state: &DaemonState, song_id: i64) -> rusqlite::Result<()> {
let conn = state.conn.lock().unwrap();
let next: i64 = conn
.query_row(
"SELECT COALESCE(MAX(position), 0) + 1 FROM queue",
[],
|row| row.get(0),
)
.unwrap_or(1);
conn.execute(
"INSERT INTO queue (position, song_id) VALUES (?1, ?2)",
rusqlite::params![next, song_id],
)?;
Ok(())
}
fn refill_queue(state: &DaemonState, snapshot: &[Song]) -> rusqlite::Result<()> {
let conn = state.conn.lock().unwrap();
conn.execute("DELETE FROM queue", [])?;
for s in snapshot {
let next: i64 = conn
.query_row(
"SELECT COALESCE(MAX(position), 0) + 1 FROM queue",
[],
|row| row.get(0),
)
.unwrap_or(1);
conn.execute(
"INSERT INTO queue (position, song_id) VALUES (?1, ?2)",
rusqlite::params![next, s.id],
)?;
}
Ok(())
}
daemon/src/db.rs
+41 -5
View File
@@ -1,4 +1,4 @@
use log::{debug, info};
use log::{debug, info, warn};
use rusqlite::Connection;
use crate::paths::{db_path, ensure_dirs};
@@ -18,9 +18,10 @@ pub fn open_db() -> Connection {
conn.execute_batch(
"CREATE TABLE IF NOT EXISTS songs (
id INTEGER PRIMARY KEY AUTOINCREMENT,
name TEXT UNIQUE NOT NULL,
path TEXT NOT NULL
id INTEGER PRIMARY KEY AUTOINCREMENT,
name TEXT UNIQUE NOT NULL,
path TEXT NOT NULL,
duration_secs INTEGER
);
CREATE TABLE IF NOT EXISTS queue (
@@ -44,11 +45,46 @@ pub fn open_db() -> Connection {
);
CREATE INDEX IF NOT EXISTS idx_playlist_songs_pl
ON playlist_songs(playlist_id, position);",
ON playlist_songs(playlist_id, position);
CREATE TABLE IF NOT EXISTS history (
id INTEGER PRIMARY KEY AUTOINCREMENT,
song_id INTEGER NOT NULL,
played_at INTEGER NOT NULL,
duration_secs_played INTEGER NOT NULL,
FOREIGN KEY(song_id) REFERENCES songs(id) ON DELETE CASCADE
);
CREATE INDEX IF NOT EXISTS idx_history_played_at
ON history(played_at DESC);
CREATE INDEX IF NOT EXISTS idx_history_song_id
ON history(song_id);",
)
.expect("Failed to create DB schema");
// Idempotent column migrations for older DBs created before
// `duration_secs` existed. SQLite raises a duplicate-column error which we
// swallow because the column already being present is exactly the intended
// post-condition of this step.
add_column_if_missing(&conn, "songs", "duration_secs", "INTEGER");
debug!("Database initialized at {}", db_path());
info!("Database ready");
conn
}
fn add_column_if_missing(conn: &Connection, table: &str, column: &str, decl: &str) {
let stmt = format!("ALTER TABLE {table} ADD COLUMN {column} {decl}");
match conn.execute_batch(&stmt) {
Ok(_) => info!("Added column {table}.{column}"),
Err(e) => {
// "duplicate column name: <name>" is the expected no-op signal.
let msg = e.to_string();
if msg.contains("duplicate column name") {
debug!("Column {table}.{column} already present");
} else {
warn!("Failed to add column {table}.{column}: {e}");
}
}
}
}
daemon/src/main.rs
+2 -1
View File
@@ -17,7 +17,8 @@ fn main() {
log::info!("Loaded configuration");
log::debug!("Config: {:?}", cfg);
paths::ensure_dirs();
paths::initialize(&cfg);
daemon::run(cfg);
}
daemon/src/paths.rs
+68 -9
View File
@@ -1,15 +1,72 @@
use std::path::PathBuf;
use std::sync::OnceLock;
pub fn base_dir() -> PathBuf {
let mut p = dirs::home_dir().unwrap();
use crate::config::Config;
/// Default base directory used when `paths::initialize` has not yet been
/// called. This is the same value the daemon used before paths became
/// configurable, so the daemon can still boot even when a config file is
/// missing or malformed.
fn default_base_dir() -> PathBuf {
let mut p = dirs::home_dir().unwrap_or_else(|| PathBuf::from("."));
p.push(".sjn/music");
p
}
fn default_songs_dir() -> PathBuf {
default_base_dir().join("songs")
}
/// Library root OnceLock. Populated by `initialize(&cfg)`; readers fall back
/// to the default base dir if `initialize` has not yet run.
static BASE: OnceLock<PathBuf> = OnceLock::new();
static SONGS: OnceLock<PathBuf> = OnceLock::new();
/// Resolves paths from `cfg` and creates all required directories. Safe to
/// call multiple times — only the first call wins because OnceLocks can't be
/// overwritten; subsequent calls with a different config are logged but
/// ignored so the daemon can keep its existing on-disk layout.
pub fn initialize(cfg: &Config) {
let base = match cfg.library.music_dir.as_deref() {
Some(dir) if !dir.is_empty() => {
let p = PathBuf::from(dir);
if p.is_absolute() {
p
} else {
let mut home = dirs::home_dir().unwrap_or_else(|| PathBuf::from("."));
home.push(p);
home
}
}
_ => default_base_dir(),
};
let songs = base.join("songs");
let _ = BASE.set(base.clone());
let _ = SONGS.set(songs.clone());
if let Err(e) = std::fs::create_dir_all(&base) {
log::warn!("Failed to create base dir {base:?}: {e}");
}
if let Err(e) = std::fs::create_dir_all(&songs) {
log::warn!("Failed to create songs dir {songs:?}: {e}");
}
}
/// Library root: configured value if `initialize` ran, otherwise the default.
/// Never panics so callers from `config::load`, `db::open_db`, and
/// `state::init` all work even before initialization.
fn base_dir_inner() -> PathBuf {
BASE.get().cloned().unwrap_or_else(default_base_dir)
}
fn songs_dir_inner() -> PathBuf {
SONGS.get().cloned().unwrap_or_else(default_songs_dir)
}
pub fn songs_dir() -> PathBuf {
let mut p = base_dir();
p.push("songs");
p
songs_dir_inner()
}
pub fn song_path(name: &str) -> String {
@@ -30,18 +87,20 @@ pub fn song_path(name: &str) -> String {
}
pub fn db_path() -> String {
let mut p = base_dir();
let mut p = base_dir_inner();
p.push("songs.db");
p.to_string_lossy().to_string()
}
pub fn config_path() -> String {
let mut p = base_dir();
let mut p = base_dir_inner();
p.push("config.toml");
p.to_string_lossy().to_string()
}
pub fn ensure_dirs() {
let _ = std::fs::create_dir_all(base_dir());
let _ = std::fs::create_dir_all(songs_dir());
let base = base_dir_inner();
let _ = std::fs::create_dir_all(&base);
let songs = base.join("songs");
let _ = std::fs::create_dir_all(&songs);
}
daemon/src/server.rs
+407 -45
View File
@@ -4,7 +4,8 @@ use log::{debug, error, info, warn};
use serde::Serialize;
use tiny_http::{Header, Method, Request, Response, Server, StatusCode};
use crate::state::DaemonState;
use crate::audio::AudioCmd;
use crate::state::{init_async, init_batch, DaemonState, RepeatMode};
#[derive(Serialize)]
struct ApiError {
@@ -60,8 +61,12 @@ pub fn start_server(state: Arc<Mutex<DaemonState>>, host: &str, port: u16) {
}
}
/// Owns the request so we can call `request.respond()` at the end. Reads the
/// body (for POST/PUT/DELETE), routes, and replies with JSON.
fn read_body(request: &mut Request) -> Result<String, std::io::Error> {
let mut s = String::new();
let _ = request.as_reader().read_to_string(&mut s)?;
Ok(s)
}
fn handle(
mut request: Request,
state: Arc<Mutex<DaemonState>>,
@@ -71,40 +76,67 @@ fn handle(
let path = url.split('?').next().unwrap_or(url);
let route = route_for(method, path);
// /help short-circuit: no body, no state.
if matches!(route, Route::Help) {
let resp = json_response(200, &serde_json::json!({
"endpoints": [
"GET /help",
"GET /songs",
"GET /search?q=... or POST /search {query}",
"GET /songs?q=",
"GET /search?q=... (best match) or POST /search {query}",
"GET /search/all?q=... (every match above threshold)",
"GET /queue",
"GET /now-playing",
"GET /downloads",
"GET /history?limit=N",
"GET /stats",
"GET /repeat",
"GET /volume",
"POST /play {query}",
"POST /add {query}",
"POST /del {query}",
"POST /init {song}",
"POST /init {song} (async, returns job id, status 202)",
"POST /init/batch {songs:[...]}",
"POST /skip",
"POST /pause",
"POST /resume",
"POST /seek {secs}",
"POST /volume {vol}",
"POST /repeat {mode}",
"POST /queue/shuffle",
"POST /queue/clear",
"GET /playlists",
"POST /playlists {name}",
"GET /playlists/{name}",
"PATCH /playlists/{name} {name}",
"DELETE /playlists/{name}",
"POST /playlists/{name}/add {query}",
"POST /playlists/{name}/play",
"DELETE /playlists/{name}",
"DELETE /playlists/{name}/songs/{id}",
"POST /playlists/{name}/reorder {from, to}",
"POST /playlists/{name}/duplicate {name}",
]
}));
return Ok(request.respond(resp)?);
}
let mut body_text = String::new();
if matches!(method, Method::Post | Method::Delete | Method::Put) {
let _ = request.as_reader().read_to_string(&mut body_text)?;
if matches!(method, Method::Post | Method::Delete | Method::Put | Method::Patch) {
match read_body(&mut request) {
Ok(s) => body_text = s,
Err(e) => {
warn!("Failed to read body: {e}");
let resp = error_response("could not read request body");
return Ok(request.respond(resp)?);
}
}
}
debug!("Body: {}", body_text);
let query_params = parse_query(url);
// Clone the Arc so async routes can spawn workers that retain their
// own handle after we lock the original Arc into a MutexGuard.
let arc_for_async = Arc::clone(&state);
let mut state = match state.lock() {
Ok(s) => s,
Err(p) => {
@@ -115,11 +147,14 @@ fn handle(
};
let response = match route {
Route::Help => unreachable!("handled above"),
Route::Help => unreachable!(),
Route::Songs => json_response(200, &serde_json::json!({
"songs": state.all_songs(),
})),
Route::Songs => {
let q = query_params.get("q").map(|s| s.as_str());
json_response(200, &serde_json::json!({
"songs": state.list_songs_filtered(q),
}))
}
Route::Search => match query_or_body(&query_params, &body_text) {
Ok(q) => match state.search(&q) {
@@ -131,16 +166,37 @@ fn handle(
Err(e) => error_response(e),
},
Route::SearchAll => match query_or_body(&query_params, &body_text) {
Ok(q) => json_response(200, &serde_json::json!({
"matches": state.search_all(&q),
})),
Err(e) => error_response(e),
},
Route::Queue => json_response(200, &serde_json::json!({
"queue": state.queue,
"current": state.current,
"repeat": state.repeat_mode.as_str(),
})),
Route::NowPlaying => json_response(200, &serde_json::json!({
"current": state.current,
"queue_len": state.queue.len(),
Route::NowPlaying => json_response_now_playing(&state),
Route::Downloads => json_response(200, &serde_json::json!({
"downloads": state.list_downloads(),
})),
Route::History => {
let limit: usize = query_params
.get("limit")
.and_then(|s| s.parse().ok())
.unwrap_or(50);
json_response(200, &serde_json::json!({
"history": state.list_history(limit),
}))
}
Route::Stats => json_response(200, &state.stats()),
Route::Play => match query_or_body(&query_params, &body_text) {
Ok(q) => match state.play(&q) {
Ok(song) => json_response(200, &song),
@@ -152,7 +208,10 @@ fn handle(
Route::Add => match query_or_body(&query_params, &body_text) {
Ok(q) => match state.add(&q) {
Ok(song) => json_response(200, &song),
Err(e) => error_response(e),
Err(e) => json_response(
if e.contains("queue full") { 413 } else { 400 },
&ApiError { error: e },
),
},
Err(e) => error_response(e),
},
@@ -165,17 +224,43 @@ fn handle(
Err(e) => error_response(e),
},
Route::Init => match name_from_body(&body_text) {
Ok(song) => match state.init(song) {
Ok(s) => json_response(200, &s),
Err(e) => error_response(e),
},
Route::InitAsync => match name_from_body(&body_text) {
Ok(song) => {
let id = init_async(&arc_for_async, song);
json_response(202, &serde_json::json!({
"job_id": id,
"status": "queued",
}))
}
Err(e) => error_response(e),
},
Route::Skip => match state.skip() {
Ok(Some(s)) => json_response(200, &s),
Ok(None) => json_response(200, &serde_json::json!({"skipped": null})),
Route::InitBatch => match songs_from_body(&body_text) {
Ok(songs) => {
let ids = init_batch(&arc_for_async, songs);
json_response(202, &serde_json::json!({
"job_ids": ids,
}))
}
Err(e) => error_response(e),
},
Route::Skip => {
state.audio_handle.send(AudioCmd::Stop);
// Always clear current — a skip is an explicit decision to
// abandon the playing song, so we don't want the daemon tick to
// later treat it as a natural end-of-track and write a bogus
// `history` row.
state.current = None;
match state.skip() {
Ok(Some(s)) => json_response(200, &s),
Ok(None) => json_response(200, &serde_json::json!({"skipped": null})),
Err(e) => error_response(e),
}
}
Route::QueueShuffle => match state.shuffle_queue() {
Ok(()) => json_response(200, &serde_json::json!({"shuffled": true, "len": state.queue.len()})),
Err(e) => error_response(e),
},
@@ -184,6 +269,44 @@ fn handle(
Err(e) => error_response(e),
},
Route::Audio(method) => handle_audio_route(&state, method, &body_text),
Route::RepeatGet => json_response(200, &serde_json::json!({
"mode": state.repeat_mode.as_str(),
})),
Route::RepeatSet => match mode_from_body_or_default(&body_text) {
Ok(mode) => {
state.repeat_mode = mode;
if matches!(mode, RepeatMode::All) && state.cycle_snapshot.is_empty() {
state.cycle_snapshot = state.queue.clone();
}
info!("Repeat mode set to {}", mode.as_str());
json_response(200, &serde_json::json!({"mode": mode.as_str()}))
}
Err(e) => error_response(e),
},
Route::VolumeGet => {
let snap = state.audio_handle.snapshot();
json_response(200, &serde_json::json!({
"volume": snap.volume,
"playing": snap.playing,
"paused": snap.paused,
}))
}
Route::VolumeSet => match vol_from_body(&body_text) {
Ok(v) => {
state.audio_handle.send(AudioCmd::SetVolume(v));
json_response(200, &serde_json::json!({
"volume": state.audio_handle.snapshot().volume,
"ok": true,
}))
}
Err(e) => error_response(e),
},
Route::PlaylistsList => json_response(200, &state.list_playlists()),
Route::PlaylistCreate => match name_from_body(&body_text) {
@@ -199,6 +322,19 @@ fn handle(
Err(e) => error_response(e),
},
Route::PlaylistRename(name) => match new_name_from_body(&body_text) {
Ok(new_name) => match state.rename_playlist(&name, &new_name) {
Ok(pl) => json_response(200, &pl),
Err(e) => error_response(e),
},
Err(e) => error_response(e),
},
Route::PlaylistDelete(name) => match state.delete_playlist(&name) {
Ok(()) => json_response(200, &serde_json::json!({"deleted": name})),
Err(e) => error_response(e),
},
Route::PlaylistAdd(name) => match query_or_body(&query_params, &body_text) {
Ok(q) => match state.add_to_playlist(&name, &q) {
Ok((pl, song)) => json_response(200, &serde_json::json!({
@@ -211,12 +347,31 @@ fn handle(
},
Route::PlaylistPlay(name) => match state.play_playlist(&name) {
Ok(pl) => {
state.cycle_snapshot = pl.songs.clone();
json_response(200, &pl)
}
Err(e) => error_response(e),
},
Route::PlaylistRemoveSong(name, id) => match state.remove_from_playlist(&name, id) {
Ok(pl) => json_response(200, &pl),
Err(e) => error_response(e),
},
Route::PlaylistDelete(name) => match state.delete_playlist(&name) {
Ok(()) => json_response(200, &serde_json::json!({"deleted": name})),
Route::PlaylistReorder(name) => match reorder_from_body(&body_text) {
Ok((from, to)) => match state.reorder_playlist(&name, from, to) {
Ok(pl) => json_response(200, &pl),
Err(e) => error_response(e),
},
Err(e) => error_response(e),
},
Route::PlaylistDuplicate(name) => match new_name_from_body(&body_text) {
Ok(new_name) => match state.duplicate_playlist(&name, &new_name) {
Ok(pl) => json_response(200, &pl),
Err(e) => error_response(e),
},
Err(e) => error_response(e),
},
@@ -228,24 +383,90 @@ fn handle(
Ok(request.respond(response)?)
}
fn json_response_now_playing(state: &DaemonState) -> Response<std::io::Cursor<Vec<u8>>> {
let snap = state.audio_handle.snapshot();
let current_duration = state.current.as_ref().and_then(|c| c.duration_secs);
let duration = snap.duration_secs.or(current_duration);
json_response(
200,
&serde_json::json!({
"current": state.current,
"queue_len": state.queue.len(),
"elapsed_secs": snap.elapsed_secs(),
"duration_secs": duration,
"paused": snap.paused,
"playing": snap.playing,
"volume": snap.volume,
"repeat": state.repeat_mode.as_str(),
}),
)
}
#[derive(Debug)]
enum AudioRoute {
Pause,
Resume,
Seek,
}
fn handle_audio_route(
state: &DaemonState,
route: AudioRoute,
body: &str,
) -> Response<std::io::Cursor<Vec<u8>>> {
match route {
AudioRoute::Pause => {
state.audio_handle.send(AudioCmd::Pause);
json_response(200, &serde_json::json!({"ok": true, "label": "pause"}))
}
AudioRoute::Resume => {
state.audio_handle.send(AudioCmd::Resume);
json_response(200, &serde_json::json!({"ok": true, "label": "resume"}))
}
AudioRoute::Seek => match secs_from_body(body) {
Ok(s) => {
state.audio_handle.send(AudioCmd::Seek(s));
json_response(200, &serde_json::json!({"ok": true, "label": "seek", "secs": s}))
}
Err(e) => error_response(e),
},
}
}
#[derive(Debug)]
enum Route {
Help,
Songs,
Search,
SearchAll,
Queue,
NowPlaying,
Downloads,
History,
Stats,
Play,
Add,
Del,
Init,
InitAsync,
InitBatch,
Skip,
QueueShuffle,
QueueClear,
Audio(AudioRoute),
RepeatGet,
RepeatSet,
VolumeGet,
VolumeSet,
PlaylistsList,
PlaylistCreate,
PlaylistGet(String),
PlaylistRename(String),
PlaylistDelete(String),
PlaylistAdd(String),
PlaylistPlay(String),
PlaylistDelete(String),
PlaylistRemoveSong(String, i64),
PlaylistReorder(String),
PlaylistDuplicate(String),
NotFound,
}
@@ -255,44 +476,93 @@ fn route_for(method: &Method, path: &str) -> Route {
(Method::Get, "") | (Method::Get, "/help") => Route::Help,
(Method::Get, "/songs") => Route::Songs,
(m, "/search") if matches!(m, Method::Get | Method::Post) => Route::Search,
(Method::Get, "/search/all") => Route::SearchAll,
(Method::Get, "/queue") => Route::Queue,
(Method::Get, "/now-playing") => Route::NowPlaying,
(Method::Get, "/downloads") => Route::Downloads,
(Method::Get, "/history") => Route::History,
(Method::Get, "/stats") => Route::Stats,
(Method::Post, "/play") => Route::Play,
(Method::Post, "/add") => Route::Add,
(Method::Post, "/del") => Route::Del,
(Method::Post, "/init") => Route::Init,
(Method::Post, "/init") => Route::InitAsync,
(Method::Post, "/init/batch") => Route::InitBatch,
(Method::Post, "/skip") => Route::Skip,
(Method::Post, "/queue/shuffle") => Route::QueueShuffle,
(Method::Post, "/queue/clear") => Route::QueueClear,
(Method::Post, "/pause") => Route::Audio(AudioRoute::Pause),
(Method::Post, "/resume") => Route::Audio(AudioRoute::Resume),
(Method::Post, "/seek") => Route::Audio(AudioRoute::Seek),
(Method::Get, "/repeat") => Route::RepeatGet,
(Method::Post, "/repeat") => Route::RepeatSet,
(Method::Get, "/volume") => Route::VolumeGet,
(Method::Post, "/volume") => Route::VolumeSet,
(Method::Get, "/playlists") => Route::PlaylistsList,
(Method::Post, "/playlists") => Route::PlaylistCreate,
(Method::Get, path) if path.starts_with("/playlists/") => {
let rest = &path["/playlists/".len()..];
let (name, suffix) = split_playlist_route(rest);
playlist_get_route(path)
}
(Method::Patch, path) if path.starts_with("/playlists/") => {
let rest = strip_prefix("/playlists/", path);
let (name, suffix) = split_playlist_suffix(rest);
match suffix {
"/add" => Route::PlaylistAdd(name),
"/play" => Route::PlaylistPlay(name),
"" => Route::PlaylistGet(name),
"" => Route::PlaylistRename(name),
_ => Route::NotFound,
}
}
(Method::Delete, path) if path.starts_with("/playlists/") => {
let name = path["/playlists/".len()..].to_string();
Route::PlaylistDelete(name)
let rest = strip_prefix("/playlists/", path);
if let Some((name, id)) = split_song_path(rest) {
return Route::PlaylistRemoveSong(name, id);
}
Route::PlaylistDelete(rest.to_string())
}
(Method::Post, path) if path.starts_with("/playlists/") => {
let rest = strip_prefix("/playlists/", path);
let (name, suffix) = split_playlist_suffix(rest);
match suffix {
"/add" => Route::PlaylistAdd(name),
"/play" => Route::PlaylistPlay(name),
"/reorder" => Route::PlaylistReorder(name),
"/duplicate" => Route::PlaylistDuplicate(name),
_ => Route::NotFound,
}
}
_ => Route::NotFound,
}
}
fn split_playlist_route(rest: &str) -> (String, &'static str) {
if let Some(name) = rest.strip_suffix("/add") {
return (name.to_string(), "/add");
fn playlist_get_route(path: &str) -> Route {
let rest = strip_prefix("/playlists/", path);
let (name, suffix) = split_playlist_suffix(rest);
match suffix {
"/add" | "/play" | "/reorder" | "/duplicate" => Route::NotFound,
"" => Route::PlaylistGet(name),
_ => Route::NotFound,
}
if let Some(name) = rest.strip_suffix("/play") {
return (name.to_string(), "/play");
}
fn strip_prefix<'a>(prefix: &str, s: &'a str) -> &'a str {
&s[prefix.len()..]
}
fn split_playlist_suffix(rest: &str) -> (String, &'static str) {
for suf in ["/duplicate", "/reorder", "/add", "/play"] {
if let Some(name) = rest.strip_suffix(suf) {
return (name.to_string(), suf);
}
}
(rest.to_string(), "")
}
fn split_song_path(rest: &str) -> Option<(String, i64)> {
let idx = rest.find("/songs/")?;
let name = &rest[..idx];
let tail = &rest[idx + "/songs/".len()..];
let id: i64 = tail.parse().ok()?;
Some((name.to_string(), id))
}
fn parse_query(url: &str) -> std::collections::HashMap<String, String> {
let mut out = std::collections::HashMap::new();
if let Some(q) = url.split('?').nth(1) {
@@ -305,8 +575,6 @@ fn parse_query(url: &str) -> std::collections::HashMap<String, String> {
out
}
/// Percent-decodes using `url` crate's `percent_encoding` so that UTF-8
/// sequences survive intact.
fn urldecode(s: &str) -> String {
use percent_encoding::percent_decode_str;
percent_decode_str(s)
@@ -350,3 +618,97 @@ fn name_from_body(body: &str) -> Result<String, String> {
}
Err("missing 'name' or 'song' in body".to_string())
}
fn songs_from_body(body: &str) -> Result<Vec<String>, String> {
let v = parse_body(body).map_err(|e| format!("invalid JSON body: {e}"))?;
if let Some(arr) = v.get("songs").and_then(|x| x.as_array()) {
let mut out = Vec::with_capacity(arr.len());
for s in arr {
if let Some(s) = s.as_str() {
if !s.is_empty() {
out.push(s.to_string());
}
}
}
if !out.is_empty() {
return Ok(out);
}
}
Err("missing 'songs' array in body".to_string())
}
fn new_name_from_body(body: &str) -> Result<String, String> {
let v = parse_body(body).map_err(|e| format!("invalid JSON body: {e}"))?;
if let Some(obj) = v.as_object() {
for key in ["name", "new_name"] {
if let Some(s) = obj.get(key).and_then(|x| x.as_str()) {
if !s.is_empty() {
return Ok(s.to_string());
}
}
}
}
Err("missing 'name' in body".to_string())
}
fn secs_from_body(body: &str) -> Result<f64, String> {
let v = parse_body(body).map_err(|e| format!("invalid JSON body: {e}"))?;
if let Some(obj) = v.as_object() {
for key in ["secs", "seconds", "pos", "position"] {
if let Some(s) = obj.get(key) {
if let Some(n) = s.as_f64() {
return Ok(n);
}
if let Some(n) = s.as_i64() {
return Ok(n as f64);
}
}
}
}
Err("missing 'secs' in body".to_string())
}
fn vol_from_body(body: &str) -> Result<f32, String> {
let v = parse_body(body).map_err(|e| format!("invalid JSON body: {e}"))?;
let obj = v.as_object().ok_or_else(|| "expected JSON object".to_string())?;
for key in ["vol", "volume", "v"] {
if let Some(s) = obj.get(key) {
if let Some(n) = s.as_f64() {
return Ok(n as f32);
}
if let Some(n) = s.as_i64() {
return Ok(n as f32);
}
}
}
Err("missing 'volume' in body".to_string())
}
fn mode_from_body_or_default(body: &str) -> Result<RepeatMode, String> {
let v = parse_body(body).map_err(|e| format!("invalid JSON body: {e}"))?;
let obj = v
.as_object()
.ok_or_else(|| "expected JSON object".to_string())?;
for key in ["mode", "name"] {
if let Some(s) = obj.get(key).and_then(|x| x.as_str()) {
return RepeatMode::parse(s).ok_or_else(|| format!("invalid mode: {s}"));
}
}
Err("missing 'mode' in body".to_string())
}
fn reorder_from_body(body: &str) -> Result<(usize, usize), String> {
let v = parse_body(body).map_err(|e| format!("invalid JSON body: {e}"))?;
let obj = v
.as_object()
.ok_or_else(|| "expected JSON object".to_string())?;
let from = obj
.get("from")
.and_then(|x| x.as_i64())
.ok_or_else(|| "missing integer 'from'".to_string())?;
let to = obj
.get("to")
.and_then(|x| x.as_i64())
.ok_or_else(|| "missing integer 'to'".to_string())?;
Ok((from as usize, to as usize))
}
daemon/src/state.rs
+626 -70
View File
@@ -1,10 +1,14 @@
use std::sync::Mutex;
use std::collections::HashMap;
use std::sync::atomic::{AtomicI64, Ordering};
use std::sync::{Arc, Mutex};
use std::time::{SystemTime, UNIX_EPOCH};
use log::{debug, error, info, warn};
use rusqlite::{params, Connection};
use serde::{Deserialize, Serialize};
use strsim::jaro_winkler;
use crate::audio::{probe_duration_secs, AudioHandle};
use crate::paths::song_path;
#[derive(Clone, Debug, Serialize, Deserialize)]
@@ -12,6 +16,11 @@ pub struct Song {
pub id: i64,
pub name: String,
pub path: String,
/// Optional playback duration in seconds (filled in on download probe or
/// before first play). `None` until the daemon has had a chance to look
/// it up.
#[serde(default, skip_serializing_if = "Option::is_none")]
pub duration_secs: Option<f64>,
}
#[derive(Clone, Debug, Serialize, Deserialize)]
@@ -21,6 +30,52 @@ pub struct Playlist {
pub songs: Vec<Song>,
}
/// Repeat behaviour: `Off` is the default; `One` re-queues the same track when
/// it ends; `All` reloads the latest snapshot of songs once the queue empties.
#[derive(Clone, Copy, Debug, Eq, PartialEq, Serialize, Deserialize)]
#[serde(rename_all = "lowercase")]
pub enum RepeatMode {
Off,
One,
All,
}
impl Default for RepeatMode {
fn default() -> Self {
RepeatMode::Off
}
}
impl RepeatMode {
pub fn parse(s: &str) -> Option<Self> {
match s.trim().to_lowercase().as_str() {
"off" | "0" | "false" => Some(RepeatMode::Off),
"one" | "1" | "single" => Some(RepeatMode::One),
"all" | "loop" => Some(RepeatMode::All),
_ => None,
}
}
pub fn as_str(&self) -> &'static str {
match self {
RepeatMode::Off => "off",
RepeatMode::One => "one",
RepeatMode::All => "all",
}
}
}
/// One in-flight or completed yt-dlp job spawned by /init or /init/batch.
#[derive(Clone, Debug, Serialize, Deserialize)]
pub struct DownloadJob {
pub id: i64,
pub name: String,
pub status: String, // "queued" | "running" | "done" | "failed"
pub song_id: Option<i64>,
pub error: Option<String>,
pub started_at: i64,
pub finished_at: Option<i64>,
}
/// The in-memory mirror of the SQLite-backed state. Wrapped in a Mutex by
/// the daemon so HTTP request handlers and the playback loop can share it.
pub struct DaemonState {
@@ -29,18 +84,52 @@ pub struct DaemonState {
pub queue: Vec<Song>,
/// Currently playing song, if any.
pub current: Option<Song>,
/// `Send + Sync` handle to the audio thread (which owns rodio's
/// `OutputStream`, a `!Send` platform type). We never touch rodio
/// directly from HTTP or the playback loop — we enqueue commands.
pub audio_handle: Arc<AudioHandle>,
/// User-selected repeat mode.
pub repeat_mode: RepeatMode,
/// Snapshot taken when entering repeat-all mode. When the queue drains in
/// this mode we restore this snapshot so playback loops without the user
/// having to refill the queue manually.
pub cycle_snapshot: Vec<Song>,
/// `true` once the queue row for `current` has been popped at start of
/// play. Reset to `false` whenever we move on (success, failure, skip,
/// or natural end). Prevents double-removal of the queue row while the
/// audio thread transitions from "pending" to "playing".
pub popped_for_current: bool,
/// In-flight and recently completed download jobs.
pub download_jobs: Mutex<HashMap<i64, DownloadJob>>,
pub next_job_id: AtomicI64,
/// Fuzzy match threshold; configurable so users can tune strictness.
pub fuzzy_threshold: f64,
/// 0 means unlimited.
pub max_queue_size: usize,
}
const FUZZY_THRESHOLD: f64 = 0.65;
impl DaemonState {
pub fn new(conn: Connection) -> Self {
pub fn new(
conn: Connection,
fuzzy_threshold: f64,
max_queue_size: usize,
repeat_mode: RepeatMode,
audio_handle: Arc<AudioHandle>,
) -> Self {
let queue = Self::load_queue_from_db(&conn);
info!("Restored queue with {} song(s) from DB", queue.len());
Self {
conn: Mutex::new(conn),
queue,
current: None,
audio_handle,
repeat_mode,
cycle_snapshot: Vec::new(),
popped_for_current: false,
download_jobs: Mutex::new(HashMap::new()),
next_job_id: AtomicI64::new(1),
fuzzy_threshold,
max_queue_size,
}
}
@@ -60,7 +149,7 @@ impl DaemonState {
fn load_queue_from_db(conn: &Connection) -> Vec<Song> {
let mut stmt = match conn.prepare(
"SELECT s.id, s.name, s.path
"SELECT s.id, s.name, s.path, s.duration_secs
FROM queue q JOIN songs s ON s.id = q.song_id
ORDER BY q.position ASC",
) {
@@ -76,21 +165,39 @@ impl DaemonState {
id: row.get(0)?,
name: row.get(1)?,
path: row.get(2)?,
duration_secs: row.get::<_, Option<f64>>(3).ok().flatten(),
})
})
.map(|it| it.filter_map(Result::ok).collect())
.unwrap_or_default()
}
/// Public wrapper around `all_songs`: returns every song in the DB, sorted
/// by name. Used by `DaemonState::search` (in-memory) and the `/songs`
/// HTTP route.
/// All songs sorted by name (used internally and by `/songs`).
pub fn all_songs(&self) -> Vec<Song> {
Self::list_songs(&self.lock_conn())
Self::list_songs(&self.lock_conn(), None)
}
fn list_songs(conn: &Connection) -> Vec<Song> {
let mut stmt = match conn.prepare("SELECT id, name, path FROM songs ORDER BY name") {
/// Used by `/songs?q=...` to filter without a separate search round-trip.
pub fn list_songs_filtered(&self, q: Option<&str>) -> Vec<Song> {
let pattern = q.map(|q| {
// Escape `%`, `_`, and `\` so users can search for those literally.
q.replace('\\', "\\\\")
.replace('%', "\\%")
.replace('_', "\\_")
});
Self::list_songs(&self.lock_conn(), pattern.as_deref())
}
fn list_songs(conn: &Connection, filter: Option<&str>) -> Vec<Song> {
let (sql, has_filter) = match filter {
Some(_) => (
"SELECT id, name, path, duration_secs FROM songs \
WHERE name LIKE ? ESCAPE '\\' ORDER BY name",
true,
),
None => ("SELECT id, name, path, duration_secs FROM songs ORDER BY name", false),
};
let mut stmt = match conn.prepare(sql) {
Ok(s) => s,
Err(e) => {
error!("Failed to list songs: {e}");
@@ -98,15 +205,22 @@ impl DaemonState {
}
};
stmt.query_map([], |row| {
let map_row = |row: &rusqlite::Row<'_>| -> rusqlite::Result<Song> {
Ok(Song {
id: row.get(0)?,
name: row.get(1)?,
path: row.get(2)?,
duration_secs: row.get::<_, Option<f64>>(3).ok().flatten(),
})
})
.map(|it| it.filter_map(Result::ok).collect())
.unwrap_or_default()
};
let mapped = if has_filter {
stmt.query_map(params![format!("%{}%", filter.unwrap_or(""))], map_row)
} else {
stmt.query_map([], map_row)
};
mapped
.map(|it| it.filter_map(Result::ok).collect())
.unwrap_or_default()
}
fn insert_song(conn: &Connection, name: &str, path: &str) -> rusqlite::Result<i64> {
@@ -122,8 +236,19 @@ impl DaemonState {
Ok(id)
}
fn update_duration_secs(
conn: &Connection,
song_id: i64,
duration_secs: f64,
) -> rusqlite::Result<()> {
conn.execute(
"UPDATE songs SET duration_secs = ?1 WHERE id = ?2",
params![duration_secs, song_id],
)?;
Ok(())
}
fn queue_append_db(conn: &Connection, song_id: i64) -> rusqlite::Result<()> {
// Append at the end. Explicit max+1 so ordering survives deletes.
let next: i64 = conn
.query_row(
"SELECT COALESCE(MAX(position), 0) + 1 FROM queue",
@@ -162,28 +287,38 @@ impl DaemonState {
Ok(())
}
fn write_history(
conn: &Connection,
song_id: i64,
played_at: i64,
duration_secs_played: i64,
) -> rusqlite::Result<()> {
conn.execute(
"INSERT INTO history (song_id, played_at, duration_secs_played) VALUES (?1, ?2, ?3)",
params![song_id, played_at, duration_secs_played],
)?;
Ok(())
}
// ------------------------------------------------------------------
// Public API used by both HTTP server and playback loop
// ------------------------------------------------------------------
/// Search the songs DB by name. Uses an exact (case-insensitive) match
/// first, then falls back to a Jaro-Winkler fuzzy match.
/// Best single match. Exact (case-insensitive) wins, then Jaro-Winkler
/// above threshold.
pub fn search(&self, query: &str) -> Option<Song> {
debug!("Searching DB for: {query}");
let query_lower = query.to_lowercase();
let q = query.to_lowercase();
let songs = self.all_songs();
// Exact match wins outright.
if let Some(s) = songs.iter().find(|s| s.name.to_lowercase() == query_lower) {
if let Some(s) = songs.iter().find(|s| s.name.to_lowercase() == q) {
debug!("Exact DB match: {} (id={})", s.name, s.id);
return Some(s.clone());
}
// Best fuzzy match above threshold.
let best = songs
.iter()
.map(|s| (s.clone(), jaro_winkler(&s.name.to_lowercase(), &query_lower)))
.filter(|(_, score)| *score > FUZZY_THRESHOLD)
.map(|s| (s.clone(), jaro_winkler(&s.name.to_lowercase(), &q)))
.filter(|(_, score)| *score > self.fuzzy_threshold)
.max_by(|a, b| a.1.partial_cmp(&b.1).unwrap_or(std::cmp::Ordering::Equal));
if let Some((song, score)) = best {
@@ -195,24 +330,57 @@ impl DaemonState {
None
}
/// All matches above the fuzzy threshold, sorted by score descending.
pub fn search_all(&self, query: &str) -> Vec<Song> {
let q = query.to_lowercase();
let songs = self.all_songs();
let mut scored: Vec<(Song, f64)> = Vec::new();
for s in songs.iter() {
let s_lower = s.name.to_lowercase();
if s_lower == q {
scored.retain(|x| x.0.id != s.id);
scored.insert(0, (s.clone(), 1.0));
} else {
let sc = jaro_winkler(&s_lower, &q);
if sc > self.fuzzy_threshold {
scored.push((s.clone(), sc));
}
}
}
scored.sort_by(|a, b| b.1.partial_cmp(&a.1).unwrap_or(std::cmp::Ordering::Equal));
scored.into_iter().map(|(s, _)| s).collect()
}
/// Insert a song at the front of the queue and start playing it.
pub fn play(&mut self, query: &str) -> Result<Song, String> {
let song = self
.search(query)
.ok_or_else(|| format!("song not found: {query}"))?;
self.queue.insert(0, song.clone());
let conn = self.lock_conn();
// Reset persisted queue to keep DB and memory consistent.
Self::queue_clear_db(&conn).map_err(|e| e.to_string())?;
for s in &self.queue {
Self::queue_append_db(&conn, s.id).map_err(|e| e.to_string())?;
// DB writes scoped so the lock guard drops before we mutate
// `cycle_snapshot` (which would otherwise re-borrow `self`).
{
let conn = self.lock_conn();
Self::queue_clear_db(&conn).map_err(|e| e.to_string())?;
for s in &self.queue {
Self::queue_append_db(&conn, s.id).map_err(|e| e.to_string())?;
}
}
self.cycle_snapshot = self.queue.clone();
info!("Play: {} (id={})", song.name, song.id);
Ok(song)
}
/// Append a song to the end of the queue.
/// Append a song to the end of the queue. Honours `max_queue_size`.
pub fn add(&mut self, query: &str) -> Result<Song, String> {
if self.max_queue_size > 0 && self.queue.len() >= self.max_queue_size {
return Err(format!(
"queue full ({} of {})",
self.queue.len(),
self.max_queue_size
));
}
let song = self
.search(query)
.ok_or_else(|| format!("song not found: {query}"))?;
@@ -230,7 +398,6 @@ impl DaemonState {
{
let conn = self.lock_conn();
// Foreign keys remove queue & playlist entries automatically.
conn.execute("DELETE FROM songs WHERE id = ?1", params![song.id])
.map_err(|e| e.to_string())?;
}
@@ -251,9 +418,37 @@ impl DaemonState {
Ok(song)
}
/// Download a song via yt-dlp and register it in the DB.
pub fn init(&self, name: String) -> Result<Song, String> {
let path = song_path(&name);
/// In-place shuffle of the current queue.
pub fn shuffle_queue(&mut self) -> Result<(), String> {
if self.queue.len() < 2 {
return Ok(());
}
use rand::seq::SliceRandom;
let mut rng = rand::thread_rng();
self.queue.shuffle(&mut rng);
{
let conn = self.lock_conn();
Self::queue_clear_db(&conn).map_err(|e| e.to_string())?;
for s in &self.queue {
Self::queue_append_db(&conn, s.id).map_err(|e| e.to_string())?;
}
}
// Keep cycle_snapshot in sync so a later repeat-all still loops over
// something sensible.
self.cycle_snapshot = self.queue.clone();
info!("Queue shuffled ({} songs)", self.queue.len());
Ok(())
}
// ------------------------------------------------------------------
// Downloads (synchronous helper used by worker thread)
// ------------------------------------------------------------------
/// Synchronous yt-dlp + DB insert. Runs in a worker thread spawned by
/// `init_async` and does NOT hold the daemon state mutex for the
/// blocking subprocess call.
pub fn init_sync(&self, name: &str) -> Result<Song, String> {
let path = song_path(name);
debug!("yt-dlp target path: {path}");
let output = std::process::Command::new("yt-dlp")
@@ -287,12 +482,20 @@ impl DaemonState {
}
let conn = self.lock_conn();
let id = Self::insert_song(&conn, &name, &path).map_err(|e| e.to_string())?;
info!("Registered song {name} (id={id}) at {path}");
let id = Self::insert_song(&conn, name, &path).map_err(|e| e.to_string())?;
if let Some(secs) = probe_duration_secs(&path) {
if let Err(e) = Self::update_duration_secs(&conn, id, secs) {
warn!("Could not update duration_secs for {name}: {e}");
}
info!("Registered song {name} (id={id}, {secs:.1}s) at {path}");
} else {
info!("Registered song {name} (id={id}) at {path}");
}
Ok(Song {
id,
name,
name: name.to_string(),
path,
duration_secs: None,
})
}
@@ -304,43 +507,40 @@ impl DaemonState {
Ok(())
}
/// Pop the song at the head of the queue as having been played. Removes
/// the same row from the in-memory queue and the `queue` table so that
/// a daemon restart does not put already-played songs back into rotation.
/// The DB is updated first — if the DELETE fails the in-memory queue
/// is left untouched and the error is returned to the caller, keeping
/// memory and DB consistent.
pub fn pop_played(&mut self) -> Result<Option<Song>, String> {
/// Pop the song at the head of the queue. Only writes history when
/// `duration_secs_played` is `Some(_)`. Pass `None` for purely
/// housekeeping pops (start-of-song, /skip, etc).
pub fn pop_played(
&mut self,
duration_secs_played: Option<f64>,
) -> Result<Option<Song>, String> {
if self.queue.is_empty() {
return Ok(None);
}
let id = self.queue[0].id;
// Scope the DB guard so it is dropped before mutating `self.queue`.
// Use a MIN(position) subquery instead of `ORDER BY ... LIMIT 1`:
// SQLite does not accept `ORDER BY` / `LIMIT` clauses on `DELETE`
// statements (parse error: "near \"ORDER\": syntax error"), so the
// standard syntax fails entirely. The subquery is portable across all
// supported SQLite versions and is equivalent: it deletes the head
// occurrence of `song_id` from the queue (oldest by position).
let result = {
let conn = self.lock_conn();
conn.execute(
"DELETE FROM queue
WHERE song_id = ?1
AND position = (
SELECT MIN(position) FROM queue WHERE song_id = ?1
)",
params![id],
)
};
result.map_err(|e| format!("failed to delete queue row from DB: {e}"))?;
let conn = self.lock_conn();
conn.execute(
"DELETE FROM queue
WHERE song_id = ?1
AND position = (
SELECT MIN(position) FROM queue WHERE song_id = ?1
)",
params![id],
)
.map_err(|e| format!("failed to delete queue row from DB: {e}"))?;
if let Some(secs) = duration_secs_played {
let secs_i = (secs.max(0.0)) as i64;
if let Err(e) = Self::write_history(&conn, id, unix_secs(), secs_i) {
warn!("failed to write history for song {id}: {e}");
}
}
drop(conn);
Ok(Some(self.queue.remove(0)))
}
/// Skip the song currently at the front of the queue. Thin wrapper around
/// `pop_played` so the queue-pop SQL lives in exactly one place.
pub fn skip(&mut self) -> Result<Option<Song>, String> {
if let Some(skipped) = self.pop_played()? {
if let Some(skipped) = self.pop_played(None)? {
self.current = None;
info!("Skipped {}", skipped.name);
Ok(Some(skipped))
} else {
@@ -348,6 +548,101 @@ impl DaemonState {
}
}
// ------------------------------------------------------------------
// History / stats
// ------------------------------------------------------------------
pub fn list_history(&self, limit: usize) -> Vec<HistoryEntry> {
let conn = self.lock_conn();
let limit = limit.max(1) as i64;
let mut stmt = match conn.prepare(
"SELECT h.id, h.song_id, s.name, h.played_at, h.duration_secs_played
FROM history h LEFT JOIN songs s ON s.id = h.song_id
ORDER BY h.played_at DESC
LIMIT ?1",
) {
Ok(s) => s,
Err(e) => {
error!("Failed to list history: {e}");
return Vec::new();
}
};
stmt.query_map(params![limit], |row| {
Ok(HistoryEntry {
id: row.get(0)?,
song_id: row.get(1)?,
song_name: row.get(2)?,
played_at: row.get(3)?,
duration_secs_played: row.get(4)?,
})
})
.map(|it| it.filter_map(Result::ok).collect())
.unwrap_or_default()
}
/// Snapshot of the current jobs (most-recently-started first).
pub fn list_downloads(&self) -> Vec<DownloadJob> {
let mut jobs: Vec<DownloadJob> = self
.download_jobs
.lock()
.unwrap_or_else(|p| p.into_inner())
.values()
.cloned()
.collect();
jobs.sort_by(|a, b| b.started_at.cmp(&a.started_at));
jobs
}
pub fn stats(&self) -> Stats {
let conn = self.lock_conn();
let total_plays: i64 = conn
.query_row("SELECT COUNT(*) FROM history", [], |r| r.get(0))
.unwrap_or(0);
let total_secs: i64 = conn
.query_row(
"SELECT COALESCE(SUM(duration_secs_played), 0) FROM history",
[],
|r| r.get(0),
)
.unwrap_or(0);
let mut stmt = match conn.prepare(
"SELECT h.song_id, s.name, COUNT(*) AS plays,
COALESCE(SUM(h.duration_secs_played), 0) AS total_secs
FROM history h LEFT JOIN songs s ON s.id = h.song_id
GROUP BY h.song_id
ORDER BY plays DESC, total_secs DESC
LIMIT 10",
) {
Ok(s) => s,
Err(e) => {
error!("Failed to compute stats: {e}");
return Stats {
total_plays,
total_secs,
top_songs: Vec::new(),
};
}
};
let top_songs: Vec<TopSong> = stmt
.query_map([], |row| {
Ok(TopSong {
song_id: row.get(0)?,
name: row.get(1)?,
plays: row.get(2)?,
total_secs: row.get(3)?,
})
})
.map(|it| it.filter_map(Result::ok).collect())
.unwrap_or_default();
Stats {
total_plays,
total_secs,
top_songs,
}
}
// ------------------------------------------------------------------
// Playlists
// ------------------------------------------------------------------
@@ -378,7 +673,7 @@ impl DaemonState {
fn load_playlist_songs(conn: &Connection, playlist_id: i64) -> Vec<Song> {
let mut stmt = match conn.prepare(
"SELECT s.id, s.name, s.path
"SELECT s.id, s.name, s.path, s.duration_secs
FROM playlist_songs ps JOIN songs s ON s.id = ps.song_id
WHERE ps.playlist_id = ?1
ORDER BY ps.position ASC",
@@ -391,6 +686,7 @@ impl DaemonState {
id: row.get(0)?,
name: row.get(1)?,
path: row.get(2)?,
duration_secs: row.get::<_, Option<f64>>(3).ok().flatten(),
})
})
.map(|it| it.filter_map(Result::ok).collect())
@@ -431,6 +727,158 @@ impl DaemonState {
Ok(())
}
pub fn remove_from_playlist(&self, playlist: &str, song_id: i64) -> Result<Playlist, String> {
let conn = self.lock_conn();
let playlist_id: i64 = conn
.query_row(
"SELECT id FROM playlists WHERE name = ?1",
params![playlist],
|row| row.get(0),
)
.map_err(|_| format!("playlist not found: {playlist}"))?;
let affected = conn
.execute(
"DELETE FROM playlist_songs WHERE playlist_id = ?1 AND song_id = ?2",
params![playlist_id, song_id],
)
.map_err(|e| e.to_string())?;
if affected == 0 {
return Err(format!("song {song_id} not in playlist {playlist}"));
}
Self::compact_playlist_positions(&conn, playlist_id).map_err(|e| e.to_string())?;
info!("Removed song {song_id} from playlist {playlist}");
Ok(Playlist {
id: playlist_id,
name: playlist.to_string(),
songs: Self::load_playlist_songs(&conn, playlist_id),
})
}
pub fn reorder_playlist(
&self,
playlist: &str,
from: usize,
to: usize,
) -> Result<Playlist, String> {
if from == 0 || to == 0 {
return Err("positions are 1-based".to_string());
}
let conn = self.lock_conn();
let playlist_id: i64 = conn
.query_row(
"SELECT id FROM playlists WHERE name = ?1",
params![playlist],
|row| row.get(0),
)
.map_err(|_| format!("playlist not found: {playlist}"))?;
let songs = Self::load_playlist_songs(&conn, playlist_id);
if from > songs.len() || to > songs.len() {
return Err(format!(
"position out of range (playlist has {} songs)",
songs.len()
));
}
let mut order = songs.clone();
let moved = order.remove(from - 1);
order.insert(to - 1, moved);
let tx = conn.unchecked_transaction().map_err(|e| e.to_string())?;
for (idx, s) in order.iter().enumerate() {
tx.execute(
"UPDATE playlist_songs SET position = ?1
WHERE playlist_id = ?2 AND song_id = ?3",
params![idx as i64 + 1, playlist_id, s.id],
)
.map_err(|e| e.to_string())?;
}
tx.commit().map_err(|e| e.to_string())?;
info!("Reordered playlist {playlist}: {from} -> {to}");
Ok(Playlist {
id: playlist_id,
name: playlist.to_string(),
songs: Self::load_playlist_songs(&conn, playlist_id),
})
}
fn compact_playlist_positions(conn: &Connection, playlist_id: i64) -> rusqlite::Result<()> {
let songs = Self::load_playlist_songs(conn, playlist_id);
for (idx, s) in songs.iter().enumerate() {
conn.execute(
"UPDATE playlist_songs SET position = ?1
WHERE playlist_id = ?2 AND song_id = ?3",
params![idx as i64 + 1, playlist_id, s.id],
)?;
}
Ok(())
}
pub fn rename_playlist(&self, old: &str, new: &str) -> Result<Playlist, String> {
let conn = self.lock_conn();
let tx = conn.unchecked_transaction().map_err(|e| e.to_string())?;
let affected = tx
.execute(
"UPDATE playlists SET name = ?1 WHERE name = ?2",
params![new, old],
)
.map_err(|e| e.to_string())?;
if affected == 0 {
tx.rollback().ok();
return Err(format!("playlist not found: {old}"));
}
let playlist_id: i64 = match tx.query_row(
"SELECT id FROM playlists WHERE name = ?1",
params![new],
|row| row.get(0),
) {
Ok(id) => id,
Err(rusqlite::Error::SqliteFailure(err, _))
if err.code == rusqlite::ErrorCode::ConstraintViolation =>
{
tx.rollback().ok();
return Err(format!("playlist name already exists: {new}"));
}
Err(e) => {
tx.rollback().ok();
return Err(e.to_string());
}
};
tx.commit().map_err(|e| e.to_string())?;
info!("Renamed playlist {old} -> {new}");
Ok(Playlist {
id: playlist_id,
name: new.to_string(),
songs: Self::load_playlist_songs(&conn, playlist_id),
})
}
pub fn duplicate_playlist(&self, src: &str, dest: &str) -> Result<Playlist, String> {
let conn = self.lock_conn();
let playlist_id: i64 = match conn.query_row(
"SELECT id FROM playlists WHERE name = ?1",
params![src],
|row| row.get(0),
) {
Ok(id) => id,
Err(_) => return Err(format!("playlist not found: {src}")),
};
let songs = Self::load_playlist_songs(&conn, playlist_id);
let new_id = match self.create_playlist(dest) {
Ok(pl) => pl.id,
Err(e) => return Err(e),
};
for s in &songs {
Self::playlist_append_db(&conn, new_id, s.id).map_err(|e| e.to_string())?;
}
info!("Duplicated playlist {src} -> {dest} ({} songs)", songs.len());
Ok(Playlist {
id: new_id,
name: dest.to_string(),
songs,
})
}
pub fn add_to_playlist(
&self,
playlist: &str,
@@ -457,7 +905,6 @@ impl DaemonState {
Ok((pl, song))
}
/// Load a playlist into the active queue (replacing the current queue).
pub fn play_playlist(&mut self, name: &str) -> Result<Playlist, String> {
let (playlist_id, songs) = {
let conn = self.lock_conn();
@@ -475,8 +922,10 @@ impl DaemonState {
}
(playlist_id, songs)
};
// Guard is dropped; safe to mutate self.queue without borrow conflict.
// DB guard dropped; safe to mutate self.queue / cycle_snapshot
// without re-borrowing self.
self.queue = songs.clone();
self.cycle_snapshot = songs.clone();
info!("Playing playlist {name} ({} songs)", songs.len());
Ok(Playlist {
id: playlist_id,
@@ -501,3 +950,110 @@ impl DaemonState {
})
}
}
#[derive(Clone, Debug, Serialize, Deserialize)]
pub struct HistoryEntry {
pub id: i64,
pub song_id: i64,
pub song_name: Option<String>,
pub played_at: i64,
pub duration_secs_played: i64,
}
#[derive(Clone, Debug, Serialize, Deserialize)]
pub struct Stats {
pub total_plays: i64,
pub total_secs: i64,
pub top_songs: Vec<TopSong>,
}
#[derive(Clone, Debug, Serialize, Deserialize)]
pub struct TopSong {
pub song_id: i64,
pub name: Option<String>,
pub plays: i64,
pub total_secs: i64,
}
// --------------------------------------------------------------------
// Free helpers / async download dispatch (outside `impl DaemonState`
// because `&Arc<Mutex<Self>>` is not a valid method-receiver type).
// --------------------------------------------------------------------
fn unix_secs() -> i64 {
SystemTime::now()
.duration_since(UNIX_EPOCH)
.map(|d| d.as_secs() as i64)
.unwrap_or(0)
}
/// Spawn a worker thread that downloads a song and updates the matching
/// `DownloadJob`. Returns the assigned job id synchronously.
pub fn init_async(state: &Arc<Mutex<DaemonState>>, name: String) -> i64 {
let job_id = state
.lock()
.unwrap()
.next_job_id
.fetch_add(1, Ordering::SeqCst);
let now = unix_secs();
let job = DownloadJob {
id: job_id,
name: name.clone(),
status: "queued".to_string(),
song_id: None,
error: None,
started_at: now,
finished_at: None,
};
// Scope the locks so both MutexGuards drop together at the end of the
// block. A temporary `state.lock()` at statement boundary would be
// dropped before the inner `download_jobs.lock()` finishes using it.
{
let guard = state.lock().unwrap();
let mut jobs = guard.download_jobs.lock().unwrap();
jobs.insert(job_id, job);
}
// Clone the name so the closure can take ownership while we still print
// the original in the debug log below.
let name_for_worker = name.clone();
let arc = Arc::clone(state);
std::thread::Builder::new()
.name(format!("sjnmusic-dl-{job_id}"))
.spawn(move || run_download_job(arc, job_id, name_for_worker))
.expect("failed to spawn download worker");
debug!("Spawned download job {job_id} for {name}");
job_id
}
/// Spawn one worker per name. Returns the assigned ids in the same order.
pub fn init_batch(state: &Arc<Mutex<DaemonState>>, names: Vec<String>) -> Vec<i64> {
names.iter().map(|n| init_async(state, n.clone())).collect()
}
fn run_download_job(state: Arc<Mutex<DaemonState>>, job_id: i64, name: String) {
{
let guard = state.lock().unwrap();
let mut jobs = guard.download_jobs.lock().unwrap();
if let Some(j) = jobs.get_mut(&job_id) {
j.status = "running".to_string();
}
}
let result = state.lock().unwrap().init_sync(&name);
{
let guard = state.lock().unwrap();
let mut jobs = guard.download_jobs.lock().unwrap();
if let Some(j) = jobs.get_mut(&job_id) {
j.finished_at = Some(unix_secs());
match &result {
Ok(song) => {
j.status = "done".to_string();
j.song_id = Some(song.id);
}
Err(e) => {
j.status = "failed".to_string();
j.error = Some(e.clone());
}
}
}
}
}