Something went wrong. Try again.
experiments of a tiny cytube-like player with yt-dlp
Something went wrong. Try again.
1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556575859606162636465666768697071727374757677787980818283848586878889909192939495969798991001011021031041051061071081091101111121131141151161171181191201211221231241251261271281291301311321331341351361371381391401411421431441451461471481491501511521531541551561571581591601611621631641651661671681691701711721731741751761771781791801811821831841851861871881891901911921931941951961971981992002012022032042052062072082092102112122132142152162172182192202212222232242252262272282292302312322332342352362372382392402412422432442452462472482492502512522532542552562572582592602612622632642652662672682692702712722732742752762772782792802812822832842852862872882892902912922932942952962972982993003013023033043053063073083093103113123133143153163173183193203213223233243253263273283293303313323333343353363373383393403413423433443453463473483493503513523533543553563573583593603613623633643653663673683693703713723733743753763773783793803813823833843853863873883893903913923933943953963973983994004014024034044054064074084094104114124134144154164174184194204214224234244254264274284294304314324334344354364374384394404414424434444454464474484494504514524534544554564574584594604614624634644654664674684694704714724734744754764774784794804814824834844854864874884894904914924934944954964974984995005015025035045055065075085095105115125135145155165175185195205215225235245255265275285295305315325335345355365375385395405415425435445455465475485495505515525535545555565575585595605615625635645655665675685695705715725735745755765775785795805815825835845855865875885895905915925935945955965975985996006016026036046056066076086096106116126136146156166176186196206216226236246256266276286296306316326336346356366376386396406416426436446456466476486496506516526536546556566576586596606616626636646656666676686696706716726736746756766776786796806816826836846856866876886896906916926936946956966976986997007017027037047057067077087097107117127137147157167177187197207217227237247257267277287297307317327337347357367377387397407417427437447457467477487497507517527537547557567577587597607617627637647657667677687697707717727737747757767777787797807817827837847857867877887897907917927937947957967977987998008018028038048058068078088098108118128138148158168178188198208218228238248258268278288298308318328338348358368378388398408418428438448458468478488498508518528538548558568578588598608618628638648658668678688698708718728738748758768778788798808818828838848858868878888898908918928938948958968978988999009019029039049059069079089099109119129139149159169179189199209219229239249259269279289299309319329339349359369379389399409419429439449459469479489499509519529539549559569579589599609619629639649659669679689699709719729739749759769779789799809819829839849859869879889899909919929939949959969979989991000100110021003100410051006100710081009101010111012101310141015101610171018101910201021102210231024102510261027102810291030103110321033103410351036103710381039104010411042104310441045104610471048104910501051105210531054105510561057105810591060106110621063106410651066106710681069107010711072107310741075107610771078107910801081108210831084108510861087108810891090109110921093109410951096109710981099110011011102110311041105110611071108110911101111111211131114111511161117111811191120112111221123112411251126112711281129113011311132113311341135113611371138113911401141114211431144114511461147114811491150115111521153115411551156115711581159116011611162116311641165116611671168116911701171117211731174117511761177117811791180118111821183118411851186118711881189119011911192119311941195119611971198119912001201120212031204120512061207120812091210121112121213121412151216121712181219122012211222122312241225122612271228122912301231123212331234123512361237123812391240124112421243124412451246124712481249125012511252125312541255125612571258125912601261126212631264126512661267126812691270127112721273127412751276127712781279128012811282128312841285128612871288128912901291129212931294129512961297129812991300130113021303130413051306130713081309131013111312131313141315131613171318131913201321132213231324132513261327132813291330133113321333133413351336133713381339134013411342134313441345134613471348134913501351135213531354135513561357135813591360136113621363136413651366136713681369137013711372137313741375137613771378137913801381138213831384138513861387138813891390139113921393139413951396139713981399140014011402140314041405140614071408140914101411141214131414141514161417141814191420142114221423142414251426142714281429143014311432143314341435143614371438143914401441144214431444144514461447144814491450145114521453145414551456145714581459146014611462146314641465146614671468146914701471147214731474147514761477147814791480148114821483148414851486148714881489149014911492149314941495149614971498149915001501150215031504150515061507150815091510151115121513151415151516151715181519152015211522152315241525152615271528152915301531153215331534153515361537153815391540154115421543154415451546154715481549155015511552155315541555155615571558155915601561156215631564156515661567156815691570157115721573157415751576157715781579158015811582158315841585158615871588158915901591159215931594159515961597159815991600160116021603160416051606160716081609161016111612161316141615161616171618161916201621162216231624162516261627162816291630163116321633163416351636163716381639164016411642164316441645164616471648164916501651165216531654165516561657165816591660166116621663166416651666166716681669167016711672167316741675167616771678167916801681168216831684168516861687168816891690169116921693169416951696169716981699170017011702170317041705170617071708170917101711171217131714171517161717171817191720172117221723172417251726172717281729173017311732173317341735173617371738173917401741174217431744174517461747174817491750175117521753175417551756175717581759176017611762176317641765176617671768176917701771177217731774177517761777177817791780178117821783178417851786178717881789179017911792179317941795179617971798179918001801180218031804180518061807180818091810181118121813181418151816181718181819182018211822182318241825182618271828182918301831183218331834183518361837183818391840184118421843184418451846184718481849185018511852185318541855185618571858185918601861186218631864186518661867186818691870187118721873187418751876187718781879188018811882188318841885188618871888188918901891189218931894189518961897189818991900190119021903190419051906190719081909191019111912191319141915191619171918191919201921192219231924192519261927192819291930193119321933193419351936193719381939194019411942194319441945194619471948194919501951195219531954195519561957195819591960196119621963196419651966196719681969197019711972197319741975197619771978197919801981198219831984198519861987198819891990199119921993199419951996199719981999200020012002200320042005200620072008200920102011201220132014//! Persistence abstraction for room data.//!//! Defines the [`RoomStore`] trait that encapsulates all SQLite operations//! behind a clean interface, along with an in-memory implementation for//! testing and a SQLite-backed implementation for production.//!//! The room actor never touches SQL directly — all persistence flows through//! a [`RoomStore`] implementation.
use std::collections::HashMap;use std::sync::Mutex;use std::time::Duration;
use async_trait::async_trait;use sqlx::{Pool, Sqlite};
use crate::state::{Effect, FinishedTrack, QueuedTrack};use crate::types::{ChatMessage, ChatMessageId, PlaylistEntryId, TrackId};
/// An entry in a room's perpetual playlist — used for auto-fill when the/// queue runs dry. Metadata is pre-resolved via the [`MEDIA_CACHE`] table so/// auto-played tracks never show a "Loading..." placeholder.////// The `title`, `duration`, `thumbnail`, `source`, and `pending` fields are/// populated at load time via a JOIN with `media_cache`; they are NOT stored/// independently in the `playlist_entries` table.#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize)]pub(crate) struct PlaylistEntry { pub id: PlaylistEntryId, pub url: String, pub title: String, pub duration: String, pub thumbnail: Option<String>, pub source: String, pub added_by: Option<String>, pub added_at: String, /// Derived from `media_cache.state` at load time. pub pending: bool, /// `Some(reason)` if the URL failed to resolve. Mutually exclusive with /// `pending: true` — a failed entry has `pending: false` so the UI can /// distinguish it from a still-loading placeholder. The reason is the /// same string the state machine wrote into `media_cache.title` as /// `"Failed to load: <reason>"`; the prefix is stripped during snapshot /// load so consumers see the raw reason. pub failed: Option<String>,}
/// Full snapshot of a room's state loaded from the store.////// Fields are split into the four logical partitions: queue (not yet playing),/// active (currently playing), history (finished), and chat messages.#[derive(Debug, Clone)]#[allow(dead_code)]pub(crate) struct RoomSnapshot { pub room_id: String, pub room_name: String, /// Tracks waiting to play (not started, not finished). pub queue: Vec<QueuedTrack>, /// The track currently playing, if any. pub active: Option<QueuedTrack>, /// Recently finished tracks (newest first). pub history: Vec<FinishedTrack>, /// Pre-resolved perpetual playlist entries. pub playlist: Vec<PlaylistEntry>, pub chat_messages: Vec<ChatMessage>, /// Room settings: when true, idle rooms auto-fill from the /// playlist (also drives the playlist panel visibility in the UI). pub autoplay_enabled: bool, /// Skip-vote threshold as a percentage of room size (0..=100). pub skip_threshold: u8,}
/// Output returned by [`RoomStore::persist`].////// Carries information the caller needs to react to a persist batch.#[derive(Debug, Clone, Default)]pub(crate) struct PersistOutput { /// Set when a [`Effect::PersistChat`] was applied. pub chat_message_id: Option<ChatMessageId>,}
/// Abstraction over room persistence.////// All implementors must be [`Send`] + [`Sync`] so they can be shared across/// async tasks and protected behind an [`Arc`](std::sync::Arc).#[async_trait]pub(crate) trait RoomStore: Send + Sync { /// Create a new room row. async fn create_room(&self, id: &str, name: &str) -> Result<(), sqlx::Error>;
/// Apply a batch of effects atomically. /// /// Effects that are not persistence-related (e.g. [`Effect::AbortPipeline`], /// [`Effect::PublishSnapshot`]) are silently ignored. async fn persist( &self, room_id: &str, effects: &[Effect], ) -> Result<PersistOutput, sqlx::Error>;
/// Load the full snapshot for a room, or [`None`] if it does not exist. async fn load(&self, room_id: &str) -> Result<Option<RoomSnapshot>, sqlx::Error>;
/// Look up a room id by its human-readable name. Returns `None` /// when the name is unknown. Used by URL routing: rooms are /// addressed by name in URLs and resolved to ids for storage. #[allow(dead_code)] async fn name_to_id(&self, name: &str) -> Result<Option<String>, sqlx::Error>;
/// Update the room's settings (autoplay toggle + skip threshold). /// Both fields are written in a single UPDATE so the row is never /// observed in a half-applied state. #[allow(dead_code)] async fn update_room_settings( &self, room_id: &str, autoplay_enabled: bool, skip_threshold: u8, ) -> Result<(), sqlx::Error>;
/// Retrieve the most recent chat messages for a room, newest first. #[allow(dead_code)] async fn recent_chat(&self, room_id: &str, limit: i64) -> Result<Vec<ChatMessage>, sqlx::Error>;
/// Delete a room and all its associated data (CASCADE). #[allow(dead_code)] async fn delete_room(&self, room_id: &str) -> Result<(), sqlx::Error>;}
/// In-memory variant of [`PlaylistEntry`] that carries extra fields needed/// for round-trip persistence but not exposed via the public type.#[cfg_attr(not(test), allow(dead_code))]#[derive(Debug, Clone)]struct InMemoryPlaylistEntry { id: PlaylistEntryId, url: String, title: String, duration: String, thumbnail: Option<String>, source: String, added_by: Option<String>, added_at: String, pending: bool, /// Newly-pushed entries are never failed; the field is set later by /// state-machine effects that mutate the playlist in place. Kept in the /// in-memory struct so the public `PlaylistEntry` shape is preserved on /// snapshot. failed: Option<String>,}
impl From<InMemoryPlaylistEntry> for PlaylistEntry { fn from(e: InMemoryPlaylistEntry) -> Self { PlaylistEntry { id: e.id, url: e.url, title: e.title, duration: e.duration, thumbnail: e.thumbnail, source: e.source, added_by: e.added_by, added_at: e.added_at, pending: e.pending, failed: e.failed, } }}
#[cfg_attr(not(test), allow(dead_code))]#[derive(Debug)]struct InMemoryRoom { name: String, queue: Vec<QueuedTrack>, active: Option<QueuedTrack>, history: Vec<FinishedTrack>, playlist: Vec<InMemoryPlaylistEntry>, chat_messages: Vec<ChatMessage>, autoplay_enabled: bool, skip_threshold: u8,}
/// Thread-safe in-memory store backed by a [`Mutex`]-protected [`HashMap`].////// Implements [`RoomStore`] without any external dependencies, making it/// suitable for unit tests and deterministic simulations.#[cfg_attr(not(test), allow(dead_code))]#[derive(Debug)]pub(crate) struct InMemoryStore { inner: Mutex<HashMap<String, InMemoryRoom>>,}
#[cfg_attr(not(test), allow(dead_code))]impl InMemoryStore { pub(crate) fn new() -> Self { Self { inner: Mutex::new(HashMap::new()), } }}
impl Default for InMemoryStore { fn default() -> Self { Self::new() }}
#[async_trait]impl RoomStore for InMemoryStore { async fn create_room(&self, id: &str, name: &str) -> Result<(), sqlx::Error> { let mut inner = self.inner.lock().expect("InMemoryStore mutex poisoned"); inner.insert( id.to_string(), InMemoryRoom { name: name.to_string(), queue: Vec::new(), active: None, history: Vec::new(), playlist: Vec::new(), chat_messages: Vec::new(), autoplay_enabled: true, skip_threshold: 34, }, ); Ok(()) }
async fn persist( &self, room_id: &str, effects: &[Effect], ) -> Result<PersistOutput, sqlx::Error> { let mut inner = self.inner.lock().expect("InMemoryStore mutex poisoned"); let room = match inner.get_mut(room_id) { Some(room) => room, None => return Ok(PersistOutput::default()), };
let mut output = PersistOutput::default();
for effect in effects { match effect { Effect::PersistQueuedTrack(track) => { room.queue.push(track.clone()); } Effect::PersistStarted(id) => { if let Some(pos) = room.queue.iter().position(|t| t.id == *id) { let track = room.queue.remove(pos); room.active = Some(track); } } Effect::PersistFinished(id) => { if let Some(active) = room.active.take() { if active.id == *id { let now = crate::util::now_iso(); room.history.insert( 0, FinishedTrack { id: active.id, title: active.title, url: active.url, duration: active.duration, thumbnail: active.thumbnail, played_at: now, }, ); } else { // ID mismatch — put the active track back. room.active = Some(active); } } } Effect::PersistMetadata { item_id, url: _, title, duration, thumbnail, source: _, } => { if let Some(track) = room.queue.iter_mut().find(|t| t.id == *item_id) { track.title.clone_from(title); track.duration.clone_from(duration); track.thumbnail.clone_from(thumbnail); track.pending = false; } if let Some(ref mut active) = room.active { if active.id == *item_id { active.title.clone_from(title); active.duration.clone_from(duration); active.thumbnail.clone_from(thumbnail); active.pending = false; } } } Effect::PersistChat { user_name, content, msg_type, created_at, } => { let id = ChatMessageId::new(); room.chat_messages.push(ChatMessage { id, user_name: user_name.clone(), content: content.clone(), msg_type: msg_type.clone(), created_at: created_at.clone(), }); output.chat_message_id = Some(id); } Effect::AddPlaylistEntry { id, url, title, duration, thumbnail, source, } => { room.playlist.push(InMemoryPlaylistEntry { id: *id, url: url.clone(), title: title.clone(), duration: duration.clone(), thumbnail: thumbnail.clone(), source: source.clone(), added_by: None, added_at: crate::util::now_iso(), pending: true, failed: None, }); } Effect::RemovePlaylistEntry(id) => { room.playlist.retain(|e| e.id != *id); } Effect::ReorderQueue { new_order } => { // The state machine's in-memory `room.queue` is already in // the new order; this arm keeps the effect exhaustively // matched. The SQLite path handles position renumbering. let _ = new_order; } Effect::RemoveQueuedTrack { track_id } => { room.queue.retain(|t| t.id != *track_id); } _ => {} } }
Ok(output) }
async fn load(&self, room_id: &str) -> Result<Option<RoomSnapshot>, sqlx::Error> { let inner = self.inner.lock().expect("InMemoryStore mutex poisoned"); let room = match inner.get(room_id) { Some(room) => room, None => return Ok(None), };
Ok(Some(RoomSnapshot { room_id: room_id.to_string(), room_name: room.name.clone(), queue: room.queue.clone(), active: room.active.clone(), history: room.history.clone(), playlist: room.playlist.iter().map(|e| e.clone().into()).collect(), chat_messages: room.chat_messages.clone(), autoplay_enabled: room.autoplay_enabled, skip_threshold: room.skip_threshold, })) }
async fn name_to_id(&self, name: &str) -> Result<Option<String>, sqlx::Error> { let inner = self.inner.lock().expect("InMemoryStore mutex poisoned"); Ok(inner .iter() .find(|(_, r)| r.name == name) .map(|(id, _)| id.clone())) }
async fn update_room_settings( &self, room_id: &str, autoplay_enabled: bool, skip_threshold: u8, ) -> Result<(), sqlx::Error> { let mut inner = self.inner.lock().expect("InMemoryStore mutex poisoned"); if let Some(room) = inner.get_mut(room_id) { room.autoplay_enabled = autoplay_enabled; room.skip_threshold = skip_threshold; } Ok(()) }
async fn recent_chat( &self, room_id: &str, limit: i64, ) -> Result<Vec<ChatMessage>, sqlx::Error> { let inner = self.inner.lock().expect("InMemoryStore mutex poisoned"); let room = match inner.get(room_id) { Some(room) => room, None => return Ok(Vec::new()), };
let limit = limit.max(0) as usize; Ok(room .chat_messages .iter() .rev() .take(limit) .cloned() .collect()) }
async fn delete_room(&self, room_id: &str) -> Result<(), sqlx::Error> { let mut inner = self.inner.lock().expect("InMemoryStore mutex poisoned"); inner.remove(room_id); Ok(()) }}
/// Production [`RoomStore`] backed by an [`sqlx`] SQLite pool.////// Schema is isolated from the legacy `db.rs` tables: this module uses/// `tracks` (not `queue_items`) and a fresh `rooms` table layout.#[derive(Debug, Clone)]pub(crate) struct SqliteStore { pool: Pool<Sqlite>,}
impl SqliteStore { /// Open a pool at `path` and run schema migrations. pub(crate) async fn connect(path: &std::path::Path) -> Result<Self, sqlx::Error> { use sqlx::sqlite::SqlitePoolOptions;
let conn_str = format!("sqlite:{}?mode=rwc", path.to_string_lossy()); let pool = SqlitePoolOptions::new() .max_connections(8) .connect(&conn_str) .await?;
for pragma in [ "PRAGMA journal_mode=wal", "PRAGMA synchronous=NORMAL", "PRAGMA foreign_keys=ON", "PRAGMA busy_timeout=5000", ] { sqlx::query(pragma).execute(&pool).await?; }
let store = Self::new(pool).await?; Ok(store) }
/// Create a new `SqliteStore` and run schema migrations. /// /// Idempotent: all tables use `CREATE TABLE IF NOT EXISTS`. New /// columns are added via `ALTER TABLE ADD COLUMN` with errors /// ignored (so re-running on an already-migrated DB is a no-op). pub(crate) async fn new(pool: Pool<Sqlite>) -> Result<Self, sqlx::Error> { sqlx::query( "CREATE TABLE IF NOT EXISTS rooms ( id TEXT PRIMARY KEY, name TEXT NOT NULL DEFAULT '', created_at TEXT NOT NULL, updated_at TEXT NOT NULL )", ) .execute(&pool) .await?;
// Migration: add room settings columns. Existing rows pick up // the schema defaults. Errors on duplicate-column are ignored. for sql in [ "ALTER TABLE rooms ADD COLUMN autoplay_enabled INTEGER NOT NULL DEFAULT 1", "ALTER TABLE rooms ADD COLUMN skip_threshold INTEGER NOT NULL DEFAULT 34", ] { if let Err(e) = sqlx::query(sql).execute(&pool).await { // "duplicate column name" means the migration already // ran. Anything else is a real error. if !e.to_string().contains("duplicate column") { return Err(e); } } }
// Canonical resolved media keyed by URL. Tracks and playlist entries // reference this table to avoid duplicating metadata and downloads. sqlx::query( "CREATE TABLE IF NOT EXISTS media_cache ( url TEXT PRIMARY KEY, title TEXT NOT NULL DEFAULT '', duration TEXT NOT NULL DEFAULT '--:--', thumbnail TEXT, source TEXT NOT NULL DEFAULT '', file_path TEXT NOT NULL DEFAULT '', state TEXT NOT NULL DEFAULT 'pending' )", ) .execute(&pool) .await?;
sqlx::query( "CREATE TABLE IF NOT EXISTS tracks ( id TEXT PRIMARY KEY, room_id TEXT NOT NULL REFERENCES rooms(id) ON DELETE CASCADE, url TEXT NOT NULL, position INTEGER NOT NULL, added_at TEXT NOT NULL, played INTEGER NOT NULL DEFAULT 0, played_at TEXT, started_at_ms INTEGER )", ) .execute(&pool) .await?;
// Composite index for the common room-scoped queries. sqlx::query( "CREATE INDEX IF NOT EXISTS idx_tracks_room_lookup ON tracks(room_id, played, started_at_ms, position)", ) .execute(&pool) .await?;
sqlx::query( "CREATE TABLE IF NOT EXISTS chat_messages ( id TEXT PRIMARY KEY, room_id TEXT NOT NULL REFERENCES rooms(id) ON DELETE CASCADE, user_name TEXT NOT NULL, content TEXT NOT NULL, msg_type TEXT NOT NULL DEFAULT 'message', created_at TEXT NOT NULL )", ) .execute(&pool) .await?;
sqlx::query( "CREATE TABLE IF NOT EXISTS playlist_entries ( id TEXT PRIMARY KEY, room_id TEXT NOT NULL REFERENCES rooms(id) ON DELETE CASCADE, url TEXT NOT NULL, added_by TEXT, added_at TEXT NOT NULL )", ) .execute(&pool) .await?;
// Enable foreign key enforcement (per-connection, best-effort). sqlx::query("PRAGMA foreign_keys = ON") .execute(&pool) .await?;
Ok(Self { pool }) }
/// Execute persistence effects inside a single transaction. /// /// Extracted so [`persist`](RoomStore::persist) can wrap it with a retry /// loop for `SQLITE_BUSY`. async fn persist_inner( &self, room_id: &str, effects: &[Effect], ) -> Result<PersistOutput, sqlx::Error> { let mut tx = self.pool.begin().await?; let mut output = PersistOutput::default();
for effect in effects { match effect { Effect::PersistQueuedTrack(track) => { let position: i64 = sqlx::query_scalar( "SELECT COALESCE(MAX(position), -1) + 1 FROM tracks WHERE room_id = ?", ) .bind(room_id) .fetch_one(&mut *tx) .await?;
// Ensure a media_cache row exists for deduplication. sqlx::query("INSERT OR IGNORE INTO media_cache (url) VALUES (?1)") .bind(&track.url) .execute(&mut *tx) .await?;
sqlx::query( "INSERT INTO tracks (id, room_id, url, position, added_at) VALUES (?1, ?2, ?3, ?4, ?5)", ) .bind(track.id.as_str_buf()) .bind(room_id) .bind(&track.url) .bind(position) .bind(crate::util::now_iso()) .execute(&mut *tx) .await?; } Effect::PersistStarted(id) => { let now_ms = chrono::Utc::now().timestamp_millis(); sqlx::query( "UPDATE tracks SET started_at_ms = ?1 WHERE id = ?2 AND played = 0", ) .bind(now_ms) .bind(id.as_str_buf()) .execute(&mut *tx) .await?; } Effect::PersistFinished(id) => { let played_at = crate::util::now_iso(); sqlx::query("UPDATE tracks SET played = 1, played_at = ?1 WHERE id = ?2") .bind(&played_at) .bind(id.as_str_buf()) .execute(&mut *tx) .await?; } Effect::PersistMetadata { item_id: _, url, title, duration, thumbnail, source, } => { sqlx::query( "UPDATE media_cache SET title = ?1, duration = ?2, thumbnail = ?3, source = ?4, state = 'ready' WHERE url = ?5", ) .bind(title.as_str()) .bind(duration.as_str()) .bind(thumbnail.as_deref()) .bind(source.as_str()) .bind(url.as_str()) .execute(&mut *tx) .await?; } Effect::PersistChat { user_name, content, msg_type, created_at, } => { let id = ChatMessageId::new(); sqlx::query( "INSERT INTO chat_messages (id, room_id, user_name, content, msg_type, created_at) VALUES (?1, ?2, ?3, ?4, ?5, ?6)", ) .bind(id.as_str_buf()) .bind(room_id) .bind(user_name.as_str()) .bind(content.as_str()) .bind(msg_type.as_str()) .bind(created_at.as_str()) .execute(&mut *tx) .await?; output.chat_message_id = Some(id); } Effect::AddPlaylistEntry { id, url, .. } => { // Ensure a media_cache row exists for deduplication. sqlx::query("INSERT OR IGNORE INTO media_cache (url) VALUES (?1)") .bind(url.as_str()) .execute(&mut *tx) .await?;
sqlx::query( "INSERT INTO playlist_entries (id, room_id, url, added_at) VALUES (?1, ?2, ?3, ?4)", ) .bind(id.as_str_buf()) .bind(room_id) .bind(url.as_str()) .bind(crate::util::now_iso()) .execute(&mut *tx) .await?; } Effect::RemovePlaylistEntry(id) => { sqlx::query("DELETE FROM playlist_entries WHERE id = ?") .bind(id.as_str_buf()) .execute(&mut *tx) .await?; } Effect::ReorderQueue { new_order } => { // Renumber every queued track's position to its index in // new_order. Limit to un-played, un-started rows so we // never touch the active or history rows. Using indexed // UPDATEs (one per row) is fine for queues of typical // size (< 100). for (i, track_id) in new_order.iter().enumerate() { sqlx::query( "UPDATE tracks SET position = ?1 WHERE id = ?2 AND room_id = ?3 AND played = 0 AND started_at_ms IS NULL", ) .bind(i as i64) .bind(track_id.as_str_buf()) .bind(room_id) .execute(&mut *tx) .await?; } } Effect::RemoveQueuedTrack { track_id } => { // Delete the track from the queue. The reorder effect // that follows renumbers the remaining tracks, but // keeping this as a separate statement means a partial // failure leaves the queue in a recoverable state. sqlx::query( "DELETE FROM tracks WHERE id = ?1 AND room_id = ?2 AND played = 0 AND started_at_ms IS NULL", ) .bind(track_id.as_str_buf()) .bind(room_id) .execute(&mut *tx) .await?; } Effect::PersistRoomSettings { autoplay_enabled, skip_threshold, } => { sqlx::query( "UPDATE rooms SET autoplay_enabled = ?1, skip_threshold = ?2 WHERE id = ?3", ) .bind(autoplay_enabled) .bind(i64::from(*skip_threshold)) .bind(room_id) .execute(&mut *tx) .await?; } _ => {} } }
tx.commit().await?; Ok(output) }}
/// Helper row type for deserialising tracks JOINed with media_cache.#[derive(sqlx::FromRow)]struct TrackRow { id: String, url: String, played_at: Option<String>, title: String, duration: String, thumbnail: Option<String>, state: String,}
/// Helper row type for deserialising playlist_entries JOINed with media_cache.#[derive(sqlx::FromRow)]struct PlaylistRow { id: String, url: String, added_by: Option<String>, added_at: String, title: String, duration: String, thumbnail: Option<String>, source: String, state: String,}
/// Helper row type for deserialising `chat_messages` rows.#[derive(sqlx::FromRow)]struct ChatMsgRow { id: String, user_name: String, content: String, msg_type: String, created_at: String,}
fn parse_track_id(s: &str) -> TrackId { TrackId(uuid::Uuid::parse_str(s).expect("valid TrackId UUID in DB"))}
fn parse_playlist_id(s: &str) -> PlaylistEntryId { PlaylistEntryId(uuid::Uuid::parse_str(s).expect("valid PlaylistEntryId UUID in DB"))}
fn parse_chat_message_id(s: &str) -> ChatMessageId { ChatMessageId(uuid::Uuid::parse_str(s).expect("valid ChatMessageId UUID in DB"))}
/// Returns `true` when the error is an SQLite `SQLITE_BUSY` that can be/// retried.fn is_sqlite_busy(e: &sqlx::Error) -> bool { match e { sqlx::Error::Database(db_err) => { db_err.message().contains("BUSY") || db_err.code().is_some_and(|c| c == "5") } _ => false, }}
#[async_trait]impl RoomStore for SqliteStore { async fn create_room(&self, id: &str, name: &str) -> Result<(), sqlx::Error> { let now = crate::util::now_iso(); sqlx::query("INSERT INTO rooms (id, name, created_at, updated_at) VALUES (?1, ?2, ?3, ?4)") .bind(id) .bind(name) .bind(&now) .bind(&now) .execute(&self.pool) .await?; Ok(()) }
async fn persist( &self, room_id: &str, effects: &[Effect], ) -> Result<PersistOutput, sqlx::Error> { let mut delay = Duration::from_millis(50);
for attempt in 1..=3 { match self.persist_inner(room_id, effects).await { Ok(out) => return Ok(out), Err(e) if attempt < 3 && is_sqlite_busy(&e) => { tracing::warn!(attempt, error = %e, "SQLITE_BUSY on persist, retrying"); tokio::time::sleep(delay).await; delay *= 2; } Err(e) => return Err(e), } }
unreachable!("retry loop always returns or errors on last iteration") }
async fn load(&self, room_id: &str) -> Result<Option<RoomSnapshot>, sqlx::Error> { // 1. Room existence check + settings. Reads in a single // SELECT so the existence, name, and settings come from the // same row snapshot. let room_row: Option<(String, bool, i64)> = sqlx::query_as( "SELECT name, autoplay_enabled, skip_threshold FROM rooms WHERE id = ?", ) .bind(room_id) .fetch_optional(&self.pool) .await?;
let (room_name, autoplay_enabled, skip_threshold) = match room_row { Some((name, ap, st)) => (name, ap, st.clamp(0, 100) as u8), None => return Ok(None), };
// 2. Queue: not played, not started, ordered by position. // JOIN media_cache for resolved metadata. NULLIF coerces the // default empty strings (from media_cache's schema defaults) // to NULL so the COALESCE fallback kicks in for pending // entries. let queue_rows: Vec<TrackRow> = sqlx::query_as( "SELECT t.id, t.url, t.played_at, COALESCE(NULLIF(m.title, ''), 'Loading...') AS title, COALESCE(NULLIF(m.duration, ''), '--:--') AS duration, m.thumbnail, m.state FROM tracks t LEFT JOIN media_cache m ON m.url = t.url WHERE t.room_id = ? AND t.played = 0 AND t.started_at_ms IS NULL ORDER BY t.position", ) .bind(room_id) .fetch_all(&self.pool) .await?;
// 3. Active: not played, started (at most one). let active_row: Option<TrackRow> = sqlx::query_as( "SELECT t.id, t.url, t.played_at, COALESCE(NULLIF(m.title, ''), 'Loading...') AS title, COALESCE(NULLIF(m.duration, ''), '--:--') AS duration, m.thumbnail, m.state FROM tracks t LEFT JOIN media_cache m ON m.url = t.url WHERE t.room_id = ? AND t.played = 0 AND t.started_at_ms IS NOT NULL ORDER BY t.position LIMIT 1", ) .bind(room_id) .fetch_optional(&self.pool) .await?;
// 4. History: played, newest first. let history_rows: Vec<TrackRow> = sqlx::query_as( "SELECT t.id, t.url, t.played_at, COALESCE(m.title, 'Unknown') AS title, COALESCE(m.duration, '--:--') AS duration, m.thumbnail, m.state FROM tracks t LEFT JOIN media_cache m ON m.url = t.url WHERE t.room_id = ? AND t.played = 1 ORDER BY t.played_at DESC", ) .bind(room_id) .fetch_all(&self.pool) .await?;
// 5. Recent chat messages (most recent 50). let chat_rows: Vec<ChatMsgRow> = sqlx::query_as( "SELECT id, user_name, content, msg_type, created_at FROM chat_messages WHERE room_id = ? ORDER BY id DESC LIMIT 50", ) .bind(room_id) .fetch_all(&self.pool) .await?;
// 6. Playlist entries with media_cache metadata. NULLIF coerces // the default empty strings (from media_cache's schema defaults) // to NULL so the COALESCE fallback kicks in for pending entries. let playlist_rows: Vec<PlaylistRow> = sqlx::query_as( "SELECT p.id, p.url, p.added_by, p.added_at, COALESCE(NULLIF(m.title, ''), 'Loading...') AS title, COALESCE(NULLIF(m.duration, ''), '--:--') AS duration, m.thumbnail, m.source, m.state FROM playlist_entries p LEFT JOIN media_cache m ON m.url = p.url WHERE p.room_id = ? ORDER BY p.id", ) .bind(room_id) .fetch_all(&self.pool) .await?;
let pending_from_state = |state: &str| state == "pending"; // The failure state is encoded in the cached title (set by // `Effect::PersistMetadata` from `handle_download_failed`). Strip the // prefix here so the public `PlaylistEntry.failed` field matches the // in-memory shape — consumers shouldn't have to parse the title. let failed_from_title = |title: &str| -> Option<String> { title .strip_prefix("Failed to load: ") .map(|rest| rest.to_string()) };
let queue: Vec<QueuedTrack> = queue_rows .into_iter() .map(|r| QueuedTrack { id: parse_track_id(&r.id), title: r.title, url: r.url, duration: r.duration, thumbnail: r.thumbnail, pending: pending_from_state(&r.state), }) .collect();
let active = active_row.map(|r| QueuedTrack { id: parse_track_id(&r.id), title: r.title, url: r.url, duration: r.duration, thumbnail: r.thumbnail, pending: pending_from_state(&r.state), });
let history: Vec<FinishedTrack> = history_rows .into_iter() .map(|r| FinishedTrack { id: parse_track_id(&r.id), title: r.title, url: r.url, duration: r.duration, thumbnail: r.thumbnail, played_at: r.played_at.unwrap_or_default(), }) .collect();
let chat_messages: Vec<ChatMessage> = chat_rows .into_iter() .map(|r| ChatMessage { id: parse_chat_message_id(&r.id), user_name: r.user_name, content: r.content, msg_type: r.msg_type, created_at: r.created_at, }) .collect();
let playlist: Vec<PlaylistEntry> = playlist_rows .into_iter() .map(|r| { let failed = failed_from_title(&r.title); PlaylistEntry { id: parse_playlist_id(&r.id), url: r.url, title: r.title, duration: r.duration, thumbnail: r.thumbnail, source: r.source, added_by: r.added_by, added_at: r.added_at, pending: pending_from_state(&r.state), failed, } }) .collect();
Ok(Some(RoomSnapshot { room_id: room_id.to_string(), room_name, queue, active, history, playlist, chat_messages, autoplay_enabled, skip_threshold, })) }
async fn name_to_id(&self, name: &str) -> Result<Option<String>, sqlx::Error> { let id: Option<String> = sqlx::query_scalar("SELECT id FROM rooms WHERE name = ?") .bind(name) .fetch_optional(&self.pool) .await?; Ok(id) }
async fn update_room_settings( &self, room_id: &str, autoplay_enabled: bool, skip_threshold: u8, ) -> Result<(), sqlx::Error> { sqlx::query( "UPDATE rooms SET autoplay_enabled = ?1, skip_threshold = ?2 WHERE id = ?3", ) .bind(autoplay_enabled) .bind(skip_threshold as i64) .bind(room_id) .execute(&self.pool) .await?; Ok(()) }
async fn recent_chat( &self, room_id: &str, limit: i64, ) -> Result<Vec<ChatMessage>, sqlx::Error> { let rows: Vec<ChatMsgRow> = sqlx::query_as( "SELECT id, user_name, content, msg_type, created_at FROM chat_messages WHERE room_id = ? ORDER BY id DESC LIMIT ?", ) .bind(room_id) .bind(limit) .fetch_all(&self.pool) .await?;
Ok(rows .into_iter() .map(|r| ChatMessage { id: parse_chat_message_id(&r.id), user_name: r.user_name, content: r.content, msg_type: r.msg_type, created_at: r.created_at, }) .collect()) }
async fn delete_room(&self, room_id: &str) -> Result<(), sqlx::Error> { sqlx::query("DELETE FROM rooms WHERE id = ?") .bind(room_id) .execute(&self.pool) .await?; Ok(()) }}
#[cfg(test)]mod tests { use super::*;
fn tid(n: u128) -> TrackId { TrackId(uuid::Uuid::from_u128(n)) }
#[tokio::test] async fn test_create_and_load_roundtrip() { let store = InMemoryStore::new(); store.create_room("test", "Test Room").await.unwrap();
let snapshot = store.load("test").await.unwrap().unwrap(); assert_eq!(snapshot.room_id, "test"); assert_eq!(snapshot.room_name, "Test Room"); assert!(snapshot.queue.is_empty()); assert!(snapshot.active.is_none()); assert!(snapshot.history.is_empty()); assert!(snapshot.chat_messages.is_empty()); }
#[tokio::test] async fn test_load_nonexistent_room() { let store = InMemoryStore::new(); let snapshot = store.load("nonexistent").await.unwrap(); assert!(snapshot.is_none()); }
#[tokio::test] async fn test_delete_room() { let store = InMemoryStore::new(); store.create_room("test", "Test Room").await.unwrap();
store.delete_room("test").await.unwrap(); assert!(store.load("test").await.unwrap().is_none()); }
#[tokio::test] async fn test_delete_nonexistent_room_is_noop() { let store = InMemoryStore::new(); store.delete_room("nonexistent").await.unwrap(); }
#[tokio::test] async fn test_recent_chat_empty_room() { let store = InMemoryStore::new(); store.create_room("test", "Test").await.unwrap();
let msgs = store.recent_chat("test", 10).await.unwrap(); assert!(msgs.is_empty()); }
#[tokio::test] async fn test_recent_chat_nonexistent_room() { let store = InMemoryStore::new(); let msgs = store.recent_chat("nobody", 10).await.unwrap(); assert!(msgs.is_empty()); }
#[tokio::test] async fn test_persist_preserves_room_order() { let store = InMemoryStore::new(); store.create_room("a", "A").await.unwrap(); store.create_room("b", "B").await.unwrap();
let snap_a = store.load("a").await.unwrap().unwrap(); let snap_b = store.load("b").await.unwrap().unwrap(); assert_eq!(snap_a.room_name, "A"); assert_eq!(snap_b.room_name, "B"); }
#[tokio::test] async fn test_persist_queued_track() { let store = InMemoryStore::new(); store.create_room("r", "R").await.unwrap();
let track = QueuedTrack { id: tid(1), title: "Loading...".into(), url: "https://example.com/t".into(), duration: "--:--".into(), thumbnail: None, pending: true, };
store .persist("r", &[Effect::PersistQueuedTrack(track)]) .await .unwrap();
let snap = store.load("r").await.unwrap().unwrap(); assert_eq!(snap.queue.len(), 1); assert_eq!(snap.queue[0].id, tid(1)); assert!(snap.queue[0].pending); assert!(snap.active.is_none()); assert!(snap.history.is_empty()); }
#[tokio::test] async fn test_persist_started_moves_track_to_active() { let store = InMemoryStore::new(); store.create_room("r", "R").await.unwrap();
let track = QueuedTrack { id: tid(1), title: "Track A".into(), url: "https://example.com/a".into(), duration: "3:45".into(), thumbnail: None, pending: false, };
store .persist("r", &[Effect::PersistQueuedTrack(track)]) .await .unwrap(); store .persist("r", &[Effect::PersistStarted(tid(1))]) .await .unwrap();
let snap = store.load("r").await.unwrap().unwrap(); assert!(snap.queue.is_empty()); assert!(snap.active.is_some()); assert_eq!(snap.active.unwrap().id, tid(1)); }
#[tokio::test] async fn test_persist_started_nonexistent_track_is_noop() { let store = InMemoryStore::new(); store.create_room("r", "R").await.unwrap();
store .persist("r", &[Effect::PersistStarted(tid(999))]) .await .unwrap(); let snap = store.load("r").await.unwrap().unwrap(); assert!(snap.active.is_none()); }
#[tokio::test] async fn test_persist_finished_moves_active_to_history() { let store = InMemoryStore::new(); store.create_room("r", "R").await.unwrap();
store .persist( "r", &[ Effect::PersistQueuedTrack(QueuedTrack { id: tid(1), title: "A".into(), url: "https://e.com/a".into(), duration: "3:00".into(), thumbnail: None, pending: false, }), Effect::PersistStarted(tid(1)), ], ) .await .unwrap(); store .persist("r", &[Effect::PersistFinished(tid(1))]) .await .unwrap();
let snap = store.load("r").await.unwrap().unwrap(); assert!(snap.active.is_none()); assert_eq!(snap.history.len(), 1); assert_eq!(snap.history[0].id, tid(1)); assert!(!snap.history[0].played_at.is_empty()); }
#[tokio::test] async fn test_persist_finished_without_active_is_noop() { let store = InMemoryStore::new(); store.create_room("r", "R").await.unwrap();
store .persist("r", &[Effect::PersistFinished(tid(1))]) .await .unwrap(); let snap = store.load("r").await.unwrap().unwrap(); assert!(snap.history.is_empty()); }
#[tokio::test] async fn test_persist_metadata_updates_queued_track() { let store = InMemoryStore::new(); store.create_room("r", "R").await.unwrap();
let track = QueuedTrack { id: tid(1), title: "Loading...".into(), url: "https://e.com/t".into(), duration: "--:--".into(), thumbnail: None, pending: true, };
store .persist("r", &[Effect::PersistQueuedTrack(track)]) .await .unwrap(); store .persist( "r", &[Effect::PersistMetadata { item_id: tid(1), url: "https://e.com/t".into(), title: "Real Title".into(), duration: "4:20".into(), thumbnail: Some("https://img.example/t.jpg".into()), source: "ytdlp".into(), }], ) .await .unwrap();
let snap = store.load("r").await.unwrap().unwrap(); assert_eq!(snap.queue[0].title, "Real Title"); assert_eq!(snap.queue[0].duration, "4:20"); assert_eq!( snap.queue[0].thumbnail, Some("https://img.example/t.jpg".into()) ); assert!(!snap.queue[0].pending); }
#[tokio::test] async fn test_persist_metadata_updates_active_track() { let store = InMemoryStore::new(); store.create_room("r", "R").await.unwrap();
store .persist( "r", &[ Effect::PersistQueuedTrack(QueuedTrack { id: tid(1), title: "Loading...".into(), url: "https://e.com/t".into(), duration: "--:--".into(), thumbnail: None, pending: true, }), Effect::PersistStarted(tid(1)), ], ) .await .unwrap(); store .persist( "r", &[Effect::PersistMetadata { item_id: tid(1), url: "https://e.com/t".into(), title: "Real Title".into(), duration: "4:20".into(), thumbnail: None, source: "direct".into(), }], ) .await .unwrap();
let snap = store.load("r").await.unwrap().unwrap(); assert!(snap.queue.is_empty()); let active = snap.active.unwrap(); assert_eq!(active.title, "Real Title"); assert!(!active.pending); }
#[tokio::test] async fn test_persist_chat() { let store = InMemoryStore::new(); store.create_room("r", "R").await.unwrap();
let output = store .persist( "r", &[Effect::PersistChat { user_name: "alice".into(), content: "hello".into(), msg_type: "message".into(), created_at: "2025-01-01T00:00:00.000Z".into(), }], ) .await .unwrap();
assert!(output.chat_message_id.is_some());
let snap = store.load("r").await.unwrap().unwrap(); assert_eq!(snap.chat_messages.len(), 1); assert_eq!(snap.chat_messages[0].user_name, "alice"); assert_eq!(snap.chat_messages[0].content, "hello"); }
#[tokio::test] async fn test_persist_chat_multiple_increments_id() { let store = InMemoryStore::new(); store.create_room("r", "R").await.unwrap();
let o1 = store .persist( "r", &[Effect::PersistChat { user_name: "a".into(), content: "1".into(), msg_type: "message".into(), created_at: "t1".into(), }], ) .await .unwrap(); let o2 = store .persist( "r", &[Effect::PersistChat { user_name: "b".into(), content: "2".into(), msg_type: "message".into(), created_at: "t2".into(), }], ) .await .unwrap();
assert!(o1.chat_message_id.is_some()); assert!(o2.chat_message_id.is_some()); assert_ne!(o1.chat_message_id, o2.chat_message_id);
let snap = store.load("r").await.unwrap().unwrap(); assert_eq!(snap.chat_messages.len(), 2); }
#[tokio::test] async fn test_persist_to_nonexistent_room_is_noop() { let store = InMemoryStore::new(); let output = store .persist( "ghost", &[Effect::PersistChat { user_name: "x".into(), content: "x".into(), msg_type: "message".into(), created_at: "x".into(), }], ) .await .unwrap(); assert!(output.chat_message_id.is_none()); }
#[tokio::test] async fn test_persist_recent_chat_ordering() { let store = InMemoryStore::new(); store.create_room("r", "R").await.unwrap();
for i in 0..5 { store .persist( "r", &[Effect::PersistChat { user_name: "user".into(), content: format!("msg {i}"), msg_type: "message".into(), created_at: format!("t{i}"), }], ) .await .unwrap(); }
let recent = store.recent_chat("r", 3).await.unwrap(); assert_eq!(recent.len(), 3); assert_eq!(recent[0].content, "msg 4"); assert_eq!(recent[2].content, "msg 2"); }
// ============================================================ // SqliteStore tests. These exercise the actual SQL paths for the // newer effects (ReorderQueue, RemoveQueuedTrack, AddPlaylistEntry, // RemovePlaylistEntry) and verify the on-disk state survives a // re-load. InMemoryStore tests above already cover the queue/active // /history round-trip; SqliteStore catches SQL bugs that the // in-memory impl can't (scoping to played=0, started_at_ms IS NULL, // referential integrity via foreign keys, etc.). // ============================================================
async fn sqlite_test_store() -> SqliteStore { use sqlx::sqlite::SqlitePoolOptions; let pool = SqlitePoolOptions::new() .max_connections(1) .connect("sqlite::memory:") .await .unwrap(); SqliteStore::new(pool).await.unwrap() }
fn queued_track(id: u128, url: &str, title: &str) -> (QueuedTrack, Effect) { let t = QueuedTrack { id: tid(id), title: title.into(), url: url.into(), duration: "5:00".into(), thumbnail: None, pending: false, }; (t.clone(), Effect::PersistQueuedTrack(t)) }
#[tokio::test] async fn sqlite_reorder_queue_renumbers_positions() { let store = sqlite_test_store().await; store.create_room("r", "R").await.unwrap();
for (i, url) in ["a", "b", "c", "d"].iter().enumerate() { let (_, eff) = queued_track(i as u128 + 1, &format!("u/{url}"), url); store.persist("r", &[eff]).await.unwrap(); } // Sanity: initial load returns them in insertion order. let snap = store.load("r").await.unwrap().unwrap(); let ids: Vec<_> = snap.queue.iter().map(|t| t.id).collect(); assert_eq!(ids, vec![tid(1), tid(2), tid(3), tid(4)]);
// Move first to last (c, d, a, b). store .persist( "r", &[Effect::ReorderQueue { new_order: vec![tid(3), tid(4), tid(1), tid(2)], }], ) .await .unwrap();
let snap = store.load("r").await.unwrap().unwrap(); let ids: Vec<_> = snap.queue.iter().map(|t| t.id).collect(); assert_eq!( ids, vec![tid(3), tid(4), tid(1), tid(2)], "ReorderQueue must renumber so load returns new_order" ); }
#[tokio::test] async fn sqlite_reorder_queue_does_not_touch_active_or_history() { let store = sqlite_test_store().await; store.create_room("r", "R").await.unwrap();
// Queue then start one track to make it active, then queue three more. let (_, a) = queued_track(100, "u/active", "active"); store.persist("r", &[a]).await.unwrap(); store .persist("r", &[Effect::PersistStarted(tid(100))]) .await .unwrap(); for (i, url) in ["x", "y", "z"].iter().enumerate() { let (_, q) = queued_track(i as u128 + 1, &format!("u/{url}"), url); store.persist("r", &[q]).await.unwrap(); }
// Reorder the queue. The active track (id 100) should be left alone. store .persist( "r", &[Effect::ReorderQueue { new_order: vec![tid(3), tid(1), tid(2)], }], ) .await .unwrap();
let snap = store.load("r").await.unwrap().unwrap(); let active_id = snap.active.as_ref().map(|t| t.id); assert_eq!(active_id, Some(tid(100)), "active track must be unchanged"); let queue_ids: Vec<_> = snap.queue.iter().map(|t| t.id).collect(); assert_eq!(queue_ids, vec![tid(3), tid(1), tid(2)]); }
#[tokio::test] async fn sqlite_reorder_queue_with_id_not_in_room_is_a_noop() { // The store's UPDATE is scoped to the current room; a stale // track id from a different room should be silently skipped. let store = sqlite_test_store().await; store.create_room("r1", "R1").await.unwrap(); store.create_room("r2", "R2").await.unwrap();
let (_, a) = queued_track(1, "u/a", "a"); store.persist("r1", &[a]).await.unwrap(); let (_, b) = queued_track(2, "u/b", "b"); store.persist("r2", &[b]).await.unwrap();
// Apply a reorder on r1 that includes r2's track id. r1's // queue should stay intact. store .persist( "r1", &[Effect::ReorderQueue { new_order: vec![tid(2), tid(1)], }], ) .await .unwrap();
let snap1 = store.load("r1").await.unwrap().unwrap(); assert_eq!(snap1.queue.len(), 1); assert_eq!(snap1.queue[0].id, tid(1)); // r2 untouched. let snap2 = store.load("r2").await.unwrap().unwrap(); assert_eq!(snap2.queue[0].id, tid(2)); }
#[tokio::test] async fn sqlite_remove_queued_track_deletes_row() { let store = sqlite_test_store().await; store.create_room("r", "R").await.unwrap();
for (i, url) in ["a", "b", "c"].iter().enumerate() { let (_, q) = queued_track(i as u128 + 1, &format!("u/{url}"), url); store.persist("r", &[q]).await.unwrap(); }
store .persist( "r", &[ Effect::RemoveQueuedTrack { track_id: tid(2) }, Effect::ReorderQueue { new_order: vec![tid(1), tid(3)], }, ], ) .await .unwrap();
let snap = store.load("r").await.unwrap().unwrap(); let ids: Vec<_> = snap.queue.iter().map(|t| t.id).collect(); assert_eq!(ids, vec![tid(1), tid(3)]); }
#[tokio::test] async fn sqlite_remove_queued_track_does_not_touch_active() { let store = sqlite_test_store().await; store.create_room("r", "R").await.unwrap();
let (_, a) = queued_track(100, "u/active", "active"); store.persist("r", &[a]).await.unwrap(); store .persist("r", &[Effect::PersistStarted(tid(100))]) .await .unwrap(); let (_, q) = queued_track(1, "u/q", "q"); store.persist("r", &[q]).await.unwrap();
// Try to remove the active track id via RemoveQueuedTrack — // should be a no-op because the SQL scopes to queued rows. store .persist("r", &[Effect::RemoveQueuedTrack { track_id: tid(100) }]) .await .unwrap();
let snap = store.load("r").await.unwrap().unwrap(); assert_eq!(snap.active.as_ref().unwrap().id, tid(100)); assert_eq!(snap.queue.len(), 1); }
#[tokio::test] async fn sqlite_add_playlist_entry_persists_and_appears_in_load() { let store = sqlite_test_store().await; store.create_room("r", "R").await.unwrap();
let entry_id = PlaylistEntryId(uuid::Uuid::from_u128(1)); store .persist( "r", &[Effect::AddPlaylistEntry { id: entry_id, url: "https://example.com/x".into(), title: "Loading...".into(), duration: "--:--".into(), thumbnail: None, source: "ytdlp".into(), }], ) .await .unwrap();
let snap = store.load("r").await.unwrap().unwrap(); assert_eq!(snap.playlist.len(), 1); assert_eq!(snap.playlist[0].id, entry_id); assert_eq!(snap.playlist[0].url, "https://example.com/x"); // Without a matching media_cache row, the load shows // 'Loading...' / pending=true via the LEFT JOIN. assert_eq!(snap.playlist[0].title, "Loading..."); assert!(snap.playlist[0].pending); }
#[tokio::test] async fn sqlite_add_playlist_entry_picks_up_resolved_metadata() { // Add the playlist entry first (this seeds media_cache with // defaults), then PersistMetadata updates the same row to the // resolved state. The load should pick up the resolved title. let store = sqlite_test_store().await; store.create_room("r", "R").await.unwrap();
let entry_id = PlaylistEntryId(uuid::Uuid::from_u128(1)); store .persist( "r", &[Effect::AddPlaylistEntry { id: entry_id, url: "https://example.com/x".into(), title: "Loading...".into(), duration: "--:--".into(), thumbnail: None, source: "ytdlp".into(), }], ) .await .unwrap();
store .persist( "r", &[Effect::PersistMetadata { item_id: TrackId::nil(), url: "https://example.com/x".into(), title: "Real Title".into(), duration: "4:20".into(), thumbnail: Some("https://img/x.jpg".into()), source: "ytdlp".into(), }], ) .await .unwrap();
let snap = store.load("r").await.unwrap().unwrap(); assert_eq!(snap.playlist.len(), 1); let entry = &snap.playlist[0]; assert_eq!(entry.title, "Real Title"); assert_eq!(entry.duration, "4:20"); assert_eq!(entry.thumbnail, Some("https://img/x.jpg".into())); assert!(!entry.pending); }
#[tokio::test] async fn sqlite_failed_playlist_entry_rehydrates_with_failed_field() { // A playlist entry whose media_cache row was rewritten to a // "Failed to load: <err>" title (via Effect::PersistMetadata from // handle_download_failed) must rehydrate with `failed = Some(err)`. // The DB schema doesn't have a `failed` column — the title prefix is // the only signal — so the snapshot loader is the place this // contract is enforced. let store = sqlite_test_store().await; store.create_room("r", "R").await.unwrap();
let entry_id = PlaylistEntryId(uuid::Uuid::from_u128(1)); store .persist( "r", &[Effect::AddPlaylistEntry { id: entry_id, url: "https://example.com/dead".into(), title: "Loading...".into(), duration: "--:--".into(), thumbnail: None, source: "ytdlp".into(), }], ) .await .unwrap();
store .persist( "r", &[Effect::PersistMetadata { item_id: TrackId::nil(), url: "https://example.com/dead".into(), title: "Failed to load: yt-dlp returned 404".into(), duration: "--:--".into(), thumbnail: None, source: "ytdlp".into(), }], ) .await .unwrap();
let snap = store.load("r").await.unwrap().unwrap(); assert_eq!(snap.playlist.len(), 1); let entry = &snap.playlist[0]; assert_eq!(entry.title, "Failed to load: yt-dlp returned 404"); assert!(!entry.pending); assert_eq!( entry.failed.as_deref(), Some("yt-dlp returned 404"), "snapshot loader must strip the title prefix into the `failed` field", ); }
#[tokio::test] async fn sqlite_resolved_playlist_entry_rehydrates_with_failed_none() { // A playlist entry with a non-failure title must rehydrate with // `failed = None`. The snapshot loader is prefix-match based, so // a title that doesn't start with "Failed to load: " must not // surface a stale `failed` reason. let store = sqlite_test_store().await; store.create_room("r", "R").await.unwrap();
let entry_id = PlaylistEntryId(uuid::Uuid::from_u128(1)); store .persist( "r", &[Effect::AddPlaylistEntry { id: entry_id, url: "https://example.com/ok".into(), title: "Loading...".into(), duration: "--:--".into(), thumbnail: None, source: "ytdlp".into(), }], ) .await .unwrap();
store .persist( "r", &[Effect::PersistMetadata { item_id: TrackId::nil(), url: "https://example.com/ok".into(), title: "Resolved Title".into(), duration: "1:23".into(), thumbnail: None, source: "ytdlp".into(), }], ) .await .unwrap();
let snap = store.load("r").await.unwrap().unwrap(); let entry = &snap.playlist[0]; assert_eq!(entry.title, "Resolved Title"); assert!(!entry.pending); assert!( entry.failed.is_none(), "resolved entry must not have a `failed` reason: {entry:?}", ); }
#[tokio::test] async fn sqlite_remove_playlist_entry_deletes_row() { let store = sqlite_test_store().await; store.create_room("r", "R").await.unwrap();
let id_a = PlaylistEntryId(uuid::Uuid::from_u128(1)); let id_b = PlaylistEntryId(uuid::Uuid::from_u128(2)); for id in [id_a, id_b] { store .persist( "r", &[Effect::AddPlaylistEntry { id, url: format!("u/{id}", id = id.as_str_buf()), title: "t".into(), duration: "--:--".into(), thumbnail: None, source: "ytdlp".into(), }], ) .await .unwrap(); } assert_eq!(store.load("r").await.unwrap().unwrap().playlist.len(), 2);
store .persist("r", &[Effect::RemovePlaylistEntry(id_a)]) .await .unwrap();
let snap = store.load("r").await.unwrap().unwrap(); assert_eq!(snap.playlist.len(), 1); assert_eq!(snap.playlist[0].id, id_b); }
#[tokio::test] async fn sqlite_cascade_delete_room_removes_tracks_and_playlist() { // foreign_keys=ON, ON DELETE CASCADE — deleting a room must // remove its tracks, playlist, and chat messages. let store = sqlite_test_store().await; store.create_room("r", "R").await.unwrap(); let (_, q) = queued_track(1, "u/a", "a"); store.persist("r", &[q]).await.unwrap(); let entry_id = PlaylistEntryId(uuid::Uuid::from_u128(1)); store .persist( "r", &[Effect::AddPlaylistEntry { id: entry_id, url: "u/a".into(), title: "t".into(), duration: "--:--".into(), thumbnail: None, source: "ytdlp".into(), }], ) .await .unwrap();
store.delete_room("r").await.unwrap(); assert!(store.load("r").await.unwrap().is_none()); }
#[tokio::test] async fn sqlite_queued_track_loads_placeholder_when_unresolved() { // Verifies the NULLIF/COALESCE fix: a queued track whose URL // has a default media_cache row (title='', state='pending') // should load as 'Loading...' / '--:--' / pending=true. Before // the fix, this returned '' / '' / pending=true. let store = sqlite_test_store().await; store.create_room("r", "R").await.unwrap(); let (_, q) = queued_track(1, "https://example.com/new", "Loading..."); store.persist("r", &[q]).await.unwrap();
let snap = store.load("r").await.unwrap().unwrap(); assert_eq!(snap.queue.len(), 1); let t = &snap.queue[0]; assert_eq!(t.title, "Loading..."); assert_eq!(t.duration, "--:--"); assert!(t.pending); }
#[tokio::test] async fn sqlite_new_room_uses_default_room_settings() { // A freshly created room picks up the column defaults // (autoplay=true, skip_threshold=34) until explicitly changed. let store = sqlite_test_store().await; store.create_room("r", "R").await.unwrap(); let snap = store.load("r").await.unwrap().unwrap(); assert!(snap.autoplay_enabled); assert_eq!(snap.skip_threshold, 34); }
#[tokio::test] async fn sqlite_update_room_settings_persists_and_loads() { let store = sqlite_test_store().await; store.create_room("r", "R").await.unwrap();
store.update_room_settings("r", false, 75).await.unwrap();
let snap = store.load("r").await.unwrap().unwrap(); assert!(!snap.autoplay_enabled); assert_eq!(snap.skip_threshold, 75); }
#[tokio::test] async fn sqlite_persist_room_settings_effect_round_trips() { // The effect-driven path: actor sends // Effect::PersistRoomSettings, store applies it, load reflects // the new values. let store = sqlite_test_store().await; store.create_room("r", "R").await.unwrap();
store .persist( "r", &[Effect::PersistRoomSettings { autoplay_enabled: false, skip_threshold: 50, }], ) .await .unwrap();
let snap = store.load("r").await.unwrap().unwrap(); assert!(!snap.autoplay_enabled); assert_eq!(snap.skip_threshold, 50); }
#[tokio::test] async fn sqlite_name_to_id_resolves_existing_room() { let store = sqlite_test_store().await; store .create_room("uuid-abc", "happy-otters-dance") .await .unwrap(); let resolved = store.name_to_id("happy-otters-dance").await.unwrap(); assert_eq!(resolved, Some("uuid-abc".to_string())); }
#[tokio::test] async fn sqlite_name_to_id_returns_none_for_unknown_name() { let store = sqlite_test_store().await; store .create_room("uuid-abc", "happy-otters-dance") .await .unwrap(); let resolved = store.name_to_id("sad-otters-mope").await.unwrap(); assert!(resolved.is_none()); }
#[tokio::test] async fn sqlite_load_clamps_out_of_range_skip_threshold() { // Defensive: if a future bug writes 200 into skip_threshold, // the load should clamp to 100 rather than blow up the // frontend's u8 JSON deserialization. let store = sqlite_test_store().await; store.create_room("r", "R").await.unwrap(); // Bypass the public API to plant a bogus value. let SqliteStore { pool } = &store; let raw_pool = pool.clone(); sqlx::query("UPDATE rooms SET skip_threshold = 200 WHERE id = 'r'") .execute(&raw_pool) .await .unwrap();
let snap = store.load("r").await.unwrap().unwrap(); assert_eq!(snap.skip_threshold, 100); }}