experiments of a tiny cytube-like player with yt-dlp
src: rewrite rooms
+1797
-1196
-297
src/db.rs
-297
src/db.rs
···
1
-
//! SQLite persistence layer for rooms, queue items, and chat messages.
2
-
3
-
use std::path::Path;
4
-
5
-
use sqlx::sqlite::SqlitePoolOptions;
6
-
use sqlx::{Pool, Sqlite};
7
-
8
-
/// Create a SQLite connection pool at `path` with WAL journal mode.
9
-
pub(crate) async fn create_pool(path: &Path) -> Result<Pool<Sqlite>, sqlx::Error> {
10
-
let conn_str = format!("sqlite:{}?mode=rwc", path.to_string_lossy());
11
-
let pool = SqlitePoolOptions::new()
12
-
.max_connections(8) // sqlite supports unlimited connections but 8 avoids contention
13
-
.connect(&conn_str)
14
-
.await?;
15
-
16
-
// Connection-safe PRAGMAs applied on each connection.
17
-
for pragma in [
18
-
"PRAGMA journal_mode=wal",
19
-
"PRAGMA synchronous=NORMAL",
20
-
"PRAGMA foreign_keys=ON",
21
-
"PRAGMA busy_timeout=5000",
22
-
] {
23
-
sqlx::query(pragma).execute(&pool).await?;
24
-
}
25
-
26
-
Ok(pool)
27
-
}
28
-
29
-
/// Run all migrations. Idempotent: uses `CREATE TABLE IF NOT EXISTS`.
30
-
pub(crate) async fn run_migrations(pool: &Pool<Sqlite>) -> Result<(), sqlx::Error> {
31
-
sqlx::query(
32
-
"CREATE TABLE IF NOT EXISTS rooms (
33
-
id TEXT PRIMARY KEY,
34
-
created_at TEXT NOT NULL,
35
-
updated_at TEXT NOT NULL,
36
-
persistent INTEGER NOT NULL DEFAULT 0
37
-
)",
38
-
)
39
-
.execute(pool)
40
-
.await?;
41
-
42
-
sqlx::query(
43
-
"CREATE TABLE IF NOT EXISTS queue_items (
44
-
id INTEGER PRIMARY KEY AUTOINCREMENT,
45
-
room_id TEXT NOT NULL REFERENCES rooms(id) ON DELETE CASCADE,
46
-
url TEXT NOT NULL,
47
-
title TEXT NOT NULL,
48
-
duration TEXT NOT NULL DEFAULT '--:--',
49
-
thumbnail TEXT,
50
-
source TEXT NOT NULL DEFAULT '',
51
-
added_by TEXT,
52
-
position INTEGER NOT NULL,
53
-
added_at TEXT NOT NULL,
54
-
played INTEGER NOT NULL DEFAULT 0,
55
-
played_at TEXT
56
-
)",
57
-
)
58
-
.execute(pool)
59
-
.await?;
60
-
61
-
sqlx::query(
62
-
"CREATE TABLE IF NOT EXISTS chat_messages (
63
-
id INTEGER PRIMARY KEY AUTOINCREMENT,
64
-
room_id TEXT NOT NULL REFERENCES rooms(id) ON DELETE CASCADE,
65
-
user_name TEXT NOT NULL,
66
-
content TEXT NOT NULL,
67
-
msg_type TEXT NOT NULL DEFAULT 'message',
68
-
created_at TEXT NOT NULL
69
-
)",
70
-
)
71
-
.execute(pool)
72
-
.await?;
73
-
74
-
// Migration: add started_playing_at to queue_items (separates "playing" from "played").
75
-
let has_column: bool = sqlx::query_scalar(
76
-
"SELECT COUNT(*) FROM pragma_table_info('queue_items') WHERE name = 'started_playing_at'",
77
-
)
78
-
.fetch_one(pool)
79
-
.await
80
-
.map(|c: i64| c > 0)
81
-
.unwrap_or(false);
82
-
83
-
if !has_column {
84
-
sqlx::query("ALTER TABLE queue_items ADD COLUMN started_playing_at TEXT")
85
-
.execute(pool)
86
-
.await?;
87
-
}
88
-
89
-
// Migration: rename `extractor` column to `source` (schema cleanup).
90
-
let has_source: bool = sqlx::query_scalar(
91
-
"SELECT COUNT(*) FROM pragma_table_info('queue_items') WHERE name = 'source'",
92
-
)
93
-
.fetch_one(pool)
94
-
.await
95
-
.map(|c: i64| c > 0)
96
-
.unwrap_or(false);
97
-
98
-
if !has_source {
99
-
// For pre-refactor databases: rename extractor → source.
100
-
let has_extractor: bool = sqlx::query_scalar(
101
-
"SELECT COUNT(*) FROM pragma_table_info('queue_items') WHERE name = 'extractor'",
102
-
)
103
-
.fetch_one(pool)
104
-
.await
105
-
.map(|c: i64| c > 0)
106
-
.unwrap_or(false);
107
-
108
-
if has_extractor {
109
-
sqlx::query("ALTER TABLE queue_items RENAME COLUMN extractor TO source")
110
-
.execute(pool)
111
-
.await?;
112
-
} else {
113
-
// Safety net: neither column exists (shouldn't happen).
114
-
sqlx::query("ALTER TABLE queue_items ADD COLUMN source TEXT NOT NULL DEFAULT ''")
115
-
.execute(pool)
116
-
.await?;
117
-
}
118
-
}
119
-
120
-
// Migration: add name column to rooms.
121
-
let has_name: bool =
122
-
sqlx::query_scalar("SELECT COUNT(*) FROM pragma_table_info('rooms') WHERE name = 'name'")
123
-
.fetch_one(pool)
124
-
.await
125
-
.map(|c: i64| c > 0)
126
-
.unwrap_or(false);
127
-
128
-
if !has_name {
129
-
sqlx::query("ALTER TABLE rooms ADD COLUMN name TEXT NOT NULL DEFAULT ''")
130
-
.execute(pool)
131
-
.await?;
132
-
}
133
-
134
-
Ok(())
135
-
}
136
-
137
-
pub(crate) struct NewChatMessage {
138
-
pub(crate) room_id: String,
139
-
pub(crate) user_name: String,
140
-
pub(crate) content: String,
141
-
pub(crate) msg_type: String,
142
-
pub(crate) created_at: String,
143
-
}
144
-
145
-
#[derive(Debug, sqlx::FromRow)]
146
-
pub(crate) struct ChatMessageRow {
147
-
pub(crate) id: i64,
148
-
pub(crate) user_name: String,
149
-
pub(crate) content: String,
150
-
pub(crate) msg_type: String,
151
-
pub(crate) created_at: String,
152
-
}
153
-
154
-
pub(crate) async fn room_insert(
155
-
pool: &Pool<Sqlite>,
156
-
id: &str,
157
-
name: &str,
158
-
persistent: bool,
159
-
) -> Result<(), sqlx::Error> {
160
-
let now = crate::util::now_iso();
161
-
sqlx::query(
162
-
"INSERT INTO rooms (id, name, created_at, updated_at, persistent) VALUES (?1, ?2, ?3, ?4, ?5)",
163
-
)
164
-
.bind(id)
165
-
.bind(name)
166
-
.bind(&now)
167
-
.bind(&now)
168
-
.bind(persistent as i32)
169
-
.execute(pool)
170
-
.await?;
171
-
Ok(())
172
-
}
173
-
174
-
pub(crate) async fn get_room_name(pool: &Pool<Sqlite>, room_id: &str) -> String {
175
-
sqlx::query_scalar::<_, String>("SELECT name FROM rooms WHERE id = ?")
176
-
.bind(room_id)
177
-
.fetch_optional(pool)
178
-
.await
179
-
.ok()
180
-
.flatten()
181
-
.unwrap_or_default()
182
-
}
183
-
184
-
pub(crate) async fn room_exists(pool: &Pool<Sqlite>, id: &str) -> Result<bool, sqlx::Error> {
185
-
let count: i64 = sqlx::query_scalar("SELECT COUNT(*) FROM rooms WHERE id = ?")
186
-
.bind(id)
187
-
.fetch_one(pool)
188
-
.await?;
189
-
Ok(count > 0)
190
-
}
191
-
192
-
pub(crate) async fn room_delete(pool: &Pool<Sqlite>, id: &str) -> Result<(), sqlx::Error> {
193
-
sqlx::query("DELETE FROM rooms WHERE id = ?")
194
-
.bind(id)
195
-
.execute(pool)
196
-
.await?;
197
-
Ok(())
198
-
}
199
-
200
-
pub(crate) async fn chat_insert(
201
-
pool: &Pool<Sqlite>,
202
-
msg: &NewChatMessage,
203
-
) -> Result<i64, sqlx::Error> {
204
-
let result = sqlx::query(
205
-
"INSERT INTO chat_messages (room_id, user_name, content, msg_type, created_at)
206
-
VALUES (?1, ?2, ?3, ?4, ?5)",
207
-
)
208
-
.bind(&msg.room_id)
209
-
.bind(&msg.user_name)
210
-
.bind(&msg.content)
211
-
.bind(&msg.msg_type)
212
-
.bind(&msg.created_at)
213
-
.execute(pool)
214
-
.await?;
215
-
Ok(result.last_insert_rowid())
216
-
}
217
-
218
-
pub(crate) async fn chat_recent(
219
-
pool: &Pool<Sqlite>,
220
-
room_id: &str,
221
-
limit: i64,
222
-
) -> Result<Vec<ChatMessageRow>, sqlx::Error> {
223
-
let rows = sqlx::query_as::<_, ChatMessageRow>(
224
-
"SELECT id, user_name, content, msg_type, created_at
225
-
FROM chat_messages WHERE room_id = ? ORDER BY id DESC LIMIT ?",
226
-
)
227
-
.bind(room_id)
228
-
.bind(limit)
229
-
.fetch_all(pool)
230
-
.await?;
231
-
Ok(rows)
232
-
}
233
-
234
-
#[cfg(test)]
235
-
mod tests {
236
-
use super::*;
237
-
use crate::util;
238
-
239
-
async fn setup() -> Pool<Sqlite> {
240
-
let pool = SqlitePoolOptions::new()
241
-
.max_connections(1)
242
-
.connect(":memory:")
243
-
.await
244
-
.unwrap();
245
-
run_migrations(&pool).await.unwrap();
246
-
pool
247
-
}
248
-
249
-
#[tokio::test]
250
-
async fn test_migrations() {
251
-
let pool = setup().await;
252
-
let count: i64 = sqlx::query_scalar(
253
-
"SELECT COUNT(*) FROM sqlite_master WHERE type='table' AND name IN ('rooms', 'queue_items', 'chat_messages')",
254
-
)
255
-
.fetch_one(&pool)
256
-
.await
257
-
.unwrap();
258
-
assert_eq!(count, 3);
259
-
}
260
-
261
-
#[tokio::test]
262
-
async fn test_room_crud() {
263
-
let pool = setup().await;
264
-
room_insert(&pool, "test-room", "test-name", false)
265
-
.await
266
-
.unwrap();
267
-
assert!(room_exists(&pool, "test-room").await.unwrap());
268
-
let name = get_room_name(&pool, "test-room").await;
269
-
assert_eq!(name, "test-name");
270
-
room_delete(&pool, "test-room").await.unwrap();
271
-
assert!(!room_exists(&pool, "test-room").await.unwrap());
272
-
}
273
-
274
-
#[tokio::test]
275
-
async fn test_chat_crud() {
276
-
let pool = setup().await;
277
-
room_insert(&pool, "test-room", "test-name", false)
278
-
.await
279
-
.unwrap();
280
-
281
-
for i in 0..5 {
282
-
let msg = NewChatMessage {
283
-
room_id: "test-room".into(),
284
-
user_name: "alice".into(),
285
-
content: format!("message {i}"),
286
-
msg_type: "message".into(),
287
-
created_at: util::now_iso(),
288
-
};
289
-
chat_insert(&pool, &msg).await.unwrap();
290
-
}
291
-
292
-
let recent = chat_recent(&pool, "test-room", 3).await.unwrap();
293
-
assert_eq!(recent.len(), 3);
294
-
assert_eq!(recent[0].content, "message 4");
295
-
assert_eq!(recent[2].content, "message 2");
296
-
}
297
-
}
+5
-6
src/main.rs
+5
-6
src/main.rs
···
3
3
#![deny(rust_2018_idioms, unsafe_code)]
4
4
#![cfg_attr(not(test), deny(clippy::unwrap_used))]
5
5
6
-
mod db;
7
6
mod media;
8
7
mod names;
9
8
mod playback;
10
-
mod playlist;
11
9
mod room;
12
10
mod source;
13
11
mod state;
12
+
mod store;
14
13
mod transport;
15
14
mod types;
16
15
mod util;
···
56
55
}
57
56
}
58
57
59
-
let pool = db::create_pool(&args.db_path).await?;
60
-
db::run_migrations(&pool).await?;
58
+
let store: Arc<dyn crate::store::RoomStore> =
59
+
Arc::new(crate::store::SqliteStore::connect(&args.db_path).await?);
61
60
tracing::info!("database ready at {}", args.db_path.display());
62
61
63
62
let mut sources = source::SourceRegistry::new();
···
65
64
sources.register(Box::new(source::direct::DirectSource::new()));
66
65
let sources = Arc::new(sources);
67
66
68
-
let rooms = room::Registry::new(pool.clone(), args.cache_dir.clone(), sources.clone());
67
+
let rooms = room::Registry::new(store, args.cache_dir.clone(), sources.clone());
69
68
70
69
// Idle room sweeper: every 60s, remove rooms idle for 600s (10 min).
71
70
let registry_clone = rooms.clone();
···
80
79
}
81
80
});
82
81
83
-
let app = web::router(rooms, pool);
82
+
let app = web::router(rooms);
84
83
let listener = tokio::net::TcpListener::bind(args.http_addr).await?;
85
84
86
85
tracing::info!("HTTP server listening on {}", args.http_addr);
-338
src/playlist.rs
-338
src/playlist.rs
···
1
-
//! Ordered track list for a room: upcoming queue and played history.
2
-
3
-
use sqlx::{Pool, Sqlite};
4
-
5
-
use crate::types::TrackMeta;
6
-
7
-
/// A single queue item, returned from all Playlist operations.
8
-
#[derive(Debug, Clone, sqlx::FromRow, serde::Serialize)]
9
-
pub(crate) struct QueueItem {
10
-
pub(crate) id: i64,
11
-
pub(crate) url: String,
12
-
pub(crate) title: String,
13
-
pub(crate) duration: String,
14
-
pub(crate) thumbnail: Option<String>,
15
-
pub(crate) source: String,
16
-
pub(crate) position: i32,
17
-
pub(crate) played: bool,
18
-
pub(crate) played_at: Option<String>,
19
-
pub(crate) started_playing_at: Option<String>,
20
-
}
21
-
22
-
#[derive(Clone)]
23
-
pub(crate) struct Playlist {
24
-
pool: Pool<Sqlite>,
25
-
room_id: String,
26
-
}
27
-
28
-
impl Playlist {
29
-
pub(crate) fn new(pool: Pool<Sqlite>, room_id: &str) -> Self {
30
-
Self {
31
-
pool,
32
-
room_id: room_id.to_string(),
33
-
}
34
-
}
35
-
36
-
pub(crate) async fn push(&self, meta: &TrackMeta) -> Result<QueueItem, sqlx::Error> {
37
-
let count: i64 = sqlx::query_scalar("SELECT COUNT(*) FROM queue_items WHERE room_id = ?")
38
-
.bind(&self.room_id)
39
-
.fetch_one(&self.pool)
40
-
.await?;
41
-
42
-
let position = count as i32;
43
-
44
-
let result = sqlx::query(
45
-
"INSERT INTO queue_items (room_id, url, title, duration, thumbnail, source, added_by, position, added_at, played)
46
-
VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, 0)",
47
-
)
48
-
.bind(&self.room_id)
49
-
.bind(&meta.url)
50
-
.bind(&meta.title)
51
-
.bind(&meta.duration)
52
-
.bind(&meta.thumbnail)
53
-
.bind(meta.source.to_string())
54
-
.bind(None::<String>) // added_by
55
-
.bind(position)
56
-
.bind(crate::util::now_iso())
57
-
.execute(&self.pool)
58
-
.await?;
59
-
60
-
let id = result.last_insert_rowid();
61
-
62
-
let item = sqlx::query_as::<_, QueueItem>(
63
-
"SELECT id, url, title, duration, thumbnail, source, position, played, played_at, started_playing_at
64
-
FROM queue_items WHERE id = ?",
65
-
)
66
-
.bind(id)
67
-
.fetch_one(&self.pool)
68
-
.await?;
69
-
70
-
Ok(item)
71
-
}
72
-
73
-
/// List queued items (not playing, not finished) ordered by position ASC.
74
-
pub(crate) async fn upcoming(&self) -> Result<Vec<QueueItem>, sqlx::Error> {
75
-
let items = sqlx::query_as::<_, QueueItem>(
76
-
"SELECT id, url, title, duration, thumbnail, source, position, played, played_at, started_playing_at
77
-
FROM queue_items WHERE room_id = ? AND played = 0 AND started_playing_at IS NULL
78
-
ORDER BY position ASC",
79
-
)
80
-
.bind(&self.room_id)
81
-
.fetch_all(&self.pool)
82
-
.await?;
83
-
Ok(items)
84
-
}
85
-
86
-
/// List played items ordered by played_at DESC (newest first), limited to `limit`.
87
-
pub(crate) async fn history(&self, limit: i64) -> Result<Vec<QueueItem>, sqlx::Error> {
88
-
let items = sqlx::query_as::<_, QueueItem>(
89
-
"SELECT id, url, title, duration, thumbnail, source, position, played, played_at, started_playing_at
90
-
FROM queue_items WHERE room_id = ? AND played = 1 ORDER BY played_at DESC, id DESC LIMIT ?",
91
-
)
92
-
.bind(&self.room_id)
93
-
.bind(limit)
94
-
.fetch_all(&self.pool)
95
-
.await?;
96
-
Ok(items)
97
-
}
98
-
99
-
/// Mark an item as started playing (sets started_playing_at timestamp).
100
-
pub(crate) async fn mark_started(&self, id: i64, started_at: &str) -> Result<(), sqlx::Error> {
101
-
sqlx::query("UPDATE queue_items SET started_playing_at = ? WHERE id = ?")
102
-
.bind(started_at)
103
-
.bind(id)
104
-
.execute(&self.pool)
105
-
.await?;
106
-
Ok(())
107
-
}
108
-
109
-
/// Mark a currently-playing item as finished (moves it from "playing" to "played").
110
-
pub(crate) async fn mark_finished(&self, id: i64) -> Result<(), sqlx::Error> {
111
-
let now = crate::util::now_iso();
112
-
sqlx::query(
113
-
"UPDATE queue_items SET played = 1, played_at = ?, started_playing_at = NULL WHERE id = ?",
114
-
)
115
-
.bind(&now)
116
-
.bind(id)
117
-
.execute(&self.pool)
118
-
.await?;
119
-
Ok(())
120
-
}
121
-
122
-
/// Update metadata fields after async extraction completes.
123
-
///
124
-
/// Updates title, duration, thumbnail, and source for the given item.
125
-
/// Used when yt-dlp extraction finishes after the item was already
126
-
/// queued with placeholder values.
127
-
pub(crate) async fn update_metadata(
128
-
&self,
129
-
id: i64,
130
-
meta: &TrackMeta,
131
-
) -> Result<(), sqlx::Error> {
132
-
sqlx::query(
133
-
"UPDATE queue_items SET title = ?, duration = ?, thumbnail = ?, source = ? WHERE id = ?",
134
-
)
135
-
.bind(&meta.title)
136
-
.bind(&meta.duration)
137
-
.bind(&meta.thumbnail)
138
-
.bind(meta.source.to_string())
139
-
.bind(id)
140
-
.execute(&self.pool)
141
-
.await?;
142
-
Ok(())
143
-
}
144
-
}
145
-
146
-
#[cfg(test)]
147
-
mod tests {
148
-
use super::*;
149
-
use crate::db;
150
-
use crate::types;
151
-
use sqlx::sqlite::SqlitePoolOptions;
152
-
153
-
async fn test_playlist() -> Playlist {
154
-
let pool = SqlitePoolOptions::new()
155
-
.max_connections(1)
156
-
.connect(":memory:")
157
-
.await
158
-
.unwrap();
159
-
db::run_migrations(&pool).await.unwrap();
160
-
db::room_insert(&pool, "test-room", "test-name", false)
161
-
.await
162
-
.unwrap();
163
-
Playlist::new(pool, "test-room")
164
-
}
165
-
166
-
fn track(title: &str) -> types::TrackMeta {
167
-
types::TrackMeta {
168
-
title: title.into(),
169
-
duration: "3:45".into(),
170
-
thumbnail: None,
171
-
url: "https://example.com/track".into(),
172
-
source: crate::types::SourceKind::Direct,
173
-
}
174
-
}
175
-
176
-
#[tokio::test]
177
-
async fn test_push_and_upcoming() {
178
-
let pl = test_playlist().await;
179
-
180
-
let a = pl.push(&track("Track A")).await.unwrap();
181
-
assert!(a.id > 0);
182
-
assert!(!a.played);
183
-
assert_eq!(a.title, "Track A");
184
-
185
-
let b = pl.push(&track("Track B")).await.unwrap();
186
-
assert!(b.id > 0);
187
-
assert_eq!(b.position, 1);
188
-
189
-
let upcoming = pl.upcoming().await.unwrap();
190
-
assert_eq!(upcoming.len(), 2);
191
-
assert_eq!(upcoming[0].title, "Track A");
192
-
assert_eq!(upcoming[1].title, "Track B");
193
-
assert!(!upcoming[0].played);
194
-
assert!(!upcoming[1].played);
195
-
}
196
-
197
-
#[tokio::test]
198
-
async fn test_push_positions() {
199
-
let pl = test_playlist().await;
200
-
201
-
let a = pl.push(&track("A")).await.unwrap();
202
-
let b = pl.push(&track("B")).await.unwrap();
203
-
let c = pl.push(&track("C")).await.unwrap();
204
-
205
-
assert_eq!(a.position, 0);
206
-
assert_eq!(b.position, 1);
207
-
assert_eq!(c.position, 2);
208
-
}
209
-
210
-
#[tokio::test]
211
-
async fn test_mark_finished_non_existent_id() {
212
-
let pl = test_playlist().await;
213
-
pl.mark_finished(99999).await.unwrap();
214
-
}
215
-
216
-
#[tokio::test]
217
-
async fn test_upcoming_excludes_playing_items() {
218
-
let pl = test_playlist().await;
219
-
let a = pl.push(&track("A")).await.unwrap();
220
-
pl.push(&track("B")).await.unwrap();
221
-
222
-
let now = chrono::Utc::now()
223
-
.format("%Y-%m-%dT%H:%M:%S%.3fZ")
224
-
.to_string();
225
-
pl.mark_started(a.id, &now).await.unwrap();
226
-
227
-
let upcoming = pl.upcoming().await.unwrap();
228
-
assert_eq!(upcoming.len(), 1);
229
-
assert_eq!(upcoming[0].title, "B");
230
-
}
231
-
232
-
#[tokio::test]
233
-
async fn test_history() {
234
-
let pl = test_playlist().await;
235
-
236
-
let a = pl.push(&track("A")).await.unwrap();
237
-
pl.push(&track("B")).await.unwrap();
238
-
let c = pl.push(&track("C")).await.unwrap();
239
-
240
-
let now_a = chrono::Utc::now()
241
-
.format("%Y-%m-%dT%H:%M:%S%.3fZ")
242
-
.to_string();
243
-
pl.mark_started(a.id, &now_a).await.unwrap();
244
-
pl.mark_finished(a.id).await.unwrap();
245
-
246
-
tokio::time::sleep(std::time::Duration::from_millis(5)).await;
247
-
let now_c = chrono::Utc::now()
248
-
.format("%Y-%m-%dT%H:%M:%S%.3fZ")
249
-
.to_string();
250
-
pl.mark_started(c.id, &now_c).await.unwrap();
251
-
pl.mark_finished(c.id).await.unwrap();
252
-
253
-
let history = pl.history(10).await.unwrap();
254
-
assert_eq!(history.len(), 2);
255
-
assert_eq!(history[0].title, "C");
256
-
assert_eq!(history[1].title, "A");
257
-
assert!(history[0].played);
258
-
assert!(history[0].played_at.is_some());
259
-
}
260
-
261
-
#[tokio::test]
262
-
async fn test_history_only_after_mark_finished() {
263
-
let pl = test_playlist().await;
264
-
let a = pl.push(&track("A")).await.unwrap();
265
-
266
-
let now = chrono::Utc::now()
267
-
.format("%Y-%m-%dT%H:%M:%S%.3fZ")
268
-
.to_string();
269
-
pl.mark_started(a.id, &now).await.unwrap();
270
-
271
-
let history = pl.history(10).await.unwrap();
272
-
assert!(history.is_empty());
273
-
274
-
pl.mark_finished(a.id).await.unwrap();
275
-
276
-
let history = pl.history(10).await.unwrap();
277
-
assert_eq!(history.len(), 1);
278
-
assert_eq!(history[0].title, "A");
279
-
assert!(history[0].played);
280
-
assert!(history[0].played_at.is_some());
281
-
}
282
-
283
-
#[tokio::test]
284
-
async fn test_push_after_mark_finished() {
285
-
let pl = test_playlist().await;
286
-
let a = pl.push(&track("A")).await.unwrap();
287
-
let now = chrono::Utc::now()
288
-
.format("%Y-%m-%dT%H:%M:%S%.3fZ")
289
-
.to_string();
290
-
pl.mark_started(a.id, &now).await.unwrap();
291
-
pl.mark_finished(a.id).await.unwrap();
292
-
293
-
pl.push(&track("B")).await.unwrap();
294
-
295
-
let upcoming = pl.upcoming().await.unwrap();
296
-
assert_eq!(upcoming.len(), 1);
297
-
assert_eq!(upcoming[0].title, "B");
298
-
299
-
let history = pl.history(10).await.unwrap();
300
-
assert_eq!(history.len(), 1);
301
-
assert_eq!(history[0].title, "A");
302
-
}
303
-
304
-
#[tokio::test]
305
-
async fn test_history_order_multiple_finished() {
306
-
let pl = test_playlist().await;
307
-
let a = pl.push(&track("A")).await.unwrap();
308
-
let b = pl.push(&track("B")).await.unwrap();
309
-
let c = pl.push(&track("C")).await.unwrap();
310
-
311
-
let now_a = chrono::Utc::now()
312
-
.format("%Y-%m-%dT%H:%M:%S%.3fZ")
313
-
.to_string();
314
-
pl.mark_started(a.id, &now_a).await.unwrap();
315
-
pl.mark_finished(a.id).await.unwrap();
316
-
317
-
tokio::time::sleep(std::time::Duration::from_millis(5)).await;
318
-
let now_b = chrono::Utc::now()
319
-
.format("%Y-%m-%dT%H:%M:%S%.3fZ")
320
-
.to_string();
321
-
pl.mark_started(b.id, &now_b).await.unwrap();
322
-
pl.mark_finished(b.id).await.unwrap();
323
-
324
-
tokio::time::sleep(std::time::Duration::from_millis(5)).await;
325
-
let now_c = chrono::Utc::now()
326
-
.format("%Y-%m-%dT%H:%M:%S%.3fZ")
327
-
.to_string();
328
-
pl.mark_started(c.id, &now_c).await.unwrap();
329
-
pl.mark_finished(c.id).await.unwrap();
330
-
331
-
let history = pl.history(10).await.unwrap();
332
-
assert_eq!(history.len(), 3);
333
-
// Most recently finished first (C, B, A).
334
-
assert_eq!(history[0].title, "C");
335
-
assert_eq!(history[1].title, "B");
336
-
assert_eq!(history[2].title, "A");
337
-
}
338
-
}
+589
-490
src/room.rs
+589
-490
src/room.rs
···
18
18
19
19
use bytes::Bytes;
20
20
use chrono::{DateTime, Utc};
21
-
use sqlx::{Pool, Sqlite};
22
21
#[cfg(test)]
23
22
use tokio::sync::oneshot;
24
23
use tokio::sync::{RwLock, mpsc, watch};
25
24
26
-
use crate::db;
27
25
use crate::names::generate_room_name;
28
26
use crate::playback::Player;
29
-
use crate::playlist::Playlist;
30
27
use crate::source::SourceRegistry;
31
-
use crate::state::{self, Effect, Event, PlaybackState};
28
+
use crate::state::{Effect, Event, PlaybackState, QueuedTrack};
29
+
use crate::store::RoomStore;
32
30
use crate::transport::TrackPublishers;
33
31
use crate::types::{
34
-
ChatMessage, HistoryEntry, QueueSummary, RoomCommand, RoomHandle, RoomId, TrackMeta, TrackState,
32
+
ActiveTrackInfo, ChatMessage, HistoryEntry, QueueSummary, RoomCommand, RoomHandle, RoomId,
33
+
TrackMeta, TrackState,
35
34
};
36
35
37
36
/// Distributed unique ID generator (Discord-style snowflake).
···
117
116
room_id: RoomId,
118
117
room_name: String,
119
118
publishers: TrackPublishers,
120
-
pool: Pool<Sqlite>,
119
+
store: Arc<dyn RoomStore>,
120
+
id_gen: Arc<SnowflakeIdGen>,
121
121
source_registry: Arc<SourceRegistry>,
122
122
cache_dir: PathBuf,
123
123
/// Clone of the command sender, used by spawned extraction tasks to
···
132
132
last_active_tx: watch::Sender<DateTime<Utc>>,
133
133
}
134
134
135
-
/// Thread-safe registry of all active rooms with SQLite persistence.
135
+
/// Thread-safe registry of all active rooms with persistent store.
136
136
#[derive(Clone)]
137
137
pub(crate) struct Registry {
138
-
pool: Pool<Sqlite>,
138
+
store: Arc<dyn RoomStore>,
139
139
id_gen: Arc<SnowflakeIdGen>,
140
140
cache_dir: PathBuf,
141
141
sources: Arc<SourceRegistry>,
···
148
148
149
149
impl Registry {
150
150
pub(crate) fn new(
151
-
pool: Pool<Sqlite>,
151
+
store: Arc<dyn RoomStore>,
152
152
cache_dir: PathBuf,
153
153
sources: Arc<SourceRegistry>,
154
154
) -> Self {
155
155
Self {
156
-
pool,
156
+
store,
157
157
id_gen: Arc::new(SnowflakeIdGen::new(1)),
158
158
inner: Arc::new(RwLock::new(Inner {
159
159
rooms: HashMap::new(),
···
163
163
}
164
164
}
165
165
166
-
/// Create a new room with a snowflake ID, persist to SQLite.
166
+
/// Create a new room with a snowflake ID, persist to store.
167
167
pub(crate) async fn create(&self) -> RoomId {
168
168
let id = RoomId(self.id_gen.next_id().await.to_string());
169
169
let room_name = generate_room_name(&id.0);
170
170
let now = Utc::now();
171
-
let _ = db::room_insert(&self.pool, &id.0, &room_name, false).await;
171
+
let _ = self.store.create_room(&id.0, &room_name).await;
172
172
173
173
let publishers = TrackPublishers::new();
174
174
let (tx, rx) = mpsc::channel(256);
···
180
180
room_id: id.clone(),
181
181
room_name: room_name.clone(),
182
182
publishers: publishers.clone(),
183
-
pool: self.pool.clone(),
183
+
store: self.store.clone(),
184
+
id_gen: self.id_gen.clone(),
184
185
source_registry: self.sources.clone(),
185
186
cache_dir: self.cache_dir.clone(),
186
187
cmd_tx: tx.clone(),
···
204
205
id
205
206
}
206
207
207
-
/// Get the room handle if the room is active.
208
+
/// Get the room handle. Rehydrates from store if the room exists but
209
+
/// its actor was swept (idle timeout).
208
210
pub(crate) async fn handle(&self, id: &RoomId) -> Option<RoomHandle> {
209
-
self.inner.read().await.rooms.get(id).cloned()
211
+
// Fast path: room is already active
212
+
if let Some(handle) = self.inner.read().await.rooms.get(id).cloned() {
213
+
return Some(handle);
214
+
}
215
+
216
+
// Slow path: try to rehydrate from store
217
+
let snapshot = self.store.load(&id.0).await.ok()??;
218
+
219
+
let (state, _effects) = PlaybackState::from_snapshot(snapshot.clone());
220
+
221
+
let publishers = TrackPublishers::new();
222
+
let (tx, rx) = mpsc::channel(256);
223
+
let (last_active_tx, last_active_rx) = watch::channel(Utc::now());
224
+
225
+
let player = Player::new(self.sources.clone(), self.cache_dir.clone(), tx.clone());
226
+
let actor = RoomActor {
227
+
rx,
228
+
room_id: id.clone(),
229
+
room_name: snapshot.room_name,
230
+
publishers: publishers.clone(),
231
+
store: self.store.clone(),
232
+
id_gen: self.id_gen.clone(),
233
+
source_registry: self.sources.clone(),
234
+
cache_dir: self.cache_dir.clone(),
235
+
cmd_tx: tx.clone(),
236
+
state,
237
+
player,
238
+
client_count: 0,
239
+
next_user_id: 1,
240
+
last_active_tx,
241
+
};
242
+
tokio::spawn(actor.run());
243
+
244
+
let handle = RoomHandle {
245
+
cmd_tx: tx,
246
+
publishers,
247
+
last_active: last_active_rx,
248
+
};
249
+
self.inner
250
+
.write()
251
+
.await
252
+
.rooms
253
+
.insert(id.clone(), handle.clone());
254
+
Some(handle)
210
255
}
211
256
212
-
/// Check whether a room exists in SQLite.
257
+
/// Check whether a room exists in the store.
213
258
#[must_use]
214
259
pub(crate) async fn exists(&self, id: &RoomId) -> bool {
215
-
db::room_exists(&self.pool, &id.0).await.unwrap_or(false)
260
+
if self.inner.read().await.rooms.contains_key(id) {
261
+
return true;
262
+
}
263
+
self.store.load(&id.0).await.ok().flatten().is_some()
216
264
}
217
265
218
-
/// Get the upcoming queue for a room from SQLite via Playlist.
266
+
/// Load room name + history (for SSR rendering).
267
+
pub(crate) async fn load_room_data(&self, id: &RoomId) -> Option<(String, Vec<TrackMeta>)> {
268
+
let snapshot = self.store.load(&id.0).await.ok()??;
269
+
let name = snapshot.room_name;
270
+
let history = snapshot
271
+
.history
272
+
.into_iter()
273
+
.map(|h| TrackMeta {
274
+
title: h.title,
275
+
duration: h.duration,
276
+
thumbnail: h.thumbnail,
277
+
url: h.url,
278
+
source: "direct".parse().unwrap(),
279
+
})
280
+
.collect();
281
+
Some((name, history))
282
+
}
283
+
284
+
/// Get the upcoming queue for a room from the store.
219
285
pub(crate) async fn queue(&self, id: &RoomId) -> Vec<TrackMeta> {
220
-
let playlist = Playlist::new(self.pool.clone(), &id.0);
221
-
let items = playlist.upcoming().await.unwrap_or_default();
222
-
items
286
+
let snapshot = match self.store.load(&id.0).await {
287
+
Ok(Some(s)) => s,
288
+
_ => return Vec::new(),
289
+
};
290
+
snapshot
291
+
.queue
223
292
.into_iter()
224
-
.map(|i| TrackMeta {
225
-
title: i.title,
226
-
duration: i.duration,
227
-
thumbnail: i.thumbnail,
228
-
url: i.url,
229
-
source: i.source.parse().unwrap_or_else(|s: String| {
230
-
tracing::warn!(source = %s, "unknown source kind in queue_items");
231
-
crate::types::SourceKind::Direct
232
-
}),
293
+
.map(|t| TrackMeta {
294
+
title: t.title,
295
+
duration: t.duration,
296
+
thumbnail: t.thumbnail,
297
+
url: t.url,
298
+
source: "direct".parse().unwrap(), // best-effort, source is not tracked in QueuedTrack
233
299
})
234
300
.collect()
235
301
}
···
244
310
let _ = handle.cmd_tx.send(RoomCommand::QueueUrl(url)).await;
245
311
}
246
312
247
-
/// Remove a room (from SQLite and in-memory state).
313
+
/// Remove a room (from store and in-memory state).
248
314
#[cfg(test)]
249
315
pub(crate) async fn remove(&self, id: &RoomId) {
250
316
if let Some(handle) = self.handle(id).await {
···
255
321
.await;
256
322
}
257
323
self.inner.write().await.rooms.remove(id);
258
-
let _ = db::room_delete(&self.pool, &id.0).await;
324
+
let _ = self.store.delete_room(&id.0).await;
259
325
}
260
326
261
327
/// Send a chat message: validate, send to actor, return best-effort ChatMessage.
···
300
366
})
301
367
}
302
368
303
-
/// Get the most recent chat messages from SQLite.
369
+
/// Get the most recent chat messages from the store.
304
370
pub(crate) async fn recent_chat(
305
371
&self,
306
372
room_id: &RoomId,
307
373
limit: i64,
308
-
) -> Result<Vec<ChatMessage>, anyhow::Error> {
309
-
let rows = db::chat_recent(&self.pool, &room_id.0, limit).await?;
310
-
Ok(rows
311
-
.into_iter()
312
-
.map(|r| ChatMessage {
313
-
id: r.id,
314
-
user_name: r.user_name,
315
-
content: r.content,
316
-
msg_type: r.msg_type,
317
-
created_at: r.created_at,
318
-
})
319
-
.collect())
374
+
) -> Result<Vec<ChatMessage>, sqlx::Error> {
375
+
self.store.recent_chat(&room_id.0, limit).await
320
376
}
321
377
322
378
/// Register a client in the room.
···
378
434
}
379
435
}
380
436
self.inner.write().await.rooms.remove(id);
381
-
let _ = db::room_delete(&self.pool, &id.0).await;
437
+
// Keep the room in the store so it can be rehydrated on rejoin.
382
438
swept.push(id.clone());
383
439
tracing::info!("swept idle room {id}");
384
440
}
···
409
465
RoomCommand::QueueUrl(url) => {
410
466
tracing::debug!(room = %self.room_id, %url, "track url queued");
411
467
412
-
let playlist = Playlist::new(self.pool.clone(), &self.room_id.0);
413
-
let placeholder = TrackMeta {
468
+
let item_id = self.id_gen.next_id().await as i64;
469
+
let now = Utc::now().timestamp_millis();
470
+
471
+
let track = QueuedTrack {
472
+
id: item_id,
414
473
title: "Loading...".into(),
474
+
url: url.clone(),
415
475
duration: "--:--".into(),
416
476
thumbnail: None,
417
-
url: url.clone(),
418
-
source: crate::types::SourceKind::Direct,
477
+
pending: true,
419
478
};
420
-
let item = match playlist.push(&placeholder).await {
421
-
Ok(item) => item,
422
-
Err(e) => {
423
-
tracing::warn!("failed to persist queue item: {e}");
424
-
return false;
425
-
}
426
-
};
427
-
let now = Utc::now().timestamp_millis();
428
479
429
-
let effects = self.state.transition(
430
-
&Event::TrackQueued(state::QueuedTrack {
431
-
id: item.id,
432
-
title: placeholder.title,
433
-
url: placeholder.url,
434
-
duration: placeholder.duration,
435
-
thumbnail: placeholder.thumbnail,
436
-
pending: true,
437
-
}),
438
-
now,
439
-
);
480
+
let effects = self.state.transition(&Event::TrackQueued(track), now);
481
+
482
+
if let Err(e) = self.store.persist(&self.room_id.0, &effects).await {
483
+
tracing::warn!("failed to persist queued track: {e}");
484
+
}
440
485
441
486
self.execute_effects(effects).await;
442
487
443
-
let item_id = item.id;
444
488
let cmd_tx = self.cmd_tx.clone();
445
489
let sources = self.source_registry.clone();
446
490
let cache_dir = self.cache_dir.clone();
···
468
512
469
513
let now = Utc::now().timestamp_millis();
470
514
471
-
let playlist = Playlist::new(self.pool.clone(), &self.room_id.0);
472
-
if let Err(e) = playlist.update_metadata(item_id, &meta).await {
473
-
tracing::warn!(item_id, error = %e, "failed to persist metadata update");
474
-
}
475
-
476
515
let effects = self.state.transition(
477
516
&Event::MetadataUpdated {
478
517
item_id,
479
518
title: meta.title,
480
519
duration: meta.duration,
481
520
thumbnail: meta.thumbnail,
521
+
source: meta.source.to_string(),
482
522
},
483
523
now,
484
524
);
485
525
526
+
if let Err(e) = self.store.persist(&self.room_id.0, &effects).await {
527
+
tracing::warn!(item_id, error = %e, "failed to persist metadata");
528
+
}
529
+
486
530
self.execute_effects(effects).await;
487
531
}
488
532
···
496
540
title: format!("Failed to load: {error}"),
497
541
duration: "--:--".into(),
498
542
thumbnail: None,
543
+
source: "ytdlp".into(),
499
544
},
500
545
now,
501
546
);
547
+
548
+
if let Err(e) = self.store.persist(&self.room_id.0, &effects).await {
549
+
tracing::warn!(item_id, error = %e, "failed to persist metadata failure");
550
+
}
551
+
502
552
self.execute_effects(effects).await;
503
553
}
504
554
···
509
559
510
560
let effects = self.state.transition(&Event::Skip, now);
511
561
562
+
if let Err(e) = self.store.persist(&self.room_id.0, &effects).await {
563
+
tracing::warn!("failed to persist skip effects: {e}");
564
+
}
565
+
512
566
self.execute_effects(effects).await;
513
567
}
514
568
···
522
576
let now = Utc::now().timestamp_millis();
523
577
524
578
let effects = self.state.transition(&Event::TrackEnded { item_id }, now);
579
+
580
+
if let Err(e) = self.store.persist(&self.room_id.0, &effects).await {
581
+
tracing::warn!("failed to persist track ended effects: {e}");
582
+
}
525
583
526
584
self.execute_effects(effects).await;
527
585
}
···
534
592
let created_at = chrono::Utc::now()
535
593
.format("%Y-%m-%dT%H:%M:%S%.3fZ")
536
594
.to_string();
537
-
let msg = db::NewChatMessage {
538
-
room_id: self.room_id.0.clone(),
595
+
let effects = vec![Effect::PersistChat {
539
596
user_name: user_name.clone(),
540
597
content: content.clone(),
541
598
msg_type: msg_type.clone(),
542
599
created_at: created_at.clone(),
543
-
};
544
-
if let Ok(msg_id) = db::chat_insert(&self.pool, &msg).await {
545
-
let json = serde_json::json!({
546
-
"user": user_name,
547
-
"content": content,
548
-
"type": msg_type,
549
-
"ts": created_at,
550
-
"id": msg_id,
551
-
});
552
-
let payload = serde_json::to_vec(&json).unwrap_or_default();
553
-
self.publishers.publish_chat(0, Bytes::from(payload));
600
+
}];
601
+
if let Ok(output) = self.store.persist(&self.room_id.0, &effects).await {
602
+
if let Some(msg_id) = output.chat_message_id {
603
+
let json = serde_json::json!({
604
+
"user": user_name,
605
+
"content": content,
606
+
"type": msg_type,
607
+
"ts": created_at,
608
+
"id": msg_id,
609
+
});
610
+
let payload = serde_json::to_vec(&json).unwrap_or_default();
611
+
self.publishers.publish_chat(0, Bytes::from(payload));
612
+
}
554
613
}
555
614
}
556
615
···
576
635
false
577
636
}
578
637
579
-
/// Execute a batch of effects from a state transition.
638
+
/// Execute actor-side effects from a state transition.
639
+
///
640
+
/// Persistence effects are handled by [`RoomStore::persist`] before this
641
+
/// is called — only pipeline and publishing effects remain.
580
642
async fn execute_effects(&mut self, effects: Vec<Effect>) {
581
643
for effect in effects {
582
644
match effect {
583
645
Effect::AbortPipeline => {
584
646
self.player.abort();
585
647
}
586
-
587
-
Effect::PersistStarted(id) => {
588
-
let now = Utc::now().format("%Y-%m-%dT%H:%M:%S%.3fZ").to_string();
589
-
let playlist = Playlist::new(self.pool.clone(), &self.room_id.0);
590
-
let _ = playlist.mark_started(id, &now).await;
591
-
}
592
-
593
648
Effect::StartPipeline(track) => {
594
649
let now = Utc::now().timestamp_millis();
595
650
self.state.resolve_started_at(now);
596
-
let info = crate::types::ActiveTrackInfo {
651
+
let info = ActiveTrackInfo {
597
652
id: track.id,
598
653
title: track.title,
599
654
url: track.url,
···
603
658
};
604
659
self.player.start(&info, self.publishers.clone());
605
660
}
606
-
607
-
Effect::PersistFinished(id) => {
608
-
let playlist = Playlist::new(self.pool.clone(), &self.room_id.0);
609
-
let _ = playlist.mark_finished(id).await;
610
-
}
611
-
612
661
Effect::PublishSnapshot => {
613
662
self.publish_state_snapshot().await;
614
663
}
664
+
// All persistence effects handled by store.persist() in process().
665
+
_ => {}
615
666
}
616
667
}
617
668
}
···
679
730
#[cfg(test)]
680
731
mod tests {
681
732
use super::*;
682
-
use sqlx::sqlite::SqlitePoolOptions;
733
+
use crate::store::{InMemoryStore, SqliteStore};
734
+
735
+
// -----------------------------------------------------------------------
736
+
// Shared test infrastructure
737
+
// -----------------------------------------------------------------------
683
738
684
739
async fn poll_until<F, Fut>(check: F, timeout: Duration, label: &str)
685
740
where
···
698
753
}
699
754
}
700
755
701
-
async fn test_registry_and_room() -> (Registry, RoomId) {
702
-
let pool = SqlitePoolOptions::new()
703
-
.max_connections(1)
704
-
.connect(":memory:")
756
+
async fn history_len(store: &Arc<dyn RoomStore>, room_id: &str) -> usize {
757
+
store
758
+
.load(room_id)
705
759
.await
706
-
.unwrap();
707
-
db::run_migrations(&pool).await.unwrap();
708
-
let reg = Registry::new(pool, PathBuf::from("/tmp/moqbox"), test_registry());
709
-
let room_id = reg.create().await;
710
-
(reg, room_id)
760
+
.ok()
761
+
.flatten()
762
+
.map(|s| s.history.len())
763
+
.unwrap_or(0)
711
764
}
712
765
713
766
struct PanaceaSource;
···
744
797
Arc::new(reg)
745
798
}
746
799
800
+
// -----------------------------------------------------------------------
801
+
// Backend-independent test: snowflake generator
802
+
// -----------------------------------------------------------------------
803
+
747
804
#[tokio::test]
748
805
async fn test_snowflake_uniqueness() {
749
806
let gen_id = Arc::new(SnowflakeIdGen::new(1));
···
768
825
);
769
826
}
770
827
771
-
#[tokio::test]
772
-
async fn test_create_and_exists() {
773
-
let (reg, room_id) = test_registry_and_room().await;
774
-
assert!(reg.exists(&room_id).await);
775
-
}
828
+
// -----------------------------------------------------------------------
829
+
// Shared scenario functions — run against any RoomStore backend
830
+
// -----------------------------------------------------------------------
776
831
777
-
#[tokio::test]
778
-
async fn test_push_and_queue() {
779
-
let (reg, room_id) = test_registry_and_room().await;
832
+
mod scenarios {
833
+
use super::*;
780
834
781
-
reg.push_url(&room_id, "https://example.com/a".into()).await;
782
-
reg.push_url(&room_id, "https://example.com/b".into()).await;
835
+
pub async fn create_and_exists(reg: &Registry, room_id: &RoomId) {
836
+
assert!(reg.exists(room_id).await);
837
+
}
783
838
784
-
poll_until(
785
-
|| async {
786
-
let q = reg.queue(&room_id).await;
787
-
q.len() == 1 && q[0].title == "b"
788
-
},
789
-
Duration::from_secs(2),
790
-
"queue should have 1 item with title 'b'",
791
-
)
792
-
.await;
793
-
}
839
+
pub async fn push_and_queue(reg: &Registry, room_id: &RoomId) {
840
+
reg.push_url(room_id, "https://example.com/a".into()).await;
841
+
reg.push_url(room_id, "https://example.com/b".into()).await;
794
842
795
-
#[tokio::test]
796
-
async fn test_remove() {
797
-
let (reg, room_id) = test_registry_and_room().await;
843
+
poll_until(
844
+
|| async {
845
+
let q = reg.queue(room_id).await;
846
+
q.len() == 1 && q[0].title == "b"
847
+
},
848
+
Duration::from_secs(2),
849
+
"queue should have 1 item with title 'b'",
850
+
)
851
+
.await;
852
+
}
798
853
799
-
reg.remove(&room_id).await;
800
-
assert!(!reg.exists(&room_id).await);
801
-
}
854
+
pub async fn push_starts_playback_when_idle(reg: &Registry, room_id: &RoomId) {
855
+
reg.push_url(room_id, "https://example.com/t".into()).await;
856
+
poll_until(
857
+
|| async { reg.queue(room_id).await.is_empty() },
858
+
Duration::from_secs(2),
859
+
"queue should be empty (item started playing)",
860
+
)
861
+
.await;
862
+
}
802
863
803
-
#[tokio::test]
804
-
async fn test_send_chat() {
805
-
let (reg, room_id) = test_registry_and_room().await;
864
+
pub async fn push_appends_when_already_playing(reg: &Registry, room_id: &RoomId) {
865
+
reg.push_url(room_id, "https://example.com/a".into()).await;
866
+
poll_until(
867
+
|| async { reg.queue(room_id).await.is_empty() },
868
+
Duration::from_secs(2),
869
+
"first push should have started playing",
870
+
)
871
+
.await;
806
872
807
-
let msg = reg
808
-
.send_chat(&room_id, "alice", "hello world", "message")
809
-
.await
810
-
.unwrap();
811
-
assert_eq!(msg.user_name, "alice");
812
-
assert_eq!(msg.content, "hello world");
813
-
assert_eq!(msg.msg_type, "message");
814
-
assert!(!msg.created_at.is_empty());
815
-
}
873
+
reg.push_url(room_id, "https://example.com/b".into()).await;
874
+
poll_until(
875
+
|| async { reg.queue(room_id).await.len() == 1 },
876
+
Duration::from_secs(2),
877
+
"second push should be queued",
878
+
)
879
+
.await;
880
+
}
816
881
817
-
#[tokio::test]
818
-
async fn test_send_chat_persistence() {
819
-
let (reg, room_id) = test_registry_and_room().await;
882
+
pub async fn skip_when_idle_does_not_panic(reg: &Registry, room_id: &RoomId) {
883
+
reg.skip(room_id).await;
884
+
}
820
885
821
-
let _ = reg
822
-
.send_chat(&room_id, "alice", "msg1", "message")
823
-
.await
824
-
.unwrap();
825
-
let _ = reg
826
-
.send_chat(&room_id, "bob", "msg2", "message")
827
-
.await
828
-
.unwrap();
886
+
pub async fn skip_moves_track_to_history(reg: &Registry, room_id: &RoomId) {
887
+
let store = reg.store.clone();
829
888
830
-
poll_until(
831
-
|| async { reg.recent_chat(&room_id, 10).await.unwrap().len() == 2 },
832
-
Duration::from_secs(2),
833
-
"recent_chat should have 2 messages",
834
-
)
835
-
.await;
889
+
reg.push_url(room_id, "https://example.com/a".into()).await;
890
+
poll_until(
891
+
|| async { reg.queue(room_id).await.is_empty() },
892
+
Duration::from_secs(2),
893
+
"first push should have started playing",
894
+
)
895
+
.await;
836
896
837
-
let recent = reg.recent_chat(&room_id, 10).await.unwrap();
838
-
assert_eq!(recent[0].user_name, "bob");
839
-
assert_eq!(recent[0].content, "msg2");
840
-
assert_eq!(recent[1].user_name, "alice");
841
-
assert_eq!(recent[1].content, "msg1");
842
-
}
897
+
reg.skip(room_id).await;
898
+
poll_until(
899
+
|| async { history_len(&store, &room_id.0).await == 1 },
900
+
Duration::from_secs(2),
901
+
"skip should move track to history",
902
+
)
903
+
.await;
843
904
844
-
#[tokio::test]
845
-
async fn test_send_chat_empty_content() {
846
-
let (reg, room_id) = test_registry_and_room().await;
905
+
assert!(reg.queue(room_id).await.is_empty());
906
+
}
847
907
848
-
let result = reg.send_chat(&room_id, "alice", "", "message").await;
849
-
result.unwrap_err();
850
-
}
908
+
pub async fn multiple_skips_sequential(reg: &Registry, room_id: &RoomId) {
909
+
for url in ["https://x/a", "https://x/b", "https://x/c"] {
910
+
reg.push_url(room_id, url.into()).await;
911
+
}
912
+
poll_until(
913
+
|| async { reg.queue(room_id).await.len() == 2 },
914
+
Duration::from_secs(2),
915
+
"3 pushes → 1 active, 2 queued",
916
+
)
917
+
.await;
851
918
852
-
#[tokio::test]
853
-
async fn test_send_chat_long_content() {
854
-
let (reg, room_id) = test_registry_and_room().await;
919
+
reg.skip(room_id).await;
920
+
poll_until(
921
+
|| async { reg.queue(room_id).await.len() == 1 },
922
+
Duration::from_secs(2),
923
+
"skip → 1 item left in queue",
924
+
)
925
+
.await;
855
926
856
-
let long = "a".repeat(2001);
857
-
let result = reg.send_chat(&room_id, "alice", &long, "message").await;
858
-
result.unwrap_err();
859
-
}
927
+
reg.skip(room_id).await;
928
+
poll_until(
929
+
|| async { reg.queue(room_id).await.is_empty() },
930
+
Duration::from_secs(2),
931
+
"skip → queue empty",
932
+
)
933
+
.await;
934
+
}
860
935
861
-
#[tokio::test]
862
-
async fn test_send_chat_long_username() {
863
-
let (reg, room_id) = test_registry_and_room().await;
936
+
pub async fn double_skip_does_not_corrupt_state(reg: &Registry, room_id: &RoomId) {
937
+
reg.push_url(room_id, "https://example.com/a".into()).await;
938
+
poll_until(
939
+
|| async { reg.queue(room_id).await.is_empty() },
940
+
Duration::from_secs(2),
941
+
"first push started playing",
942
+
)
943
+
.await;
864
944
865
-
let long = "a".repeat(33);
866
-
let result = reg.send_chat(&room_id, &long, "hello", "message").await;
867
-
result.unwrap_err();
868
-
}
945
+
reg.push_url(room_id, "https://example.com/b".into()).await;
946
+
poll_until(
947
+
|| async { reg.queue(room_id).await.len() == 1 },
948
+
Duration::from_secs(2),
949
+
"second push queued",
950
+
)
951
+
.await;
869
952
870
-
#[tokio::test]
871
-
async fn test_register_client() {
872
-
let (reg, room_id) = test_registry_and_room().await;
953
+
reg.skip(room_id).await;
954
+
reg.skip(room_id).await;
955
+
poll_until(
956
+
|| async {
957
+
match reg.store.load(&room_id.0).await.ok().flatten() {
958
+
Some(s) => s.history.first().map(|h| h.title.as_str()) == Some("b"),
959
+
None => false,
960
+
}
961
+
},
962
+
Duration::from_secs(2),
963
+
"history should have 'b' as most recent",
964
+
)
965
+
.await;
873
966
874
-
let id1 = reg.register_client(&room_id).await;
875
-
assert_eq!(id1, Some(1));
967
+
let snapshot = reg.store.load(&room_id.0).await.ok().flatten().unwrap();
968
+
assert!(!snapshot.history.is_empty());
969
+
assert_eq!(snapshot.history[0].title, "b");
970
+
}
876
971
877
-
let id2 = reg.register_client(&room_id).await;
878
-
assert_eq!(id2, Some(2));
879
-
}
972
+
pub async fn send_chat(reg: &Registry, room_id: &RoomId) {
973
+
let msg = reg
974
+
.send_chat(room_id, "alice", "hello world", "message")
975
+
.await
976
+
.unwrap();
977
+
assert_eq!(msg.user_name, "alice");
978
+
assert_eq!(msg.content, "hello world");
979
+
assert_eq!(msg.msg_type, "message");
980
+
assert!(!msg.created_at.is_empty());
981
+
}
880
982
881
-
#[tokio::test]
882
-
async fn test_register_client_nonexistent_room() {
883
-
let (reg, _) = test_registry_and_room().await;
884
-
let fake_id = RoomId("nonexistent".into());
885
-
assert_eq!(reg.register_client(&fake_id).await, None);
886
-
}
983
+
pub async fn send_chat_persistence(reg: &Registry, room_id: &RoomId) {
984
+
let _ = reg
985
+
.send_chat(room_id, "alice", "msg1", "message")
986
+
.await
987
+
.unwrap();
988
+
let _ = reg
989
+
.send_chat(room_id, "bob", "msg2", "message")
990
+
.await
991
+
.unwrap();
887
992
888
-
#[tokio::test]
889
-
async fn test_unregister_client() {
890
-
let (reg, room_id) = test_registry_and_room().await;
993
+
poll_until(
994
+
|| async { reg.recent_chat(room_id, 10).await.unwrap().len() == 2 },
995
+
Duration::from_secs(2),
996
+
"recent_chat should have 2 messages",
997
+
)
998
+
.await;
891
999
892
-
reg.register_client(&room_id).await;
893
-
reg.unregister_client(&room_id).await;
1000
+
let recent = reg.recent_chat(room_id, 10).await.unwrap();
1001
+
assert_eq!(recent[0].user_name, "bob");
1002
+
assert_eq!(recent[0].content, "msg2");
1003
+
assert_eq!(recent[1].user_name, "alice");
1004
+
assert_eq!(recent[1].content, "msg1");
1005
+
}
894
1006
895
-
let id = reg.register_client(&room_id).await;
896
-
assert_eq!(id, Some(2));
897
-
}
1007
+
pub async fn remove_room(reg: &Registry, room_id: &RoomId) {
1008
+
reg.remove(room_id).await;
1009
+
assert!(!reg.exists(room_id).await);
1010
+
}
898
1011
899
-
#[tokio::test]
900
-
async fn test_skip_when_idle_does_not_panic() {
901
-
let (reg, room_id) = test_registry_and_room().await;
902
-
reg.skip(&room_id).await;
903
-
}
1012
+
pub async fn remove_while_playing(reg: &Registry, room_id: &RoomId) {
1013
+
reg.push_url(room_id, "https://example.com/a".into()).await;
1014
+
poll_until(
1015
+
|| async { reg.queue(room_id).await.is_empty() },
1016
+
Duration::from_secs(2),
1017
+
"push should have started playing",
1018
+
)
1019
+
.await;
904
1020
905
-
#[tokio::test]
906
-
async fn test_double_skip_does_not_deadlock() {
907
-
let (reg, room_id) = test_registry_and_room().await;
1021
+
reg.remove(room_id).await;
1022
+
assert!(!reg.exists(room_id).await);
1023
+
}
908
1024
909
-
reg.push_url(&room_id, "https://example.com/a".into()).await;
910
-
poll_until(
911
-
|| async { reg.queue(&room_id).await.is_empty() },
912
-
Duration::from_secs(2),
913
-
"queue should be empty after push starts playing",
914
-
)
915
-
.await;
1025
+
pub async fn sweep_keeps_room_in_store(reg: &Registry, room_id: &RoomId) {
1026
+
tokio::time::sleep(Duration::from_millis(1)).await;
916
1027
917
-
let r1 = reg.clone();
918
-
let rid1 = room_id.clone();
919
-
let h1 = tokio::spawn(async move { r1.skip(&rid1).await });
920
-
let r2 = reg.clone();
921
-
let rid2 = room_id.clone();
922
-
let h2 = tokio::spawn(async move { r2.skip(&rid2).await });
923
-
let _ = tokio::join!(h1, h2);
924
-
}
1028
+
let swept = reg.sweep_idle(Duration::from_secs(0)).await;
1029
+
assert!(swept.contains(room_id));
925
1030
926
-
#[tokio::test]
927
-
async fn test_send_chat_returns_id_zero() {
928
-
let (reg, room_id) = test_registry_and_room().await;
929
-
let msg = reg
930
-
.send_chat(&room_id, "alice", "hello", "message")
931
-
.await
932
-
.unwrap();
933
-
assert_eq!(msg.id, 0);
934
-
}
1031
+
assert!(
1032
+
reg.exists(room_id).await,
1033
+
"room must remain in store after sweep"
1034
+
);
1035
+
assert!(
1036
+
reg.handle(room_id).await.is_some(),
1037
+
"handle() rehydrates after sweep"
1038
+
);
1039
+
}
935
1040
936
-
#[tokio::test]
937
-
async fn test_sweep_idle_removes_inactive_rooms() {
938
-
let (reg, room_id) = test_registry_and_room().await;
1041
+
pub async fn sweep_preserves_active_rooms(reg: &Registry, room_id: &RoomId) {
1042
+
reg.register_client(room_id).await;
1043
+
let swept = reg.sweep_idle(Duration::from_millis(100)).await;
1044
+
assert!(!swept.contains(room_id));
1045
+
}
939
1046
940
-
tokio::time::sleep(Duration::from_millis(1)).await;
1047
+
pub async fn multiple_rooms_dont_interfere(reg: &Registry, room_a: &RoomId) {
1048
+
let room_b = reg.create().await;
941
1049
942
-
let swept = reg.sweep_idle(Duration::from_secs(0)).await;
943
-
assert!(swept.contains(&room_id));
944
-
assert!(!reg.exists(&room_id).await);
945
-
}
1050
+
assert_ne!(room_a, &room_b);
1051
+
assert!(reg.exists(room_a).await);
1052
+
assert!(reg.exists(&room_b).await);
946
1053
947
-
#[tokio::test]
948
-
async fn test_sweep_idle_preserves_active_rooms() {
949
-
let (reg, room_id) = test_registry_and_room().await;
1054
+
reg.push_url(room_a, "https://example.com/a".into()).await;
1055
+
reg.push_url(&room_b, "https://example.com/b".into()).await;
950
1056
951
-
reg.register_client(&room_id).await;
952
-
let swept = reg.sweep_idle(Duration::from_millis(100)).await;
953
-
assert!(!swept.contains(&room_id));
954
-
}
1057
+
poll_until(
1058
+
|| async { reg.queue(room_a).await.is_empty() },
1059
+
Duration::from_secs(2),
1060
+
"room_a queue empty",
1061
+
)
1062
+
.await;
1063
+
poll_until(
1064
+
|| async { reg.queue(&room_b).await.is_empty() },
1065
+
Duration::from_secs(2),
1066
+
"room_b queue empty",
1067
+
)
1068
+
.await;
1069
+
}
955
1070
956
-
#[tokio::test]
957
-
async fn test_push_starts_playback_when_idle() {
958
-
let (reg, room_id) = test_registry_and_room().await;
1071
+
pub async fn push_after_skip_works(reg: &Registry, room_id: &RoomId) {
1072
+
let store = reg.store.clone();
959
1073
960
-
reg.push_url(&room_id, "https://example.com/t".into()).await;
961
-
poll_until(
962
-
|| async { reg.queue(&room_id).await.is_empty() },
963
-
Duration::from_secs(2),
964
-
"queue should be empty (item started playing)",
965
-
)
966
-
.await;
967
-
}
1074
+
reg.push_url(room_id, "https://a".into()).await;
1075
+
poll_until(
1076
+
|| async { reg.queue(room_id).await.is_empty() },
1077
+
Duration::from_secs(2),
1078
+
"first push started playing",
1079
+
)
1080
+
.await;
968
1081
969
-
#[tokio::test]
970
-
async fn test_push_appends_when_already_playing() {
971
-
let (reg, room_id) = test_registry_and_room().await;
1082
+
reg.skip(room_id).await;
1083
+
poll_until(
1084
+
|| async { history_len(&store, &room_id.0).await > 0 },
1085
+
Duration::from_secs(2),
1086
+
"skip should move track to history",
1087
+
)
1088
+
.await;
972
1089
973
-
reg.push_url(&room_id, "https://example.com/a".into()).await;
974
-
poll_until(
975
-
|| async { reg.queue(&room_id).await.is_empty() },
976
-
Duration::from_secs(2),
977
-
"first push should have started playing",
978
-
)
979
-
.await;
1090
+
reg.push_url(room_id, "https://b".into()).await;
1091
+
poll_until(
1092
+
|| async { reg.queue(room_id).await.is_empty() },
1093
+
Duration::from_secs(2),
1094
+
"second push should start playing immediately",
1095
+
)
1096
+
.await;
980
1097
981
-
reg.push_url(&room_id, "https://example.com/b".into()).await;
982
-
poll_until(
983
-
|| async { reg.queue(&room_id).await.len() == 1 },
984
-
Duration::from_secs(2),
985
-
"second push should be queued",
986
-
)
987
-
.await;
1098
+
assert!(history_len(&store, &room_id.0).await > 0);
1099
+
}
1100
+
}
988
1101
989
-
let queue = reg.queue(&room_id).await;
990
-
assert_eq!(queue[0].title, "b");
1102
+
// -----------------------------------------------------------------------
1103
+
// InMemoryStore backend
1104
+
// -----------------------------------------------------------------------
991
1105
992
-
let playlist = crate::playlist::Playlist::new(reg.pool.clone(), &room_id.0);
993
-
let history = playlist.history(10).await.unwrap();
994
-
assert!(history.is_empty());
995
-
}
1106
+
mod inmem {
1107
+
use super::*;
996
1108
997
-
#[tokio::test]
998
-
async fn test_skip_when_queue_empty_after_track_ends() {
999
-
let (reg, room_id) = test_registry_and_room().await;
1109
+
async fn setup() -> (Registry, RoomId) {
1110
+
let store = Arc::new(InMemoryStore::new());
1111
+
let reg = Registry::new(store, PathBuf::from("/tmp/moqbox"), test_registry());
1112
+
let room_id = reg.create().await;
1113
+
(reg, room_id)
1114
+
}
1000
1115
1001
-
reg.push_url(&room_id, "https://example.com/a".into()).await;
1002
-
poll_until(
1003
-
|| async { reg.queue(&room_id).await.is_empty() },
1004
-
Duration::from_secs(2),
1005
-
"first push should have started playing",
1006
-
)
1007
-
.await;
1116
+
#[tokio::test]
1117
+
async fn test_create_and_exists() {
1118
+
let (reg, id) = setup().await;
1119
+
scenarios::create_and_exists(®, &id).await;
1120
+
}
1008
1121
1009
-
reg.skip(&room_id).await;
1010
-
poll_until(
1011
-
|| async {
1012
-
let playlist = crate::playlist::Playlist::new(reg.pool.clone(), &room_id.0);
1013
-
playlist.history(10).await.unwrap().len() == 1
1014
-
},
1015
-
Duration::from_secs(2),
1016
-
"skip should move track to history",
1017
-
)
1018
-
.await;
1122
+
#[tokio::test]
1123
+
async fn test_push_and_queue() {
1124
+
let (reg, id) = setup().await;
1125
+
scenarios::push_and_queue(®, &id).await;
1126
+
}
1019
1127
1020
-
let queue = reg.queue(&room_id).await;
1021
-
assert!(queue.is_empty());
1022
-
}
1128
+
#[tokio::test]
1129
+
async fn test_push_starts_playback_when_idle() {
1130
+
let (reg, id) = setup().await;
1131
+
scenarios::push_starts_playback_when_idle(®, &id).await;
1132
+
}
1023
1133
1024
-
#[tokio::test]
1025
-
async fn test_double_skip_does_not_corrupt_state() {
1026
-
let (reg, room_id) = test_registry_and_room().await;
1134
+
#[tokio::test]
1135
+
async fn test_push_appends_when_already_playing() {
1136
+
let (reg, id) = setup().await;
1137
+
scenarios::push_appends_when_already_playing(®, &id).await;
1138
+
}
1027
1139
1028
-
reg.push_url(&room_id, "https://example.com/a".into()).await;
1029
-
poll_until(
1030
-
|| async { reg.queue(&room_id).await.is_empty() },
1031
-
Duration::from_secs(2),
1032
-
"first push started playing",
1033
-
)
1034
-
.await;
1140
+
#[tokio::test]
1141
+
async fn test_skip_when_idle_does_not_panic() {
1142
+
let (reg, id) = setup().await;
1143
+
scenarios::skip_when_idle_does_not_panic(®, &id).await;
1144
+
}
1035
1145
1036
-
reg.push_url(&room_id, "https://example.com/b".into()).await;
1037
-
poll_until(
1038
-
|| async { reg.queue(&room_id).await.len() == 1 },
1039
-
Duration::from_secs(2),
1040
-
"second push queued",
1041
-
)
1042
-
.await;
1146
+
#[tokio::test]
1147
+
async fn test_skip_moves_track_to_history() {
1148
+
let (reg, id) = setup().await;
1149
+
scenarios::skip_moves_track_to_history(®, &id).await;
1150
+
}
1043
1151
1044
-
reg.skip(&room_id).await;
1045
-
reg.skip(&room_id).await;
1046
-
poll_until(
1047
-
|| async {
1048
-
let playlist = crate::playlist::Playlist::new(reg.pool.clone(), &room_id.0);
1049
-
let h = playlist.history(10).await.unwrap();
1050
-
!h.is_empty() && h[0].title == "b"
1051
-
},
1052
-
Duration::from_secs(2),
1053
-
"history should have 'b' as most recent",
1054
-
)
1055
-
.await;
1152
+
#[tokio::test]
1153
+
async fn test_multiple_skips_sequential() {
1154
+
let (reg, id) = setup().await;
1155
+
scenarios::multiple_skips_sequential(®, &id).await;
1156
+
}
1056
1157
1057
-
let playlist = crate::playlist::Playlist::new(reg.pool.clone(), &room_id.0);
1058
-
let history = playlist.history(10).await.unwrap();
1059
-
assert!(!history.is_empty());
1060
-
assert_eq!(history[0].title, "b");
1061
-
}
1158
+
#[tokio::test]
1159
+
async fn test_double_skip_does_not_corrupt_state() {
1160
+
let (reg, id) = setup().await;
1161
+
scenarios::double_skip_does_not_corrupt_state(®, &id).await;
1162
+
}
1062
1163
1063
-
#[tokio::test]
1064
-
async fn test_register_client_runs_three_times() {
1065
-
let (reg, room_id) = test_registry_and_room().await;
1164
+
#[tokio::test]
1165
+
async fn test_send_chat() {
1166
+
let (reg, id) = setup().await;
1167
+
scenarios::send_chat(®, &id).await;
1168
+
}
1066
1169
1067
-
for _ in 0..3 {
1068
-
let id = reg.register_client(&room_id).await;
1069
-
assert!(id.is_some());
1170
+
#[tokio::test]
1171
+
async fn test_send_chat_persistence() {
1172
+
let (reg, id) = setup().await;
1173
+
scenarios::send_chat_persistence(®, &id).await;
1070
1174
}
1071
1175
1072
-
assert!(reg.exists(&room_id).await);
1073
-
}
1074
-
1075
-
#[tokio::test]
1076
-
async fn test_remove_while_playing_does_not_panic() {
1077
-
let (reg, room_id) = test_registry_and_room().await;
1176
+
#[tokio::test]
1177
+
async fn test_remove() {
1178
+
let (reg, id) = setup().await;
1179
+
scenarios::remove_room(®, &id).await;
1180
+
}
1078
1181
1079
-
reg.push_url(&room_id, "https://example.com/a".into()).await;
1080
-
poll_until(
1081
-
|| async { reg.queue(&room_id).await.is_empty() },
1082
-
Duration::from_secs(2),
1083
-
"push should have started playing",
1084
-
)
1085
-
.await;
1182
+
#[tokio::test]
1183
+
async fn test_remove_while_playing_does_not_panic() {
1184
+
let (reg, id) = setup().await;
1185
+
scenarios::remove_while_playing(®, &id).await;
1186
+
}
1086
1187
1087
-
reg.remove(&room_id).await;
1088
-
assert!(!reg.exists(&room_id).await);
1089
-
}
1188
+
#[tokio::test]
1189
+
async fn test_sweep_idle_removes_actor_but_keeps_room() {
1190
+
let (reg, id) = setup().await;
1191
+
scenarios::sweep_keeps_room_in_store(®, &id).await;
1192
+
}
1090
1193
1091
-
#[tokio::test]
1092
-
async fn test_sweep_idle_keeps_active_room() {
1093
-
let (reg, room_id) = test_registry_and_room().await;
1194
+
#[tokio::test]
1195
+
async fn test_sweep_idle_preserves_active_rooms() {
1196
+
let (reg, id) = setup().await;
1197
+
scenarios::sweep_preserves_active_rooms(®, &id).await;
1198
+
}
1094
1199
1095
-
reg.register_client(&room_id).await;
1200
+
#[tokio::test]
1201
+
async fn test_multiple_rooms_dont_interfere() {
1202
+
let (reg, id) = setup().await;
1203
+
scenarios::multiple_rooms_dont_interfere(®, &id).await;
1204
+
}
1096
1205
1097
-
let swept = reg.sweep_idle(Duration::from_millis(10_000)).await;
1098
-
assert!(!swept.contains(&room_id));
1206
+
#[tokio::test]
1207
+
async fn test_push_after_skip_works() {
1208
+
let (reg, id) = setup().await;
1209
+
scenarios::push_after_skip_works(®, &id).await;
1210
+
}
1099
1211
}
1100
1212
1101
-
#[tokio::test]
1102
-
async fn test_client_count_tracking() {
1103
-
let (reg, room_id) = test_registry_and_room().await;
1213
+
// -----------------------------------------------------------------------
1214
+
// SqliteStore backend — runs same scenarios via :memory: SQLite
1215
+
// -----------------------------------------------------------------------
1104
1216
1105
-
let id1 = reg.register_client(&room_id).await;
1106
-
let id2 = reg.register_client(&room_id).await;
1107
-
let id3 = reg.register_client(&room_id).await;
1108
-
assert!(id1.is_some(), "id1 should be Some");
1109
-
assert!(id2.is_some(), "id2 should be Some");
1110
-
assert!(id3.is_some(), "id3 should be Some");
1217
+
mod sqlite {
1218
+
use super::*;
1111
1219
1112
-
reg.unregister_client(&room_id).await;
1220
+
async fn setup() -> (Registry, RoomId) {
1221
+
let pool = sqlx::sqlite::SqlitePoolOptions::new()
1222
+
.max_connections(1)
1223
+
.connect(":memory:")
1224
+
.await
1225
+
.unwrap();
1226
+
let store = Arc::new(SqliteStore::new(pool).await.unwrap());
1227
+
let reg = Registry::new(store, PathBuf::from("/tmp/moqbox"), test_registry());
1228
+
let room_id = reg.create().await;
1229
+
(reg, room_id)
1230
+
}
1113
1231
1114
-
let id4 = reg.register_client(&room_id).await;
1115
-
assert!(id4.is_some());
1116
-
assert_ne!(id4, id1);
1117
-
}
1232
+
#[tokio::test]
1233
+
async fn test_create_and_exists() {
1234
+
let (reg, id) = setup().await;
1235
+
scenarios::create_and_exists(®, &id).await;
1236
+
}
1118
1237
1119
-
#[tokio::test]
1120
-
async fn test_multiple_rooms_dont_interfere() {
1121
-
let (reg, room_a) = test_registry_and_room().await;
1122
-
let room_b = reg.create().await;
1238
+
#[tokio::test]
1239
+
async fn test_push_and_queue() {
1240
+
let (reg, id) = setup().await;
1241
+
scenarios::push_and_queue(®, &id).await;
1242
+
}
1123
1243
1124
-
assert_ne!(room_a, room_b);
1125
-
assert!(reg.exists(&room_a).await);
1126
-
assert!(reg.exists(&room_b).await);
1244
+
#[tokio::test]
1245
+
async fn test_push_starts_playback_when_idle() {
1246
+
let (reg, id) = setup().await;
1247
+
scenarios::push_starts_playback_when_idle(®, &id).await;
1248
+
}
1127
1249
1128
-
reg.push_url(&room_a, "https://example.com/a".into()).await;
1129
-
reg.push_url(&room_b, "https://example.com/b".into()).await;
1250
+
#[tokio::test]
1251
+
async fn test_push_appends_when_already_playing() {
1252
+
let (reg, id) = setup().await;
1253
+
scenarios::push_appends_when_already_playing(®, &id).await;
1254
+
}
1130
1255
1131
-
poll_until(
1132
-
|| async { reg.queue(&room_a).await.is_empty() },
1133
-
Duration::from_secs(2),
1134
-
"room_a queue empty",
1135
-
)
1136
-
.await;
1137
-
poll_until(
1138
-
|| async { reg.queue(&room_b).await.is_empty() },
1139
-
Duration::from_secs(2),
1140
-
"room_b queue empty",
1141
-
)
1142
-
.await;
1143
-
}
1256
+
#[tokio::test]
1257
+
async fn test_skip_when_idle_does_not_panic() {
1258
+
let (reg, id) = setup().await;
1259
+
scenarios::skip_when_idle_does_not_panic(®, &id).await;
1260
+
}
1144
1261
1145
-
#[tokio::test]
1146
-
async fn test_push_after_skip_works() {
1147
-
let (reg, room_id) = test_registry_and_room().await;
1262
+
#[tokio::test]
1263
+
async fn test_skip_moves_track_to_history() {
1264
+
let (reg, id) = setup().await;
1265
+
scenarios::skip_moves_track_to_history(®, &id).await;
1266
+
}
1148
1267
1149
-
reg.push_url(&room_id, "https://a".into()).await;
1150
-
poll_until(
1151
-
|| async { reg.queue(&room_id).await.is_empty() },
1152
-
Duration::from_secs(2),
1153
-
"first push started playing",
1154
-
)
1155
-
.await;
1268
+
#[tokio::test]
1269
+
async fn test_multiple_skips_sequential() {
1270
+
let (reg, id) = setup().await;
1271
+
scenarios::multiple_skips_sequential(®, &id).await;
1272
+
}
1156
1273
1157
-
reg.skip(&room_id).await;
1158
-
poll_until(
1159
-
|| async {
1160
-
let playlist = crate::playlist::Playlist::new(reg.pool.clone(), &room_id.0);
1161
-
!playlist.history(10).await.unwrap().is_empty()
1162
-
},
1163
-
Duration::from_secs(2),
1164
-
"skip should move track to history",
1165
-
)
1166
-
.await;
1274
+
#[tokio::test]
1275
+
async fn test_double_skip_does_not_corrupt_state() {
1276
+
let (reg, id) = setup().await;
1277
+
scenarios::double_skip_does_not_corrupt_state(®, &id).await;
1278
+
}
1167
1279
1168
-
reg.push_url(&room_id, "https://b".into()).await;
1169
-
poll_until(
1170
-
|| async { reg.queue(&room_id).await.is_empty() },
1171
-
Duration::from_secs(2),
1172
-
"second push should start playing immediately (idle after skip)",
1173
-
)
1174
-
.await;
1280
+
#[tokio::test]
1281
+
async fn test_send_chat() {
1282
+
let (reg, id) = setup().await;
1283
+
scenarios::send_chat(®, &id).await;
1284
+
}
1175
1285
1176
-
let playlist = crate::playlist::Playlist::new(reg.pool.clone(), &room_id.0);
1177
-
let history = playlist.history(10).await.unwrap();
1178
-
assert!(!history.is_empty());
1179
-
}
1180
-
1181
-
#[tokio::test]
1182
-
async fn test_client_count_does_not_underflow() {
1183
-
let (reg, room_id) = test_registry_and_room().await;
1184
-
1185
-
reg.unregister_client(&room_id).await;
1186
-
1187
-
let re_registered = reg.register_client(&room_id).await;
1188
-
assert!(
1189
-
re_registered.is_some(),
1190
-
"room should accept new clients after underflow"
1191
-
);
1192
-
}
1286
+
#[tokio::test]
1287
+
async fn test_send_chat_persistence() {
1288
+
let (reg, id) = setup().await;
1289
+
scenarios::send_chat_persistence(®, &id).await;
1290
+
}
1193
1291
1194
-
#[tokio::test]
1195
-
async fn test_multiple_skips_sequential() {
1196
-
let (reg, room_id) = test_registry_and_room().await;
1292
+
#[tokio::test]
1293
+
async fn test_remove() {
1294
+
let (reg, id) = setup().await;
1295
+
scenarios::remove_room(®, &id).await;
1296
+
}
1197
1297
1198
-
for url in ["https://x/a", "https://x/b", "https://x/c"] {
1199
-
reg.push_url(&room_id, url.into()).await;
1298
+
#[tokio::test]
1299
+
async fn test_remove_while_playing_does_not_panic() {
1300
+
let (reg, id) = setup().await;
1301
+
scenarios::remove_while_playing(®, &id).await;
1200
1302
}
1201
-
poll_until(
1202
-
|| async { reg.queue(&room_id).await.len() == 2 },
1203
-
Duration::from_secs(2),
1204
-
"3 pushes → 1 active, 2 queued",
1205
-
)
1206
-
.await;
1207
1303
1208
-
reg.skip(&room_id).await;
1209
-
poll_until(
1210
-
|| async { reg.queue(&room_id).await.len() == 1 },
1211
-
Duration::from_secs(2),
1212
-
"skip → 1 item left in queue",
1213
-
)
1214
-
.await;
1304
+
#[tokio::test]
1305
+
async fn test_sweep_idle_removes_actor_but_keeps_room() {
1306
+
let (reg, id) = setup().await;
1307
+
scenarios::sweep_keeps_room_in_store(®, &id).await;
1308
+
}
1215
1309
1216
-
let queue = reg.queue(&room_id).await;
1217
-
assert_eq!(queue[0].title, "c");
1310
+
#[tokio::test]
1311
+
async fn test_sweep_idle_preserves_active_rooms() {
1312
+
let (reg, id) = setup().await;
1313
+
scenarios::sweep_preserves_active_rooms(®, &id).await;
1314
+
}
1218
1315
1219
-
reg.skip(&room_id).await;
1220
-
poll_until(
1221
-
|| async { reg.queue(&room_id).await.is_empty() },
1222
-
Duration::from_secs(2),
1223
-
"skip → queue empty",
1224
-
)
1225
-
.await;
1316
+
#[tokio::test]
1317
+
async fn test_multiple_rooms_dont_interfere() {
1318
+
let (reg, id) = setup().await;
1319
+
scenarios::multiple_rooms_dont_interfere(®, &id).await;
1320
+
}
1226
1321
1227
-
reg.skip(&room_id).await;
1322
+
#[tokio::test]
1323
+
async fn test_push_after_skip_works() {
1324
+
let (reg, id) = setup().await;
1325
+
scenarios::push_after_skip_works(®, &id).await;
1326
+
}
1228
1327
}
1229
1328
}
+142
-40
src/state.rs
+142
-40
src/state.rs
···
43
43
title: String,
44
44
duration: String,
45
45
thumbnail: Option<String>,
46
+
source: String,
46
47
},
47
48
}
48
49
···
61
62
PersistFinished(i64),
62
63
/// Broadcast the current state snapshot to all clients.
63
64
PublishSnapshot,
65
+
/// Persist a newly queued track to the store.
66
+
PersistQueuedTrack(QueuedTrack),
67
+
/// Persist metadata update after async extraction completes.
68
+
PersistMetadata {
69
+
item_id: i64,
70
+
title: String,
71
+
duration: String,
72
+
thumbnail: Option<String>,
73
+
source: String,
74
+
},
75
+
/// Persist a chat message.
76
+
PersistChat {
77
+
user_name: String,
78
+
content: String,
79
+
msg_type: String,
80
+
created_at: String,
81
+
},
64
82
}
65
83
66
84
/// Pure playback state: no IO handles, no DB connections.
···
89
107
title,
90
108
duration,
91
109
thumbnail,
92
-
} => self.handle_metadata_updated(*item_id, title, duration, thumbnail),
110
+
source,
111
+
} => self.handle_metadata_updated(*item_id, title, duration, thumbnail, source),
93
112
}
94
113
}
95
114
···
98
117
self.queue.push(track.clone());
99
118
100
119
if self.active.is_some() {
101
-
return vec![Effect::PublishSnapshot];
120
+
return vec![
121
+
Effect::PersistQueuedTrack(track.clone()),
122
+
Effect::PublishSnapshot,
123
+
];
102
124
}
103
125
104
-
self.advance(track.clone())
126
+
// Idle: persist the track first, then advance it to playing.
127
+
let mut effects = vec![Effect::PersistQueuedTrack(track.clone())];
128
+
effects.extend(self.advance(track.clone()));
129
+
effects
105
130
}
106
131
107
132
/// User requested skip.
···
193
218
title: &str,
194
219
duration: &str,
195
220
thumbnail: &Option<String>,
221
+
source: &str,
196
222
) -> Vec<Effect> {
197
223
if let Some(item) = self.queue.iter_mut().find(|t| t.id == item_id) {
198
224
item.title = title.to_string();
···
207
233
active.thumbnail.clone_from(thumbnail);
208
234
}
209
235
}
210
-
vec![Effect::PublishSnapshot]
236
+
vec![
237
+
Effect::PersistMetadata {
238
+
item_id,
239
+
title: title.to_string(),
240
+
duration: duration.to_string(),
241
+
thumbnail: thumbnail.clone(),
242
+
source: source.to_string(),
243
+
},
244
+
Effect::PublishSnapshot,
245
+
]
211
246
}
212
247
213
248
/// Pop the first track from the queue and start playing it.
···
242
277
active.started_at_wall = real_started_at;
243
278
}
244
279
}
280
+
281
+
/// Reconstruct state from a [`RoomSnapshot`] loaded from the store.
282
+
///
283
+
/// If there's a stale active track (server died mid-playback), it's moved
284
+
/// to history. The queue is restored as-is. No playback auto-starts on
285
+
/// rehydration — users explicitly trigger playback.
286
+
pub fn from_snapshot(snapshot: crate::store::RoomSnapshot) -> (Self, Vec<Effect>) {
287
+
let mut state = PlaybackState::default();
288
+
let mut effects = Vec::new();
289
+
290
+
state.queue = snapshot.queue;
291
+
292
+
// Stale active track → move to history front
293
+
if let Some(stale) = snapshot.active {
294
+
state.history.push(FinishedTrack {
295
+
id: stale.id,
296
+
title: stale.title,
297
+
url: stale.url,
298
+
duration: stale.duration,
299
+
thumbnail: stale.thumbnail,
300
+
played_at: String::new(),
301
+
});
302
+
effects.push(Effect::PersistFinished(stale.id));
303
+
}
304
+
305
+
// Append existing history
306
+
state.history.extend(snapshot.history);
307
+
308
+
if !effects.is_empty() {
309
+
effects.push(Effect::PublishSnapshot);
310
+
}
311
+
312
+
(state, effects)
313
+
}
245
314
}
246
315
247
316
/// # Pure unit tests (no async, no DB, no IO)
···
268
337
assert_eq!(
269
338
effects,
270
339
vec![
340
+
Effect::PersistQueuedTrack(queued("A", 1)),
271
341
Effect::PersistStarted(1),
272
342
Effect::StartPipeline(queued("A", 1)),
273
343
Effect::PublishSnapshot,
274
344
],
275
-
"idle → first track: persist started_at, spawn pipeline, notify clients"
345
+
"idle → first track: persist queued, started_at, spawn pipeline, notify clients"
276
346
);
277
347
}
278
348
···
284
354
285
355
assert_eq!(
286
356
effects,
287
-
vec![Effect::PublishSnapshot],
288
-
"playing -> queue another: just notify, pipeline untouched"
357
+
vec![
358
+
Effect::PersistQueuedTrack(queued("B", 2)),
359
+
Effect::PublishSnapshot
360
+
],
361
+
"playing -> queue another: persist track, notify, pipeline untouched"
289
362
);
290
363
}
291
364
···
303
376
Effect::PersistFinished(1),
304
377
Effect::PersistStarted(2),
305
378
],
306
-
"skip with next: abort current, finalize A, persist B started_at"
379
+
"skip with next: abort current, finalize A, start B"
307
380
);
308
381
assert!(
309
382
effects
···
335
408
}
336
409
337
410
#[test]
338
-
fn test_track_ended_with_next_effect_order() {
339
-
let mut s = PlaybackState::default();
340
-
s.transition(&Event::TrackQueued(queued("A", 1)), 0);
341
-
s.transition(&Event::TrackQueued(queued("B", 2)), 0);
342
-
let effects = s.transition(&Event::TrackEnded { item_id: 1 }, 0);
343
-
344
-
assert_eq!(
345
-
&effects[..2],
346
-
&[Effect::PersistFinished(1), Effect::PersistStarted(2),],
347
-
"track ended (next): finalize A, persist B started_at, no AbortPipeline"
348
-
);
349
-
assert!(
350
-
effects
351
-
.iter()
352
-
.any(|e| matches!(e, Effect::StartPipeline(t) if t.title == "B")),
353
-
"track ended (next): must start B"
354
-
);
355
-
assert!(
356
-
effects.contains(&Effect::PublishSnapshot),
357
-
"track ended (next): must notify"
358
-
);
359
-
}
360
-
361
-
#[test]
362
411
fn test_track_ended_last_goes_idle_effect_order() {
363
412
let mut s = PlaybackState::default();
364
413
s.transition(&Event::TrackQueued(queued("A", 1)), 0);
···
426
475
}
427
476
428
477
#[test]
429
-
fn test_track_ended_advances_to_next() {
478
+
fn test_track_ended_with_next_effect_order() {
430
479
let mut s = PlaybackState::default();
431
480
s.transition(&Event::TrackQueued(queued("A", 1)), 0);
432
481
s.transition(&Event::TrackQueued(queued("B", 2)), 0);
433
-
s.transition(&Event::TrackEnded { item_id: 1 }, 0);
482
+
let effects = s.transition(&Event::TrackEnded { item_id: 1 }, 0);
434
483
435
-
assert_eq!(s.active.as_ref().unwrap().title, "B");
436
-
assert_eq!(s.history.len(), 1);
437
-
assert_eq!(s.history[0].title, "A");
484
+
assert_eq!(
485
+
&effects[..2],
486
+
&[Effect::PersistFinished(1), Effect::PersistStarted(2),],
487
+
"track ended (next): finalize A, start B, no AbortPipeline, no re-persist queued"
488
+
);
489
+
assert!(
490
+
effects
491
+
.iter()
492
+
.any(|e| matches!(e, Effect::StartPipeline(t) if t.title == "B")),
493
+
"track ended (next): must start B"
494
+
);
495
+
assert!(
496
+
effects.contains(&Effect::PublishSnapshot),
497
+
"track ended (next): must notify"
498
+
);
438
499
}
439
500
440
501
#[test]
···
492
553
fn test_metadata_update_updates_queue_item() {
493
554
let mut s = PlaybackState::default();
494
555
s.transition(&Event::TrackQueued(queued("A", 1)), 0);
495
-
s.transition(
556
+
let effects = s.transition(
496
557
&Event::MetadataUpdated {
497
558
item_id: 1,
498
559
title: "Track A (Real)".into(),
499
560
duration: "4:20".into(),
500
561
thumbnail: Some("https://img.example/a.jpg".into()),
562
+
source: "ytdlp".into(),
501
563
},
502
564
0,
503
565
);
504
566
567
+
assert_eq!(
568
+
effects,
569
+
vec![
570
+
Effect::PersistMetadata {
571
+
item_id: 1,
572
+
title: "Track A (Real)".into(),
573
+
duration: "4:20".into(),
574
+
thumbnail: Some("https://img.example/a.jpg".into()),
575
+
source: "ytdlp".into(),
576
+
},
577
+
Effect::PublishSnapshot,
578
+
]
579
+
);
580
+
505
581
// Queue should have empty (track was advanced to active), but active
506
582
// should have updated metadata.
507
583
assert_eq!(s.active.as_ref().unwrap().title, "Track A (Real)");
···
525
601
title: "Track B (Real)".into(),
526
602
duration: "5:55".into(),
527
603
thumbnail: None,
604
+
source: "ytdlp".into(),
528
605
},
529
606
0,
530
607
);
531
-
assert_eq!(effects, vec![Effect::PublishSnapshot]);
608
+
assert_eq!(
609
+
effects,
610
+
vec![
611
+
Effect::PersistMetadata {
612
+
item_id: 2,
613
+
title: "Track B (Real)".into(),
614
+
duration: "5:55".into(),
615
+
thumbnail: None,
616
+
source: "ytdlp".into(),
617
+
},
618
+
Effect::PublishSnapshot,
619
+
]
620
+
);
532
621
assert_eq!(s.queue[0].title, "Track B (Real)");
533
622
assert!(!s.queue[0].pending);
534
623
}
···
542
631
title: "Ghost".into(),
543
632
duration: "0:00".into(),
544
633
thumbnail: None,
634
+
source: "ytdlp".into(),
545
635
},
546
636
0,
547
637
);
548
638
// Still publishes snapshot even if nothing changed: clients may want
549
639
// to know the update was processed.
550
-
assert_eq!(effects, vec![Effect::PublishSnapshot]);
640
+
assert_eq!(
641
+
effects,
642
+
vec![
643
+
Effect::PersistMetadata {
644
+
item_id: 999,
645
+
title: "Ghost".into(),
646
+
duration: "0:00".into(),
647
+
thumbnail: None,
648
+
source: "ytdlp".into(),
649
+
},
650
+
Effect::PublishSnapshot,
651
+
]
652
+
);
551
653
assert!(s.active.is_none());
552
654
assert!(s.queue.is_empty());
553
655
}
+1053
src/store.rs
+1053
src/store.rs
···
1
+
//! Persistence abstraction for room data.
2
+
//!
3
+
//! Defines the [`RoomStore`] trait that encapsulates all SQLite operations
4
+
//! behind a clean interface, along with an in-memory implementation for
5
+
//! testing and a SQLite-backed implementation for production.
6
+
//!
7
+
//! The room actor never touches SQL directly — all persistence flows through
8
+
//! a [`RoomStore`] implementation.
9
+
10
+
use std::collections::HashMap;
11
+
use std::sync::Mutex;
12
+
13
+
use async_trait::async_trait;
14
+
use sqlx::{Pool, Sqlite};
15
+
16
+
use crate::state::{Effect, FinishedTrack, QueuedTrack};
17
+
use crate::types::ChatMessage;
18
+
19
+
/// Full snapshot of a room's state loaded from the store.
20
+
///
21
+
/// Fields are split into the four logical partitions: queue (not yet playing),
22
+
/// active (currently playing), history (finished), and chat messages.
23
+
#[derive(Debug, Clone)]
24
+
#[allow(dead_code)]
25
+
pub(crate) struct RoomSnapshot {
26
+
pub room_id: String,
27
+
pub room_name: String,
28
+
/// Tracks waiting to play (not started, not finished).
29
+
pub queue: Vec<QueuedTrack>,
30
+
/// The track currently playing, if any.
31
+
pub active: Option<QueuedTrack>,
32
+
/// Recently finished tracks (newest first).
33
+
pub history: Vec<FinishedTrack>,
34
+
pub chat_messages: Vec<ChatMessage>,
35
+
}
36
+
37
+
/// Output returned by [`RoomStore::persist`].
38
+
///
39
+
/// Carries information the caller needs to react to a persist batch.
40
+
#[derive(Debug, Clone, Default)]
41
+
pub(crate) struct PersistOutput {
42
+
/// Set when a [`Effect::PersistChat`] was applied.
43
+
pub chat_message_id: Option<i64>,
44
+
}
45
+
46
+
/// Abstraction over room persistence.
47
+
///
48
+
/// All implementors must be [`Send`] + [`Sync`] so they can be shared across
49
+
/// async tasks and protected behind an [`Arc`](std::sync::Arc).
50
+
#[async_trait]
51
+
pub(crate) trait RoomStore: Send + Sync {
52
+
/// Create a new room row.
53
+
async fn create_room(&self, id: &str, name: &str) -> Result<(), sqlx::Error>;
54
+
55
+
/// Apply a batch of effects atomically.
56
+
///
57
+
/// Effects that are not persistence-related (e.g. [`Effect::AbortPipeline`],
58
+
/// [`Effect::PublishSnapshot`]) are silently ignored.
59
+
async fn persist(
60
+
&self,
61
+
room_id: &str,
62
+
effects: &[Effect],
63
+
) -> Result<PersistOutput, sqlx::Error>;
64
+
65
+
/// Load the full snapshot for a room, or [`None`] if it does not exist.
66
+
async fn load(&self, room_id: &str) -> Result<Option<RoomSnapshot>, sqlx::Error>;
67
+
68
+
/// Retrieve the most recent chat messages for a room, newest first.
69
+
async fn recent_chat(&self, room_id: &str, limit: i64)
70
+
-> Result<Vec<ChatMessage>, sqlx::Error>;
71
+
72
+
/// Delete a room and all its associated data (CASCADE).
73
+
#[allow(dead_code)]
74
+
async fn delete_room(&self, room_id: &str) -> Result<(), sqlx::Error>;
75
+
}
76
+
77
+
// ---------------------------------------------------------------------------
78
+
// In-memory implementation (for testing)
79
+
// ---------------------------------------------------------------------------
80
+
81
+
#[cfg_attr(not(test), allow(dead_code))]
82
+
#[derive(Debug)]
83
+
struct InMemoryRoom {
84
+
name: String,
85
+
queue: Vec<QueuedTrack>,
86
+
active: Option<QueuedTrack>,
87
+
history: Vec<FinishedTrack>,
88
+
chat_messages: Vec<ChatMessage>,
89
+
next_chat_id: i64,
90
+
}
91
+
92
+
/// Thread-safe in-memory store backed by a [`Mutex`]-protected [`HashMap`].
93
+
///
94
+
/// Implements [`RoomStore`] without any external dependencies, making it
95
+
/// suitable for unit tests and deterministic simulations.
96
+
#[cfg_attr(not(test), allow(dead_code))]
97
+
#[derive(Debug)]
98
+
pub(crate) struct InMemoryStore {
99
+
inner: Mutex<HashMap<String, InMemoryRoom>>,
100
+
}
101
+
102
+
#[cfg_attr(not(test), allow(dead_code))]
103
+
impl InMemoryStore {
104
+
pub(crate) fn new() -> Self {
105
+
Self {
106
+
inner: Mutex::new(HashMap::new()),
107
+
}
108
+
}
109
+
}
110
+
111
+
impl Default for InMemoryStore {
112
+
fn default() -> Self {
113
+
Self::new()
114
+
}
115
+
}
116
+
117
+
#[async_trait]
118
+
impl RoomStore for InMemoryStore {
119
+
async fn create_room(&self, id: &str, name: &str) -> Result<(), sqlx::Error> {
120
+
let mut inner = self.inner.lock().expect("InMemoryStore mutex poisoned");
121
+
inner.insert(
122
+
id.to_string(),
123
+
InMemoryRoom {
124
+
name: name.to_string(),
125
+
queue: Vec::new(),
126
+
active: None,
127
+
history: Vec::new(),
128
+
chat_messages: Vec::new(),
129
+
next_chat_id: 1,
130
+
},
131
+
);
132
+
Ok(())
133
+
}
134
+
135
+
async fn persist(
136
+
&self,
137
+
room_id: &str,
138
+
effects: &[Effect],
139
+
) -> Result<PersistOutput, sqlx::Error> {
140
+
let mut inner = self.inner.lock().expect("InMemoryStore mutex poisoned");
141
+
let room = match inner.get_mut(room_id) {
142
+
Some(room) => room,
143
+
None => return Ok(PersistOutput::default()),
144
+
};
145
+
146
+
let mut output = PersistOutput::default();
147
+
148
+
for effect in effects {
149
+
match effect {
150
+
Effect::PersistQueuedTrack(track) => {
151
+
room.queue.push(track.clone());
152
+
}
153
+
Effect::PersistStarted(id) => {
154
+
if let Some(pos) = room.queue.iter().position(|t| t.id == *id) {
155
+
let track = room.queue.remove(pos);
156
+
room.active = Some(track);
157
+
}
158
+
}
159
+
Effect::PersistFinished(id) => {
160
+
if let Some(active) = room.active.take() {
161
+
if active.id == *id {
162
+
let now = crate::util::now_iso();
163
+
room.history.insert(
164
+
0,
165
+
FinishedTrack {
166
+
id: active.id,
167
+
title: active.title,
168
+
url: active.url,
169
+
duration: active.duration,
170
+
thumbnail: active.thumbnail,
171
+
played_at: now,
172
+
},
173
+
);
174
+
} else {
175
+
// ID mismatch — put the active track back.
176
+
room.active = Some(active);
177
+
}
178
+
}
179
+
}
180
+
Effect::PersistMetadata {
181
+
item_id,
182
+
title,
183
+
duration,
184
+
thumbnail,
185
+
source: _,
186
+
} => {
187
+
// Update in the queue.
188
+
if let Some(track) = room.queue.iter_mut().find(|t| t.id == *item_id) {
189
+
track.title.clone_from(title);
190
+
track.duration.clone_from(duration);
191
+
track.thumbnail.clone_from(thumbnail);
192
+
track.pending = false;
193
+
}
194
+
// Update in the active track if it matches.
195
+
if let Some(ref mut active) = room.active {
196
+
if active.id == *item_id {
197
+
active.title.clone_from(title);
198
+
active.duration.clone_from(duration);
199
+
active.thumbnail.clone_from(thumbnail);
200
+
active.pending = false;
201
+
}
202
+
}
203
+
}
204
+
Effect::PersistChat {
205
+
user_name,
206
+
content,
207
+
msg_type,
208
+
created_at,
209
+
} => {
210
+
let id = room.next_chat_id;
211
+
room.next_chat_id += 1;
212
+
room.chat_messages.push(ChatMessage {
213
+
id,
214
+
user_name: user_name.clone(),
215
+
content: content.clone(),
216
+
msg_type: msg_type.clone(),
217
+
created_at: created_at.clone(),
218
+
});
219
+
output.chat_message_id = Some(id);
220
+
}
221
+
_ => {}
222
+
}
223
+
}
224
+
225
+
Ok(output)
226
+
}
227
+
228
+
async fn load(&self, room_id: &str) -> Result<Option<RoomSnapshot>, sqlx::Error> {
229
+
let inner = self.inner.lock().expect("InMemoryStore mutex poisoned");
230
+
let room = match inner.get(room_id) {
231
+
Some(room) => room,
232
+
None => return Ok(None),
233
+
};
234
+
235
+
Ok(Some(RoomSnapshot {
236
+
room_id: room_id.to_string(),
237
+
room_name: room.name.clone(),
238
+
queue: room.queue.clone(),
239
+
active: room.active.clone(),
240
+
history: room.history.clone(),
241
+
chat_messages: room.chat_messages.clone(),
242
+
}))
243
+
}
244
+
245
+
async fn recent_chat(
246
+
&self,
247
+
room_id: &str,
248
+
limit: i64,
249
+
) -> Result<Vec<ChatMessage>, sqlx::Error> {
250
+
let inner = self.inner.lock().expect("InMemoryStore mutex poisoned");
251
+
let room = match inner.get(room_id) {
252
+
Some(room) => room,
253
+
None => return Ok(Vec::new()),
254
+
};
255
+
256
+
let limit = limit.max(0) as usize;
257
+
Ok(room
258
+
.chat_messages
259
+
.iter()
260
+
.rev()
261
+
.take(limit)
262
+
.cloned()
263
+
.collect())
264
+
}
265
+
266
+
async fn delete_room(&self, room_id: &str) -> Result<(), sqlx::Error> {
267
+
let mut inner = self.inner.lock().expect("InMemoryStore mutex poisoned");
268
+
inner.remove(room_id);
269
+
Ok(())
270
+
}
271
+
}
272
+
273
+
// ---------------------------------------------------------------------------
274
+
// SQLite-backed implementation (production)
275
+
// ---------------------------------------------------------------------------
276
+
277
+
/// Production [`RoomStore`] backed by an [`sqlx`] SQLite pool.
278
+
///
279
+
/// Schema is isolated from the legacy `db.rs` tables: this module uses
280
+
/// `tracks` (not `queue_items`) and a fresh `rooms` table layout.
281
+
#[derive(Debug, Clone)]
282
+
pub(crate) struct SqliteStore {
283
+
pool: Pool<Sqlite>,
284
+
}
285
+
286
+
impl SqliteStore {
287
+
/// Open a pool at `path` and run schema migrations.
288
+
pub(crate) async fn connect(path: &std::path::Path) -> Result<Self, sqlx::Error> {
289
+
use sqlx::sqlite::SqlitePoolOptions;
290
+
291
+
let conn_str = format!("sqlite:{}?mode=rwc", path.to_string_lossy());
292
+
let pool = SqlitePoolOptions::new()
293
+
.max_connections(8)
294
+
.connect(&conn_str)
295
+
.await?;
296
+
297
+
for pragma in [
298
+
"PRAGMA journal_mode=wal",
299
+
"PRAGMA synchronous=NORMAL",
300
+
"PRAGMA foreign_keys=ON",
301
+
"PRAGMA busy_timeout=5000",
302
+
] {
303
+
sqlx::query(pragma).execute(&pool).await?;
304
+
}
305
+
306
+
let store = Self::new(pool).await?;
307
+
Ok(store)
308
+
}
309
+
310
+
/// Create a new `SqliteStore` and run schema migrations.
311
+
///
312
+
/// Idempotent: all tables use `CREATE TABLE IF NOT EXISTS`.
313
+
pub(crate) async fn new(pool: Pool<Sqlite>) -> Result<Self, sqlx::Error> {
314
+
sqlx::query(
315
+
"CREATE TABLE IF NOT EXISTS rooms (
316
+
id TEXT PRIMARY KEY,
317
+
name TEXT NOT NULL DEFAULT '',
318
+
created_at TEXT NOT NULL,
319
+
updated_at TEXT NOT NULL
320
+
)",
321
+
)
322
+
.execute(&pool)
323
+
.await?;
324
+
325
+
sqlx::query(
326
+
"CREATE TABLE IF NOT EXISTS tracks (
327
+
id INTEGER PRIMARY KEY,
328
+
room_id TEXT NOT NULL REFERENCES rooms(id) ON DELETE CASCADE,
329
+
url TEXT NOT NULL,
330
+
title TEXT NOT NULL,
331
+
duration TEXT NOT NULL DEFAULT '--:--',
332
+
thumbnail TEXT,
333
+
source TEXT NOT NULL DEFAULT '',
334
+
position INTEGER NOT NULL,
335
+
added_at TEXT NOT NULL,
336
+
played INTEGER NOT NULL DEFAULT 0,
337
+
played_at TEXT,
338
+
started_at_ms INTEGER,
339
+
pending INTEGER NOT NULL DEFAULT 1
340
+
)",
341
+
)
342
+
.execute(&pool)
343
+
.await?;
344
+
345
+
// Composite index for the common room-scoped queries.
346
+
sqlx::query(
347
+
"CREATE INDEX IF NOT EXISTS idx_tracks_room_lookup
348
+
ON tracks(room_id, played, started_at_ms, position)",
349
+
)
350
+
.execute(&pool)
351
+
.await?;
352
+
353
+
sqlx::query(
354
+
"CREATE TABLE IF NOT EXISTS chat_messages (
355
+
id INTEGER PRIMARY KEY AUTOINCREMENT,
356
+
room_id TEXT NOT NULL REFERENCES rooms(id) ON DELETE CASCADE,
357
+
user_name TEXT NOT NULL,
358
+
content TEXT NOT NULL,
359
+
msg_type TEXT NOT NULL DEFAULT 'message',
360
+
created_at TEXT NOT NULL
361
+
)",
362
+
)
363
+
.execute(&pool)
364
+
.await?;
365
+
366
+
// Enable foreign key enforcement (per-connection, best-effort).
367
+
sqlx::query("PRAGMA foreign_keys = ON")
368
+
.execute(&pool)
369
+
.await?;
370
+
371
+
Ok(Self { pool })
372
+
}
373
+
}
374
+
375
+
/// Helper row type for deserialising `tracks` rows.
376
+
#[derive(sqlx::FromRow)]
377
+
#[allow(dead_code)]
378
+
struct TrackRow {
379
+
id: i64,
380
+
title: String,
381
+
url: String,
382
+
duration: String,
383
+
thumbnail: Option<String>,
384
+
played: i64,
385
+
played_at: Option<String>,
386
+
started_at_ms: Option<i64>,
387
+
pending: i64,
388
+
}
389
+
390
+
/// Helper row type for deserialising `chat_messages` rows.
391
+
#[derive(sqlx::FromRow)]
392
+
struct ChatMsgRow {
393
+
id: i64,
394
+
user_name: String,
395
+
content: String,
396
+
msg_type: String,
397
+
created_at: String,
398
+
}
399
+
400
+
#[async_trait]
401
+
impl RoomStore for SqliteStore {
402
+
async fn create_room(&self, id: &str, name: &str) -> Result<(), sqlx::Error> {
403
+
let now = crate::util::now_iso();
404
+
sqlx::query("INSERT INTO rooms (id, name, created_at, updated_at) VALUES (?1, ?2, ?3, ?4)")
405
+
.bind(id)
406
+
.bind(name)
407
+
.bind(&now)
408
+
.bind(&now)
409
+
.execute(&self.pool)
410
+
.await?;
411
+
Ok(())
412
+
}
413
+
414
+
async fn persist(
415
+
&self,
416
+
room_id: &str,
417
+
effects: &[Effect],
418
+
) -> Result<PersistOutput, sqlx::Error> {
419
+
let mut tx = self.pool.begin().await?;
420
+
let mut output = PersistOutput::default();
421
+
422
+
for effect in effects {
423
+
match effect {
424
+
Effect::PersistQueuedTrack(track) => {
425
+
// Compute the next position at the end of the room's queue.
426
+
let position: i64 = sqlx::query_scalar(
427
+
"SELECT COALESCE(MAX(position), -1) + 1 FROM tracks WHERE room_id = ?",
428
+
)
429
+
.bind(room_id)
430
+
.fetch_one(&mut *tx)
431
+
.await?;
432
+
433
+
sqlx::query(
434
+
"INSERT INTO tracks
435
+
(id, room_id, url, title, duration, thumbnail, source,
436
+
position, added_at, pending)
437
+
VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10)",
438
+
)
439
+
.bind(track.id)
440
+
.bind(room_id)
441
+
.bind(&track.url)
442
+
.bind(&track.title)
443
+
.bind(&track.duration)
444
+
.bind(track.thumbnail.as_deref())
445
+
.bind("") // source is resolved later via PersistMetadata
446
+
.bind(position)
447
+
.bind(crate::util::now_iso())
448
+
.bind(track.pending as i64)
449
+
.execute(&mut *tx)
450
+
.await?;
451
+
}
452
+
Effect::PersistStarted(id) => {
453
+
let now_ms = chrono::Utc::now().timestamp_millis();
454
+
sqlx::query(
455
+
"UPDATE tracks SET started_at_ms = ?1 WHERE id = ?2 AND played = 0",
456
+
)
457
+
.bind(now_ms)
458
+
.bind(id)
459
+
.execute(&mut *tx)
460
+
.await?;
461
+
}
462
+
Effect::PersistFinished(id) => {
463
+
let played_at = crate::util::now_iso();
464
+
sqlx::query("UPDATE tracks SET played = 1, played_at = ?1 WHERE id = ?2")
465
+
.bind(&played_at)
466
+
.bind(id)
467
+
.execute(&mut *tx)
468
+
.await?;
469
+
}
470
+
Effect::PersistMetadata {
471
+
item_id,
472
+
title,
473
+
duration,
474
+
thumbnail,
475
+
source,
476
+
} => {
477
+
let src_str = source.to_string();
478
+
sqlx::query(
479
+
"UPDATE tracks
480
+
SET title = ?1, duration = ?2, thumbnail = ?3,
481
+
source = ?4, pending = 0
482
+
WHERE id = ?5",
483
+
)
484
+
.bind(title.as_str())
485
+
.bind(duration.as_str())
486
+
.bind(thumbnail.as_deref())
487
+
.bind(&src_str)
488
+
.bind(item_id)
489
+
.execute(&mut *tx)
490
+
.await?;
491
+
}
492
+
Effect::PersistChat {
493
+
user_name,
494
+
content,
495
+
msg_type,
496
+
created_at,
497
+
} => {
498
+
let result = sqlx::query(
499
+
"INSERT INTO chat_messages
500
+
(room_id, user_name, content, msg_type, created_at)
501
+
VALUES (?1, ?2, ?3, ?4, ?5)",
502
+
)
503
+
.bind(room_id)
504
+
.bind(user_name.as_str())
505
+
.bind(content.as_str())
506
+
.bind(msg_type.as_str())
507
+
.bind(created_at.as_str())
508
+
.execute(&mut *tx)
509
+
.await?;
510
+
output.chat_message_id = Some(result.last_insert_rowid());
511
+
}
512
+
// Non-persistence effects are ignored.
513
+
_ => {}
514
+
}
515
+
}
516
+
517
+
tx.commit().await?;
518
+
Ok(output)
519
+
}
520
+
521
+
async fn load(&self, room_id: &str) -> Result<Option<RoomSnapshot>, sqlx::Error> {
522
+
// 1. Room existence check.
523
+
let room_name: Option<String> = sqlx::query_scalar("SELECT name FROM rooms WHERE id = ?")
524
+
.bind(room_id)
525
+
.fetch_optional(&self.pool)
526
+
.await?;
527
+
528
+
let room_name = match room_name {
529
+
Some(name) => name,
530
+
None => return Ok(None),
531
+
};
532
+
533
+
// 2. Queue: not played, not started, ordered by position.
534
+
let queue_rows: Vec<TrackRow> = sqlx::query_as(
535
+
"SELECT id, url, title, duration, thumbnail, played, played_at,
536
+
started_at_ms, pending
537
+
FROM tracks
538
+
WHERE room_id = ? AND played = 0 AND started_at_ms IS NULL
539
+
ORDER BY position",
540
+
)
541
+
.bind(room_id)
542
+
.fetch_all(&self.pool)
543
+
.await?;
544
+
545
+
// 3. Active: not played, started (at most one).
546
+
let active_row: Option<TrackRow> = sqlx::query_as(
547
+
"SELECT id, url, title, duration, thumbnail, played, played_at,
548
+
started_at_ms, pending
549
+
FROM tracks
550
+
WHERE room_id = ? AND played = 0 AND started_at_ms IS NOT NULL
551
+
ORDER BY position
552
+
LIMIT 1",
553
+
)
554
+
.bind(room_id)
555
+
.fetch_optional(&self.pool)
556
+
.await?;
557
+
558
+
// 4. History: played, newest first.
559
+
let history_rows: Vec<TrackRow> = sqlx::query_as(
560
+
"SELECT id, url, title, duration, thumbnail, played, played_at,
561
+
started_at_ms, pending
562
+
FROM tracks
563
+
WHERE room_id = ? AND played = 1
564
+
ORDER BY played_at DESC",
565
+
)
566
+
.bind(room_id)
567
+
.fetch_all(&self.pool)
568
+
.await?;
569
+
570
+
// 5. Recent chat messages (most recent 50).
571
+
let chat_rows: Vec<ChatMsgRow> = sqlx::query_as(
572
+
"SELECT id, user_name, content, msg_type, created_at
573
+
FROM chat_messages
574
+
WHERE room_id = ?
575
+
ORDER BY id DESC
576
+
LIMIT 50",
577
+
)
578
+
.bind(room_id)
579
+
.fetch_all(&self.pool)
580
+
.await?;
581
+
582
+
let queue: Vec<QueuedTrack> = queue_rows
583
+
.into_iter()
584
+
.map(|r| QueuedTrack {
585
+
id: r.id,
586
+
title: r.title,
587
+
url: r.url,
588
+
duration: r.duration,
589
+
thumbnail: r.thumbnail,
590
+
pending: r.pending != 0,
591
+
})
592
+
.collect();
593
+
594
+
let active = active_row.map(|r| QueuedTrack {
595
+
id: r.id,
596
+
title: r.title,
597
+
url: r.url,
598
+
duration: r.duration,
599
+
thumbnail: r.thumbnail,
600
+
pending: r.pending != 0,
601
+
});
602
+
603
+
let history: Vec<FinishedTrack> = history_rows
604
+
.into_iter()
605
+
.map(|r| FinishedTrack {
606
+
id: r.id,
607
+
title: r.title,
608
+
url: r.url,
609
+
duration: r.duration,
610
+
thumbnail: r.thumbnail,
611
+
played_at: r.played_at.unwrap_or_default(),
612
+
})
613
+
.collect();
614
+
615
+
let chat_messages: Vec<ChatMessage> = chat_rows
616
+
.into_iter()
617
+
.map(|r| ChatMessage {
618
+
id: r.id,
619
+
user_name: r.user_name,
620
+
content: r.content,
621
+
msg_type: r.msg_type,
622
+
created_at: r.created_at,
623
+
})
624
+
.collect();
625
+
626
+
Ok(Some(RoomSnapshot {
627
+
room_id: room_id.to_string(),
628
+
room_name,
629
+
queue,
630
+
active,
631
+
history,
632
+
chat_messages,
633
+
}))
634
+
}
635
+
636
+
async fn recent_chat(
637
+
&self,
638
+
room_id: &str,
639
+
limit: i64,
640
+
) -> Result<Vec<ChatMessage>, sqlx::Error> {
641
+
let rows: Vec<ChatMsgRow> = sqlx::query_as(
642
+
"SELECT id, user_name, content, msg_type, created_at
643
+
FROM chat_messages
644
+
WHERE room_id = ?
645
+
ORDER BY id DESC
646
+
LIMIT ?",
647
+
)
648
+
.bind(room_id)
649
+
.bind(limit)
650
+
.fetch_all(&self.pool)
651
+
.await?;
652
+
653
+
Ok(rows
654
+
.into_iter()
655
+
.map(|r| ChatMessage {
656
+
id: r.id,
657
+
user_name: r.user_name,
658
+
content: r.content,
659
+
msg_type: r.msg_type,
660
+
created_at: r.created_at,
661
+
})
662
+
.collect())
663
+
}
664
+
665
+
async fn delete_room(&self, room_id: &str) -> Result<(), sqlx::Error> {
666
+
sqlx::query("DELETE FROM rooms WHERE id = ?")
667
+
.bind(room_id)
668
+
.execute(&self.pool)
669
+
.await?;
670
+
Ok(())
671
+
}
672
+
}
673
+
674
+
#[cfg(test)]
675
+
mod tests {
676
+
use super::*;
677
+
678
+
// ------------------------------------------------------------------
679
+
// InMemoryStore tests
680
+
// ------------------------------------------------------------------
681
+
682
+
#[tokio::test]
683
+
async fn test_create_and_load_roundtrip() {
684
+
let store = InMemoryStore::new();
685
+
store.create_room("test", "Test Room").await.unwrap();
686
+
687
+
let snapshot = store.load("test").await.unwrap().unwrap();
688
+
assert_eq!(snapshot.room_id, "test");
689
+
assert_eq!(snapshot.room_name, "Test Room");
690
+
assert!(snapshot.queue.is_empty());
691
+
assert!(snapshot.active.is_none());
692
+
assert!(snapshot.history.is_empty());
693
+
assert!(snapshot.chat_messages.is_empty());
694
+
}
695
+
696
+
#[tokio::test]
697
+
async fn test_load_nonexistent_room() {
698
+
let store = InMemoryStore::new();
699
+
let snapshot = store.load("nonexistent").await.unwrap();
700
+
assert!(snapshot.is_none());
701
+
}
702
+
703
+
#[tokio::test]
704
+
async fn test_delete_room() {
705
+
let store = InMemoryStore::new();
706
+
store.create_room("test", "Test Room").await.unwrap();
707
+
708
+
store.delete_room("test").await.unwrap();
709
+
assert!(store.load("test").await.unwrap().is_none());
710
+
}
711
+
712
+
#[tokio::test]
713
+
async fn test_delete_nonexistent_room_is_noop() {
714
+
let store = InMemoryStore::new();
715
+
store.delete_room("nonexistent").await.unwrap();
716
+
}
717
+
718
+
#[tokio::test]
719
+
async fn test_recent_chat_empty_room() {
720
+
let store = InMemoryStore::new();
721
+
store.create_room("test", "Test").await.unwrap();
722
+
723
+
let msgs = store.recent_chat("test", 10).await.unwrap();
724
+
assert!(msgs.is_empty());
725
+
}
726
+
727
+
#[tokio::test]
728
+
async fn test_recent_chat_nonexistent_room() {
729
+
let store = InMemoryStore::new();
730
+
let msgs = store.recent_chat("nobody", 10).await.unwrap();
731
+
assert!(msgs.is_empty());
732
+
}
733
+
734
+
#[tokio::test]
735
+
async fn test_persist_preserves_room_order() {
736
+
let store = InMemoryStore::new();
737
+
store.create_room("a", "A").await.unwrap();
738
+
store.create_room("b", "B").await.unwrap();
739
+
740
+
let snap_a = store.load("a").await.unwrap().unwrap();
741
+
let snap_b = store.load("b").await.unwrap().unwrap();
742
+
assert_eq!(snap_a.room_name, "A");
743
+
assert_eq!(snap_b.room_name, "B");
744
+
}
745
+
746
+
#[tokio::test]
747
+
async fn test_persist_queued_track() {
748
+
let store = InMemoryStore::new();
749
+
store.create_room("r", "R").await.unwrap();
750
+
751
+
let track = QueuedTrack {
752
+
id: 1,
753
+
title: "Loading...".into(),
754
+
url: "https://example.com/t".into(),
755
+
duration: "--:--".into(),
756
+
thumbnail: None,
757
+
pending: true,
758
+
};
759
+
760
+
store
761
+
.persist("r", &[Effect::PersistQueuedTrack(track)])
762
+
.await
763
+
.unwrap();
764
+
765
+
let snap = store.load("r").await.unwrap().unwrap();
766
+
assert_eq!(snap.queue.len(), 1);
767
+
assert_eq!(snap.queue[0].id, 1);
768
+
assert!(snap.queue[0].pending);
769
+
assert!(snap.active.is_none());
770
+
assert!(snap.history.is_empty());
771
+
}
772
+
773
+
#[tokio::test]
774
+
async fn test_persist_started_moves_track_to_active() {
775
+
let store = InMemoryStore::new();
776
+
store.create_room("r", "R").await.unwrap();
777
+
778
+
let track = QueuedTrack {
779
+
id: 1,
780
+
title: "Track A".into(),
781
+
url: "https://example.com/a".into(),
782
+
duration: "3:45".into(),
783
+
thumbnail: None,
784
+
pending: false,
785
+
};
786
+
787
+
store
788
+
.persist("r", &[Effect::PersistQueuedTrack(track)])
789
+
.await
790
+
.unwrap();
791
+
store
792
+
.persist("r", &[Effect::PersistStarted(1)])
793
+
.await
794
+
.unwrap();
795
+
796
+
let snap = store.load("r").await.unwrap().unwrap();
797
+
assert!(snap.queue.is_empty());
798
+
assert!(snap.active.is_some());
799
+
assert_eq!(snap.active.unwrap().id, 1);
800
+
}
801
+
802
+
#[tokio::test]
803
+
async fn test_persist_started_nonexistent_track_is_noop() {
804
+
let store = InMemoryStore::new();
805
+
store.create_room("r", "R").await.unwrap();
806
+
807
+
store
808
+
.persist("r", &[Effect::PersistStarted(999)])
809
+
.await
810
+
.unwrap();
811
+
let snap = store.load("r").await.unwrap().unwrap();
812
+
assert!(snap.active.is_none());
813
+
}
814
+
815
+
#[tokio::test]
816
+
async fn test_persist_finished_moves_active_to_history() {
817
+
let store = InMemoryStore::new();
818
+
store.create_room("r", "R").await.unwrap();
819
+
820
+
store
821
+
.persist(
822
+
"r",
823
+
&[
824
+
Effect::PersistQueuedTrack(QueuedTrack {
825
+
id: 1,
826
+
title: "A".into(),
827
+
url: "https://e.com/a".into(),
828
+
duration: "3:00".into(),
829
+
thumbnail: None,
830
+
pending: false,
831
+
}),
832
+
Effect::PersistStarted(1),
833
+
],
834
+
)
835
+
.await
836
+
.unwrap();
837
+
store
838
+
.persist("r", &[Effect::PersistFinished(1)])
839
+
.await
840
+
.unwrap();
841
+
842
+
let snap = store.load("r").await.unwrap().unwrap();
843
+
assert!(snap.active.is_none());
844
+
assert_eq!(snap.history.len(), 1);
845
+
assert_eq!(snap.history[0].id, 1);
846
+
assert!(!snap.history[0].played_at.is_empty());
847
+
}
848
+
849
+
#[tokio::test]
850
+
async fn test_persist_finished_without_active_is_noop() {
851
+
let store = InMemoryStore::new();
852
+
store.create_room("r", "R").await.unwrap();
853
+
854
+
store
855
+
.persist("r", &[Effect::PersistFinished(1)])
856
+
.await
857
+
.unwrap();
858
+
let snap = store.load("r").await.unwrap().unwrap();
859
+
assert!(snap.history.is_empty());
860
+
}
861
+
862
+
#[tokio::test]
863
+
async fn test_persist_metadata_updates_queued_track() {
864
+
let store = InMemoryStore::new();
865
+
store.create_room("r", "R").await.unwrap();
866
+
867
+
let track = QueuedTrack {
868
+
id: 1,
869
+
title: "Loading...".into(),
870
+
url: "https://e.com/t".into(),
871
+
duration: "--:--".into(),
872
+
thumbnail: None,
873
+
pending: true,
874
+
};
875
+
876
+
store
877
+
.persist("r", &[Effect::PersistQueuedTrack(track)])
878
+
.await
879
+
.unwrap();
880
+
store
881
+
.persist(
882
+
"r",
883
+
&[Effect::PersistMetadata {
884
+
item_id: 1,
885
+
title: "Real Title".into(),
886
+
duration: "4:20".into(),
887
+
thumbnail: Some("https://img.example/t.jpg".into()),
888
+
source: "ytdlp".into(),
889
+
}],
890
+
)
891
+
.await
892
+
.unwrap();
893
+
894
+
let snap = store.load("r").await.unwrap().unwrap();
895
+
assert_eq!(snap.queue[0].title, "Real Title");
896
+
assert_eq!(snap.queue[0].duration, "4:20");
897
+
assert_eq!(
898
+
snap.queue[0].thumbnail,
899
+
Some("https://img.example/t.jpg".into())
900
+
);
901
+
assert!(!snap.queue[0].pending);
902
+
}
903
+
904
+
#[tokio::test]
905
+
async fn test_persist_metadata_updates_active_track() {
906
+
let store = InMemoryStore::new();
907
+
store.create_room("r", "R").await.unwrap();
908
+
909
+
store
910
+
.persist(
911
+
"r",
912
+
&[
913
+
Effect::PersistQueuedTrack(QueuedTrack {
914
+
id: 1,
915
+
title: "Loading...".into(),
916
+
url: "https://e.com/t".into(),
917
+
duration: "--:--".into(),
918
+
thumbnail: None,
919
+
pending: true,
920
+
}),
921
+
Effect::PersistStarted(1),
922
+
],
923
+
)
924
+
.await
925
+
.unwrap();
926
+
store
927
+
.persist(
928
+
"r",
929
+
&[Effect::PersistMetadata {
930
+
item_id: 1,
931
+
title: "Real Title".into(),
932
+
duration: "4:20".into(),
933
+
thumbnail: None,
934
+
source: "direct".into(),
935
+
}],
936
+
)
937
+
.await
938
+
.unwrap();
939
+
940
+
let snap = store.load("r").await.unwrap().unwrap();
941
+
assert!(snap.queue.is_empty());
942
+
let active = snap.active.unwrap();
943
+
assert_eq!(active.title, "Real Title");
944
+
assert!(!active.pending);
945
+
}
946
+
947
+
#[tokio::test]
948
+
async fn test_persist_chat() {
949
+
let store = InMemoryStore::new();
950
+
store.create_room("r", "R").await.unwrap();
951
+
952
+
let output = store
953
+
.persist(
954
+
"r",
955
+
&[Effect::PersistChat {
956
+
user_name: "alice".into(),
957
+
content: "hello".into(),
958
+
msg_type: "message".into(),
959
+
created_at: "2025-01-01T00:00:00.000Z".into(),
960
+
}],
961
+
)
962
+
.await
963
+
.unwrap();
964
+
965
+
assert_eq!(output.chat_message_id, Some(1));
966
+
967
+
let snap = store.load("r").await.unwrap().unwrap();
968
+
assert_eq!(snap.chat_messages.len(), 1);
969
+
assert_eq!(snap.chat_messages[0].user_name, "alice");
970
+
assert_eq!(snap.chat_messages[0].content, "hello");
971
+
}
972
+
973
+
#[tokio::test]
974
+
async fn test_persist_chat_multiple_increments_id() {
975
+
let store = InMemoryStore::new();
976
+
store.create_room("r", "R").await.unwrap();
977
+
978
+
let o1 = store
979
+
.persist(
980
+
"r",
981
+
&[Effect::PersistChat {
982
+
user_name: "a".into(),
983
+
content: "1".into(),
984
+
msg_type: "message".into(),
985
+
created_at: "t1".into(),
986
+
}],
987
+
)
988
+
.await
989
+
.unwrap();
990
+
let o2 = store
991
+
.persist(
992
+
"r",
993
+
&[Effect::PersistChat {
994
+
user_name: "b".into(),
995
+
content: "2".into(),
996
+
msg_type: "message".into(),
997
+
created_at: "t2".into(),
998
+
}],
999
+
)
1000
+
.await
1001
+
.unwrap();
1002
+
1003
+
assert_eq!(o1.chat_message_id, Some(1));
1004
+
assert_eq!(o2.chat_message_id, Some(2));
1005
+
1006
+
let snap = store.load("r").await.unwrap().unwrap();
1007
+
assert_eq!(snap.chat_messages.len(), 2);
1008
+
}
1009
+
1010
+
#[tokio::test]
1011
+
async fn test_persist_to_nonexistent_room_is_noop() {
1012
+
let store = InMemoryStore::new();
1013
+
let output = store
1014
+
.persist(
1015
+
"ghost",
1016
+
&[Effect::PersistChat {
1017
+
user_name: "x".into(),
1018
+
content: "x".into(),
1019
+
msg_type: "message".into(),
1020
+
created_at: "x".into(),
1021
+
}],
1022
+
)
1023
+
.await
1024
+
.unwrap();
1025
+
assert!(output.chat_message_id.is_none());
1026
+
}
1027
+
1028
+
#[tokio::test]
1029
+
async fn test_persist_recent_chat_ordering() {
1030
+
let store = InMemoryStore::new();
1031
+
store.create_room("r", "R").await.unwrap();
1032
+
1033
+
for i in 0..5 {
1034
+
store
1035
+
.persist(
1036
+
"r",
1037
+
&[Effect::PersistChat {
1038
+
user_name: "user".into(),
1039
+
content: format!("msg {i}"),
1040
+
msg_type: "message".into(),
1041
+
created_at: format!("t{i}"),
1042
+
}],
1043
+
)
1044
+
.await
1045
+
.unwrap();
1046
+
}
1047
+
1048
+
let recent = store.recent_chat("r", 3).await.unwrap();
1049
+
assert_eq!(recent.len(), 3);
1050
+
assert_eq!(recent[0].content, "msg 4");
1051
+
assert_eq!(recent[2].content, "msg 2");
1052
+
}
1053
+
}
+8
-25
src/web.rs
+8
-25
src/web.rs
···
12
12
response::{Html, IntoResponse, Json},
13
13
routing::{get, post},
14
14
};
15
-
use sqlx::{Pool, Sqlite};
16
15
use tokio::sync::Mutex;
17
16
18
-
use crate::db;
19
17
use crate::room;
20
18
use crate::types::{RoomId, TrackMeta};
21
19
···
55
53
#[derive(Clone)]
56
54
pub(crate) struct AppState {
57
55
pub rooms: room::Registry,
58
-
pub pool: Pool<Sqlite>,
59
56
pub rate_limiter: RateLimiter,
60
57
}
61
58
62
-
pub(crate) fn router(rooms: room::Registry, pool: Pool<Sqlite>) -> Router {
59
+
pub(crate) fn router(rooms: room::Registry) -> Router {
63
60
let rate_limiter = RateLimiter::new();
64
61
65
62
let cleanup_limiter = rate_limiter.clone();
···
73
70
74
71
let state = AppState {
75
72
rooms,
76
-
pool,
77
73
rate_limiter,
78
74
};
79
75
···
126
122
let queue = state.rooms.queue(&room_id).await;
127
123
let queue_empty = queue.is_empty();
128
124
129
-
let history = {
130
-
let playlist = crate::playlist::Playlist::new(state.pool.clone(), &id);
131
-
playlist
132
-
.history(20)
133
-
.await
134
-
.unwrap_or_default()
135
-
.into_iter()
136
-
.map(|i| TrackMeta {
137
-
title: i.title,
138
-
duration: i.duration,
139
-
thumbnail: i.thumbnail,
140
-
url: i.url,
141
-
source: i.source.parse().unwrap_or_else(|s: String| {
142
-
tracing::warn!(source = %s, "unknown source kind in queue_items");
143
-
crate::types::SourceKind::Direct
144
-
}),
145
-
})
146
-
.collect::<Vec<_>>()
125
+
// Load snapshot once for room name + history.
126
+
let (room_name, history) = {
127
+
let rid = RoomId(id.clone());
128
+
match state.rooms.load_room_data(&rid).await {
129
+
Some((name, hist)) => (name, hist),
130
+
None => (String::new(), Vec::new()),
131
+
}
147
132
};
148
-
149
-
let room_name = db::get_room_name(&state.pool, &id).await;
150
133
151
134
let tpl = RoomTemplate {
152
135
room_id: id,