From 60146b51eab0f1e0e8126bbaf2d34a46938fbfa5 Mon Sep 17 00:00:00 2001 From: karitham Date: Mon, 18 May 2026 00:10:34 +0200 Subject: [PATCH] src: rewrite rooms --- src/db.rs | 297 ------------- src/main.rs | 11 +- src/playlist.rs | 338 -------------- src/room.rs | 1129 ++++++++++++++++++++++++++--------------------- src/state.rs | 182 ++++++-- src/store.rs | 1053 +++++++++++++++++++++++++++++++++++++++++++ src/web.rs | 33 +- 7 files changed, 1822 insertions(+), 1221 deletions(-) delete mode 100644 src/db.rs delete mode 100644 src/playlist.rs create mode 100644 src/store.rs diff --git a/src/db.rs b/src/db.rs deleted file mode 100644 index d9bcb68..0000000 --- a/src/db.rs +++ /dev/null @@ -1,297 +0,0 @@ -//! SQLite persistence layer for rooms, queue items, and chat messages. - -use std::path::Path; - -use sqlx::sqlite::SqlitePoolOptions; -use sqlx::{Pool, Sqlite}; - -/// Create a SQLite connection pool at `path` with WAL journal mode. -pub(crate) async fn create_pool(path: &Path) -> Result, sqlx::Error> { - let conn_str = format!("sqlite:{}?mode=rwc", path.to_string_lossy()); - let pool = SqlitePoolOptions::new() - .max_connections(8) // sqlite supports unlimited connections but 8 avoids contention - .connect(&conn_str) - .await?; - - // Connection-safe PRAGMAs applied on each connection. - for pragma in [ - "PRAGMA journal_mode=wal", - "PRAGMA synchronous=NORMAL", - "PRAGMA foreign_keys=ON", - "PRAGMA busy_timeout=5000", - ] { - sqlx::query(pragma).execute(&pool).await?; - } - - Ok(pool) -} - -/// Run all migrations. Idempotent: uses `CREATE TABLE IF NOT EXISTS`. -pub(crate) async fn run_migrations(pool: &Pool) -> Result<(), sqlx::Error> { - sqlx::query( - "CREATE TABLE IF NOT EXISTS rooms ( - id TEXT PRIMARY KEY, - created_at TEXT NOT NULL, - updated_at TEXT NOT NULL, - persistent INTEGER NOT NULL DEFAULT 0 - )", - ) - .execute(pool) - .await?; - - sqlx::query( - "CREATE TABLE IF NOT EXISTS queue_items ( - id INTEGER PRIMARY KEY AUTOINCREMENT, - room_id TEXT NOT NULL REFERENCES rooms(id) ON DELETE CASCADE, - url TEXT NOT NULL, - title TEXT NOT NULL, - duration TEXT NOT NULL DEFAULT '--:--', - thumbnail TEXT, - source TEXT NOT NULL DEFAULT '', - added_by TEXT, - position INTEGER NOT NULL, - added_at TEXT NOT NULL, - played INTEGER NOT NULL DEFAULT 0, - played_at TEXT - )", - ) - .execute(pool) - .await?; - - sqlx::query( - "CREATE TABLE IF NOT EXISTS chat_messages ( - id INTEGER PRIMARY KEY AUTOINCREMENT, - 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?; - - // Migration: add started_playing_at to queue_items (separates "playing" from "played"). - let has_column: bool = sqlx::query_scalar( - "SELECT COUNT(*) FROM pragma_table_info('queue_items') WHERE name = 'started_playing_at'", - ) - .fetch_one(pool) - .await - .map(|c: i64| c > 0) - .unwrap_or(false); - - if !has_column { - sqlx::query("ALTER TABLE queue_items ADD COLUMN started_playing_at TEXT") - .execute(pool) - .await?; - } - - // Migration: rename `extractor` column to `source` (schema cleanup). - let has_source: bool = sqlx::query_scalar( - "SELECT COUNT(*) FROM pragma_table_info('queue_items') WHERE name = 'source'", - ) - .fetch_one(pool) - .await - .map(|c: i64| c > 0) - .unwrap_or(false); - - if !has_source { - // For pre-refactor databases: rename extractor → source. - let has_extractor: bool = sqlx::query_scalar( - "SELECT COUNT(*) FROM pragma_table_info('queue_items') WHERE name = 'extractor'", - ) - .fetch_one(pool) - .await - .map(|c: i64| c > 0) - .unwrap_or(false); - - if has_extractor { - sqlx::query("ALTER TABLE queue_items RENAME COLUMN extractor TO source") - .execute(pool) - .await?; - } else { - // Safety net: neither column exists (shouldn't happen). - sqlx::query("ALTER TABLE queue_items ADD COLUMN source TEXT NOT NULL DEFAULT ''") - .execute(pool) - .await?; - } - } - - // Migration: add name column to rooms. - let has_name: bool = - sqlx::query_scalar("SELECT COUNT(*) FROM pragma_table_info('rooms') WHERE name = 'name'") - .fetch_one(pool) - .await - .map(|c: i64| c > 0) - .unwrap_or(false); - - if !has_name { - sqlx::query("ALTER TABLE rooms ADD COLUMN name TEXT NOT NULL DEFAULT ''") - .execute(pool) - .await?; - } - - Ok(()) -} - -pub(crate) struct NewChatMessage { - pub(crate) room_id: String, - pub(crate) user_name: String, - pub(crate) content: String, - pub(crate) msg_type: String, - pub(crate) created_at: String, -} - -#[derive(Debug, sqlx::FromRow)] -pub(crate) struct ChatMessageRow { - pub(crate) id: i64, - pub(crate) user_name: String, - pub(crate) content: String, - pub(crate) msg_type: String, - pub(crate) created_at: String, -} - -pub(crate) async fn room_insert( - pool: &Pool, - id: &str, - name: &str, - persistent: bool, -) -> Result<(), sqlx::Error> { - let now = crate::util::now_iso(); - sqlx::query( - "INSERT INTO rooms (id, name, created_at, updated_at, persistent) VALUES (?1, ?2, ?3, ?4, ?5)", - ) - .bind(id) - .bind(name) - .bind(&now) - .bind(&now) - .bind(persistent as i32) - .execute(pool) - .await?; - Ok(()) -} - -pub(crate) async fn get_room_name(pool: &Pool, room_id: &str) -> String { - sqlx::query_scalar::<_, String>("SELECT name FROM rooms WHERE id = ?") - .bind(room_id) - .fetch_optional(pool) - .await - .ok() - .flatten() - .unwrap_or_default() -} - -pub(crate) async fn room_exists(pool: &Pool, id: &str) -> Result { - let count: i64 = sqlx::query_scalar("SELECT COUNT(*) FROM rooms WHERE id = ?") - .bind(id) - .fetch_one(pool) - .await?; - Ok(count > 0) -} - -pub(crate) async fn room_delete(pool: &Pool, id: &str) -> Result<(), sqlx::Error> { - sqlx::query("DELETE FROM rooms WHERE id = ?") - .bind(id) - .execute(pool) - .await?; - Ok(()) -} - -pub(crate) async fn chat_insert( - pool: &Pool, - msg: &NewChatMessage, -) -> Result { - let result = sqlx::query( - "INSERT INTO chat_messages (room_id, user_name, content, msg_type, created_at) - VALUES (?1, ?2, ?3, ?4, ?5)", - ) - .bind(&msg.room_id) - .bind(&msg.user_name) - .bind(&msg.content) - .bind(&msg.msg_type) - .bind(&msg.created_at) - .execute(pool) - .await?; - Ok(result.last_insert_rowid()) -} - -pub(crate) async fn chat_recent( - pool: &Pool, - room_id: &str, - limit: i64, -) -> Result, sqlx::Error> { - let rows = sqlx::query_as::<_, ChatMessageRow>( - "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(pool) - .await?; - Ok(rows) -} - -#[cfg(test)] -mod tests { - use super::*; - use crate::util; - - async fn setup() -> Pool { - let pool = SqlitePoolOptions::new() - .max_connections(1) - .connect(":memory:") - .await - .unwrap(); - run_migrations(&pool).await.unwrap(); - pool - } - - #[tokio::test] - async fn test_migrations() { - let pool = setup().await; - let count: i64 = sqlx::query_scalar( - "SELECT COUNT(*) FROM sqlite_master WHERE type='table' AND name IN ('rooms', 'queue_items', 'chat_messages')", - ) - .fetch_one(&pool) - .await - .unwrap(); - assert_eq!(count, 3); - } - - #[tokio::test] - async fn test_room_crud() { - let pool = setup().await; - room_insert(&pool, "test-room", "test-name", false) - .await - .unwrap(); - assert!(room_exists(&pool, "test-room").await.unwrap()); - let name = get_room_name(&pool, "test-room").await; - assert_eq!(name, "test-name"); - room_delete(&pool, "test-room").await.unwrap(); - assert!(!room_exists(&pool, "test-room").await.unwrap()); - } - - #[tokio::test] - async fn test_chat_crud() { - let pool = setup().await; - room_insert(&pool, "test-room", "test-name", false) - .await - .unwrap(); - - for i in 0..5 { - let msg = NewChatMessage { - room_id: "test-room".into(), - user_name: "alice".into(), - content: format!("message {i}"), - msg_type: "message".into(), - created_at: util::now_iso(), - }; - chat_insert(&pool, &msg).await.unwrap(); - } - - let recent = chat_recent(&pool, "test-room", 3).await.unwrap(); - assert_eq!(recent.len(), 3); - assert_eq!(recent[0].content, "message 4"); - assert_eq!(recent[2].content, "message 2"); - } -} diff --git a/src/main.rs b/src/main.rs index 3eb2934..fe11e40 100644 --- a/src/main.rs +++ b/src/main.rs @@ -3,14 +3,13 @@ #![deny(rust_2018_idioms, unsafe_code)] #![cfg_attr(not(test), deny(clippy::unwrap_used))] -mod db; mod media; mod names; mod playback; -mod playlist; mod room; mod source; mod state; +mod store; mod transport; mod types; mod util; @@ -56,8 +55,8 @@ async fn main() -> anyhow::Result<()> { } } - let pool = db::create_pool(&args.db_path).await?; - db::run_migrations(&pool).await?; + let store: Arc = + Arc::new(crate::store::SqliteStore::connect(&args.db_path).await?); tracing::info!("database ready at {}", args.db_path.display()); let mut sources = source::SourceRegistry::new(); @@ -65,7 +64,7 @@ async fn main() -> anyhow::Result<()> { sources.register(Box::new(source::direct::DirectSource::new())); let sources = Arc::new(sources); - let rooms = room::Registry::new(pool.clone(), args.cache_dir.clone(), sources.clone()); + let rooms = room::Registry::new(store, args.cache_dir.clone(), sources.clone()); // Idle room sweeper: every 60s, remove rooms idle for 600s (10 min). let registry_clone = rooms.clone(); @@ -80,7 +79,7 @@ async fn main() -> anyhow::Result<()> { } }); - let app = web::router(rooms, pool); + let app = web::router(rooms); let listener = tokio::net::TcpListener::bind(args.http_addr).await?; tracing::info!("HTTP server listening on {}", args.http_addr); diff --git a/src/playlist.rs b/src/playlist.rs deleted file mode 100644 index 06c28cf..0000000 --- a/src/playlist.rs +++ /dev/null @@ -1,338 +0,0 @@ -//! Ordered track list for a room: upcoming queue and played history. - -use sqlx::{Pool, Sqlite}; - -use crate::types::TrackMeta; - -/// A single queue item, returned from all Playlist operations. -#[derive(Debug, Clone, sqlx::FromRow, serde::Serialize)] -pub(crate) struct QueueItem { - pub(crate) id: i64, - pub(crate) url: String, - pub(crate) title: String, - pub(crate) duration: String, - pub(crate) thumbnail: Option, - pub(crate) source: String, - pub(crate) position: i32, - pub(crate) played: bool, - pub(crate) played_at: Option, - pub(crate) started_playing_at: Option, -} - -#[derive(Clone)] -pub(crate) struct Playlist { - pool: Pool, - room_id: String, -} - -impl Playlist { - pub(crate) fn new(pool: Pool, room_id: &str) -> Self { - Self { - pool, - room_id: room_id.to_string(), - } - } - - pub(crate) async fn push(&self, meta: &TrackMeta) -> Result { - let count: i64 = sqlx::query_scalar("SELECT COUNT(*) FROM queue_items WHERE room_id = ?") - .bind(&self.room_id) - .fetch_one(&self.pool) - .await?; - - let position = count as i32; - - let result = sqlx::query( - "INSERT INTO queue_items (room_id, url, title, duration, thumbnail, source, added_by, position, added_at, played) - VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, 0)", - ) - .bind(&self.room_id) - .bind(&meta.url) - .bind(&meta.title) - .bind(&meta.duration) - .bind(&meta.thumbnail) - .bind(meta.source.to_string()) - .bind(None::) // added_by - .bind(position) - .bind(crate::util::now_iso()) - .execute(&self.pool) - .await?; - - let id = result.last_insert_rowid(); - - let item = sqlx::query_as::<_, QueueItem>( - "SELECT id, url, title, duration, thumbnail, source, position, played, played_at, started_playing_at - FROM queue_items WHERE id = ?", - ) - .bind(id) - .fetch_one(&self.pool) - .await?; - - Ok(item) - } - - /// List queued items (not playing, not finished) ordered by position ASC. - pub(crate) async fn upcoming(&self) -> Result, sqlx::Error> { - let items = sqlx::query_as::<_, QueueItem>( - "SELECT id, url, title, duration, thumbnail, source, position, played, played_at, started_playing_at - FROM queue_items WHERE room_id = ? AND played = 0 AND started_playing_at IS NULL - ORDER BY position ASC", - ) - .bind(&self.room_id) - .fetch_all(&self.pool) - .await?; - Ok(items) - } - - /// List played items ordered by played_at DESC (newest first), limited to `limit`. - pub(crate) async fn history(&self, limit: i64) -> Result, sqlx::Error> { - let items = sqlx::query_as::<_, QueueItem>( - "SELECT id, url, title, duration, thumbnail, source, position, played, played_at, started_playing_at - FROM queue_items WHERE room_id = ? AND played = 1 ORDER BY played_at DESC, id DESC LIMIT ?", - ) - .bind(&self.room_id) - .bind(limit) - .fetch_all(&self.pool) - .await?; - Ok(items) - } - - /// Mark an item as started playing (sets started_playing_at timestamp). - pub(crate) async fn mark_started(&self, id: i64, started_at: &str) -> Result<(), sqlx::Error> { - sqlx::query("UPDATE queue_items SET started_playing_at = ? WHERE id = ?") - .bind(started_at) - .bind(id) - .execute(&self.pool) - .await?; - Ok(()) - } - - /// Mark a currently-playing item as finished (moves it from "playing" to "played"). - pub(crate) async fn mark_finished(&self, id: i64) -> Result<(), sqlx::Error> { - let now = crate::util::now_iso(); - sqlx::query( - "UPDATE queue_items SET played = 1, played_at = ?, started_playing_at = NULL WHERE id = ?", - ) - .bind(&now) - .bind(id) - .execute(&self.pool) - .await?; - Ok(()) - } - - /// Update metadata fields after async extraction completes. - /// - /// Updates title, duration, thumbnail, and source for the given item. - /// Used when yt-dlp extraction finishes after the item was already - /// queued with placeholder values. - pub(crate) async fn update_metadata( - &self, - id: i64, - meta: &TrackMeta, - ) -> Result<(), sqlx::Error> { - sqlx::query( - "UPDATE queue_items SET title = ?, duration = ?, thumbnail = ?, source = ? WHERE id = ?", - ) - .bind(&meta.title) - .bind(&meta.duration) - .bind(&meta.thumbnail) - .bind(meta.source.to_string()) - .bind(id) - .execute(&self.pool) - .await?; - Ok(()) - } -} - -#[cfg(test)] -mod tests { - use super::*; - use crate::db; - use crate::types; - use sqlx::sqlite::SqlitePoolOptions; - - async fn test_playlist() -> Playlist { - let pool = SqlitePoolOptions::new() - .max_connections(1) - .connect(":memory:") - .await - .unwrap(); - db::run_migrations(&pool).await.unwrap(); - db::room_insert(&pool, "test-room", "test-name", false) - .await - .unwrap(); - Playlist::new(pool, "test-room") - } - - fn track(title: &str) -> types::TrackMeta { - types::TrackMeta { - title: title.into(), - duration: "3:45".into(), - thumbnail: None, - url: "https://example.com/track".into(), - source: crate::types::SourceKind::Direct, - } - } - - #[tokio::test] - async fn test_push_and_upcoming() { - let pl = test_playlist().await; - - let a = pl.push(&track("Track A")).await.unwrap(); - assert!(a.id > 0); - assert!(!a.played); - assert_eq!(a.title, "Track A"); - - let b = pl.push(&track("Track B")).await.unwrap(); - assert!(b.id > 0); - assert_eq!(b.position, 1); - - let upcoming = pl.upcoming().await.unwrap(); - assert_eq!(upcoming.len(), 2); - assert_eq!(upcoming[0].title, "Track A"); - assert_eq!(upcoming[1].title, "Track B"); - assert!(!upcoming[0].played); - assert!(!upcoming[1].played); - } - - #[tokio::test] - async fn test_push_positions() { - let pl = test_playlist().await; - - let a = pl.push(&track("A")).await.unwrap(); - let b = pl.push(&track("B")).await.unwrap(); - let c = pl.push(&track("C")).await.unwrap(); - - assert_eq!(a.position, 0); - assert_eq!(b.position, 1); - assert_eq!(c.position, 2); - } - - #[tokio::test] - async fn test_mark_finished_non_existent_id() { - let pl = test_playlist().await; - pl.mark_finished(99999).await.unwrap(); - } - - #[tokio::test] - async fn test_upcoming_excludes_playing_items() { - let pl = test_playlist().await; - let a = pl.push(&track("A")).await.unwrap(); - pl.push(&track("B")).await.unwrap(); - - let now = chrono::Utc::now() - .format("%Y-%m-%dT%H:%M:%S%.3fZ") - .to_string(); - pl.mark_started(a.id, &now).await.unwrap(); - - let upcoming = pl.upcoming().await.unwrap(); - assert_eq!(upcoming.len(), 1); - assert_eq!(upcoming[0].title, "B"); - } - - #[tokio::test] - async fn test_history() { - let pl = test_playlist().await; - - let a = pl.push(&track("A")).await.unwrap(); - pl.push(&track("B")).await.unwrap(); - let c = pl.push(&track("C")).await.unwrap(); - - let now_a = chrono::Utc::now() - .format("%Y-%m-%dT%H:%M:%S%.3fZ") - .to_string(); - pl.mark_started(a.id, &now_a).await.unwrap(); - pl.mark_finished(a.id).await.unwrap(); - - tokio::time::sleep(std::time::Duration::from_millis(5)).await; - let now_c = chrono::Utc::now() - .format("%Y-%m-%dT%H:%M:%S%.3fZ") - .to_string(); - pl.mark_started(c.id, &now_c).await.unwrap(); - pl.mark_finished(c.id).await.unwrap(); - - let history = pl.history(10).await.unwrap(); - assert_eq!(history.len(), 2); - assert_eq!(history[0].title, "C"); - assert_eq!(history[1].title, "A"); - assert!(history[0].played); - assert!(history[0].played_at.is_some()); - } - - #[tokio::test] - async fn test_history_only_after_mark_finished() { - let pl = test_playlist().await; - let a = pl.push(&track("A")).await.unwrap(); - - let now = chrono::Utc::now() - .format("%Y-%m-%dT%H:%M:%S%.3fZ") - .to_string(); - pl.mark_started(a.id, &now).await.unwrap(); - - let history = pl.history(10).await.unwrap(); - assert!(history.is_empty()); - - pl.mark_finished(a.id).await.unwrap(); - - let history = pl.history(10).await.unwrap(); - assert_eq!(history.len(), 1); - assert_eq!(history[0].title, "A"); - assert!(history[0].played); - assert!(history[0].played_at.is_some()); - } - - #[tokio::test] - async fn test_push_after_mark_finished() { - let pl = test_playlist().await; - let a = pl.push(&track("A")).await.unwrap(); - let now = chrono::Utc::now() - .format("%Y-%m-%dT%H:%M:%S%.3fZ") - .to_string(); - pl.mark_started(a.id, &now).await.unwrap(); - pl.mark_finished(a.id).await.unwrap(); - - pl.push(&track("B")).await.unwrap(); - - let upcoming = pl.upcoming().await.unwrap(); - assert_eq!(upcoming.len(), 1); - assert_eq!(upcoming[0].title, "B"); - - let history = pl.history(10).await.unwrap(); - assert_eq!(history.len(), 1); - assert_eq!(history[0].title, "A"); - } - - #[tokio::test] - async fn test_history_order_multiple_finished() { - let pl = test_playlist().await; - let a = pl.push(&track("A")).await.unwrap(); - let b = pl.push(&track("B")).await.unwrap(); - let c = pl.push(&track("C")).await.unwrap(); - - let now_a = chrono::Utc::now() - .format("%Y-%m-%dT%H:%M:%S%.3fZ") - .to_string(); - pl.mark_started(a.id, &now_a).await.unwrap(); - pl.mark_finished(a.id).await.unwrap(); - - tokio::time::sleep(std::time::Duration::from_millis(5)).await; - let now_b = chrono::Utc::now() - .format("%Y-%m-%dT%H:%M:%S%.3fZ") - .to_string(); - pl.mark_started(b.id, &now_b).await.unwrap(); - pl.mark_finished(b.id).await.unwrap(); - - tokio::time::sleep(std::time::Duration::from_millis(5)).await; - let now_c = chrono::Utc::now() - .format("%Y-%m-%dT%H:%M:%S%.3fZ") - .to_string(); - pl.mark_started(c.id, &now_c).await.unwrap(); - pl.mark_finished(c.id).await.unwrap(); - - let history = pl.history(10).await.unwrap(); - assert_eq!(history.len(), 3); - // Most recently finished first (C, B, A). - assert_eq!(history[0].title, "C"); - assert_eq!(history[1].title, "B"); - assert_eq!(history[2].title, "A"); - } -} diff --git a/src/room.rs b/src/room.rs index 3c62a55..d3710ee 100644 --- a/src/room.rs +++ b/src/room.rs @@ -18,20 +18,19 @@ use std::time::{Duration, SystemTime, UNIX_EPOCH}; use bytes::Bytes; use chrono::{DateTime, Utc}; -use sqlx::{Pool, Sqlite}; #[cfg(test)] use tokio::sync::oneshot; use tokio::sync::{RwLock, mpsc, watch}; -use crate::db; use crate::names::generate_room_name; use crate::playback::Player; -use crate::playlist::Playlist; use crate::source::SourceRegistry; -use crate::state::{self, Effect, Event, PlaybackState}; +use crate::state::{Effect, Event, PlaybackState, QueuedTrack}; +use crate::store::RoomStore; use crate::transport::TrackPublishers; use crate::types::{ - ChatMessage, HistoryEntry, QueueSummary, RoomCommand, RoomHandle, RoomId, TrackMeta, TrackState, + ActiveTrackInfo, ChatMessage, HistoryEntry, QueueSummary, RoomCommand, RoomHandle, RoomId, + TrackMeta, TrackState, }; /// Distributed unique ID generator (Discord-style snowflake). @@ -117,7 +116,8 @@ struct RoomActor { room_id: RoomId, room_name: String, publishers: TrackPublishers, - pool: Pool, + store: Arc, + id_gen: Arc, source_registry: Arc, cache_dir: PathBuf, /// Clone of the command sender, used by spawned extraction tasks to @@ -132,10 +132,10 @@ struct RoomActor { last_active_tx: watch::Sender>, } -/// Thread-safe registry of all active rooms with SQLite persistence. +/// Thread-safe registry of all active rooms with persistent store. #[derive(Clone)] pub(crate) struct Registry { - pool: Pool, + store: Arc, id_gen: Arc, cache_dir: PathBuf, sources: Arc, @@ -148,12 +148,12 @@ struct Inner { impl Registry { pub(crate) fn new( - pool: Pool, + store: Arc, cache_dir: PathBuf, sources: Arc, ) -> Self { Self { - pool, + store, id_gen: Arc::new(SnowflakeIdGen::new(1)), inner: Arc::new(RwLock::new(Inner { rooms: HashMap::new(), @@ -163,12 +163,12 @@ impl Registry { } } - /// Create a new room with a snowflake ID, persist to SQLite. + /// Create a new room with a snowflake ID, persist to store. pub(crate) async fn create(&self) -> RoomId { let id = RoomId(self.id_gen.next_id().await.to_string()); let room_name = generate_room_name(&id.0); let now = Utc::now(); - let _ = db::room_insert(&self.pool, &id.0, &room_name, false).await; + let _ = self.store.create_room(&id.0, &room_name).await; let publishers = TrackPublishers::new(); let (tx, rx) = mpsc::channel(256); @@ -180,7 +180,8 @@ impl Registry { room_id: id.clone(), room_name: room_name.clone(), publishers: publishers.clone(), - pool: self.pool.clone(), + store: self.store.clone(), + id_gen: self.id_gen.clone(), source_registry: self.sources.clone(), cache_dir: self.cache_dir.clone(), cmd_tx: tx.clone(), @@ -204,32 +205,97 @@ impl Registry { id } - /// Get the room handle if the room is active. + /// Get the room handle. Rehydrates from store if the room exists but + /// its actor was swept (idle timeout). pub(crate) async fn handle(&self, id: &RoomId) -> Option { - self.inner.read().await.rooms.get(id).cloned() + // Fast path: room is already active + if let Some(handle) = self.inner.read().await.rooms.get(id).cloned() { + return Some(handle); + } + + // Slow path: try to rehydrate from store + let snapshot = self.store.load(&id.0).await.ok()??; + + let (state, _effects) = PlaybackState::from_snapshot(snapshot.clone()); + + let publishers = TrackPublishers::new(); + let (tx, rx) = mpsc::channel(256); + let (last_active_tx, last_active_rx) = watch::channel(Utc::now()); + + let player = Player::new(self.sources.clone(), self.cache_dir.clone(), tx.clone()); + let actor = RoomActor { + rx, + room_id: id.clone(), + room_name: snapshot.room_name, + publishers: publishers.clone(), + store: self.store.clone(), + id_gen: self.id_gen.clone(), + source_registry: self.sources.clone(), + cache_dir: self.cache_dir.clone(), + cmd_tx: tx.clone(), + state, + player, + client_count: 0, + next_user_id: 1, + last_active_tx, + }; + tokio::spawn(actor.run()); + + let handle = RoomHandle { + cmd_tx: tx, + publishers, + last_active: last_active_rx, + }; + self.inner + .write() + .await + .rooms + .insert(id.clone(), handle.clone()); + Some(handle) } - /// Check whether a room exists in SQLite. + /// Check whether a room exists in the store. #[must_use] pub(crate) async fn exists(&self, id: &RoomId) -> bool { - db::room_exists(&self.pool, &id.0).await.unwrap_or(false) + if self.inner.read().await.rooms.contains_key(id) { + return true; + } + self.store.load(&id.0).await.ok().flatten().is_some() + } + + /// Load room name + history (for SSR rendering). + pub(crate) async fn load_room_data(&self, id: &RoomId) -> Option<(String, Vec)> { + let snapshot = self.store.load(&id.0).await.ok()??; + let name = snapshot.room_name; + let history = snapshot + .history + .into_iter() + .map(|h| TrackMeta { + title: h.title, + duration: h.duration, + thumbnail: h.thumbnail, + url: h.url, + source: "direct".parse().unwrap(), + }) + .collect(); + Some((name, history)) } - /// Get the upcoming queue for a room from SQLite via Playlist. + /// Get the upcoming queue for a room from the store. pub(crate) async fn queue(&self, id: &RoomId) -> Vec { - let playlist = Playlist::new(self.pool.clone(), &id.0); - let items = playlist.upcoming().await.unwrap_or_default(); - items + let snapshot = match self.store.load(&id.0).await { + Ok(Some(s)) => s, + _ => return Vec::new(), + }; + snapshot + .queue .into_iter() - .map(|i| TrackMeta { - title: i.title, - duration: i.duration, - thumbnail: i.thumbnail, - url: i.url, - source: i.source.parse().unwrap_or_else(|s: String| { - tracing::warn!(source = %s, "unknown source kind in queue_items"); - crate::types::SourceKind::Direct - }), + .map(|t| TrackMeta { + title: t.title, + duration: t.duration, + thumbnail: t.thumbnail, + url: t.url, + source: "direct".parse().unwrap(), // best-effort, source is not tracked in QueuedTrack }) .collect() } @@ -244,7 +310,7 @@ impl Registry { let _ = handle.cmd_tx.send(RoomCommand::QueueUrl(url)).await; } - /// Remove a room (from SQLite and in-memory state). + /// Remove a room (from store and in-memory state). #[cfg(test)] pub(crate) async fn remove(&self, id: &RoomId) { if let Some(handle) = self.handle(id).await { @@ -255,7 +321,7 @@ impl Registry { .await; } self.inner.write().await.rooms.remove(id); - let _ = db::room_delete(&self.pool, &id.0).await; + let _ = self.store.delete_room(&id.0).await; } /// Send a chat message: validate, send to actor, return best-effort ChatMessage. @@ -300,23 +366,13 @@ impl Registry { }) } - /// Get the most recent chat messages from SQLite. + /// Get the most recent chat messages from the store. pub(crate) async fn recent_chat( &self, room_id: &RoomId, limit: i64, - ) -> Result, anyhow::Error> { - let rows = db::chat_recent(&self.pool, &room_id.0, limit).await?; - Ok(rows - .into_iter() - .map(|r| ChatMessage { - id: r.id, - user_name: r.user_name, - content: r.content, - msg_type: r.msg_type, - created_at: r.created_at, - }) - .collect()) + ) -> Result, sqlx::Error> { + self.store.recent_chat(&room_id.0, limit).await } /// Register a client in the room. @@ -378,7 +434,7 @@ impl Registry { } } self.inner.write().await.rooms.remove(id); - let _ = db::room_delete(&self.pool, &id.0).await; + // Keep the room in the store so it can be rehydrated on rejoin. swept.push(id.clone()); tracing::info!("swept idle room {id}"); } @@ -409,38 +465,26 @@ impl RoomActor { RoomCommand::QueueUrl(url) => { tracing::debug!(room = %self.room_id, %url, "track url queued"); - let playlist = Playlist::new(self.pool.clone(), &self.room_id.0); - let placeholder = TrackMeta { + let item_id = self.id_gen.next_id().await as i64; + let now = Utc::now().timestamp_millis(); + + let track = QueuedTrack { + id: item_id, title: "Loading...".into(), + url: url.clone(), duration: "--:--".into(), thumbnail: None, - url: url.clone(), - source: crate::types::SourceKind::Direct, + pending: true, }; - let item = match playlist.push(&placeholder).await { - Ok(item) => item, - Err(e) => { - tracing::warn!("failed to persist queue item: {e}"); - return false; - } - }; - let now = Utc::now().timestamp_millis(); - let effects = self.state.transition( - &Event::TrackQueued(state::QueuedTrack { - id: item.id, - title: placeholder.title, - url: placeholder.url, - duration: placeholder.duration, - thumbnail: placeholder.thumbnail, - pending: true, - }), - now, - ); + let effects = self.state.transition(&Event::TrackQueued(track), now); + + if let Err(e) = self.store.persist(&self.room_id.0, &effects).await { + tracing::warn!("failed to persist queued track: {e}"); + } self.execute_effects(effects).await; - let item_id = item.id; let cmd_tx = self.cmd_tx.clone(); let sources = self.source_registry.clone(); let cache_dir = self.cache_dir.clone(); @@ -468,21 +512,21 @@ impl RoomActor { let now = Utc::now().timestamp_millis(); - let playlist = Playlist::new(self.pool.clone(), &self.room_id.0); - if let Err(e) = playlist.update_metadata(item_id, &meta).await { - tracing::warn!(item_id, error = %e, "failed to persist metadata update"); - } - let effects = self.state.transition( &Event::MetadataUpdated { item_id, title: meta.title, duration: meta.duration, thumbnail: meta.thumbnail, + source: meta.source.to_string(), }, now, ); + if let Err(e) = self.store.persist(&self.room_id.0, &effects).await { + tracing::warn!(item_id, error = %e, "failed to persist metadata"); + } + self.execute_effects(effects).await; } @@ -496,9 +540,15 @@ impl RoomActor { title: format!("Failed to load: {error}"), duration: "--:--".into(), thumbnail: None, + source: "ytdlp".into(), }, now, ); + + if let Err(e) = self.store.persist(&self.room_id.0, &effects).await { + tracing::warn!(item_id, error = %e, "failed to persist metadata failure"); + } + self.execute_effects(effects).await; } @@ -509,6 +559,10 @@ impl RoomActor { let effects = self.state.transition(&Event::Skip, now); + if let Err(e) = self.store.persist(&self.room_id.0, &effects).await { + tracing::warn!("failed to persist skip effects: {e}"); + } + self.execute_effects(effects).await; } @@ -523,6 +577,10 @@ impl RoomActor { let effects = self.state.transition(&Event::TrackEnded { item_id }, now); + if let Err(e) = self.store.persist(&self.room_id.0, &effects).await { + tracing::warn!("failed to persist track ended effects: {e}"); + } + self.execute_effects(effects).await; } @@ -534,23 +592,24 @@ impl RoomActor { let created_at = chrono::Utc::now() .format("%Y-%m-%dT%H:%M:%S%.3fZ") .to_string(); - let msg = db::NewChatMessage { - room_id: self.room_id.0.clone(), + let effects = vec![Effect::PersistChat { user_name: user_name.clone(), content: content.clone(), msg_type: msg_type.clone(), created_at: created_at.clone(), - }; - if let Ok(msg_id) = db::chat_insert(&self.pool, &msg).await { - let json = serde_json::json!({ - "user": user_name, - "content": content, - "type": msg_type, - "ts": created_at, - "id": msg_id, - }); - let payload = serde_json::to_vec(&json).unwrap_or_default(); - self.publishers.publish_chat(0, Bytes::from(payload)); + }]; + if let Ok(output) = self.store.persist(&self.room_id.0, &effects).await { + if let Some(msg_id) = output.chat_message_id { + let json = serde_json::json!({ + "user": user_name, + "content": content, + "type": msg_type, + "ts": created_at, + "id": msg_id, + }); + let payload = serde_json::to_vec(&json).unwrap_or_default(); + self.publishers.publish_chat(0, Bytes::from(payload)); + } } } @@ -576,24 +635,20 @@ impl RoomActor { false } - /// Execute a batch of effects from a state transition. + /// Execute actor-side effects from a state transition. + /// + /// Persistence effects are handled by [`RoomStore::persist`] before this + /// is called — only pipeline and publishing effects remain. async fn execute_effects(&mut self, effects: Vec) { for effect in effects { match effect { Effect::AbortPipeline => { self.player.abort(); } - - Effect::PersistStarted(id) => { - let now = Utc::now().format("%Y-%m-%dT%H:%M:%S%.3fZ").to_string(); - let playlist = Playlist::new(self.pool.clone(), &self.room_id.0); - let _ = playlist.mark_started(id, &now).await; - } - Effect::StartPipeline(track) => { let now = Utc::now().timestamp_millis(); self.state.resolve_started_at(now); - let info = crate::types::ActiveTrackInfo { + let info = ActiveTrackInfo { id: track.id, title: track.title, url: track.url, @@ -603,15 +658,11 @@ impl RoomActor { }; self.player.start(&info, self.publishers.clone()); } - - Effect::PersistFinished(id) => { - let playlist = Playlist::new(self.pool.clone(), &self.room_id.0); - let _ = playlist.mark_finished(id).await; - } - Effect::PublishSnapshot => { self.publish_state_snapshot().await; } + // All persistence effects handled by store.persist() in process(). + _ => {} } } } @@ -679,7 +730,11 @@ impl RoomActor { #[cfg(test)] mod tests { use super::*; - use sqlx::sqlite::SqlitePoolOptions; + use crate::store::{InMemoryStore, SqliteStore}; + + // ----------------------------------------------------------------------- + // Shared test infrastructure + // ----------------------------------------------------------------------- async fn poll_until(check: F, timeout: Duration, label: &str) where @@ -698,16 +753,14 @@ mod tests { } } - async fn test_registry_and_room() -> (Registry, RoomId) { - let pool = SqlitePoolOptions::new() - .max_connections(1) - .connect(":memory:") + async fn history_len(store: &Arc, room_id: &str) -> usize { + store + .load(room_id) .await - .unwrap(); - db::run_migrations(&pool).await.unwrap(); - let reg = Registry::new(pool, PathBuf::from("/tmp/moqbox"), test_registry()); - let room_id = reg.create().await; - (reg, room_id) + .ok() + .flatten() + .map(|s| s.history.len()) + .unwrap_or(0) } struct PanaceaSource; @@ -744,6 +797,10 @@ mod tests { Arc::new(reg) } + // ----------------------------------------------------------------------- + // Backend-independent test: snowflake generator + // ----------------------------------------------------------------------- + #[tokio::test] async fn test_snowflake_uniqueness() { let gen_id = Arc::new(SnowflakeIdGen::new(1)); @@ -768,462 +825,504 @@ mod tests { ); } - #[tokio::test] - async fn test_create_and_exists() { - let (reg, room_id) = test_registry_and_room().await; - assert!(reg.exists(&room_id).await); - } + // ----------------------------------------------------------------------- + // Shared scenario functions — run against any RoomStore backend + // ----------------------------------------------------------------------- - #[tokio::test] - async fn test_push_and_queue() { - let (reg, room_id) = test_registry_and_room().await; + mod scenarios { + use super::*; - reg.push_url(&room_id, "https://example.com/a".into()).await; - reg.push_url(&room_id, "https://example.com/b".into()).await; + pub async fn create_and_exists(reg: &Registry, room_id: &RoomId) { + assert!(reg.exists(room_id).await); + } - poll_until( - || async { - let q = reg.queue(&room_id).await; - q.len() == 1 && q[0].title == "b" - }, - Duration::from_secs(2), - "queue should have 1 item with title 'b'", - ) - .await; - } + pub async fn push_and_queue(reg: &Registry, room_id: &RoomId) { + reg.push_url(room_id, "https://example.com/a".into()).await; + reg.push_url(room_id, "https://example.com/b".into()).await; + + poll_until( + || async { + let q = reg.queue(room_id).await; + q.len() == 1 && q[0].title == "b" + }, + Duration::from_secs(2), + "queue should have 1 item with title 'b'", + ) + .await; + } - #[tokio::test] - async fn test_remove() { - let (reg, room_id) = test_registry_and_room().await; + pub async fn push_starts_playback_when_idle(reg: &Registry, room_id: &RoomId) { + reg.push_url(room_id, "https://example.com/t".into()).await; + poll_until( + || async { reg.queue(room_id).await.is_empty() }, + Duration::from_secs(2), + "queue should be empty (item started playing)", + ) + .await; + } - reg.remove(&room_id).await; - assert!(!reg.exists(&room_id).await); - } + pub async fn push_appends_when_already_playing(reg: &Registry, room_id: &RoomId) { + reg.push_url(room_id, "https://example.com/a".into()).await; + poll_until( + || async { reg.queue(room_id).await.is_empty() }, + Duration::from_secs(2), + "first push should have started playing", + ) + .await; - #[tokio::test] - async fn test_send_chat() { - let (reg, room_id) = test_registry_and_room().await; + reg.push_url(room_id, "https://example.com/b".into()).await; + poll_until( + || async { reg.queue(room_id).await.len() == 1 }, + Duration::from_secs(2), + "second push should be queued", + ) + .await; + } - let msg = reg - .send_chat(&room_id, "alice", "hello world", "message") - .await - .unwrap(); - assert_eq!(msg.user_name, "alice"); - assert_eq!(msg.content, "hello world"); - assert_eq!(msg.msg_type, "message"); - assert!(!msg.created_at.is_empty()); - } + pub async fn skip_when_idle_does_not_panic(reg: &Registry, room_id: &RoomId) { + reg.skip(room_id).await; + } - #[tokio::test] - async fn test_send_chat_persistence() { - let (reg, room_id) = test_registry_and_room().await; + pub async fn skip_moves_track_to_history(reg: &Registry, room_id: &RoomId) { + let store = reg.store.clone(); - let _ = reg - .send_chat(&room_id, "alice", "msg1", "message") - .await - .unwrap(); - let _ = reg - .send_chat(&room_id, "bob", "msg2", "message") - .await - .unwrap(); - - poll_until( - || async { reg.recent_chat(&room_id, 10).await.unwrap().len() == 2 }, - Duration::from_secs(2), - "recent_chat should have 2 messages", - ) - .await; - - let recent = reg.recent_chat(&room_id, 10).await.unwrap(); - assert_eq!(recent[0].user_name, "bob"); - assert_eq!(recent[0].content, "msg2"); - assert_eq!(recent[1].user_name, "alice"); - assert_eq!(recent[1].content, "msg1"); - } + reg.push_url(room_id, "https://example.com/a".into()).await; + poll_until( + || async { reg.queue(room_id).await.is_empty() }, + Duration::from_secs(2), + "first push should have started playing", + ) + .await; - #[tokio::test] - async fn test_send_chat_empty_content() { - let (reg, room_id) = test_registry_and_room().await; + reg.skip(room_id).await; + poll_until( + || async { history_len(&store, &room_id.0).await == 1 }, + Duration::from_secs(2), + "skip should move track to history", + ) + .await; - let result = reg.send_chat(&room_id, "alice", "", "message").await; - result.unwrap_err(); - } + assert!(reg.queue(room_id).await.is_empty()); + } - #[tokio::test] - async fn test_send_chat_long_content() { - let (reg, room_id) = test_registry_and_room().await; + pub async fn multiple_skips_sequential(reg: &Registry, room_id: &RoomId) { + for url in ["https://x/a", "https://x/b", "https://x/c"] { + reg.push_url(room_id, url.into()).await; + } + poll_until( + || async { reg.queue(room_id).await.len() == 2 }, + Duration::from_secs(2), + "3 pushes → 1 active, 2 queued", + ) + .await; - let long = "a".repeat(2001); - let result = reg.send_chat(&room_id, "alice", &long, "message").await; - result.unwrap_err(); - } + reg.skip(room_id).await; + poll_until( + || async { reg.queue(room_id).await.len() == 1 }, + Duration::from_secs(2), + "skip → 1 item left in queue", + ) + .await; - #[tokio::test] - async fn test_send_chat_long_username() { - let (reg, room_id) = test_registry_and_room().await; + reg.skip(room_id).await; + poll_until( + || async { reg.queue(room_id).await.is_empty() }, + Duration::from_secs(2), + "skip → queue empty", + ) + .await; + } - let long = "a".repeat(33); - let result = reg.send_chat(&room_id, &long, "hello", "message").await; - result.unwrap_err(); - } + pub async fn double_skip_does_not_corrupt_state(reg: &Registry, room_id: &RoomId) { + reg.push_url(room_id, "https://example.com/a".into()).await; + poll_until( + || async { reg.queue(room_id).await.is_empty() }, + Duration::from_secs(2), + "first push started playing", + ) + .await; - #[tokio::test] - async fn test_register_client() { - let (reg, room_id) = test_registry_and_room().await; + reg.push_url(room_id, "https://example.com/b".into()).await; + poll_until( + || async { reg.queue(room_id).await.len() == 1 }, + Duration::from_secs(2), + "second push queued", + ) + .await; - let id1 = reg.register_client(&room_id).await; - assert_eq!(id1, Some(1)); + reg.skip(room_id).await; + reg.skip(room_id).await; + poll_until( + || async { + match reg.store.load(&room_id.0).await.ok().flatten() { + Some(s) => s.history.first().map(|h| h.title.as_str()) == Some("b"), + None => false, + } + }, + Duration::from_secs(2), + "history should have 'b' as most recent", + ) + .await; - let id2 = reg.register_client(&room_id).await; - assert_eq!(id2, Some(2)); - } + let snapshot = reg.store.load(&room_id.0).await.ok().flatten().unwrap(); + assert!(!snapshot.history.is_empty()); + assert_eq!(snapshot.history[0].title, "b"); + } - #[tokio::test] - async fn test_register_client_nonexistent_room() { - let (reg, _) = test_registry_and_room().await; - let fake_id = RoomId("nonexistent".into()); - assert_eq!(reg.register_client(&fake_id).await, None); - } + pub async fn send_chat(reg: &Registry, room_id: &RoomId) { + let msg = reg + .send_chat(room_id, "alice", "hello world", "message") + .await + .unwrap(); + assert_eq!(msg.user_name, "alice"); + assert_eq!(msg.content, "hello world"); + assert_eq!(msg.msg_type, "message"); + assert!(!msg.created_at.is_empty()); + } - #[tokio::test] - async fn test_unregister_client() { - let (reg, room_id) = test_registry_and_room().await; + pub async fn send_chat_persistence(reg: &Registry, room_id: &RoomId) { + let _ = reg + .send_chat(room_id, "alice", "msg1", "message") + .await + .unwrap(); + let _ = reg + .send_chat(room_id, "bob", "msg2", "message") + .await + .unwrap(); - reg.register_client(&room_id).await; - reg.unregister_client(&room_id).await; + poll_until( + || async { reg.recent_chat(room_id, 10).await.unwrap().len() == 2 }, + Duration::from_secs(2), + "recent_chat should have 2 messages", + ) + .await; - let id = reg.register_client(&room_id).await; - assert_eq!(id, Some(2)); - } + let recent = reg.recent_chat(room_id, 10).await.unwrap(); + assert_eq!(recent[0].user_name, "bob"); + assert_eq!(recent[0].content, "msg2"); + assert_eq!(recent[1].user_name, "alice"); + assert_eq!(recent[1].content, "msg1"); + } - #[tokio::test] - async fn test_skip_when_idle_does_not_panic() { - let (reg, room_id) = test_registry_and_room().await; - reg.skip(&room_id).await; - } + pub async fn remove_room(reg: &Registry, room_id: &RoomId) { + reg.remove(room_id).await; + assert!(!reg.exists(room_id).await); + } - #[tokio::test] - async fn test_double_skip_does_not_deadlock() { - let (reg, room_id) = test_registry_and_room().await; - - reg.push_url(&room_id, "https://example.com/a".into()).await; - poll_until( - || async { reg.queue(&room_id).await.is_empty() }, - Duration::from_secs(2), - "queue should be empty after push starts playing", - ) - .await; - - let r1 = reg.clone(); - let rid1 = room_id.clone(); - let h1 = tokio::spawn(async move { r1.skip(&rid1).await }); - let r2 = reg.clone(); - let rid2 = room_id.clone(); - let h2 = tokio::spawn(async move { r2.skip(&rid2).await }); - let _ = tokio::join!(h1, h2); - } + pub async fn remove_while_playing(reg: &Registry, room_id: &RoomId) { + reg.push_url(room_id, "https://example.com/a".into()).await; + poll_until( + || async { reg.queue(room_id).await.is_empty() }, + Duration::from_secs(2), + "push should have started playing", + ) + .await; - #[tokio::test] - async fn test_send_chat_returns_id_zero() { - let (reg, room_id) = test_registry_and_room().await; - let msg = reg - .send_chat(&room_id, "alice", "hello", "message") - .await - .unwrap(); - assert_eq!(msg.id, 0); - } + reg.remove(room_id).await; + assert!(!reg.exists(room_id).await); + } - #[tokio::test] - async fn test_sweep_idle_removes_inactive_rooms() { - let (reg, room_id) = test_registry_and_room().await; + pub async fn sweep_keeps_room_in_store(reg: &Registry, room_id: &RoomId) { + tokio::time::sleep(Duration::from_millis(1)).await; - tokio::time::sleep(Duration::from_millis(1)).await; + let swept = reg.sweep_idle(Duration::from_secs(0)).await; + assert!(swept.contains(room_id)); - let swept = reg.sweep_idle(Duration::from_secs(0)).await; - assert!(swept.contains(&room_id)); - assert!(!reg.exists(&room_id).await); - } + assert!( + reg.exists(room_id).await, + "room must remain in store after sweep" + ); + assert!( + reg.handle(room_id).await.is_some(), + "handle() rehydrates after sweep" + ); + } - #[tokio::test] - async fn test_sweep_idle_preserves_active_rooms() { - let (reg, room_id) = test_registry_and_room().await; + pub async fn sweep_preserves_active_rooms(reg: &Registry, room_id: &RoomId) { + reg.register_client(room_id).await; + let swept = reg.sweep_idle(Duration::from_millis(100)).await; + assert!(!swept.contains(room_id)); + } - reg.register_client(&room_id).await; - let swept = reg.sweep_idle(Duration::from_millis(100)).await; - assert!(!swept.contains(&room_id)); - } + pub async fn multiple_rooms_dont_interfere(reg: &Registry, room_a: &RoomId) { + let room_b = reg.create().await; - #[tokio::test] - async fn test_push_starts_playback_when_idle() { - let (reg, room_id) = test_registry_and_room().await; - - reg.push_url(&room_id, "https://example.com/t".into()).await; - poll_until( - || async { reg.queue(&room_id).await.is_empty() }, - Duration::from_secs(2), - "queue should be empty (item started playing)", - ) - .await; - } + assert_ne!(room_a, &room_b); + assert!(reg.exists(room_a).await); + assert!(reg.exists(&room_b).await); - #[tokio::test] - async fn test_push_appends_when_already_playing() { - let (reg, room_id) = test_registry_and_room().await; - - reg.push_url(&room_id, "https://example.com/a".into()).await; - poll_until( - || async { reg.queue(&room_id).await.is_empty() }, - Duration::from_secs(2), - "first push should have started playing", - ) - .await; - - reg.push_url(&room_id, "https://example.com/b".into()).await; - poll_until( - || async { reg.queue(&room_id).await.len() == 1 }, - Duration::from_secs(2), - "second push should be queued", - ) - .await; - - let queue = reg.queue(&room_id).await; - assert_eq!(queue[0].title, "b"); - - let playlist = crate::playlist::Playlist::new(reg.pool.clone(), &room_id.0); - let history = playlist.history(10).await.unwrap(); - assert!(history.is_empty()); - } + reg.push_url(room_a, "https://example.com/a".into()).await; + reg.push_url(&room_b, "https://example.com/b".into()).await; - #[tokio::test] - async fn test_skip_when_queue_empty_after_track_ends() { - let (reg, room_id) = test_registry_and_room().await; - - reg.push_url(&room_id, "https://example.com/a".into()).await; - poll_until( - || async { reg.queue(&room_id).await.is_empty() }, - Duration::from_secs(2), - "first push should have started playing", - ) - .await; - - reg.skip(&room_id).await; - poll_until( - || async { - let playlist = crate::playlist::Playlist::new(reg.pool.clone(), &room_id.0); - playlist.history(10).await.unwrap().len() == 1 - }, - Duration::from_secs(2), - "skip should move track to history", - ) - .await; + poll_until( + || async { reg.queue(room_a).await.is_empty() }, + Duration::from_secs(2), + "room_a queue empty", + ) + .await; + poll_until( + || async { reg.queue(&room_b).await.is_empty() }, + Duration::from_secs(2), + "room_b queue empty", + ) + .await; + } - let queue = reg.queue(&room_id).await; - assert!(queue.is_empty()); - } + pub async fn push_after_skip_works(reg: &Registry, room_id: &RoomId) { + let store = reg.store.clone(); - #[tokio::test] - async fn test_double_skip_does_not_corrupt_state() { - let (reg, room_id) = test_registry_and_room().await; - - reg.push_url(&room_id, "https://example.com/a".into()).await; - poll_until( - || async { reg.queue(&room_id).await.is_empty() }, - Duration::from_secs(2), - "first push started playing", - ) - .await; - - reg.push_url(&room_id, "https://example.com/b".into()).await; - poll_until( - || async { reg.queue(&room_id).await.len() == 1 }, - Duration::from_secs(2), - "second push queued", - ) - .await; - - reg.skip(&room_id).await; - reg.skip(&room_id).await; - poll_until( - || async { - let playlist = crate::playlist::Playlist::new(reg.pool.clone(), &room_id.0); - let h = playlist.history(10).await.unwrap(); - !h.is_empty() && h[0].title == "b" - }, - Duration::from_secs(2), - "history should have 'b' as most recent", - ) - .await; - - let playlist = crate::playlist::Playlist::new(reg.pool.clone(), &room_id.0); - let history = playlist.history(10).await.unwrap(); - assert!(!history.is_empty()); - assert_eq!(history[0].title, "b"); + reg.push_url(room_id, "https://a".into()).await; + poll_until( + || async { reg.queue(room_id).await.is_empty() }, + Duration::from_secs(2), + "first push started playing", + ) + .await; + + reg.skip(room_id).await; + poll_until( + || async { history_len(&store, &room_id.0).await > 0 }, + Duration::from_secs(2), + "skip should move track to history", + ) + .await; + + reg.push_url(room_id, "https://b".into()).await; + poll_until( + || async { reg.queue(room_id).await.is_empty() }, + Duration::from_secs(2), + "second push should start playing immediately", + ) + .await; + + assert!(history_len(&store, &room_id.0).await > 0); + } } - #[tokio::test] - async fn test_register_client_runs_three_times() { - let (reg, room_id) = test_registry_and_room().await; + // ----------------------------------------------------------------------- + // InMemoryStore backend + // ----------------------------------------------------------------------- - for _ in 0..3 { - let id = reg.register_client(&room_id).await; - assert!(id.is_some()); + mod inmem { + use super::*; + + async fn setup() -> (Registry, RoomId) { + let store = Arc::new(InMemoryStore::new()); + let reg = Registry::new(store, PathBuf::from("/tmp/moqbox"), test_registry()); + let room_id = reg.create().await; + (reg, room_id) } - assert!(reg.exists(&room_id).await); - } + #[tokio::test] + async fn test_create_and_exists() { + let (reg, id) = setup().await; + scenarios::create_and_exists(®, &id).await; + } - #[tokio::test] - async fn test_remove_while_playing_does_not_panic() { - let (reg, room_id) = test_registry_and_room().await; - - reg.push_url(&room_id, "https://example.com/a".into()).await; - poll_until( - || async { reg.queue(&room_id).await.is_empty() }, - Duration::from_secs(2), - "push should have started playing", - ) - .await; - - reg.remove(&room_id).await; - assert!(!reg.exists(&room_id).await); - } + #[tokio::test] + async fn test_push_and_queue() { + let (reg, id) = setup().await; + scenarios::push_and_queue(®, &id).await; + } - #[tokio::test] - async fn test_sweep_idle_keeps_active_room() { - let (reg, room_id) = test_registry_and_room().await; + #[tokio::test] + async fn test_push_starts_playback_when_idle() { + let (reg, id) = setup().await; + scenarios::push_starts_playback_when_idle(®, &id).await; + } - reg.register_client(&room_id).await; + #[tokio::test] + async fn test_push_appends_when_already_playing() { + let (reg, id) = setup().await; + scenarios::push_appends_when_already_playing(®, &id).await; + } - let swept = reg.sweep_idle(Duration::from_millis(10_000)).await; - assert!(!swept.contains(&room_id)); - } + #[tokio::test] + async fn test_skip_when_idle_does_not_panic() { + let (reg, id) = setup().await; + scenarios::skip_when_idle_does_not_panic(®, &id).await; + } - #[tokio::test] - async fn test_client_count_tracking() { - let (reg, room_id) = test_registry_and_room().await; + #[tokio::test] + async fn test_skip_moves_track_to_history() { + let (reg, id) = setup().await; + scenarios::skip_moves_track_to_history(®, &id).await; + } - let id1 = reg.register_client(&room_id).await; - let id2 = reg.register_client(&room_id).await; - let id3 = reg.register_client(&room_id).await; - assert!(id1.is_some(), "id1 should be Some"); - assert!(id2.is_some(), "id2 should be Some"); - assert!(id3.is_some(), "id3 should be Some"); + #[tokio::test] + async fn test_multiple_skips_sequential() { + let (reg, id) = setup().await; + scenarios::multiple_skips_sequential(®, &id).await; + } - reg.unregister_client(&room_id).await; + #[tokio::test] + async fn test_double_skip_does_not_corrupt_state() { + let (reg, id) = setup().await; + scenarios::double_skip_does_not_corrupt_state(®, &id).await; + } - let id4 = reg.register_client(&room_id).await; - assert!(id4.is_some()); - assert_ne!(id4, id1); - } + #[tokio::test] + async fn test_send_chat() { + let (reg, id) = setup().await; + scenarios::send_chat(®, &id).await; + } - #[tokio::test] - async fn test_multiple_rooms_dont_interfere() { - let (reg, room_a) = test_registry_and_room().await; - let room_b = reg.create().await; - - assert_ne!(room_a, room_b); - assert!(reg.exists(&room_a).await); - assert!(reg.exists(&room_b).await); - - reg.push_url(&room_a, "https://example.com/a".into()).await; - reg.push_url(&room_b, "https://example.com/b".into()).await; - - poll_until( - || async { reg.queue(&room_a).await.is_empty() }, - Duration::from_secs(2), - "room_a queue empty", - ) - .await; - poll_until( - || async { reg.queue(&room_b).await.is_empty() }, - Duration::from_secs(2), - "room_b queue empty", - ) - .await; - } + #[tokio::test] + async fn test_send_chat_persistence() { + let (reg, id) = setup().await; + scenarios::send_chat_persistence(®, &id).await; + } - #[tokio::test] - async fn test_push_after_skip_works() { - let (reg, room_id) = test_registry_and_room().await; - - reg.push_url(&room_id, "https://a".into()).await; - poll_until( - || async { reg.queue(&room_id).await.is_empty() }, - Duration::from_secs(2), - "first push started playing", - ) - .await; - - reg.skip(&room_id).await; - poll_until( - || async { - let playlist = crate::playlist::Playlist::new(reg.pool.clone(), &room_id.0); - !playlist.history(10).await.unwrap().is_empty() - }, - Duration::from_secs(2), - "skip should move track to history", - ) - .await; - - reg.push_url(&room_id, "https://b".into()).await; - poll_until( - || async { reg.queue(&room_id).await.is_empty() }, - Duration::from_secs(2), - "second push should start playing immediately (idle after skip)", - ) - .await; - - let playlist = crate::playlist::Playlist::new(reg.pool.clone(), &room_id.0); - let history = playlist.history(10).await.unwrap(); - assert!(!history.is_empty()); - } + #[tokio::test] + async fn test_remove() { + let (reg, id) = setup().await; + scenarios::remove_room(®, &id).await; + } - #[tokio::test] - async fn test_client_count_does_not_underflow() { - let (reg, room_id) = test_registry_and_room().await; + #[tokio::test] + async fn test_remove_while_playing_does_not_panic() { + let (reg, id) = setup().await; + scenarios::remove_while_playing(®, &id).await; + } - reg.unregister_client(&room_id).await; + #[tokio::test] + async fn test_sweep_idle_removes_actor_but_keeps_room() { + let (reg, id) = setup().await; + scenarios::sweep_keeps_room_in_store(®, &id).await; + } - let re_registered = reg.register_client(&room_id).await; - assert!( - re_registered.is_some(), - "room should accept new clients after underflow" - ); + #[tokio::test] + async fn test_sweep_idle_preserves_active_rooms() { + let (reg, id) = setup().await; + scenarios::sweep_preserves_active_rooms(®, &id).await; + } + + #[tokio::test] + async fn test_multiple_rooms_dont_interfere() { + let (reg, id) = setup().await; + scenarios::multiple_rooms_dont_interfere(®, &id).await; + } + + #[tokio::test] + async fn test_push_after_skip_works() { + let (reg, id) = setup().await; + scenarios::push_after_skip_works(®, &id).await; + } } - #[tokio::test] - async fn test_multiple_skips_sequential() { - let (reg, room_id) = test_registry_and_room().await; - - for url in ["https://x/a", "https://x/b", "https://x/c"] { - reg.push_url(&room_id, url.into()).await; - } - poll_until( - || async { reg.queue(&room_id).await.len() == 2 }, - Duration::from_secs(2), - "3 pushes → 1 active, 2 queued", - ) - .await; - - reg.skip(&room_id).await; - poll_until( - || async { reg.queue(&room_id).await.len() == 1 }, - Duration::from_secs(2), - "skip → 1 item left in queue", - ) - .await; - - let queue = reg.queue(&room_id).await; - assert_eq!(queue[0].title, "c"); - - reg.skip(&room_id).await; - poll_until( - || async { reg.queue(&room_id).await.is_empty() }, - Duration::from_secs(2), - "skip → queue empty", - ) - .await; - - reg.skip(&room_id).await; + // ----------------------------------------------------------------------- + // SqliteStore backend — runs same scenarios via :memory: SQLite + // ----------------------------------------------------------------------- + + mod sqlite { + use super::*; + + async fn setup() -> (Registry, RoomId) { + let pool = sqlx::sqlite::SqlitePoolOptions::new() + .max_connections(1) + .connect(":memory:") + .await + .unwrap(); + let store = Arc::new(SqliteStore::new(pool).await.unwrap()); + let reg = Registry::new(store, PathBuf::from("/tmp/moqbox"), test_registry()); + let room_id = reg.create().await; + (reg, room_id) + } + + #[tokio::test] + async fn test_create_and_exists() { + let (reg, id) = setup().await; + scenarios::create_and_exists(®, &id).await; + } + + #[tokio::test] + async fn test_push_and_queue() { + let (reg, id) = setup().await; + scenarios::push_and_queue(®, &id).await; + } + + #[tokio::test] + async fn test_push_starts_playback_when_idle() { + let (reg, id) = setup().await; + scenarios::push_starts_playback_when_idle(®, &id).await; + } + + #[tokio::test] + async fn test_push_appends_when_already_playing() { + let (reg, id) = setup().await; + scenarios::push_appends_when_already_playing(®, &id).await; + } + + #[tokio::test] + async fn test_skip_when_idle_does_not_panic() { + let (reg, id) = setup().await; + scenarios::skip_when_idle_does_not_panic(®, &id).await; + } + + #[tokio::test] + async fn test_skip_moves_track_to_history() { + let (reg, id) = setup().await; + scenarios::skip_moves_track_to_history(®, &id).await; + } + + #[tokio::test] + async fn test_multiple_skips_sequential() { + let (reg, id) = setup().await; + scenarios::multiple_skips_sequential(®, &id).await; + } + + #[tokio::test] + async fn test_double_skip_does_not_corrupt_state() { + let (reg, id) = setup().await; + scenarios::double_skip_does_not_corrupt_state(®, &id).await; + } + + #[tokio::test] + async fn test_send_chat() { + let (reg, id) = setup().await; + scenarios::send_chat(®, &id).await; + } + + #[tokio::test] + async fn test_send_chat_persistence() { + let (reg, id) = setup().await; + scenarios::send_chat_persistence(®, &id).await; + } + + #[tokio::test] + async fn test_remove() { + let (reg, id) = setup().await; + scenarios::remove_room(®, &id).await; + } + + #[tokio::test] + async fn test_remove_while_playing_does_not_panic() { + let (reg, id) = setup().await; + scenarios::remove_while_playing(®, &id).await; + } + + #[tokio::test] + async fn test_sweep_idle_removes_actor_but_keeps_room() { + let (reg, id) = setup().await; + scenarios::sweep_keeps_room_in_store(®, &id).await; + } + + #[tokio::test] + async fn test_sweep_idle_preserves_active_rooms() { + let (reg, id) = setup().await; + scenarios::sweep_preserves_active_rooms(®, &id).await; + } + + #[tokio::test] + async fn test_multiple_rooms_dont_interfere() { + let (reg, id) = setup().await; + scenarios::multiple_rooms_dont_interfere(®, &id).await; + } + + #[tokio::test] + async fn test_push_after_skip_works() { + let (reg, id) = setup().await; + scenarios::push_after_skip_works(®, &id).await; + } } } diff --git a/src/state.rs b/src/state.rs index 9aa6490..942a3f6 100644 --- a/src/state.rs +++ b/src/state.rs @@ -43,6 +43,7 @@ pub(crate) enum Event { title: String, duration: String, thumbnail: Option, + source: String, }, } @@ -61,6 +62,23 @@ pub(crate) enum Effect { PersistFinished(i64), /// Broadcast the current state snapshot to all clients. PublishSnapshot, + /// Persist a newly queued track to the store. + PersistQueuedTrack(QueuedTrack), + /// Persist metadata update after async extraction completes. + PersistMetadata { + item_id: i64, + title: String, + duration: String, + thumbnail: Option, + source: String, + }, + /// Persist a chat message. + PersistChat { + user_name: String, + content: String, + msg_type: String, + created_at: String, + }, } /// Pure playback state: no IO handles, no DB connections. @@ -89,7 +107,8 @@ impl PlaybackState { title, duration, thumbnail, - } => self.handle_metadata_updated(*item_id, title, duration, thumbnail), + source, + } => self.handle_metadata_updated(*item_id, title, duration, thumbnail, source), } } @@ -98,10 +117,16 @@ impl PlaybackState { self.queue.push(track.clone()); if self.active.is_some() { - return vec![Effect::PublishSnapshot]; + return vec![ + Effect::PersistQueuedTrack(track.clone()), + Effect::PublishSnapshot, + ]; } - self.advance(track.clone()) + // Idle: persist the track first, then advance it to playing. + let mut effects = vec![Effect::PersistQueuedTrack(track.clone())]; + effects.extend(self.advance(track.clone())); + effects } /// User requested skip. @@ -193,6 +218,7 @@ impl PlaybackState { title: &str, duration: &str, thumbnail: &Option, + source: &str, ) -> Vec { if let Some(item) = self.queue.iter_mut().find(|t| t.id == item_id) { item.title = title.to_string(); @@ -207,7 +233,16 @@ impl PlaybackState { active.thumbnail.clone_from(thumbnail); } } - vec![Effect::PublishSnapshot] + vec![ + Effect::PersistMetadata { + item_id, + title: title.to_string(), + duration: duration.to_string(), + thumbnail: thumbnail.clone(), + source: source.to_string(), + }, + Effect::PublishSnapshot, + ] } /// Pop the first track from the queue and start playing it. @@ -242,6 +277,40 @@ impl PlaybackState { active.started_at_wall = real_started_at; } } + + /// Reconstruct state from a [`RoomSnapshot`] loaded from the store. + /// + /// If there's a stale active track (server died mid-playback), it's moved + /// to history. The queue is restored as-is. No playback auto-starts on + /// rehydration — users explicitly trigger playback. + pub fn from_snapshot(snapshot: crate::store::RoomSnapshot) -> (Self, Vec) { + let mut state = PlaybackState::default(); + let mut effects = Vec::new(); + + state.queue = snapshot.queue; + + // Stale active track → move to history front + if let Some(stale) = snapshot.active { + state.history.push(FinishedTrack { + id: stale.id, + title: stale.title, + url: stale.url, + duration: stale.duration, + thumbnail: stale.thumbnail, + played_at: String::new(), + }); + effects.push(Effect::PersistFinished(stale.id)); + } + + // Append existing history + state.history.extend(snapshot.history); + + if !effects.is_empty() { + effects.push(Effect::PublishSnapshot); + } + + (state, effects) + } } /// # Pure unit tests (no async, no DB, no IO) @@ -268,11 +337,12 @@ mod tests { assert_eq!( effects, vec![ + Effect::PersistQueuedTrack(queued("A", 1)), Effect::PersistStarted(1), Effect::StartPipeline(queued("A", 1)), Effect::PublishSnapshot, ], - "idle → first track: persist started_at, spawn pipeline, notify clients" + "idle → first track: persist queued, started_at, spawn pipeline, notify clients" ); } @@ -284,8 +354,11 @@ mod tests { assert_eq!( effects, - vec![Effect::PublishSnapshot], - "playing -> queue another: just notify, pipeline untouched" + vec![ + Effect::PersistQueuedTrack(queued("B", 2)), + Effect::PublishSnapshot + ], + "playing -> queue another: persist track, notify, pipeline untouched" ); } @@ -303,7 +376,7 @@ mod tests { Effect::PersistFinished(1), Effect::PersistStarted(2), ], - "skip with next: abort current, finalize A, persist B started_at" + "skip with next: abort current, finalize A, start B" ); assert!( effects @@ -334,30 +407,6 @@ mod tests { ); } - #[test] - fn test_track_ended_with_next_effect_order() { - let mut s = PlaybackState::default(); - s.transition(&Event::TrackQueued(queued("A", 1)), 0); - s.transition(&Event::TrackQueued(queued("B", 2)), 0); - let effects = s.transition(&Event::TrackEnded { item_id: 1 }, 0); - - assert_eq!( - &effects[..2], - &[Effect::PersistFinished(1), Effect::PersistStarted(2),], - "track ended (next): finalize A, persist B started_at, no AbortPipeline" - ); - assert!( - effects - .iter() - .any(|e| matches!(e, Effect::StartPipeline(t) if t.title == "B")), - "track ended (next): must start B" - ); - assert!( - effects.contains(&Effect::PublishSnapshot), - "track ended (next): must notify" - ); - } - #[test] fn test_track_ended_last_goes_idle_effect_order() { let mut s = PlaybackState::default(); @@ -426,15 +475,27 @@ mod tests { } #[test] - fn test_track_ended_advances_to_next() { + fn test_track_ended_with_next_effect_order() { let mut s = PlaybackState::default(); s.transition(&Event::TrackQueued(queued("A", 1)), 0); s.transition(&Event::TrackQueued(queued("B", 2)), 0); - s.transition(&Event::TrackEnded { item_id: 1 }, 0); + let effects = s.transition(&Event::TrackEnded { item_id: 1 }, 0); - assert_eq!(s.active.as_ref().unwrap().title, "B"); - assert_eq!(s.history.len(), 1); - assert_eq!(s.history[0].title, "A"); + assert_eq!( + &effects[..2], + &[Effect::PersistFinished(1), Effect::PersistStarted(2),], + "track ended (next): finalize A, start B, no AbortPipeline, no re-persist queued" + ); + assert!( + effects + .iter() + .any(|e| matches!(e, Effect::StartPipeline(t) if t.title == "B")), + "track ended (next): must start B" + ); + assert!( + effects.contains(&Effect::PublishSnapshot), + "track ended (next): must notify" + ); } #[test] @@ -492,16 +553,31 @@ mod tests { fn test_metadata_update_updates_queue_item() { let mut s = PlaybackState::default(); s.transition(&Event::TrackQueued(queued("A", 1)), 0); - s.transition( + let effects = s.transition( &Event::MetadataUpdated { item_id: 1, title: "Track A (Real)".into(), duration: "4:20".into(), thumbnail: Some("https://img.example/a.jpg".into()), + source: "ytdlp".into(), }, 0, ); + assert_eq!( + effects, + vec![ + Effect::PersistMetadata { + item_id: 1, + title: "Track A (Real)".into(), + duration: "4:20".into(), + thumbnail: Some("https://img.example/a.jpg".into()), + source: "ytdlp".into(), + }, + Effect::PublishSnapshot, + ] + ); + // Queue should have empty (track was advanced to active), but active // should have updated metadata. assert_eq!(s.active.as_ref().unwrap().title, "Track A (Real)"); @@ -525,10 +601,23 @@ mod tests { title: "Track B (Real)".into(), duration: "5:55".into(), thumbnail: None, + source: "ytdlp".into(), }, 0, ); - assert_eq!(effects, vec![Effect::PublishSnapshot]); + assert_eq!( + effects, + vec![ + Effect::PersistMetadata { + item_id: 2, + title: "Track B (Real)".into(), + duration: "5:55".into(), + thumbnail: None, + source: "ytdlp".into(), + }, + Effect::PublishSnapshot, + ] + ); assert_eq!(s.queue[0].title, "Track B (Real)"); assert!(!s.queue[0].pending); } @@ -542,12 +631,25 @@ mod tests { title: "Ghost".into(), duration: "0:00".into(), thumbnail: None, + source: "ytdlp".into(), }, 0, ); // Still publishes snapshot even if nothing changed: clients may want // to know the update was processed. - assert_eq!(effects, vec![Effect::PublishSnapshot]); + assert_eq!( + effects, + vec![ + Effect::PersistMetadata { + item_id: 999, + title: "Ghost".into(), + duration: "0:00".into(), + thumbnail: None, + source: "ytdlp".into(), + }, + Effect::PublishSnapshot, + ] + ); assert!(s.active.is_none()); assert!(s.queue.is_empty()); } diff --git a/src/store.rs b/src/store.rs new file mode 100644 index 0000000..b72874d --- /dev/null +++ b/src/store.rs @@ -0,0 +1,1053 @@ +//! 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 async_trait::async_trait; +use sqlx::{Pool, Sqlite}; + +use crate::state::{Effect, FinishedTrack, QueuedTrack}; +use crate::types::ChatMessage; + +/// 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, + /// The track currently playing, if any. + pub active: Option, + /// Recently finished tracks (newest first). + pub history: Vec, + pub chat_messages: Vec, +} + +/// 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, +} + +/// 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; + + /// Load the full snapshot for a room, or [`None`] if it does not exist. + async fn load(&self, room_id: &str) -> Result, sqlx::Error>; + + /// Retrieve the most recent chat messages for a room, newest first. + async fn recent_chat(&self, room_id: &str, limit: i64) + -> Result, 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 implementation (for testing) +// --------------------------------------------------------------------------- + +#[cfg_attr(not(test), allow(dead_code))] +#[derive(Debug)] +struct InMemoryRoom { + name: String, + queue: Vec, + active: Option, + history: Vec, + chat_messages: Vec, + next_chat_id: i64, +} + +/// 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>, +} + +#[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(), + chat_messages: Vec::new(), + next_chat_id: 1, + }, + ); + Ok(()) + } + + async fn persist( + &self, + room_id: &str, + effects: &[Effect], + ) -> Result { + 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, + title, + duration, + thumbnail, + source: _, + } => { + // Update in the queue. + 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; + } + // Update in the active track if it matches. + 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 = room.next_chat_id; + room.next_chat_id += 1; + 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); + } + _ => {} + } + } + + Ok(output) + } + + async fn load(&self, room_id: &str) -> Result, 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(), + chat_messages: room.chat_messages.clone(), + })) + } + + async fn recent_chat( + &self, + room_id: &str, + limit: i64, + ) -> Result, 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(()) + } +} + +// --------------------------------------------------------------------------- +// SQLite-backed implementation (production) +// --------------------------------------------------------------------------- + +/// 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, +} + +impl SqliteStore { + /// Open a pool at `path` and run schema migrations. + pub(crate) async fn connect(path: &std::path::Path) -> Result { + 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`. + pub(crate) async fn new(pool: Pool) -> Result { + 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?; + + sqlx::query( + "CREATE TABLE IF NOT EXISTS tracks ( + id INTEGER PRIMARY KEY, + room_id TEXT NOT NULL REFERENCES rooms(id) ON DELETE CASCADE, + url TEXT NOT NULL, + title TEXT NOT NULL, + duration TEXT NOT NULL DEFAULT '--:--', + thumbnail TEXT, + source TEXT NOT NULL DEFAULT '', + position INTEGER NOT NULL, + added_at TEXT NOT NULL, + played INTEGER NOT NULL DEFAULT 0, + played_at TEXT, + started_at_ms INTEGER, + pending INTEGER NOT NULL DEFAULT 1 + )", + ) + .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 INTEGER PRIMARY KEY AUTOINCREMENT, + 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?; + + // Enable foreign key enforcement (per-connection, best-effort). + sqlx::query("PRAGMA foreign_keys = ON") + .execute(&pool) + .await?; + + Ok(Self { pool }) + } +} + +/// Helper row type for deserialising `tracks` rows. +#[derive(sqlx::FromRow)] +#[allow(dead_code)] +struct TrackRow { + id: i64, + title: String, + url: String, + duration: String, + thumbnail: Option, + played: i64, + played_at: Option, + started_at_ms: Option, + pending: i64, +} + +/// Helper row type for deserialising `chat_messages` rows. +#[derive(sqlx::FromRow)] +struct ChatMsgRow { + id: i64, + user_name: String, + content: String, + msg_type: String, + created_at: String, +} + +#[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 { + let mut tx = self.pool.begin().await?; + let mut output = PersistOutput::default(); + + for effect in effects { + match effect { + Effect::PersistQueuedTrack(track) => { + // Compute the next position at the end of the room's queue. + 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?; + + sqlx::query( + "INSERT INTO tracks + (id, room_id, url, title, duration, thumbnail, source, + position, added_at, pending) + VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10)", + ) + .bind(track.id) + .bind(room_id) + .bind(&track.url) + .bind(&track.title) + .bind(&track.duration) + .bind(track.thumbnail.as_deref()) + .bind("") // source is resolved later via PersistMetadata + .bind(position) + .bind(crate::util::now_iso()) + .bind(track.pending as i64) + .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) + .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) + .execute(&mut *tx) + .await?; + } + Effect::PersistMetadata { + item_id, + title, + duration, + thumbnail, + source, + } => { + let src_str = source.to_string(); + sqlx::query( + "UPDATE tracks + SET title = ?1, duration = ?2, thumbnail = ?3, + source = ?4, pending = 0 + WHERE id = ?5", + ) + .bind(title.as_str()) + .bind(duration.as_str()) + .bind(thumbnail.as_deref()) + .bind(&src_str) + .bind(item_id) + .execute(&mut *tx) + .await?; + } + Effect::PersistChat { + user_name, + content, + msg_type, + created_at, + } => { + let result = sqlx::query( + "INSERT INTO chat_messages + (room_id, user_name, content, msg_type, created_at) + VALUES (?1, ?2, ?3, ?4, ?5)", + ) + .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(result.last_insert_rowid()); + } + // Non-persistence effects are ignored. + _ => {} + } + } + + tx.commit().await?; + Ok(output) + } + + async fn load(&self, room_id: &str) -> Result, sqlx::Error> { + // 1. Room existence check. + let room_name: Option = sqlx::query_scalar("SELECT name FROM rooms WHERE id = ?") + .bind(room_id) + .fetch_optional(&self.pool) + .await?; + + let room_name = match room_name { + Some(name) => name, + None => return Ok(None), + }; + + // 2. Queue: not played, not started, ordered by position. + let queue_rows: Vec = sqlx::query_as( + "SELECT id, url, title, duration, thumbnail, played, played_at, + started_at_ms, pending + FROM tracks + WHERE room_id = ? AND played = 0 AND started_at_ms IS NULL + ORDER BY position", + ) + .bind(room_id) + .fetch_all(&self.pool) + .await?; + + // 3. Active: not played, started (at most one). + let active_row: Option = sqlx::query_as( + "SELECT id, url, title, duration, thumbnail, played, played_at, + started_at_ms, pending + FROM tracks + WHERE room_id = ? AND played = 0 AND started_at_ms IS NOT NULL + ORDER BY position + LIMIT 1", + ) + .bind(room_id) + .fetch_optional(&self.pool) + .await?; + + // 4. History: played, newest first. + let history_rows: Vec = sqlx::query_as( + "SELECT id, url, title, duration, thumbnail, played, played_at, + started_at_ms, pending + FROM tracks + WHERE room_id = ? AND played = 1 + ORDER BY played_at DESC", + ) + .bind(room_id) + .fetch_all(&self.pool) + .await?; + + // 5. Recent chat messages (most recent 50). + let chat_rows: Vec = 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?; + + let queue: Vec = queue_rows + .into_iter() + .map(|r| QueuedTrack { + id: r.id, + title: r.title, + url: r.url, + duration: r.duration, + thumbnail: r.thumbnail, + pending: r.pending != 0, + }) + .collect(); + + let active = active_row.map(|r| QueuedTrack { + id: r.id, + title: r.title, + url: r.url, + duration: r.duration, + thumbnail: r.thumbnail, + pending: r.pending != 0, + }); + + let history: Vec = history_rows + .into_iter() + .map(|r| FinishedTrack { + 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 = chat_rows + .into_iter() + .map(|r| ChatMessage { + id: r.id, + user_name: r.user_name, + content: r.content, + msg_type: r.msg_type, + created_at: r.created_at, + }) + .collect(); + + Ok(Some(RoomSnapshot { + room_id: room_id.to_string(), + room_name, + queue, + active, + history, + chat_messages, + })) + } + + async fn recent_chat( + &self, + room_id: &str, + limit: i64, + ) -> Result, sqlx::Error> { + let rows: Vec = 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: 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::*; + + // ------------------------------------------------------------------ + // InMemoryStore tests + // ------------------------------------------------------------------ + + #[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: 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, 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: 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(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, 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(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: 1, + title: "A".into(), + url: "https://e.com/a".into(), + duration: "3:00".into(), + thumbnail: None, + pending: false, + }), + Effect::PersistStarted(1), + ], + ) + .await + .unwrap(); + store + .persist("r", &[Effect::PersistFinished(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, 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(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: 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: 1, + 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: 1, + title: "Loading...".into(), + url: "https://e.com/t".into(), + duration: "--:--".into(), + thumbnail: None, + pending: true, + }), + Effect::PersistStarted(1), + ], + ) + .await + .unwrap(); + store + .persist( + "r", + &[Effect::PersistMetadata { + item_id: 1, + 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_eq!(output.chat_message_id, Some(1)); + + 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_eq!(o1.chat_message_id, Some(1)); + assert_eq!(o2.chat_message_id, Some(2)); + + 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"); + } +} diff --git a/src/web.rs b/src/web.rs index 3341c51..7307604 100644 --- a/src/web.rs +++ b/src/web.rs @@ -12,10 +12,8 @@ use axum::{ response::{Html, IntoResponse, Json}, routing::{get, post}, }; -use sqlx::{Pool, Sqlite}; use tokio::sync::Mutex; -use crate::db; use crate::room; use crate::types::{RoomId, TrackMeta}; @@ -55,11 +53,10 @@ impl RateLimiter { #[derive(Clone)] pub(crate) struct AppState { pub rooms: room::Registry, - pub pool: Pool, pub rate_limiter: RateLimiter, } -pub(crate) fn router(rooms: room::Registry, pool: Pool) -> Router { +pub(crate) fn router(rooms: room::Registry) -> Router { let rate_limiter = RateLimiter::new(); let cleanup_limiter = rate_limiter.clone(); @@ -73,7 +70,6 @@ pub(crate) fn router(rooms: room::Registry, pool: Pool) -> Router { let state = AppState { rooms, - pool, rate_limiter, }; @@ -126,28 +122,15 @@ async fn room_view( let queue = state.rooms.queue(&room_id).await; let queue_empty = queue.is_empty(); - let history = { - let playlist = crate::playlist::Playlist::new(state.pool.clone(), &id); - playlist - .history(20) - .await - .unwrap_or_default() - .into_iter() - .map(|i| TrackMeta { - title: i.title, - duration: i.duration, - thumbnail: i.thumbnail, - url: i.url, - source: i.source.parse().unwrap_or_else(|s: String| { - tracing::warn!(source = %s, "unknown source kind in queue_items"); - crate::types::SourceKind::Direct - }), - }) - .collect::>() + // Load snapshot once for room name + history. + let (room_name, history) = { + let rid = RoomId(id.clone()); + match state.rooms.load_room_data(&rid).await { + Some((name, hist)) => (name, hist), + None => (String::new(), Vec::new()), + } }; - let room_name = db::get_room_name(&state.pool, &id).await; - let tpl = RoomTemplate { room_id: id, room_name, -- 2.51.2