From 5e14a9e30884e44c57dec2a260bbf946f81fbfd2 Mon Sep 17 00:00:00 2001 From: "jonas h." Date: Wed, 24 Jun 2026 20:02:07 +0200 Subject: [PATCH] add a bunch of qol features --- cli/sjnmusic | 202 +++++++++++-- daemon/Cargo.lock | 40 +++ daemon/Cargo.toml | 1 + daemon/src/audio.rs | 353 ++++++++++++++++++++-- daemon/src/config.rs | 49 +++ daemon/src/daemon.rs | 208 +++++++++---- daemon/src/db.rs | 46 ++- daemon/src/main.rs | 3 +- daemon/src/paths.rs | 77 ++++- daemon/src/server.rs | 452 +++++++++++++++++++++++++--- daemon/src/state.rs | 696 ++++++++++++++++++++++++++++++++++++++----- 11 files changed, 1881 insertions(+), 246 deletions(-) diff --git a/cli/sjnmusic b/cli/sjnmusic index df90e3f..4d9c791 100755 --- a/cli/sjnmusic +++ b/cli/sjnmusic @@ -9,6 +9,7 @@ import argparse import json import os import sys +import time import urllib.error import urllib.parse import urllib.request @@ -39,17 +40,34 @@ def build_parser() -> argparse.ArgumentParser: add_search_fields(sub.add_parser("play", help="search & start playing immediately")) add_search_fields(sub.add_parser("add", help="search & append to queue")) add_search_fields(sub.add_parser("del", help="search & delete from disk+db")) - add_search_fields(sub.add_parser("init", help="download via yt-dlp and register")) - add_search_fields(sub.add_parser("search", help="search the song database")) + add_search_fields(sub.add_parser("init", help="download via yt-dlp and register (async)")) + add_search_fields(sub.add_parser("search", help="search the song database (best match)")) + add_search_fields(sub.add_parser("search-all", help="search every match above threshold")) sub.add_parser("queue", help="show the current queue") sub.add_parser("now", help="alias for /now-playing") + sub.add_parser("status", help="one-line summary of current playback") sub.add_parser("skip", help="skip the next song") + sub.add_parser("pause", help="pause playback") + sub.add_parser("resume", help="resume playback") + sub.add_parser("shuffle", help="shuffle the queue in place") sub.add_parser("clear", help="empty the queue") sub.add_parser("songs", help="list all known songs") sub.add_parser("playlists", help="list all playlists") + sub.add_parser("downloads", help="list recent yt-dlp download jobs") + sub.add_parser("history", help="show recently played songs") + sub.add_parser("stats", help="aggregate stats from history") sub.add_parser("help", help="list HTTP endpoints") + vol = sub.add_parser("volume", help="get/set volume (0-100 or floating 0.0-2.0)") + vol.add_argument("value", nargs="?", type=float, help="0-100 (interpreted as percent)") + + seek = sub.add_parser("seek", help="jump to position within the current track") + seek.add_argument("secs", type=float, help="position in seconds") + + repeat = sub.add_parser("repeat", help="get/set repeat mode (off/one/all)") + repeat.add_argument("mode", nargs="?", help="off, one, all") + pl_new = sub.add_parser("pl-new", help="create a playlist") pl_new.add_argument("name") @@ -67,6 +85,26 @@ def build_parser() -> argparse.ArgumentParser: pl_rm = sub.add_parser("pl-rm", help="delete a playlist") pl_rm.add_argument("name") + pl_rm_song = sub.add_parser("pl-rm-song", help="remove a song from a playlist") + pl_rm_song.add_argument("name") + pl_rm_song.add_argument("song_id", type=int) + + pl_rename = sub.add_parser("pl-rename", help="rename a playlist") + pl_rename.add_argument("old") + pl_rename.add_argument("new") + + pl_dup = sub.add_parser("pl-dup", help="duplicate a playlist under a new name") + pl_dup.add_argument("src") + pl_dup.add_argument("dest") + + pl_reorder = sub.add_parser("pl-reorder", help="reorder songs in a playlist") + pl_reorder.add_argument("name") + pl_reorder.add_argument("from_pos", type=int) + pl_reorder.add_argument("to_pos", type=int) + + init_batch = sub.add_parser("init-batch", help="batch-download several songs at once") + init_batch.add_argument("queries", nargs="+") + return p @@ -103,20 +141,91 @@ def call(host, port, method, route, body=None, query=None): sys.exit(2) +def fmt_secs(s): + if s is None: + return "--:--" + try: + s = float(s) + except (TypeError, ValueError): + return "--:--" + if s < 0: + return "--:--" + m, sec = divmod(int(s), 60) + return f"{m:d}:{sec:02d}" + + +def try_parse(payload): + try: + return json.loads(payload) + except json.JSONDecodeError: + return None + + def print_response(args, status, payload): if args.json: print(f"HTTP {status}\n{payload}") return - try: - obj = json.loads(payload) - except json.JSONDecodeError: + obj = try_parse(payload) + if obj is None: print(f"HTTP {status}: {payload}") return if isinstance(obj, dict) and "error" in obj: print(f"error: {obj['error']}", file=sys.stderr) sys.exit(1) - if "queue" in obj and "current" in obj: - print("queue:") + # Specialised formatters used by status / history / stats / downloads. + if ( + isinstance(obj, dict) + and "current" in obj + and "queue_len" in obj + and "elapsed_secs" in obj + ): + cur = obj.get("current") or {} + name = cur.get("name") or "(idle)" + elapsed = fmt_secs(obj.get("elapsed_secs")) + total = fmt_secs(obj.get("duration_secs")) + print( + f"[{cur.get('id', '-')}] {name} {elapsed}/{total} " + f"q={obj.get('queue_len', 0)} " + f"vol={obj.get('volume', 0):.2f} " + f"paused={obj.get('paused', False)} " + f"repeat={obj.get('repeat', 'off')}" + ) + return + if isinstance(obj, dict) and "history" in obj: + for h in obj["history"]: + ts = h.get("played_at", 0) + when = time.strftime("%Y-%m-%d %H:%M:%S", time.localtime(ts)) + print( + f"{when} [{h.get('song_id', '-')}] {h.get('song_name') or '(missing)'} " + f"{fmt_secs(h.get('duration_secs_played', 0))}" + ) + return + if isinstance(obj, dict) and "top_songs" in obj: + print(f"total plays: {obj.get('total_plays', 0)}") + print(f"total time: {fmt_secs(obj.get('total_secs', 0))}") + for t in obj.get("top_songs", []): + print( + f" - [{t.get('song_id', '-')}] {t.get('name') or '(missing)'} " + f"plays={t.get('plays', 0)} time={fmt_secs(t.get('total_secs', 0))}" + ) + return + if isinstance(obj, dict) and "downloads" in obj: + for d in obj["downloads"]: + extras = [] + if d.get("song_id") is not None: + extras.append(f"song={d['song_id']}") + if d.get("error"): + extras.append(f"err={d['error']}") + extra_str = (" " + " ".join(extras)) if extras else "" + print(f"[{d.get('id', '-')}] {d.get('status', '?')} {d.get('name', '?')}{extra_str}") + return + if isinstance(obj, dict) and "matches" in obj: + for s in obj["matches"]: + print(f"[{s['id']}] {s['name']}") + return + if isinstance(obj, dict) and "queue" in obj and "current" in obj: + repeat = obj.get("repeat", "off") + print(f"queue (repeat={repeat}):") for s in obj.get("queue", []): print(f" - [{s['id']}] {s['name']}") cur = obj.get("current") @@ -125,15 +234,7 @@ def print_response(args, status, payload): else: print("now playing: (idle)") return - if "current" in obj and "queue_len" in obj: - cur = obj.get("current") - if cur: - print(f"now playing: [{cur['id']}] {cur['name']}") - else: - print("now playing: (idle)") - print(f"queue length: {obj['queue_len']}") - return - if "songs" in obj: + if isinstance(obj, dict) and "songs" in obj: for s in obj["songs"]: print(f"[{s['id']}] {s['name']}") return @@ -174,6 +275,9 @@ def main(argv=None): body.setdefault("query", q) return body + def pl_route(name): + return f"/playlists/{urllib.parse.quote(name)}" + if args.cmd == "play": status, payload = call(host, port, "POST", "/play", body=with_q(None)) elif args.cmd == "add": @@ -188,21 +292,58 @@ def main(argv=None): print("error: 'init' requires a song name", file=sys.stderr) sys.exit(2) status, payload = call(host, port, "POST", "/init", body={"song": q}) + elif args.cmd == "init-batch": + status, payload = call( + host, port, "POST", "/init/batch", body={"songs": args.queries} + ) elif args.cmd == "search": if not q: print("error: 'search' requires a query", file=sys.stderr) sys.exit(2) status, payload = call(host, port, "GET", "/search", query={"q": q}) + elif args.cmd == "search-all": + if not q: + print("error: 'search-all' requires a query", file=sys.stderr) + sys.exit(2) + status, payload = call(host, port, "GET", "/search/all", query={"q": q}) elif args.cmd == "queue": status, payload = call(host, port, "GET", "/queue") - elif args.cmd == "now": + elif args.cmd in ("now", "status"): status, payload = call(host, port, "GET", "/now-playing") elif args.cmd == "skip": status, payload = call(host, port, "POST", "/skip") + elif args.cmd == "pause": + status, payload = call(host, port, "POST", "/pause") + elif args.cmd == "resume": + status, payload = call(host, port, "POST", "/resume") + elif args.cmd == "seek": + status, payload = call(host, port, "POST", "/seek", body={"secs": args.secs}) + elif args.cmd == "shuffle": + status, payload = call(host, port, "POST", "/queue/shuffle") elif args.cmd == "clear": status, payload = call(host, port, "POST", "/queue/clear") elif args.cmd == "songs": status, payload = call(host, port, "GET", "/songs") + elif args.cmd == "downloads": + status, payload = call(host, port, "GET", "/downloads") + elif args.cmd == "history": + status, payload = call(host, port, "GET", "/history") + elif args.cmd == "stats": + status, payload = call(host, port, "GET", "/stats") + elif args.cmd == "volume": + if args.value is None: + status, payload = call(host, port, "GET", "/volume") + else: + # Treat 0..100 as a percentage of rodio's default 1.0. + v = args.value + if v > 2.0: + v = v / 100.0 + status, payload = call(host, port, "POST", "/volume", body={"vol": v}) + elif args.cmd == "repeat": + if args.mode is None: + status, payload = call(host, port, "GET", "/repeat") + else: + status, payload = call(host, port, "POST", "/repeat", body={"mode": args.mode}) elif args.cmd == "playlists": status, payload = call(host, port, "GET", "/playlists") elif args.cmd == "help": @@ -210,20 +351,35 @@ def main(argv=None): elif args.cmd == "pl-new": status, payload = call(host, port, "POST", "/playlists", body={"name": args.name}) elif args.cmd == "pl-show": - status, payload = call(host, port, "GET", f"/playlists/{urllib.parse.quote(args.name)}") + status, payload = call(host, port, "GET", pl_route(args.name)) elif args.cmd == "pl-add": if not q: print("error: 'pl-add' requires a song query", file=sys.stderr) sys.exit(2) status, payload = call( - host, port, "POST", - f"/playlists/{urllib.parse.quote(args.name)}/add", - body={"query": q}, + host, port, "POST", f"{pl_route(args.name)}/add", body={"query": q} ) elif args.cmd == "pl-play": - status, payload = call(host, port, "POST", f"/playlists/{urllib.parse.quote(args.name)}/play") + status, payload = call(host, port, "POST", f"{pl_route(args.name)}/play") elif args.cmd == "pl-rm": - status, payload = call(host, port, "DELETE", f"/playlists/{urllib.parse.quote(args.name)}") + status, payload = call(host, port, "DELETE", pl_route(args.name)) + elif args.cmd == "pl-rm-song": + status, payload = call( + host, port, "DELETE", f"{pl_route(args.name)}/songs/{args.song_id}" + ) + elif args.cmd == "pl-rename": + status, payload = call( + host, port, "PATCH", pl_route(args.old), body={"name": args.new} + ) + elif args.cmd == "pl-dup": + status, payload = call( + host, port, "POST", f"{pl_route(args.src)}/duplicate", body={"name": args.dest} + ) + elif args.cmd == "pl-reorder": + status, payload = call( + host, port, "POST", f"{pl_route(args.name)}/reorder", + body={"from": args.from_pos, "to": args.to_pos}, + ) else: status, payload = call(host, port, "GET", "/help") diff --git a/daemon/Cargo.lock b/daemon/Cargo.lock index 951b6f1..140c1f3 100644 --- a/daemon/Cargo.lock +++ b/daemon/Cargo.lock @@ -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", diff --git a/daemon/Cargo.toml b/daemon/Cargo.toml index 686e49f..2b6153d 100644 --- a/daemon/Cargo.toml +++ b/daemon/Cargo.toml @@ -15,3 +15,4 @@ log = "0.4" env_logger = "0.11" toml = "0.8" percent-encoding = "2" +rand = "0.8" diff --git a/daemon/src/audio.rs b/daemon/src/audio.rs index 3046a37..a63edd4 100644 --- a/daemon/src/audio.rs +++ b/daemon/src/audio.rs @@ -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` for commands (Play, Pause, Resume, ...) +//! - `Arc>` 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, - _stream: Option, +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, + /// Source-reported duration in seconds; `None` when unknown. + pub duration_secs: Option, + /// `Instant` of the most recent play-start or seek reset. + pub play_started_at: Option, + /// 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, + /// `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, + status: Arc>, +} + +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 { + 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, status: Arc>) { + let mut sink: Option = None; + let mut _stream: Option = 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>) -> bool { + status.lock().unwrap().paused +} + +fn status_snapshot_path(status: &Arc>) -> Option { + status.lock().unwrap().current_path.clone() +} diff --git a/daemon/src/config.rs b/daemon/src/config.rs index b4949cb..3a6d4c0 100644 --- a/daemon/src/config.rs +++ b/daemon/src/config.rs @@ -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, + /// 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 { diff --git a/daemon/src/daemon.rs b/daemon/src/daemon.rs index a964f20..1d164b6 100644 --- a/daemon/src/daemon.rs +++ b/daemon/src/daemon.rs @@ -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>) { + 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(()) +} diff --git a/daemon/src/db.rs b/daemon/src/db.rs index a10d097..8bde4e9 100644 --- a/daemon/src/db.rs +++ b/daemon/src/db.rs @@ -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: " 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}"); + } + } + } +} diff --git a/daemon/src/main.rs b/daemon/src/main.rs index fd10c19..3204392 100644 --- a/daemon/src/main.rs +++ b/daemon/src/main.rs @@ -17,7 +17,8 @@ fn main() { log::info!("Loaded configuration"); log::debug!("Config: {:?}", cfg); - paths::ensure_dirs(); + paths::initialize(&cfg); + daemon::run(cfg); } diff --git a/daemon/src/paths.rs b/daemon/src/paths.rs index 29ad075..211d06c 100644 --- a/daemon/src/paths.rs +++ b/daemon/src/paths.rs @@ -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 = OnceLock::new(); +static SONGS: OnceLock = 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); } diff --git a/daemon/src/server.rs b/daemon/src/server.rs index f955f66..856b4a5 100644 --- a/daemon/src/server.rs +++ b/daemon/src/server.rs @@ -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>, 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 { + let mut s = String::new(); + let _ = request.as_reader().read_to_string(&mut s)?; + Ok(s) +} + fn handle( mut request: Request, state: Arc>, @@ -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>> { + 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>> { + 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 { 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 { 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 { } Err("missing 'name' or 'song' in body".to_string()) } + +fn songs_from_body(body: &str) -> Result, 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 { + 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 { + 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 { + 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 { + 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)) +} diff --git a/daemon/src/state.rs b/daemon/src/state.rs index ffd321e..82797f1 100644 --- a/daemon/src/state.rs +++ b/daemon/src/state.rs @@ -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, } #[derive(Clone, Debug, Serialize, Deserialize)] @@ -21,6 +30,52 @@ pub struct Playlist { pub songs: Vec, } +/// 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 { + 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, + pub error: Option, + pub started_at: i64, + pub finished_at: Option, +} + /// 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, /// Currently playing song, if any. pub current: Option, + /// `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, + /// 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, + /// `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>, + 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, + ) -> 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 { 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>(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 { - Self::list_songs(&self.lock_conn()) + Self::list_songs(&self.lock_conn(), None) } - fn list_songs(conn: &Connection) -> Vec { - 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 { + 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 { + 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 { Ok(Song { id: row.get(0)?, name: row.get(1)?, path: row.get(2)?, + duration_secs: row.get::<_, Option>(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 { @@ -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 { - 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 { + 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 { 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 { + 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 { - 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 { + 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, 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, + ) -> Result, 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, 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 { + 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 { + let mut jobs: Vec = 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 = 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 { 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>(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 { + 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 { + 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 { + 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 { + 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 { 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, + 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, +} + +#[derive(Clone, Debug, Serialize, Deserialize)] +pub struct TopSong { + pub song_id: i64, + pub name: Option, + pub plays: i64, + pub total_secs: i64, +} + +// -------------------------------------------------------------------- +// Free helpers / async download dispatch (outside `impl DaemonState` +// because `&Arc>` 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>, 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>, names: Vec) -> Vec { + names.iter().map(|n| init_async(state, n.clone())).collect() +} + +fn run_download_job(state: Arc>, 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()); + } + } + } + } +}