···66license = "MIT OR Apache-2.0"
7788[dependencies]
99-# ── Async ──────────────────────────────────────────────────────────
109tokio = { version = "1", features = ["full"] }
1110anyhow = "1"
1211thiserror = "2"
1313-1414-# ── HTTP / SSR ─────────────────────────────────────────────────────
1512axum = { version = "0.8", features = ["ws"] }
1616-tower = "0.5"
1717-tower-http = { version = "0.6", features = ["cors", "trace", "fs", "compression-gzip"] }
1813tracing = "0.1"
1914tracing-subscriber = { version = "0.3", features = ["env-filter"] }
2015serde = { version = "1", features = ["derive"] }
2116serde_json = "1"
2222-2323-# ── Templates ──────────────────────────────────────────────────────
2417askama = "0.13"
2525-2626-# ── CLI ────────────────────────────────────────────────────────────
2718clap = { version = "4", features = ["derive", "env"] }
2828-2929-# ── WebSocket / MoQ Transport ───────────────────────────────────────
3030-tokio-tungstenite = "0.24"
3119futures-util = "0.3"
3232-3333-# ── TLS (reserved for future use) ─────────────────────────────────────
3434-rustls = { version = "0.23", features = ["ring"] }
3535-rcgen = "0.13"
3636-3737-# ── Async ─────────────────────────────────────────────────────────
3820async-trait = "0.1"
3939-4040-# ── Identifiers ────────────────────────────────────────────────────
4141-uuid = { version = "1", features = ["v4", "serde"] }
4242-4343-# ── Database ────────────────────────────────────────────────────────
4421sqlx = { version = "0.8", features = ["sqlite", "runtime-tokio", "chrono"] }
4522chrono = { version = "0.4", features = ["serde"] }
4623bytes = "1"
+44-6
README.md
···7788One Axum binary serving SSR pages, REST API, and a custom media streaming
99protocol over WebSocket on the same port. Per-room mpsc actor processes commands
1010-(queue, skip, chat, register) sequentially — no locks. Room state persisted to
1111-SQLite (WAL mode).
1010+(queue, skip, chat, register, metadata) sequentially -- no locks. Room state
1111+persisted to SQLite (WAL mode).
1212+1313+### Async metadata extraction
1414+1515+The ingest HTTP handler returns immediately after queuing a URL. Metadata
1616+extraction (yt-dlp) runs in a spawned task within the actor. When extraction
1717+completes, the result is sent back to the actor which updates the DB and pushes
1818+the resolved metadata to clients via the state broadcast channel.
1919+2020+```mermaid
2121+sequenceDiagram
2222+ participant Client
2323+ participant HTTP as HTTP Handler
2424+ participant Actor as Room Actor
2525+ participant Extract as Extraction Task
2626+2727+ Client->>HTTP: POST /api/ingest
2828+ HTTP->>Actor: QueueUrl(url)
2929+ HTTP-->>Client: {ok: true}
3030+ Actor->>Actor: insert placeholder
3131+ Actor->>Extract: spawn extraction
3232+ Extract->>Actor: MetadataReady
3333+ Actor->>Client: PublishSnapshot via WS
3434+```
3535+3636+### Playback pipeline
12371338yt-dlp resolves URLs to direct stream addresses. FFmpeg transcodes to fragmented
1439MP4 at real-time speed (`-re`). fMP4 boxes are read from stdout as moof+mdat
1540pairs and published over per-room broadcast channels. The video init segment
1616-(ftyp+moov) is cached separately. Forward tasks send cached init first, then
1717-catch up to the broadcast live edge before forwarding, so late joiners start at
1818-the live position.
4141+(ftyp+moov) is cached for late-joining clients.
4242+4343+### State machine
4444+4545+`PlaybackState` is pure -- all transitions are deterministic. The actor follows
4646+the impure-pure-impure sandwich:
4747+4848+1. **Gather** -- DB queries, wall clock
4949+2. **Transition** -- `PlaybackState::transition(event, now)` -- returns `Effect`s
5050+3. **Commit** -- actor executes effects
5151+5252+Events: `TrackQueued`, `Skip`, `TrackEnded`, `MetadataUpdated`.
5353+Effects: `AbortPipeline`, `StartPipeline`, `PersistStarted`, `PersistFinished`, `PublishSnapshot`.
5454+5555+### Client delivery
19562057Clients use MSE with a single SourceBuffer in segments mode. A state track
2158(watch channel) delivers current playback position, queue, and history as JSON.
2222-Chat is persisted to SQLite, replayed on subscribe (last N), then live.
5959+Queue items carry a `pending` flag while metadata is unresolved. Chat is
6060+persisted to SQLite, replayed on subscribe, then live.
···44//! real-time speed (`-re`), reads fMP4 boxes from stdout as moof+mdat pairs,
55//! and publishes them over per-room broadcast channels.
66//!
77-//! This module does **not** resolve URLs or extract metadata — that's the
77+//! This module does **not** resolve URLs or extract metadata: that is the
88//! responsibility of [`crate::source`].
991010use tokio::io::AsyncReadExt;
1111use tokio::process::Command;
12121313use crate::transport::TrackPublishers;
1414-1515-// ---------------------------------------------------------------------------
1616-// Error
1717-// ---------------------------------------------------------------------------
18141915/// Errors that can occur during FFmpeg transcoding.
2016#[derive(Debug, thiserror::Error)]
···2723 Aborted,
2824}
29253030-// ---------------------------------------------------------------------------
3131-// Public API
3232-// ---------------------------------------------------------------------------
3333-3426/// Transcode a playable stream URL to fMP4, publishing raw boxes as MoQ objects.
3527///
3628/// Spawns `ffmpeg` with the stream URL as input, transcodes to fragmented MP4
···3931/// can initialise their MediaSource.
4032///
4133/// Returns `Ok(())` on normal EOF, `Err` on failure.
4242-///
4343-/// ## Abort behaviour
4434///
4535/// When `abort` is signalled, the pipeline exits early with
4636/// [`TranscodeError::Aborted`] and the ffmpeg process is killed.
···113103 // Publishing complete boxes ensures MSE can consume them directly.
114104 read_and_publish_boxes(&mut stdout, &publishers, &mut abort, group_id).await?;
115105116116- // 2. Wait for FFmpeg to exit
117106 let status = ffmpeg
118107 .wait()
119108 .await
···128117129118 if !status.success() {
130119 tracing::warn!("ffmpeg exited with status {status}; stderr: {stderr_output}");
131131- // Still return Ok — EOF is EOF even if ffmpeg reported an error at the end
120120+ // Still return Ok: EOF is EOF even if ffmpeg reported an error at the end
132121 }
133122134123 tracing::info!(status = %status, stderr_len = %stderr_output.len(), "ffmpeg pipeline finished");
135124136125 Ok(())
137126}
138138-139139-// ---------------------------------------------------------------------------
140140-// fMP4 box parsing
141141-// ---------------------------------------------------------------------------
142127143128/// Read fMP4 boxes from FFmpeg stdout, buffer moof+mdat pairs, and publish
144129/// each complete segment as a single MoQ object.
···163148 let mut pending_moof: Option<Vec<u8>> = None;
164149165150 loop {
166166- // Read 8-byte box header
167151 let mut header = [0u8; 8];
168152 tokio::select! {
169153 _ = abort.changed() => return Err(TranscodeError::Aborted),
···229213 let size = u32::from_be_bytes([header[0], header[1], header[2], header[3]]);
230214231215 if size == 0 {
232232- // Last box — read all remaining data.
216216+ // Last box: read all remaining data.
233217 let mut remaining = Vec::new();
234218 reader.read_to_end(&mut remaining).await?;
235219 let mut full = header.to_vec();
···260244 full.extend_from_slice(&payload);
261245 Ok(full)
262246}
263263-264264-// ---------------------------------------------------------------------------
265265-// Formatting
266266-// ---------------------------------------------------------------------------
267247268248/// Format a duration in seconds to a human-readable `M:SS` or `H:MM:SS` string.
269249pub(crate) fn format_duration(total_secs: f64) -> String {
···279259 }
280260}
281261282282-// ---------------------------------------------------------------------------
283283-// Tests
284284-// ---------------------------------------------------------------------------
285285-286262#[cfg(test)]
287263mod tests {
288264 use super::*;
···323299 let mut slice: &[u8] = &input;
324300 read_and_publish_boxes(&mut slice, &publishers, &mut abort_rx, 0)
325301 .await
326326- .expect("read_and_publish_boxes should succeed on valid fMP4");
302302+ .unwrap();
327303328328- // Init is cached (for late joiners) AND published to broadcast (for track changes).
329304 let cached = publishers.get_init_segment();
330330- assert!(cached.is_some(), "init cache should contain ftyp+moov");
305305+ assert!(cached.is_some());
331306 let (cached_group, cached_data) = cached.unwrap();
332307 assert_eq!(cached_group, 0);
333308 let mut expected_init = ftyp.clone();
334309 expected_init.extend_from_slice(&moov);
335310 assert_eq!(cached_data.to_vec(), expected_init);
336311337337- // Object 0: init (ftyp+moov) published to broadcast.
338338- let init_obj = rx.recv().await.expect("should receive init via broadcast");
312312+ let init_obj = rx.recv().await.unwrap();
339313 assert_eq!(init_obj.object_id, 0);
340314 let mut expected_init_broadcast = ftyp.clone();
341315 expected_init_broadcast.extend_from_slice(&moov);
342316 assert_eq!(init_obj.payload.to_vec(), expected_init_broadcast);
343317344344- // Object 1: moof+mdat combined.
345345- let obj1 = rx.recv().await.expect("should receive combined moof+mdat");
318318+ let obj1 = rx.recv().await.unwrap();
346319 assert_eq!(obj1.object_id, 1);
347320 let mut expected_media = moof.clone();
348321 expected_media.extend_from_slice(&mdat);
···350323 }
351324352325 #[tokio::test]
353353- async fn test_init_broadcast_and_cache() {
326326+ async fn test_init_cache_has_group_id() {
354327 let ftyp = make_box(b"ftyp", b"iso5");
355328 let moov = make_box(b"moov", b"moovdata");
356356- let moof = make_box(b"moof", b"moofdata");
357357- let mdat = make_box(b"mdat", b"datadata");
358329359330 let mut input = Vec::new();
360331 input.extend_from_slice(&ftyp);
361332 input.extend_from_slice(&moov);
362362- input.extend_from_slice(&moof);
363363- input.extend_from_slice(&mdat);
364333365334 let publishers = TrackPublishers::new();
366366- let mut rx = publishers.video.subscribe();
367335 let (_tx, mut abort_rx) = tokio::sync::watch::channel(false);
368336369337 let mut slice: &[u8] = &input;
370370- read_and_publish_boxes(&mut slice, &publishers, &mut abort_rx, 0)
338338+ read_and_publish_boxes(&mut slice, &publishers, &mut abort_rx, 42)
371339 .await
372372- .expect("read_and_publish_boxes should succeed");
340340+ .unwrap();
373341374374- // Init is both cached and broadcast.
375342 let cached = publishers.get_init_segment();
376376- assert!(cached.is_some(), "init cache should contain ftyp+moov");
377377-378378- // Object 0: init via broadcast.
379379- let init = rx.recv().await.expect("should receive init via broadcast");
380380- assert_eq!(init.object_id, 0);
381381- assert_eq!(init.group_id, 0);
382382- let mut expected_init = ftyp.clone();
383383- expected_init.extend_from_slice(&moov);
384384- assert_eq!(init.payload.to_vec(), expected_init);
385385-386386- // Object 1: moof+mdat media segment.
387387- let media = rx.recv().await.expect("should receive media segment");
388388- assert_eq!(media.object_id, 1);
389389- let mut expected_media = moof.clone();
390390- expected_media.extend_from_slice(&mdat);
391391- assert_eq!(media.payload.to_vec(), expected_media);
343343+ assert!(cached.is_some());
344344+ let (group_id, _data) = cached.unwrap();
345345+ assert_eq!(group_id, 42);
392346 }
393347394348 #[tokio::test]
395395- async fn test_init_cache_has_group_id() {
396396- let ftyp = make_box(b"ftyp", b"iso5");
397397- let moov = make_box(b"moov", b"moovdata");
398398-399399- let mut input = Vec::new();
400400- input.extend_from_slice(&ftyp);
401401- input.extend_from_slice(&moov);
402402-349349+ async fn test_empty_input_no_crash() {
403350 let publishers = TrackPublishers::new();
404351 let (_tx, mut abort_rx) = tokio::sync::watch::channel(false);
405405-406406- let mut slice: &[u8] = &input;
407407- read_and_publish_boxes(&mut slice, &publishers, &mut abort_rx, 42)
352352+ let mut empty: &[u8] = &[];
353353+ read_and_publish_boxes(&mut empty, &publishers, &mut abort_rx, 0)
408354 .await
409409- .expect("read_and_publish_boxes should succeed");
355355+ .unwrap();
356356+ }
410357411411- let cached = publishers.get_init_segment();
412412- assert!(cached.is_some(), "init should be cached");
413413- let (group_id, _data) = cached.unwrap();
414414- assert_eq!(
415415- group_id, 42,
416416- "group_id in cache should match the input group_id"
417417- );
358358+ #[tokio::test]
359359+ async fn test_partial_box_returns_error() {
360360+ let publishers = TrackPublishers::new();
361361+ let (_tx, mut abort_rx) = tokio::sync::watch::channel(false);
362362+ let mut partial: &[u8] = &[0, 0, 0, 100, b'f', b't', b'y', b'p'];
363363+ read_and_publish_boxes(&mut partial, &publishers, &mut abort_rx, 0)
364364+ .await
365365+ .unwrap_err();
418366 }
419367}
+27
src/names.rs
···337337 "yacht",
338338 "zebra",
339339];
340340+341341+#[cfg(test)]
342342+mod tests {
343343+ use super::*;
344344+345345+ #[test]
346346+ fn test_generate_room_name_is_deterministic() {
347347+ let name1 = generate_room_name("12345");
348348+ let name2 = generate_room_name("12345");
349349+ assert_eq!(name1, name2);
350350+ }
351351+352352+ #[test]
353353+ fn test_generate_room_name_differs_for_diff_ids() {
354354+ let name1 = generate_room_name("11111");
355355+ let name2 = generate_room_name("22222");
356356+ assert_ne!(name1, name2);
357357+ }
358358+359359+ #[test]
360360+ fn test_generate_room_name_has_hyphen() {
361361+ let name = generate_room_name("99999");
362362+ assert!(name.contains('-'), "room name should be adjective-noun");
363363+ let parts: Vec<&str> = name.split('-').collect();
364364+ assert_eq!(parts.len(), 2);
365365+ }
366366+}
+16-61
src/playback.rs
···11//! Playback lifecycle management.
22//!
33//! The [`Player`] struct owns the abort handle for the currently-running
44-//! transcoding pipeline. It does **not** track what's playing — that's the
44+//! transcoding pipeline. It does **not** track what is playing: that is the
55//! [`PlaybackState`](crate::state::PlaybackState) machine's job.
66-//!
77-//! # Lifecycle
88-//!
99-//! 1. **Start**: [`Player::start`] receives a fully-formed [`ActiveTrackInfo`],
1010-//! resolves the URL via the source registry, and spawns ffmpeg transcoding.
1111-//! 2. **End**: The spawned task sends [`RoomCommand::TrackEnded`] when ffmpeg
1212-//! finishes or errors. The room actor calls [`Player::on_track_ended`] to
1313-//! acknowledge.
1414-//! 3. **Skip**: [`Player::abort`] signals the transcoding task to stop.
156167use std::path::PathBuf;
178use std::sync::Arc;
···26172718/// Manages the abort handle for the currently-running transcoding pipeline.
2819///
2929-/// The player is **not** responsible for queue or state management — that
2020+/// The player is **not** responsible for queue or state management: that
3021/// belongs to the [`PlaybackState`](crate::state::PlaybackState) machine in the
3122/// room actor.
3223pub(crate) struct Player {
···6960 let pipeline_tx = self.pipeline_tx.clone();
70617162 tokio::spawn(async move {
7272- // 1. Resolve URL to a playable stream.
7363 let stream_url = match source_registry.resolve(&url, &cache_dir).await {
7464 Ok(url) => url,
7565 Err(e) => {
···7969 }
8070 };
81718282- // 2. Transcode to fMP4 via ffmpeg.
8372 let group_id = track_id as u64;
8473 let result =
8574 media::transcode_to_fmp4(&stream_url, publishers, abort_rx, group_id).await;
···10998 ///
11099 /// Returns `true` if `item_id` matched the currently-tracked pipeline,
111100 /// `false` if it was a stale event from an older pipeline.
101101+ #[must_use]
112102 pub(crate) fn on_track_ended(&mut self, item_id: i64) -> bool {
113103 if self.track_id == Some(item_id) {
114104 self.track_id = None;
···125115 tracing::warn!(item_id, error = %e, "player: failed to send TrackEnded");
126116 }
127117}
128128-129129-// ---------------------------------------------------------------------------
130130-// Tests
131131-// ---------------------------------------------------------------------------
132118133119#[cfg(test)]
134120mod tests {
135121 use super::*;
136136- use crate::db;
137137- use crate::playlist::Playlist;
138138- use crate::types::TrackMeta;
139139- use sqlx::sqlite::SqlitePoolOptions;
140140- use sqlx::Pool;
141122 use std::path::Path;
142123 use std::sync::Arc;
124124+ use std::sync::atomic::{AtomicI64, Ordering};
143125144126 struct MockPlayerSource;
145127···149131 &self,
150132 url: &str,
151133 _cache_dir: &Path,
152152- ) -> Result<TrackMeta, crate::source::SourceError> {
134134+ ) -> Result<crate::types::TrackMeta, crate::source::SourceError> {
153135 Err(crate::source::SourceError::Unsupported(url.into()))
154136 }
155137···168150 Arc::new(reg)
169151 }
170152171171- async fn test_pool() -> Pool<sqlx::Sqlite> {
172172- let pool = SqlitePoolOptions::new()
173173- .max_connections(1)
174174- .connect(":memory:")
175175- .await
176176- .unwrap();
177177- db::run_migrations(&pool).await.unwrap();
178178- db::room_insert(&pool, "test-room", "test-name", false)
179179- .await
180180- .unwrap();
181181- pool
182182- }
183183-184184- async fn make_queue_item(pool: &Pool<sqlx::Sqlite>) -> i64 {
185185- let pl = Playlist::new(pool.clone(), "test-room");
186186- pl.push(&TrackMeta {
187187- title: "Test".into(),
188188- duration: "3:45".into(),
189189- thumbnail: None,
190190- url: "https://example.com/track".into(),
191191- source: crate::types::SourceKind::Direct,
192192- })
193193- .await
194194- .unwrap()
195195- .id
153153+ fn make_item_id() -> i64 {
154154+ static NEXT_ID: AtomicI64 = AtomicI64::new(1);
155155+ NEXT_ID.fetch_add(1, Ordering::Relaxed)
196156 }
197157198158 fn test_info(id: i64) -> ActiveTrackInfo {
···208168209169 #[tokio::test]
210170 async fn test_player_start_then_abort() {
211211- let pool = test_pool().await;
212212- let item_id = make_queue_item(&pool).await;
171171+ let item_id = make_item_id();
213172 let (tx, _rx) = mpsc::channel(256);
214173215174 let mut player = Player::new(test_registry(), PathBuf::from("/tmp"), tx);
···217176218177 player.abort();
219178220220- // on_track_ended with a stale id should return false.
221179 assert!(!player.on_track_ended(99999));
222180 }
223181224182 #[tokio::test]
225183 async fn test_player_abort_idempotent() {
226226- let pool = test_pool().await;
227227- let item_id = make_queue_item(&pool).await;
184184+ let item_id = make_item_id();
228185 let (tx, _rx) = mpsc::channel(256);
229186230187 let mut player = Player::new(test_registry(), PathBuf::from("/tmp"), tx);
231188 player.start(&test_info(item_id), TrackPublishers::new());
232189233190 player.abort();
234234- player.abort(); // second abort is a no-op
235235- // Did not panic = pass
191191+ player.abort();
192192+ assert!(!player.on_track_ended(99999));
236193 }
237194238195 #[tokio::test]
239196 async fn test_player_abort_when_idle() {
240197 let (tx, _rx) = mpsc::channel(256);
241198 let mut player = Player::new(test_registry(), PathBuf::from("/tmp"), tx);
242242- player.abort(); // no-op
199199+ player.abort();
200200+ assert!(!player.on_track_ended(99999));
243201 }
244202245203 #[tokio::test]
246204 async fn test_player_on_track_ended_matches_current() {
247247- let pool = test_pool().await;
248248- let item_id = make_queue_item(&pool).await;
205205+ let item_id = make_item_id();
249206 let (tx, _rx) = mpsc::channel(256);
250207251208 let mut player = Player::new(test_registry(), PathBuf::from("/tmp"), tx);
···256213257214 #[tokio::test]
258215 async fn test_player_on_track_ended_ignores_old_id() {
259259- let pool = test_pool().await;
260260- let item_id = make_queue_item(&pool).await;
216216+ let item_id = make_item_id();
261217 let (tx, _rx) = mpsc::channel(256);
262218263219 let mut player = Player::new(test_registry(), PathBuf::from("/tmp"), tx);
264220 player.start(&test_info(item_id), TrackPublishers::new());
265221266266- // Stale id should be ignored.
267222 assert!(!player.on_track_ended(99999));
268223 }
269224}
+70-166
src/playlist.rs
···11-//! Ordered track list for a room — upcoming queue and played history.
22-//!
33-//! Wraps SQLite queue operations with room-scoped queries, providing
44-//! push, pop-next, upcoming-list, and history-list operations.
11+//! Ordered track list for a room: upcoming queue and played history.
5263use sqlx::{Pool, Sqlite};
7485use crate::types::TrackMeta;
99-1010-fn now_iso() -> String {
1111- chrono::Utc::now()
1212- .format("%Y-%m-%dT%H:%M:%S%.3fZ")
1313- .to_string()
1414-}
156167/// A single queue item, returned from all Playlist operations.
178#[derive(Debug, Clone, sqlx::FromRow, serde::Serialize)]
···6253 .bind(meta.source.to_string())
6354 .bind(None::<String>) // added_by
6455 .bind(position)
6565- .bind(now_iso())
5656+ .bind(crate::util::now_iso())
6657 .execute(&self.pool)
6758 .await?;
6859···7970 Ok(item)
8071 }
81728282- /// Atomically fetch and mark the first unplayed item as playing.
8383- ///
8484- /// Returns the item with `started_playing_at` set, or `None` when the
8585- /// queue is empty. Unlike the old `pop_next`, this does NOT set
8686- /// `played = 1` — that happens later via `mark_finished`.
8787- #[allow(dead_code)]
8888- pub(crate) async fn pop_next(&self) -> Result<Option<QueueItem>, sqlx::Error> {
8989- let mut tx = self.pool.begin().await?;
9090-9191- let item = sqlx::query_as::<_, QueueItem>(
9292- "SELECT id, url, title, duration, thumbnail, source, position, played, played_at, started_playing_at
9393- FROM queue_items WHERE room_id = ? AND played = 0 AND started_playing_at IS NULL
9494- ORDER BY position ASC LIMIT 1",
9595- )
9696- .bind(&self.room_id)
9797- .fetch_optional(&mut *tx)
9898- .await?;
9999-100100- let result = if let Some(item) = item {
101101- let now = now_iso();
102102- sqlx::query("UPDATE queue_items SET started_playing_at = ? WHERE id = ?")
103103- .bind(&now)
104104- .bind(item.id)
105105- .execute(&mut *tx)
106106- .await?;
107107-108108- Some(QueueItem {
109109- started_playing_at: Some(now),
110110- ..item
111111- })
112112- } else {
113113- None
114114- };
115115-116116- tx.commit().await?;
117117- Ok(result)
118118- }
119119-12073 /// List queued items (not playing, not finished) ordered by position ASC.
12174 pub(crate) async fn upcoming(&self) -> Result<Vec<QueueItem>, sqlx::Error> {
12275 let items = sqlx::query_as::<_, QueueItem>(
···155108156109 /// Mark a currently-playing item as finished (moves it from "playing" to "played").
157110 pub(crate) async fn mark_finished(&self, id: i64) -> Result<(), sqlx::Error> {
158158- let now = now_iso();
111111+ let now = crate::util::now_iso();
159112 sqlx::query(
160113 "UPDATE queue_items SET played = 1, played_at = ?, started_playing_at = NULL WHERE id = ?",
161114 )
···204157 .await
205158 .unwrap();
206159 db::run_migrations(&pool).await.unwrap();
207207- // Create a room row so foreign-key constraints are satisfied.
208208- db::room_insert(&pool, "test-room", "test-name", false).await.unwrap();
160160+ db::room_insert(&pool, "test-room", "test-name", false)
161161+ .await
162162+ .unwrap();
209163 Playlist::new(pool, "test-room")
210164 }
211165···241195 }
242196243197 #[tokio::test]
244244- async fn test_pop_next() {
245245- let pl = test_playlist().await;
246246-247247- pl.push(&track("First")).await.unwrap();
248248- pl.push(&track("Second")).await.unwrap();
249249-250250- let popped = pl.pop_next().await.unwrap().expect("should pop first");
251251- assert_eq!(popped.title, "First");
252252- assert!(!popped.played, "pop_next should NOT mark as played");
253253- assert!(
254254- popped.played_at.is_none(),
255255- "pop_next should NOT set played_at"
256256- );
257257- assert!(
258258- popped.started_playing_at.is_some(),
259259- "pop_next should set started_playing_at"
260260- );
261261-262262- // Upcoming should exclude items that have started playing.
263263- let upcoming = pl.upcoming().await.unwrap();
264264- assert_eq!(upcoming.len(), 1);
265265- assert_eq!(upcoming[0].title, "Second");
266266- }
267267-268268- #[tokio::test]
269269- async fn test_history() {
270270- let pl = test_playlist().await;
271271-272272- pl.push(&track("A")).await.unwrap();
273273- pl.push(&track("B")).await.unwrap();
274274- pl.push(&track("C")).await.unwrap();
275275-276276- let a = pl.pop_next().await.unwrap().unwrap();
277277- let b = pl.pop_next().await.unwrap().unwrap();
278278-279279- // Items only appear in history after mark_finished.
280280- let history = pl.history(10).await.unwrap();
281281- assert_eq!(
282282- history.len(),
283283- 0,
284284- "items should not appear in history until mark_finished"
285285- );
286286-287287- pl.mark_finished(a.id).await.unwrap();
288288- pl.mark_finished(b.id).await.unwrap();
289289-290290- let history = pl.history(10).await.unwrap();
291291- assert_eq!(history.len(), 2);
292292- // Most recently played first.
293293- assert_eq!(history[0].title, "B");
294294- assert_eq!(history[1].title, "A");
295295- assert!(history[0].played);
296296- assert!(history[0].played_at.is_some());
297297- }
298298-299299- #[tokio::test]
300300- async fn test_pop_empty() {
301301- let pl = test_playlist().await;
302302- let popped = pl.pop_next().await.unwrap();
303303- assert!(popped.is_none());
304304- }
305305-306306- #[tokio::test]
307198 async fn test_push_positions() {
308199 let pl = test_playlist().await;
309200···319210 #[tokio::test]
320211 async fn test_mark_finished_non_existent_id() {
321212 let pl = test_playlist().await;
322322- // Should not panic or error.
323323- let result = pl.mark_finished(99999).await;
324324- assert!(
325325- result.is_ok(),
326326- "mark_finished on non-existent id should be Ok"
327327- );
213213+ pl.mark_finished(99999).await.unwrap();
328214 }
329215330216 #[tokio::test]
331331- async fn test_pop_next_returns_none_when_all_played() {
217217+ async fn test_upcoming_excludes_playing_items() {
332218 let pl = test_playlist().await;
333333- pl.push(&track("A")).await.unwrap();
334334- pl.pop_next().await.unwrap();
335335- let next = pl.pop_next().await.unwrap();
336336- assert!(
337337- next.is_none(),
338338- "pop_next should return None when queue is empty"
339339- );
219219+ let a = pl.push(&track("A")).await.unwrap();
220220+ pl.push(&track("B")).await.unwrap();
221221+222222+ let now = chrono::Utc::now()
223223+ .format("%Y-%m-%dT%H:%M:%S%.3fZ")
224224+ .to_string();
225225+ pl.mark_started(a.id, &now).await.unwrap();
226226+227227+ let upcoming = pl.upcoming().await.unwrap();
228228+ assert_eq!(upcoming.len(), 1);
229229+ assert_eq!(upcoming[0].title, "B");
340230 }
341231342232 #[tokio::test]
343343- async fn test_upcoming_excludes_playing_items() {
233233+ async fn test_history() {
344234 let pl = test_playlist().await;
345345- pl.push(&track("A")).await.unwrap();
235235+236236+ let a = pl.push(&track("A")).await.unwrap();
346237 pl.push(&track("B")).await.unwrap();
238238+ let c = pl.push(&track("C")).await.unwrap();
347239348348- // Pop A (marks as playing, not finished).
349349- let a = pl.pop_next().await.unwrap().unwrap();
350350- assert_eq!(a.title, "A");
351351- assert!(!a.played, "popped item should not be marked played");
352352- assert!(
353353- a.started_playing_at.is_some(),
354354- "popped item should have started_playing_at"
355355- );
240240+ let now_a = chrono::Utc::now()
241241+ .format("%Y-%m-%dT%H:%M:%S%.3fZ")
242242+ .to_string();
243243+ pl.mark_started(a.id, &now_a).await.unwrap();
244244+ pl.mark_finished(a.id).await.unwrap();
356245357357- // Upcoming should only have B.
358358- let upcoming = pl.upcoming().await.unwrap();
359359- assert_eq!(upcoming.len(), 1, "upcoming should exclude playing item");
360360- assert_eq!(upcoming[0].title, "B");
246246+ tokio::time::sleep(std::time::Duration::from_millis(5)).await;
247247+ let now_c = chrono::Utc::now()
248248+ .format("%Y-%m-%dT%H:%M:%S%.3fZ")
249249+ .to_string();
250250+ pl.mark_started(c.id, &now_c).await.unwrap();
251251+ pl.mark_finished(c.id).await.unwrap();
252252+253253+ let history = pl.history(10).await.unwrap();
254254+ assert_eq!(history.len(), 2);
255255+ assert_eq!(history[0].title, "C");
256256+ assert_eq!(history[1].title, "A");
257257+ assert!(history[0].played);
258258+ assert!(history[0].played_at.is_some());
361259 }
362260363261 #[tokio::test]
364262 async fn test_history_only_after_mark_finished() {
365263 let pl = test_playlist().await;
366366- pl.push(&track("A")).await.unwrap();
367367- let a = pl.pop_next().await.unwrap().unwrap();
264264+ let a = pl.push(&track("A")).await.unwrap();
368265369369- // A is playing — should NOT be in history.
266266+ let now = chrono::Utc::now()
267267+ .format("%Y-%m-%dT%H:%M:%S%.3fZ")
268268+ .to_string();
269269+ pl.mark_started(a.id, &now).await.unwrap();
270270+370271 let history = pl.history(10).await.unwrap();
371371- assert!(
372372- history.is_empty(),
373373- "playing items should not appear in history"
374374- );
272272+ assert!(history.is_empty());
375273376376- // Mark as finished.
377274 pl.mark_finished(a.id).await.unwrap();
378275379379- // Now A should be in history.
380276 let history = pl.history(10).await.unwrap();
381277 assert_eq!(history.len(), 1);
382278 assert_eq!(history[0].title, "A");
383279 assert!(history[0].played);
384384- assert!(
385385- history[0].played_at.is_some(),
386386- "finished items should have played_at"
387387- );
280280+ assert!(history[0].played_at.is_some());
388281 }
389282390283 #[tokio::test]
391284 async fn test_push_after_mark_finished() {
392285 let pl = test_playlist().await;
393393- pl.push(&track("A")).await.unwrap();
394394- let a = pl.pop_next().await.unwrap().unwrap();
286286+ let a = pl.push(&track("A")).await.unwrap();
287287+ let now = chrono::Utc::now()
288288+ .format("%Y-%m-%dT%H:%M:%S%.3fZ")
289289+ .to_string();
290290+ pl.mark_started(a.id, &now).await.unwrap();
395291 pl.mark_finished(a.id).await.unwrap();
396292397397- // Push another track
398293 pl.push(&track("B")).await.unwrap();
399294400295 let upcoming = pl.upcoming().await.unwrap();
···409304 #[tokio::test]
410305 async fn test_history_order_multiple_finished() {
411306 let pl = test_playlist().await;
412412- pl.push(&track("A")).await.unwrap();
413413- pl.push(&track("B")).await.unwrap();
414414- pl.push(&track("C")).await.unwrap();
307307+ let a = pl.push(&track("A")).await.unwrap();
308308+ let b = pl.push(&track("B")).await.unwrap();
309309+ let c = pl.push(&track("C")).await.unwrap();
415310416416- let a = pl.pop_next().await.unwrap().unwrap();
417417- tokio::time::sleep(std::time::Duration::from_millis(5)).await;
311311+ let now_a = chrono::Utc::now()
312312+ .format("%Y-%m-%dT%H:%M:%S%.3fZ")
313313+ .to_string();
314314+ pl.mark_started(a.id, &now_a).await.unwrap();
418315 pl.mark_finished(a.id).await.unwrap();
419316420420- let b = pl.pop_next().await.unwrap().unwrap();
421317 tokio::time::sleep(std::time::Duration::from_millis(5)).await;
318318+ let now_b = chrono::Utc::now()
319319+ .format("%Y-%m-%dT%H:%M:%S%.3fZ")
320320+ .to_string();
321321+ pl.mark_started(b.id, &now_b).await.unwrap();
422322 pl.mark_finished(b.id).await.unwrap();
423323424424- let c = pl.pop_next().await.unwrap().unwrap();
324324+ tokio::time::sleep(std::time::Duration::from_millis(5)).await;
325325+ let now_c = chrono::Utc::now()
326326+ .format("%Y-%m-%dT%H:%M:%S%.3fZ")
327327+ .to_string();
328328+ pl.mark_started(c.id, &now_c).await.unwrap();
425329 pl.mark_finished(c.id).await.unwrap();
426330427331 let history = pl.history(10).await.unwrap();
+283-500
src/room.rs
···44//! SQLite, and holds ephemeral playback state in memory via a pure
55//! [`PlaybackState`] machine.
66//!
77-//! All mutations flow through a per-room mpsc event-loop actor,
88-//! following the impure-pure-impure sandwich:
77+//! All mutations flow through a per-room mpsc event-loop actor:
98//!
1010-//! 1. **Gather** — fetch data (DB, wall clock)
1111-//! 2. **Transition** — `PlaybackState::transition()` (pure)
1212-//! 3. **Commit** — execute returned [`Effect`]s
99+//! 1. Gather: fetch data (DB, wall clock)
1010+//! 2. Transition: `PlaybackState::transition()` (pure)
1111+//! 3. Commit: execute returned [`Effect`]s
13121413use std::collections::HashMap;
1514use std::path::PathBuf;
···3231use crate::state::{self, Effect, Event, PlaybackState};
3332use crate::transport::TrackPublishers;
3433use crate::types::{
3535- ChatMessage, HistoryEntry, QueueSummary, RoomCommand, RoomHandle, RoomId, TrackMeta,
3636- TrackState,
3434+ ChatMessage, HistoryEntry, QueueSummary, RoomCommand, RoomHandle, RoomId, TrackMeta, TrackState,
3735};
38363937/// Distributed unique ID generator (Discord-style snowflake).
4038///
4139/// Bit layout:
4242-/// 1 bit — unused (always 0)
4343-/// 41 bits — timestamp in milliseconds since custom epoch
4444-/// 10 bits — node ID (0–1023)
4545-/// 12 bits — sequence number (0–4095)
4040+/// 1 bit unused (always 0)
4141+/// 41 bits timestamp in milliseconds since custom epoch
4242+/// 10 bits node ID (0-1023)
4343+/// 12 bits sequence number (0-4095)
4644///
4745/// Thread-safety is provided by a single `AtomicU64` that packs
4846/// `(timestamp_ms << 12) | sequence`, allowing lock-free CAS updates.
···5149 state: AtomicU64, // (timestamp_ms << 12) | sequence (12 bits)
5250}
53515454-const CUSTOM_EPOCH_MS: u64 = 173_568_960_0000; // 2025-01-01T00:00:00Z
5252+const CUSTOM_EPOCH_MS: u64 = 1_735_689_600_000; // 2025-01-01T00:00:00Z
55535654impl SnowflakeIdGen {
5755 pub(crate) fn new(node_id: u16) -> Self {
···6765 /// Thread-safe. Under extreme load (≥4096 IDs in the same
6866 /// millisecond) the generator yields to the tokio runtime until
6967 /// the next millisecond tick.
6868+ #[must_use]
7069 pub(crate) async fn next_id(&self) -> u64 {
7170 loop {
7271 let now = SystemTime::now()
···8988 {
9089 return (ts << 22) | ((self.node_id as u64) << 12);
9190 }
9292- // CAS lost — another thread already advanced or incremented.
9393- // Retry with fresh state.
9491 continue;
9592 }
9693···105102 {
106103 return (last_ts << 22) | ((self.node_id as u64) << 12) | seq as u64;
107104 }
108108- // CAS lost — another thread claimed this seq first; retry.
109105 continue;
110106 }
111107···124120 pool: Pool<Sqlite>,
125121 source_registry: Arc<SourceRegistry>,
126122 cache_dir: PathBuf,
127127- /// Clone of the command sender — used by spawned extraction tasks to
123123+ /// Clone of the command sender, used by spawned extraction tasks to
128124 /// deliver MetadataReady / MetadataFailed back to the actor.
129125 cmd_tx: mpsc::Sender<RoomCommand>,
130126···168164 }
169165170166 /// Create a new room with a snowflake ID, persist to SQLite.
171171- pub(crate) async fn create(&self) -> RoomRef {
167167+ pub(crate) async fn create(&self) -> RoomId {
172168 let id = RoomId(self.id_gen.next_id().await.to_string());
173169 let room_name = generate_room_name(&id.0);
174170 let now = Utc::now();
···178174 let (tx, rx) = mpsc::channel(256);
179175 let (last_active_tx, last_active_rx) = watch::channel(now);
180176181181- let player = Player::new(
182182- self.sources.clone(),
183183- self.cache_dir.clone(),
184184- tx.clone(),
185185- );
177177+ let player = Player::new(self.sources.clone(), self.cache_dir.clone(), tx.clone());
186178 let actor = RoomActor {
187179 rx,
188180 room_id: id.clone(),
···209201 },
210202 );
211203212212- RoomRef {
213213- id,
214214- _registry: self.clone(),
215215- }
204204+ id
216205 }
217206218207 /// Get the room handle if the room is active.
···221210 }
222211223212 /// Check whether a room exists in SQLite.
213213+ #[must_use]
224214 pub(crate) async fn exists(&self, id: &RoomId) -> bool {
225215 db::room_exists(&self.pool, &id.0).await.unwrap_or(false)
226216 }
···236226 duration: i.duration,
237227 thumbnail: i.thumbnail,
238228 url: i.url,
239239- source: i
240240- .source
241241- .parse()
242242- .unwrap_or_else(|s: String| {
243243- tracing::warn!(source = %s, "unknown source kind in queue_items");
244244- crate::types::SourceKind::Direct
245245- }),
229229+ source: i.source.parse().unwrap_or_else(|s: String| {
230230+ tracing::warn!(source = %s, "unknown source kind in queue_items");
231231+ crate::types::SourceKind::Direct
232232+ }),
246233 })
247234 .collect()
248248- }
249249-250250- /// Push a track onto a room's queue (persisted to SQLite via Playlist).
251251- #[allow(dead_code)] // only used by tests
252252- pub(crate) async fn push(&self, id: &RoomId, meta: TrackMeta) {
253253- if let Some(handle) = self.handle(id).await {
254254- let _ = handle.cmd_tx.send(RoomCommand::QueueTrack(meta)).await;
255255- }
256235 }
257236258237 /// Push a raw URL onto a room's queue. Metadata extraction (yt-dlp) runs
259259- /// asynchronously in a spawned task — the caller returns immediately.
238238+ /// asynchronously in a spawned task. The caller returns immediately.
260239 pub(crate) async fn push_url(&self, id: &RoomId, url: String) {
261261- let t0 = std::time::Instant::now();
262240 let handle = match self.handle(id).await {
263241 Some(h) => h,
264242 None => return,
265243 };
266266- let t1 = t0.elapsed();
267244 let _ = handle.cmd_tx.send(RoomCommand::QueueUrl(url)).await;
268268- let t2 = t0.elapsed();
269269- tracing::info!(
270270- room = %id,
271271- "push_url timing: handle={:.0?} send={:.0?} total={:.0?}",
272272- t1, t2 - t1, t2
273273- );
274245 }
275246276247 /// Remove a room (from SQLite and in-memory state).
···278249 pub(crate) async fn remove(&self, id: &RoomId) {
279250 if let Some(handle) = self.handle(id).await {
280251 let _ = tokio::time::timeout(
281281- Duration::from_secs(1),
252252+ Duration::from_secs(1), // generous 1s window for graceful shutdown
282253 handle.cmd_tx.send(RoomCommand::Shutdown),
283254 )
284255 .await;
···397368 for id in &idle_rooms {
398369 if let Some(handle) = self.handle(id).await {
399370 if tokio::time::timeout(
400400- Duration::from_secs(1),
371371+ Duration::from_secs(1), // generous 1s window for graceful shutdown
401372 handle.cmd_tx.send(RoomCommand::Shutdown),
402373 )
403374 .await
···416387 }
417388}
418389419419-/// An owned handle to a room, returned by [`Registry::create`].
420420-pub(crate) struct RoomRef {
421421- pub id: RoomId,
422422- _registry: Registry,
423423-}
424424-425390impl RoomActor {
426391 /// Run the event loop. Returns when rx is closed or Shutdown received.
427392 async fn run(mut self) {
···431396 break;
432397 }
433398 }
434434- // On shutdown, abort any active pipeline.
435399 self.player.abort();
436400 tracing::info!(room = %self.room_id, "room actor shut down");
437401 }
···439403 /// Process a single command.
440404 ///
441405 /// Returns `true` if the actor should shut down.
406406+ #[must_use]
442407 async fn process(&mut self, cmd: RoomCommand) -> bool {
443408 match cmd {
444444- RoomCommand::QueueTrack(meta) => {
445445- tracing::debug!(room = %self.room_id, title = %meta.title, "track queued");
446446-447447- // ── Gather ──────────────────────────────────────────
448448- let playlist = Playlist::new(self.pool.clone(), &self.room_id.0);
449449- let item = match playlist.push(&meta).await {
450450- Ok(item) => item,
451451- Err(e) => {
452452- tracing::warn!("failed to persist queue item: {e}");
453453- return false;
454454- }
455455- };
456456- let now = Utc::now().timestamp_millis();
457457-458458- // ── Transition (pure) ───────────────────────────────
459459- let effects = self.state.transition(
460460- &Event::TrackQueued(state::QueuedTrack {
461461- id: item.id,
462462- title: item.title,
463463- url: item.url,
464464- duration: item.duration,
465465- thumbnail: item.thumbnail,
466466- pending: false,
467467- }),
468468- now,
469469- );
470470-471471- // ── Commit ──────────────────────────────────────────
472472- self.execute_effects(effects).await;
473473- }
474474-475409 RoomCommand::QueueUrl(url) => {
476476- let t0 = std::time::Instant::now();
477410 tracing::debug!(room = %self.room_id, %url, "track url queued");
478411479479- // ── Gather ──────────────────────────────────────────
480412 let playlist = Playlist::new(self.pool.clone(), &self.room_id.0);
481413 let placeholder = TrackMeta {
482482- title: "Loading…".into(),
414414+ title: "Loading...".into(),
483415 duration: "--:--".into(),
484416 thumbnail: None,
485417 url: url.clone(),
···492424 return false;
493425 }
494426 };
495495- let t1 = t0.elapsed();
496427 let now = Utc::now().timestamp_millis();
497428498498- // ── Transition (pure) ───────────────────────────────
499429 let effects = self.state.transition(
500430 &Event::TrackQueued(state::QueuedTrack {
501431 id: item.id,
···507437 }),
508438 now,
509439 );
510510- let t2 = t0.elapsed();
511440512512- // ── Commit ──────────────────────────────────────────
513441 self.execute_effects(effects).await;
514514- let t3 = t0.elapsed();
515442516516- // ── Spawn async extraction (non-blocking) ───────────
517443 let item_id = item.id;
518444 let cmd_tx = self.cmd_tx.clone();
519445 let sources = self.source_registry.clone();
···535461 }
536462 }
537463 });
538538-539539- tracing::info!(
540540- room = %self.room_id,
541541- "QueueUrl actor timing: db_push={:.0?} transition={:.0?} effects={:.0?} total={:.0?}",
542542- t1, t2 - t1, t3 - t2, t3
543543- );
544464 }
545465546466 RoomCommand::MetadataReady { item_id, meta } => {
547467 tracing::debug!(room = %self.room_id, item_id, title = %meta.title, "metadata ready");
548468549549- // ── Gather ──────────────────────────────────────────
550469 let now = Utc::now().timestamp_millis();
551470552552- // ── DB update (synchronous commit, actor-safe) ──────
553471 let playlist = Playlist::new(self.pool.clone(), &self.room_id.0);
554472 if let Err(e) = playlist.update_metadata(item_id, &meta).await {
555473 tracing::warn!(item_id, error = %e, "failed to persist metadata update");
556474 }
557475558558- // ── Transition (pure) ───────────────────────────────
559476 let effects = self.state.transition(
560477 &Event::MetadataUpdated {
561478 item_id,
···566483 now,
567484 );
568485569569- // ── Commit ──────────────────────────────────────────
570486 self.execute_effects(effects).await;
571487 }
572488573489 RoomCommand::MetadataFailed { item_id, error } => {
574490 tracing::debug!(room = %self.room_id, item_id, %error, "metadata extraction failed");
575491576576- // ── Transition (pure) — set error state in title ────
577492 let now = Utc::now().timestamp_millis();
578493 let effects = self.state.transition(
579494 &Event::MetadataUpdated {
580495 item_id,
581581- title: format!("Failed to load — {error}"),
496496+ title: format!("Failed to load: {error}"),
582497 duration: "--:--".into(),
583498 thumbnail: None,
584499 },
···590505 RoomCommand::Skip => {
591506 tracing::info!(room = %self.room_id, "skip requested");
592507593593- // ── Gather ──────────────────────────────────────────
594508 let now = Utc::now().timestamp_millis();
595509596596- // ── Transition (pure) ───────────────────────────────
597510 let effects = self.state.transition(&Event::Skip, now);
598511599599- // ── Commit ──────────────────────────────────────────
600512 self.execute_effects(effects).await;
601513 }
602514603515 RoomCommand::TrackEnded { item_id } => {
604516 tracing::debug!(room = %self.room_id, item_id, "track ended");
605517606606- // Gate on player first — stale events are silently dropped.
607518 if !self.player.on_track_ended(item_id) {
608519 return false;
609520 }
610521611611- // ── Gather ──────────────────────────────────────────
612522 let now = Utc::now().timestamp_millis();
613523614614- // ── Transition (pure) ───────────────────────────────
615524 let effects = self.state.transition(&Event::TrackEnded { item_id }, now);
616525617617- // ── Commit ──────────────────────────────────────────
618526 self.execute_effects(effects).await;
619527 }
620528···677585 }
678586679587 Effect::PersistStarted(id) => {
680680- let t0 = std::time::Instant::now();
681588 let now = Utc::now().format("%Y-%m-%dT%H:%M:%S%.3fZ").to_string();
682589 let playlist = Playlist::new(self.pool.clone(), &self.room_id.0);
683590 let _ = playlist.mark_started(id, &now).await;
684684- tracing::debug!(room = %self.room_id, id, "PersistStarted: {:.0?}", t0.elapsed());
685591 }
686592687593 Effect::StartPipeline(track) => {
···712618713619 /// Publish the current state snapshot to all connected clients.
714620 ///
715715- /// Reads from in-memory [`PlaybackState`] — no SQLite queries.
621621+ /// Reads from in-memory [`PlaybackState`], no SQLite queries.
716622 async fn publish_state_snapshot(&self) {
717623 let current = self.state.active.as_ref().map(|t| TrackState {
718624 id: t.id,
719625 title: t.title.clone(),
720626 url: t.url.clone(),
721627 duration: t.duration.clone(),
628628+ thumbnail: t.thumbnail.clone(),
722629 started_at: t.started_at_wall,
723630 });
724631···774681 use super::*;
775682 use sqlx::sqlite::SqlitePoolOptions;
776683777777- async fn test_pool() -> Pool<Sqlite> {
684684+ async fn poll_until<F, Fut>(check: F, timeout: Duration, label: &str)
685685+ where
686686+ F: Fn() -> Fut,
687687+ Fut: std::future::Future<Output = bool>,
688688+ {
689689+ let start = std::time::Instant::now();
690690+ loop {
691691+ if check().await {
692692+ return;
693693+ }
694694+ if start.elapsed() > timeout {
695695+ panic!("timed out: {label}");
696696+ }
697697+ tokio::time::sleep(Duration::from_millis(10)).await;
698698+ }
699699+ }
700700+701701+ async fn test_registry_and_room() -> (Registry, RoomId) {
778702 let pool = SqlitePoolOptions::new()
779703 .max_connections(1)
780704 .connect(":memory:")
781705 .await
782706 .unwrap();
783707 db::run_migrations(&pool).await.unwrap();
784784- pool
708708+ let reg = Registry::new(pool, PathBuf::from("/tmp/moqbox"), test_registry());
709709+ let room_id = reg.create().await;
710710+ (reg, room_id)
785711 }
786712787787- /// A mock source that handles all URLs.
788713 struct PanaceaSource;
789714790715 #[async_trait::async_trait]
···794719 url: &str,
795720 _cache_dir: &std::path::Path,
796721 ) -> Result<TrackMeta, crate::source::SourceError> {
722722+ let title = url.rsplit('/').next().unwrap_or("test");
797723 Ok(TrackMeta {
798798- title: "test".into(),
724724+ title: title.into(),
799725 duration: "3:45".into(),
800726 thumbnail: None,
801727 url: url.into(),
···844770845771 #[tokio::test]
846772 async fn test_create_and_exists() {
847847- let pool = test_pool().await;
848848- let reg = Registry::new(pool, PathBuf::from("/tmp/moqbox"), test_registry());
849849- let room = reg.create().await;
850850- assert!(reg.exists(&room.id).await);
773773+ let (reg, room_id) = test_registry_and_room().await;
774774+ assert!(reg.exists(&room_id).await);
851775 }
852776853777 #[tokio::test]
854778 async fn test_push_and_queue() {
855855- let pool = test_pool().await;
856856- let reg = Registry::new(pool, PathBuf::from("/tmp/moqbox"), test_registry());
857857- let room = reg.create().await;
858858-859859- let meta_a = TrackMeta {
860860- title: "Track A".into(),
861861- duration: "3:45".into(),
862862- thumbnail: None,
863863- url: "https://example.com/a".into(),
864864- source: crate::types::SourceKind::Direct,
865865- };
866866- let meta_b = TrackMeta {
867867- title: "Track B".into(),
868868- duration: "4:20".into(),
869869- thumbnail: None,
870870- url: "https://example.com/b".into(),
871871- source: crate::types::SourceKind::Direct,
872872- };
779779+ let (reg, room_id) = test_registry_and_room().await;
873780874874- // Push first track — it gets popped and starts playing immediately.
875875- reg.push(&room.id, meta_a.clone()).await;
876876- // Push second track — stays in the upcoming queue.
877877- reg.push(&room.id, meta_b.clone()).await;
781781+ reg.push_url(&room_id, "https://example.com/a".into()).await;
782782+ reg.push_url(&room_id, "https://example.com/b".into()).await;
878783879879- // Give the actor time to process both commands.
880880- tokio::time::sleep(Duration::from_millis(100)).await;
881881-882882- let queue = reg.queue(&room.id).await;
883883- assert_eq!(queue.len(), 1, "only unplayed track should appear in queue");
884884- assert_eq!(queue[0].title, "Track B");
784784+ poll_until(
785785+ || async {
786786+ let q = reg.queue(&room_id).await;
787787+ q.len() == 1 && q[0].title == "b"
788788+ },
789789+ Duration::from_secs(2),
790790+ "queue should have 1 item with title 'b'",
791791+ )
792792+ .await;
885793 }
886794887795 #[tokio::test]
888796 async fn test_remove() {
889889- let pool = test_pool().await;
890890- let reg = Registry::new(pool, PathBuf::from("/tmp/moqbox"), test_registry());
891891- let room = reg.create().await;
797797+ let (reg, room_id) = test_registry_and_room().await;
892798893893- reg.remove(&room.id).await;
894894- assert!(!reg.exists(&room.id).await);
799799+ reg.remove(&room_id).await;
800800+ assert!(!reg.exists(&room_id).await);
895801 }
896802897803 #[tokio::test]
898804 async fn test_send_chat() {
899899- let pool = test_pool().await;
900900- let reg = Registry::new(pool, PathBuf::from("/tmp/moqbox"), test_registry());
901901- let room = reg.create().await;
805805+ let (reg, room_id) = test_registry_and_room().await;
902806903807 let msg = reg
904904- .send_chat(&room.id, "alice", "hello world", "message")
808808+ .send_chat(&room_id, "alice", "hello world", "message")
905809 .await
906810 .unwrap();
907811 assert_eq!(msg.user_name, "alice");
···912816913817 #[tokio::test]
914818 async fn test_send_chat_persistence() {
915915- let pool = test_pool().await;
916916- let reg = Registry::new(pool, PathBuf::from("/tmp/moqbox"), test_registry());
917917- let room = reg.create().await;
819819+ let (reg, room_id) = test_registry_and_room().await;
918820919821 let _ = reg
920920- .send_chat(&room.id, "alice", "msg1", "message")
822822+ .send_chat(&room_id, "alice", "msg1", "message")
921823 .await
922824 .unwrap();
923825 let _ = reg
924924- .send_chat(&room.id, "bob", "msg2", "message")
826826+ .send_chat(&room_id, "bob", "msg2", "message")
925827 .await
926828 .unwrap();
927829928928- tokio::time::sleep(Duration::from_millis(100)).await;
830830+ poll_until(
831831+ || async { reg.recent_chat(&room_id, 10).await.unwrap().len() == 2 },
832832+ Duration::from_secs(2),
833833+ "recent_chat should have 2 messages",
834834+ )
835835+ .await;
929836930930- let recent = reg.recent_chat(&room.id, 10).await.unwrap();
931931- assert_eq!(recent.len(), 2);
837837+ let recent = reg.recent_chat(&room_id, 10).await.unwrap();
932838 assert_eq!(recent[0].user_name, "bob");
933839 assert_eq!(recent[0].content, "msg2");
934840 assert_eq!(recent[1].user_name, "alice");
···937843938844 #[tokio::test]
939845 async fn test_send_chat_empty_content() {
940940- let pool = test_pool().await;
941941- let reg = Registry::new(pool, PathBuf::from("/tmp/moqbox"), test_registry());
942942- let room = reg.create().await;
846846+ let (reg, room_id) = test_registry_and_room().await;
943847944944- let result = reg.send_chat(&room.id, "alice", "", "message").await;
945945- assert!(result.is_err());
848848+ let result = reg.send_chat(&room_id, "alice", "", "message").await;
849849+ result.unwrap_err();
946850 }
947851948852 #[tokio::test]
949853 async fn test_send_chat_long_content() {
950950- let pool = test_pool().await;
951951- let reg = Registry::new(pool, PathBuf::from("/tmp/moqbox"), test_registry());
952952- let room = reg.create().await;
854854+ let (reg, room_id) = test_registry_and_room().await;
953855954856 let long = "a".repeat(2001);
955955- let result = reg.send_chat(&room.id, "alice", &long, "message").await;
956956- assert!(result.is_err());
857857+ let result = reg.send_chat(&room_id, "alice", &long, "message").await;
858858+ result.unwrap_err();
957859 }
958860959861 #[tokio::test]
960862 async fn test_send_chat_long_username() {
961961- let pool = test_pool().await;
962962- let reg = Registry::new(pool, PathBuf::from("/tmp/moqbox"), test_registry());
963963- let room = reg.create().await;
863863+ let (reg, room_id) = test_registry_and_room().await;
964864965865 let long = "a".repeat(33);
966966- let result = reg.send_chat(&room.id, &long, "hello", "message").await;
967967- assert!(result.is_err());
866866+ let result = reg.send_chat(&room_id, &long, "hello", "message").await;
867867+ result.unwrap_err();
968868 }
969869970870 #[tokio::test]
971871 async fn test_register_client() {
972972- let pool = test_pool().await;
973973- let reg = Registry::new(pool, PathBuf::from("/tmp/moqbox"), test_registry());
974974- let room = reg.create().await;
872872+ let (reg, room_id) = test_registry_and_room().await;
975873976976- let id1 = reg.register_client(&room.id).await;
874874+ let id1 = reg.register_client(&room_id).await;
977875 assert_eq!(id1, Some(1));
978876979979- let id2 = reg.register_client(&room.id).await;
877877+ let id2 = reg.register_client(&room_id).await;
980878 assert_eq!(id2, Some(2));
981879 }
982880983881 #[tokio::test]
984882 async fn test_register_client_nonexistent_room() {
985985- let pool = test_pool().await;
986986- let reg = Registry::new(pool, PathBuf::from("/tmp/moqbox"), test_registry());
883883+ let (reg, _) = test_registry_and_room().await;
987884 let fake_id = RoomId("nonexistent".into());
988885 assert_eq!(reg.register_client(&fake_id).await, None);
989886 }
990887991888 #[tokio::test]
992889 async fn test_unregister_client() {
993993- let pool = test_pool().await;
994994- let reg = Registry::new(pool, PathBuf::from("/tmp/moqbox"), test_registry());
995995- let room = reg.create().await;
890890+ let (reg, room_id) = test_registry_and_room().await;
996891997997- reg.register_client(&room.id).await;
998998- reg.unregister_client(&room.id).await;
892892+ reg.register_client(&room_id).await;
893893+ reg.unregister_client(&room_id).await;
99989410001000- let id = reg.register_client(&room.id).await;
895895+ let id = reg.register_client(&room_id).await;
1001896 assert_eq!(id, Some(2));
1002897 }
10038981004899 #[tokio::test]
1005900 async fn test_skip_when_idle_does_not_panic() {
10061006- let pool = test_pool().await;
10071007- let reg = Registry::new(pool, PathBuf::from("/tmp/moqbox"), test_registry());
10081008- let room = reg.create().await;
10091009- reg.skip(&room.id).await;
901901+ let (reg, room_id) = test_registry_and_room().await;
902902+ reg.skip(&room_id).await;
1010903 }
10119041012905 #[tokio::test]
1013906 async fn test_double_skip_does_not_deadlock() {
10141014- let pool = test_pool().await;
10151015- let reg = Registry::new(pool, PathBuf::from("/tmp/moqbox"), test_registry());
10161016- let room = reg.create().await;
907907+ let (reg, room_id) = test_registry_and_room().await;
101790810181018- let meta = TrackMeta {
10191019- title: "A".into(),
10201020- duration: "3:45".into(),
10211021- thumbnail: None,
10221022- url: "https://example.com/a".into(),
10231023- source: crate::types::SourceKind::Direct,
10241024- };
10251025- reg.push(&room.id, meta).await;
10261026- tokio::time::sleep(Duration::from_millis(100)).await;
909909+ reg.push_url(&room_id, "https://example.com/a".into()).await;
910910+ poll_until(
911911+ || async { reg.queue(&room_id).await.is_empty() },
912912+ Duration::from_secs(2),
913913+ "queue should be empty after push starts playing",
914914+ )
915915+ .await;
10279161028917 let r1 = reg.clone();
10291029- let rid1 = room.id.clone();
918918+ let rid1 = room_id.clone();
1030919 let h1 = tokio::spawn(async move { r1.skip(&rid1).await });
1031920 let r2 = reg.clone();
10321032- let rid2 = room.id.clone();
921921+ let rid2 = room_id.clone();
1033922 let h2 = tokio::spawn(async move { r2.skip(&rid2).await });
1034923 let _ = tokio::join!(h1, h2);
1035924 }
10369251037926 #[tokio::test]
1038927 async fn test_send_chat_returns_id_zero() {
10391039- let pool = test_pool().await;
10401040- let reg = Registry::new(pool, PathBuf::from("/tmp/moqbox"), test_registry());
10411041- let room = reg.create().await;
928928+ let (reg, room_id) = test_registry_and_room().await;
1042929 let msg = reg
10431043- .send_chat(&room.id, "alice", "hello", "message")
930930+ .send_chat(&room_id, "alice", "hello", "message")
1044931 .await
1045932 .unwrap();
10461046- assert_eq!(msg.id, 0, "send_chat should return id 0 in actor mode");
933933+ assert_eq!(msg.id, 0);
1047934 }
10489351049936 #[tokio::test]
1050937 async fn test_sweep_idle_removes_inactive_rooms() {
10511051- let pool = test_pool().await;
10521052- let reg = Registry::new(pool, PathBuf::from("/tmp/moqbox"), test_registry());
10531053- let room = reg.create().await;
938938+ let (reg, room_id) = test_registry_and_room().await;
10549391055940 tokio::time::sleep(Duration::from_millis(1)).await;
10569411057942 let swept = reg.sweep_idle(Duration::from_secs(0)).await;
10581058- assert!(swept.contains(&room.id), "room should be swept immediately");
10591059- assert!(
10601060- !reg.exists(&room.id).await,
10611061- "room should no longer exist after sweep"
10621062- );
943943+ assert!(swept.contains(&room_id));
944944+ assert!(!reg.exists(&room_id).await);
1063945 }
10649461065947 #[tokio::test]
1066948 async fn test_sweep_idle_preserves_active_rooms() {
10671067- let pool = test_pool().await;
10681068- let reg = Registry::new(pool, PathBuf::from("/tmp/moqbox"), test_registry());
10691069- let room = reg.create().await;
949949+ let (reg, room_id) = test_registry_and_room().await;
107095010711071- reg.register_client(&room.id).await;
10721072- tokio::time::sleep(Duration::from_millis(50)).await;
951951+ reg.register_client(&room_id).await;
1073952 let swept = reg.sweep_idle(Duration::from_millis(100)).await;
10741074- assert!(!swept.contains(&room.id), "active room should not be swept");
953953+ assert!(!swept.contains(&room_id));
1075954 }
10769551077956 #[tokio::test]
1078957 async fn test_push_starts_playback_when_idle() {
10791079- let pool = test_pool().await;
10801080- let reg = Registry::new(pool, PathBuf::from("/tmp/moqbox"), test_registry());
10811081- let room = reg.create().await;
958958+ let (reg, room_id) = test_registry_and_room().await;
108295910831083- let meta = TrackMeta {
10841084- title: "Test".into(),
10851085- duration: "3:45".into(),
10861086- thumbnail: None,
10871087- url: "https://example.com/t".into(),
10881088- source: crate::types::SourceKind::Direct,
10891089- };
10901090- reg.push(&room.id, meta).await;
10911091-10921092- tokio::time::sleep(Duration::from_millis(100)).await;
10931093-10941094- let queue = reg.queue(&room.id).await;
10951095- assert!(
10961096- queue.is_empty(),
10971097- "track should be popped when playback starts"
10981098- );
960960+ reg.push_url(&room_id, "https://example.com/t".into()).await;
961961+ poll_until(
962962+ || async { reg.queue(&room_id).await.is_empty() },
963963+ Duration::from_secs(2),
964964+ "queue should be empty (item started playing)",
965965+ )
966966+ .await;
1099967 }
11009681101969 #[tokio::test]
1102970 async fn test_push_appends_when_already_playing() {
11031103- let pool = test_pool().await;
11041104- let reg = Registry::new(pool, PathBuf::from("/tmp/moqbox"), test_registry());
11051105- let room = reg.create().await;
971971+ let (reg, room_id) = test_registry_and_room().await;
110697211071107- reg.push(
11081108- &room.id,
11091109- TrackMeta {
11101110- title: "A".into(),
11111111- duration: "3:45".into(),
11121112- thumbnail: None,
11131113- url: "https://example.com/a".into(),
11141114- source: crate::types::SourceKind::Direct,
11151115- },
973973+ reg.push_url(&room_id, "https://example.com/a".into()).await;
974974+ poll_until(
975975+ || async { reg.queue(&room_id).await.is_empty() },
976976+ Duration::from_secs(2),
977977+ "first push should have started playing",
1116978 )
1117979 .await;
11181118- tokio::time::sleep(Duration::from_millis(100)).await;
111998011201120- reg.push(
11211121- &room.id,
11221122- TrackMeta {
11231123- title: "B".into(),
11241124- duration: "3:45".into(),
11251125- thumbnail: None,
11261126- url: "https://example.com/b".into(),
11271127- source: crate::types::SourceKind::Direct,
11281128- },
981981+ reg.push_url(&room_id, "https://example.com/b".into()).await;
982982+ poll_until(
983983+ || async { reg.queue(&room_id).await.len() == 1 },
984984+ Duration::from_secs(2),
985985+ "second push should be queued",
1129986 )
1130987 .await;
11311131- tokio::time::sleep(Duration::from_millis(100)).await;
113298811331133- let queue = reg.queue(&room.id).await;
11341134- assert_eq!(queue.len(), 1, "second track should be queued");
11351135- assert_eq!(queue[0].title, "B");
989989+ let queue = reg.queue(&room_id).await;
990990+ assert_eq!(queue[0].title, "b");
113699111371137- // History should be empty (first track is still "playing", not finished).
11381138- let playlist = crate::playlist::Playlist::new(reg.pool.clone(), &room.id.0);
992992+ let playlist = crate::playlist::Playlist::new(reg.pool.clone(), &room_id.0);
1139993 let history = playlist.history(10).await.unwrap();
11401140- assert!(
11411141- history.is_empty(),
11421142- "no history yet — first track is still playing"
11431143- );
994994+ assert!(history.is_empty());
1144995 }
11459961146997 #[tokio::test]
1147998 async fn test_skip_when_queue_empty_after_track_ends() {
11481148- let pool = test_pool().await;
11491149- let reg = Registry::new(pool, PathBuf::from("/tmp/moqbox"), test_registry());
11501150- let room = reg.create().await;
999999+ let (reg, room_id) = test_registry_and_room().await;
1151100011521152- reg.push(
11531153- &room.id,
11541154- TrackMeta {
11551155- title: "A".into(),
11561156- duration: "3:45".into(),
11571157- thumbnail: None,
11581158- url: "https://example.com/a".into(),
11591159- source: crate::types::SourceKind::Direct,
10011001+ reg.push_url(&room_id, "https://example.com/a".into()).await;
10021002+ poll_until(
10031003+ || async { reg.queue(&room_id).await.is_empty() },
10041004+ Duration::from_secs(2),
10051005+ "first push should have started playing",
10061006+ )
10071007+ .await;
10081008+10091009+ reg.skip(&room_id).await;
10101010+ poll_until(
10111011+ || async {
10121012+ let playlist = crate::playlist::Playlist::new(reg.pool.clone(), &room_id.0);
10131013+ playlist.history(10).await.unwrap().len() == 1
11601014 },
10151015+ Duration::from_secs(2),
10161016+ "skip should move track to history",
11611017 )
11621018 .await;
11631163- tokio::time::sleep(Duration::from_millis(100)).await;
1164101911651165- reg.skip(&room.id).await;
11661166- tokio::time::sleep(Duration::from_millis(100)).await;
11671167-11681168- let queue = reg.queue(&room.id).await;
11691169- assert!(
11701170- queue.is_empty(),
11711171- "queue should be empty after skip with no next track"
11721172- );
10201020+ let queue = reg.queue(&room_id).await;
10211021+ assert!(queue.is_empty());
11731022 }
1174102311751024 #[tokio::test]
11761025 async fn test_double_skip_does_not_corrupt_state() {
11771177- let pool = test_pool().await;
11781178- let reg = Registry::new(pool, PathBuf::from("/tmp/moqbox"), test_registry());
11791179- let room = reg.create().await;
10261026+ let (reg, room_id) = test_registry_and_room().await;
1180102711811181- reg.push(
11821182- &room.id,
11831183- TrackMeta {
11841184- title: "A".into(),
11851185- duration: "3:45".into(),
11861186- thumbnail: None,
11871187- url: "https://example.com/a".into(),
11881188- source: crate::types::SourceKind::Direct,
11891189- },
10281028+ reg.push_url(&room_id, "https://example.com/a".into()).await;
10291029+ poll_until(
10301030+ || async { reg.queue(&room_id).await.is_empty() },
10311031+ Duration::from_secs(2),
10321032+ "first push started playing",
11901033 )
11911034 .await;
11921192- tokio::time::sleep(Duration::from_millis(100)).await;
1193103511941194- reg.push(
11951195- &room.id,
11961196- TrackMeta {
11971197- title: "B".into(),
11981198- duration: "3:45".into(),
11991199- thumbnail: None,
12001200- url: "https://example.com/b".into(),
12011201- source: crate::types::SourceKind::Direct,
10361036+ reg.push_url(&room_id, "https://example.com/b".into()).await;
10371037+ poll_until(
10381038+ || async { reg.queue(&room_id).await.len() == 1 },
10391039+ Duration::from_secs(2),
10401040+ "second push queued",
10411041+ )
10421042+ .await;
10431043+10441044+ reg.skip(&room_id).await;
10451045+ reg.skip(&room_id).await;
10461046+ poll_until(
10471047+ || async {
10481048+ let playlist = crate::playlist::Playlist::new(reg.pool.clone(), &room_id.0);
10491049+ let h = playlist.history(10).await.unwrap();
10501050+ !h.is_empty() && h[0].title == "b"
12021051 },
10521052+ Duration::from_secs(2),
10531053+ "history should have 'b' as most recent",
12031054 )
12041055 .await;
12051205- tokio::time::sleep(Duration::from_millis(50)).await;
1206105612071207- reg.skip(&room.id).await;
12081208- reg.skip(&room.id).await;
12091209- tokio::time::sleep(Duration::from_millis(100)).await;
12101210-12111211- let playlist = crate::playlist::Playlist::new(reg.pool.clone(), &room.id.0);
10571057+ let playlist = crate::playlist::Playlist::new(reg.pool.clone(), &room_id.0);
12121058 let history = playlist.history(10).await.unwrap();
12131213- assert!(
12141214- !history.is_empty(),
12151215- "at least one track should be in history after double skip"
12161216- );
12171217- assert_eq!(history[0].title, "B", "last skipped track should be B");
10591059+ assert!(!history.is_empty());
10601060+ assert_eq!(history[0].title, "b");
12181061 }
1219106212201063 #[tokio::test]
12211221- async fn test_register_client_overflow() {
12221222- let pool = test_pool().await;
12231223- let reg = Registry::new(pool, PathBuf::from("/tmp/moqbox"), test_registry());
12241224- let room = reg.create().await;
10641064+ async fn test_register_client_runs_three_times() {
10651065+ let (reg, room_id) = test_registry_and_room().await;
1225106612261226- for _ in 0..100 {
12271227- let id = reg.register_client(&room.id).await;
12281228- assert!(id.is_some(), "register_client should return Some(id)");
10671067+ for _ in 0..3 {
10681068+ let id = reg.register_client(&room_id).await;
10691069+ assert!(id.is_some());
12291070 }
12301230- tokio::time::sleep(Duration::from_millis(50)).await;
1231107112321232- let exists = reg.exists(&room.id).await;
12331233- assert!(exists, "room should still exist");
10721072+ assert!(reg.exists(&room_id).await);
12341073 }
1235107412361075 #[tokio::test]
12371076 async fn test_remove_while_playing_does_not_panic() {
12381238- let pool = test_pool().await;
12391239- let reg = Registry::new(pool, PathBuf::from("/tmp/moqbox"), test_registry());
12401240- let room = reg.create().await;
10771077+ let (reg, room_id) = test_registry_and_room().await;
1241107812421242- reg.push(
12431243- &room.id,
12441244- TrackMeta {
12451245- title: "A".into(),
12461246- duration: "3:45".into(),
12471247- thumbnail: None,
12481248- url: "https://example.com/a".into(),
12491249- source: crate::types::SourceKind::Direct,
12501250- },
10791079+ reg.push_url(&room_id, "https://example.com/a".into()).await;
10801080+ poll_until(
10811081+ || async { reg.queue(&room_id).await.is_empty() },
10821082+ Duration::from_secs(2),
10831083+ "push should have started playing",
12511084 )
12521085 .await;
12531253- tokio::time::sleep(Duration::from_millis(100)).await;
1254108612551255- reg.remove(&room.id).await;
12561256- tokio::time::sleep(Duration::from_millis(100)).await;
12571257-12581258- assert!(
12591259- !reg.exists(&room.id).await,
12601260- "room should not exist after remove"
12611261- );
10871087+ reg.remove(&room_id).await;
10881088+ assert!(!reg.exists(&room_id).await);
12621089 }
1263109012641091 #[tokio::test]
12651092 async fn test_sweep_idle_keeps_active_room() {
12661266- let pool = test_pool().await;
12671267- let reg = Registry::new(pool, PathBuf::from("/tmp/moqbox"), test_registry());
12681268- let room = reg.create().await;
10931093+ let (reg, room_id) = test_registry_and_room().await;
1269109412701270- reg.register_client(&room.id).await;
10951095+ reg.register_client(&room_id).await;
1271109612721097 let swept = reg.sweep_idle(Duration::from_millis(10_000)).await;
12731273- assert!(
12741274- !swept.contains(&room.id),
12751275- "active room should not be swept while clients are connected"
12761276- );
10981098+ assert!(!swept.contains(&room_id));
12771099 }
1278110012791101 #[tokio::test]
12801102 async fn test_client_count_tracking() {
12811281- let pool = test_pool().await;
12821282- let reg = Registry::new(pool, PathBuf::from("/tmp/moqbox"), test_registry());
12831283- let room = reg.create().await;
11031103+ let (reg, room_id) = test_registry_and_room().await;
1284110412851285- let id1 = reg.register_client(&room.id).await;
12861286- let id2 = reg.register_client(&room.id).await;
12871287- let id3 = reg.register_client(&room.id).await;
12881288- assert!(id1.is_some() && id2.is_some() && id3.is_some());
11051105+ let id1 = reg.register_client(&room_id).await;
11061106+ let id2 = reg.register_client(&room_id).await;
11071107+ let id3 = reg.register_client(&room_id).await;
11081108+ assert!(id1.is_some(), "id1 should be Some");
11091109+ assert!(id2.is_some(), "id2 should be Some");
11101110+ assert!(id3.is_some(), "id3 should be Some");
1289111112901290- reg.unregister_client(&room.id).await;
12911291- tokio::time::sleep(Duration::from_millis(50)).await;
11121112+ reg.unregister_client(&room_id).await;
1292111312931293- let id4 = reg.register_client(&room.id).await;
12941294- assert!(id4.is_some(), "new client should get an id");
12951295- assert_ne!(id4, id1, "ids should be unique");
11141114+ let id4 = reg.register_client(&room_id).await;
11151115+ assert!(id4.is_some());
11161116+ assert_ne!(id4, id1);
12961117 }
1297111812981119 #[tokio::test]
12991120 async fn test_multiple_rooms_dont_interfere() {
13001300- let pool = test_pool().await;
13011301- let reg = Registry::new(pool, PathBuf::from("/tmp/moqbox"), test_registry());
13021302- let room_a = reg.create().await;
11211121+ let (reg, room_a) = test_registry_and_room().await;
13031122 let room_b = reg.create().await;
1304112313051305- assert_ne!(room_a.id, room_b.id, "room IDs should be unique");
13061306- assert!(reg.exists(&room_a.id).await);
13071307- assert!(reg.exists(&room_b.id).await);
11241124+ assert_ne!(room_a, room_b);
11251125+ assert!(reg.exists(&room_a).await);
11261126+ assert!(reg.exists(&room_b).await);
11271127+11281128+ reg.push_url(&room_a, "https://example.com/a".into()).await;
11291129+ reg.push_url(&room_b, "https://example.com/b".into()).await;
1308113013091309- reg.push(
13101310- &room_a.id,
13111311- TrackMeta {
13121312- title: "A".into(),
13131313- duration: "1:00".into(),
13141314- thumbnail: None,
13151315- url: "https://example.com/a".into(),
13161316- source: crate::types::SourceKind::Direct,
13171317- },
11311131+ poll_until(
11321132+ || async { reg.queue(&room_a).await.is_empty() },
11331133+ Duration::from_secs(2),
11341134+ "room_a queue empty",
13181135 )
13191136 .await;
13201320- reg.push(
13211321- &room_b.id,
13221322- TrackMeta {
13231323- title: "B".into(),
13241324- duration: "2:00".into(),
13251325- thumbnail: None,
13261326- url: "https://example.com/b".into(),
13271327- source: crate::types::SourceKind::Direct,
13281328- },
11371137+ poll_until(
11381138+ || async { reg.queue(&room_b).await.is_empty() },
11391139+ Duration::from_secs(2),
11401140+ "room_b queue empty",
13291141 )
13301142 .await;
13311331- tokio::time::sleep(Duration::from_millis(100)).await;
13321332-13331333- let queue_a = reg.queue(&room_a.id).await;
13341334- let queue_b = reg.queue(&room_b.id).await;
13351335- assert!(
13361336- queue_a.is_empty(),
13371337- "room A should be empty after first track starts"
13381338- );
13391339- assert!(
13401340- queue_b.is_empty(),
13411341- "room B should be empty after first track starts"
13421342- );
13431343- }
13441344-13451345- #[tokio::test]
13461346- async fn test_skip_on_empty_queue_no_crash() {
13471347- let pool = test_pool().await;
13481348- let reg = Registry::new(pool, PathBuf::from("/tmp/moqbox"), test_registry());
13491349- let room = reg.create().await;
13501350- reg.skip(&room.id).await;
13511351- tokio::time::sleep(Duration::from_millis(50)).await;
13521143 }
1353114413541145 #[tokio::test]
13551146 async fn test_push_after_skip_works() {
13561356- let pool = test_pool().await;
13571357- let reg = Registry::new(pool, PathBuf::from("/tmp/moqbox"), test_registry());
13581358- let room = reg.create().await;
11471147+ let (reg, room_id) = test_registry_and_room().await;
1359114813601360- reg.push(
13611361- &room.id,
13621362- TrackMeta {
13631363- title: "A".into(),
13641364- duration: "3:45".into(),
13651365- thumbnail: None,
13661366- url: "https://a".into(),
13671367- source: crate::types::SourceKind::Direct,
13681368- },
11491149+ reg.push_url(&room_id, "https://a".into()).await;
11501150+ poll_until(
11511151+ || async { reg.queue(&room_id).await.is_empty() },
11521152+ Duration::from_secs(2),
11531153+ "first push started playing",
13691154 )
13701155 .await;
13711371- tokio::time::sleep(Duration::from_millis(100)).await;
13721372- reg.skip(&room.id).await;
13731373- tokio::time::sleep(Duration::from_millis(100)).await;
1374115613751375- reg.push(
13761376- &room.id,
13771377- TrackMeta {
13781378- title: "B".into(),
13791379- duration: "3:45".into(),
13801380- thumbnail: None,
13811381- url: "https://b".into(),
13821382- source: crate::types::SourceKind::Direct,
11571157+ reg.skip(&room_id).await;
11581158+ poll_until(
11591159+ || async {
11601160+ let playlist = crate::playlist::Playlist::new(reg.pool.clone(), &room_id.0);
11611161+ !playlist.history(10).await.unwrap().is_empty()
13831162 },
11631163+ Duration::from_secs(2),
11641164+ "skip should move track to history",
13841165 )
13851166 .await;
13861386- tokio::time::sleep(Duration::from_millis(100)).await;
1387116713881388- let queue = reg.queue(&room.id).await;
13891389- assert!(queue.is_empty(), "new track should be playing");
11681168+ reg.push_url(&room_id, "https://b".into()).await;
11691169+ poll_until(
11701170+ || async { reg.queue(&room_id).await.is_empty() },
11711171+ Duration::from_secs(2),
11721172+ "second push should start playing immediately (idle after skip)",
11731173+ )
11741174+ .await;
1390117513911391- let playlist = crate::playlist::Playlist::new(reg.pool.clone(), &room.id.0);
11761176+ let playlist = crate::playlist::Playlist::new(reg.pool.clone(), &room_id.0);
13921177 let history = playlist.history(10).await.unwrap();
13931393- assert!(!history.is_empty(), "skipped track should be in history");
11781178+ assert!(!history.is_empty());
13941179 }
1395118013961181 #[tokio::test]
13971182 async fn test_client_count_does_not_underflow() {
13981398- let pool = test_pool().await;
13991399- let reg = Registry::new(pool, PathBuf::from("/tmp/moqbox"), test_registry());
14001400- let room = reg.create().await;
11831183+ let (reg, room_id) = test_registry_and_room().await;
1401118414021402- reg.unregister_client(&room.id).await;
14031403- tokio::time::sleep(Duration::from_millis(50)).await;
11851185+ reg.unregister_client(&room_id).await;
1404118614051405- assert!(reg.exists(&room.id).await);
11871187+ let re_registered = reg.register_client(&room_id).await;
11881188+ assert!(
11891189+ re_registered.is_some(),
11901190+ "room should accept new clients after underflow"
11911191+ );
14061192 }
1407119314081194 #[tokio::test]
14091195 async fn test_multiple_skips_sequential() {
14101410- let pool = test_pool().await;
14111411- let reg = Registry::new(pool, PathBuf::from("/tmp/moqbox"), test_registry());
14121412- let room = reg.create().await;
11961196+ let (reg, room_id) = test_registry_and_room().await;
1413119714141414- for title in ["A", "B", "C"] {
14151415- reg.push(
14161416- &room.id,
14171417- TrackMeta {
14181418- title: title.into(),
14191419- duration: "1:00".into(),
14201420- thumbnail: None,
14211421- url: "https://x".into(),
14221422- source: crate::types::SourceKind::Direct,
14231423- },
14241424- )
14251425- .await;
11981198+ for url in ["https://x/a", "https://x/b", "https://x/c"] {
11991199+ reg.push_url(&room_id, url.into()).await;
14261200 }
14271427- tokio::time::sleep(Duration::from_millis(100)).await;
12011201+ poll_until(
12021202+ || async { reg.queue(&room_id).await.len() == 2 },
12031203+ Duration::from_secs(2),
12041204+ "3 pushes → 1 active, 2 queued",
12051205+ )
12061206+ .await;
1428120714291429- let mut queue = reg.queue(&room.id).await;
14301430- assert_eq!(queue.len(), 2, "B and C should be queued");
12081208+ reg.skip(&room_id).await;
12091209+ poll_until(
12101210+ || async { reg.queue(&room_id).await.len() == 1 },
12111211+ Duration::from_secs(2),
12121212+ "skip → 1 item left in queue",
12131213+ )
12141214+ .await;
1431121514321432- reg.skip(&room.id).await;
14331433- tokio::time::sleep(Duration::from_millis(100)).await;
14341434- queue = reg.queue(&room.id).await;
14351435- assert_eq!(queue.len(), 1, "only C should remain after skipping A");
14361436- assert_eq!(queue[0].title, "C");
12161216+ let queue = reg.queue(&room_id).await;
12171217+ assert_eq!(queue[0].title, "c");
1437121814381438- reg.skip(&room.id).await;
14391439- tokio::time::sleep(Duration::from_millis(100)).await;
14401440- queue = reg.queue(&room.id).await;
14411441- assert!(queue.is_empty(), "nothing should remain after skipping B");
12191219+ reg.skip(&room_id).await;
12201220+ poll_until(
12211221+ || async { reg.queue(&room_id).await.is_empty() },
12221222+ Duration::from_secs(2),
12231223+ "skip → queue empty",
12241224+ )
12251225+ .await;
1442122614431443- reg.skip(&room.id).await;
14441444- tokio::time::sleep(Duration::from_millis(100)).await;
12271227+ reg.skip(&room_id).await;
14451228 }
14461229}
+30-35
src/source/direct.rs
···22//!
33//! Handles URLs that point directly to media files (`.mp4`, `.webm`, `.mp3`,
44//! etc.) without going through yt-dlp. Metadata is extracted heuristically
55-//! from the URL path — no network requests are made during extraction.
55+//! from the URL path. No network requests are made during extraction.
6677use std::path::Path;
88···11111212/// Media file extensions that DirectSource can handle.
1313const MEDIA_EXTENSIONS: &[&str] = &[
1414- "mp4", "webm", "mkv", "avi", "mov", "m4v",
1515- "mp3", "ogg", "flac", "wav", "m4a", "aac", "opus", "wma",
1414+ "mp4", "webm", "mkv", "avi", "mov", "m4v", "mp3", "ogg", "flac", "wav", "m4a", "aac", "opus",
1515+ "wma",
1616];
17171818-// ---------------------------------------------------------------------------
1919-// Source
2020-// ---------------------------------------------------------------------------
2121-2218/// A media source that handles direct media file URLs.
2319///
2420/// Detects media files by URL path extension. If the URL has a recognised
···2925/// replaces separators with spaces)
3026/// - **resolve** returns the URL unchanged
3127///
3232-/// This source is intended as a fallback after [`YtdlpSource`](super::ytdlp::YtdlpSource)
3333-/// — it only matches URLs with media extensions whose filenames look like
2828+/// This source is intended as a fallback after [`YtdlpSource`](super::ytdlp::YtdlpSource).
2929+/// It only matches URLs with media extensions whose filenames look like
3430/// media files.
3531pub(crate) struct DirectSource;
3632···4440impl MediaSource for DirectSource {
4541 async fn extract(&self, url: &str, _cache_dir: &Path) -> Result<TrackMeta, SourceError> {
4642 let Some(ext) = extension(url) else {
4747- return Err(SourceError::Unsupported(format!("no file extension in URL: {url}")));
4343+ return Err(SourceError::Unsupported(format!(
4444+ "no file extension in URL: {url}"
4545+ )));
4846 };
49475048 if !MEDIA_EXTENSIONS.contains(&ext.as_str()) {
···66646765 async fn resolve(&self, url: &str, _cache_dir: &Path) -> Result<String, SourceError> {
6866 if !url.starts_with("http://") && !url.starts_with("https://") {
6969- return Err(SourceError::Unsupported(format!("not an HTTP(S) URL: {url}")));
6767+ return Err(SourceError::Unsupported(format!(
6868+ "not an HTTP(S) URL: {url}"
6969+ )));
7070 }
71717272- // Direct URLs are already playable as-is.
7372 Ok(url.to_string())
7473 }
7574}
7676-7777-// ---------------------------------------------------------------------------
7878-// Helpers
7979-// ---------------------------------------------------------------------------
80758176/// Extract the file extension (without dot) from a URL path, if any.
8277fn extension(url: &str) -> Option<String> {
···123118 .collect::<Vec<_>>()
124119 .join(" ");
125120126126- if result.is_empty() { "Untitled".into() } else { result }
121121+ if result.is_empty() {
122122+ "Untitled".into()
123123+ } else {
124124+ result
125125+ }
127126}
128127129129-// ---------------------------------------------------------------------------
130130-// Tests
131131-// ---------------------------------------------------------------------------
132132-133128#[cfg(test)]
134129mod tests {
135130 use super::*;
136136-137137- // -- Extension detection -------------------------------------------------
138131139132 #[test]
140133 fn test_extension_mp4() {
141141- assert_eq!(extension("https://example.com/video.mp4"), Some("mp4".into()));
134134+ assert_eq!(
135135+ extension("https://example.com/video.mp4"),
136136+ Some("mp4".into())
137137+ );
142138 }
143139144140 #[test]
···161157162158 #[test]
163159 fn test_extension_uppercase() {
164164- assert_eq!(extension("https://example.com/video.MP4"), Some("mp4".into()));
160160+ assert_eq!(
161161+ extension("https://example.com/video.MP4"),
162162+ Some("mp4".into())
163163+ );
165164 }
166165167167- // -- Title extraction ----------------------------------------------------
168168-169166 #[test]
170167 fn test_title_basic() {
171171- assert_eq!(extract_title("https://example.com/my-video.mp4", "mp4"), "my video");
168168+ assert_eq!(
169169+ extract_title("https://example.com/my-video.mp4", "mp4"),
170170+ "my video"
171171+ );
172172 }
173173174174 #[test]
···212212213213 #[test]
214214 fn test_title_multi_space_collapse() {
215215- assert_eq!(
216216- extract_title("https://example.com/a---b.mp4", "mp4"),
217217- "a b"
218218- );
215215+ assert_eq!(extract_title("https://example.com/a---b.mp4", "mp4"), "a b");
219216 }
220220-221221- // -- MediaSource implementation ------------------------------------------
222217223218 #[tokio::test]
224219 async fn test_direct_source_accepts_mp4() {
···270265 assert!(matches!(result, Err(SourceError::Unsupported(_))));
271266 }
272267273273- // Integration test with yt-dlp is in src/tests/ — this file focuses on
268268+ // Integration test with yt-dlp is in src/tests/. This file focuses on
274269 // DirectSource unit tests without network dependencies.
275270}
+5-25
src/source/mod.rs
···11//! Pluggable media source resolution.
22//!
33-//! Defines [`MediaSource`] — the seam between URL-based track selection and
33+//! Defines [`MediaSource`]: the seam between URL-based track selection and
44//! playable stream production. Different implementations handle different URL
55//! patterns (yt-dlp, direct media URLs, etc.).
66//!
···88//!
99//! 1. Implement [`MediaSource`] for your source type.
1010//! 2. Register it with [`SourceRegistry::register`].
1111-//! 3. Sources are tried in registration order — first match wins.
1111+//! 3. Sources are tried in registration order: first match wins.
12121313pub(crate) mod direct;
1414pub(crate) mod ytdlp;
···1616use std::path::Path;
17171818use crate::types::TrackMeta;
1919-2020-// ---------------------------------------------------------------------------
2121-// Error
2222-// ---------------------------------------------------------------------------
23192420/// Errors from media source resolution.
2521#[derive(Debug, thiserror::Error)]
···4137 Io(#[from] std::io::Error),
4238}
43394444-// ---------------------------------------------------------------------------
4545-// Trait
4646-// ---------------------------------------------------------------------------
4747-4840/// A media source knows how to extract metadata from a URL and resolve it to a
4941/// playable stream URL.
5042#[async_trait::async_trait]
···5244 /// Extract metadata from `url`.
5345 ///
5446 /// Returns [`SourceError::Unsupported`] if this source cannot handle the
5555- /// URL — the registry will try the next source. Any other error is
4747+ /// URL. The registry will try the next source. Any other error is
5648 /// terminal for this URL.
5749 async fn extract(&self, url: &str, cache_dir: &Path) -> Result<TrackMeta, SourceError>;
58505951 /// Resolve `url` to a playable stream URL.
6052 ///
6153 /// Returns [`SourceError::Unsupported`] if this source cannot handle the
6262- /// URL — the registry will try the next source.
5454+ /// URL. The registry will try the next source.
6355 async fn resolve(&self, url: &str, cache_dir: &Path) -> Result<String, SourceError>;
6456}
6565-6666-// ---------------------------------------------------------------------------
6767-// Registry
6868-// ---------------------------------------------------------------------------
69577058/// A registry of media sources tried in registration order.
7159///
···127115 }
128116129117 /// Resolve a URL to a playable stream URL by trying each registered source.
130130- pub(crate) async fn resolve(
131131- &self,
132132- url: &str,
133133- cache_dir: &Path,
134134- ) -> Result<String, SourceError> {
118118+ pub(crate) async fn resolve(&self, url: &str, cache_dir: &Path) -> Result<String, SourceError> {
135119 let mut unsupported_hint = String::new();
136120 for source in &self.sources {
137121 match source.resolve(url, cache_dir).await {
···148132 )))
149133 }
150134}
151151-152152-// ---------------------------------------------------------------------------
153153-// Tests
154154-// ---------------------------------------------------------------------------
155135156136#[cfg(test)]
157137mod tests {
+9-17
src/source/ytdlp.rs
···1212use crate::source::{MediaSource, SourceError};
1313use crate::types::TrackMeta;
14141515-// ---------------------------------------------------------------------------
1616-// yt-dlp JSON output fields
1717-// ---------------------------------------------------------------------------
1818-1915#[derive(Debug, Deserialize)]
2016#[serde(rename_all = "snake_case")]
2117struct YtdlpMetadata {
···2521 webpage_url: String,
2622}
27232828-// ---------------------------------------------------------------------------
2929-// Source
3030-// ---------------------------------------------------------------------------
3131-3224/// A media source backed by yt-dlp.
3325///
3426/// Handles URLs from YouTube, SoundCloud, Bandcamp, and hundreds of other
···4537impl MediaSource for YtdlpSource {
4638 async fn extract(&self, url: &str, cache_dir: &Path) -> Result<TrackMeta, SourceError> {
4739 if !url.starts_with("http://") && !url.starts_with("https://") {
4848- return Err(SourceError::Unsupported(format!("not an HTTP(S) URL: {url}")));
4040+ return Err(SourceError::Unsupported(format!(
4141+ "not an HTTP(S) URL: {url}"
4242+ )));
4943 }
50445145 let cache_dir = cache_dir.to_path_buf();
···6357 .output()
6458 })
6559 .await
6666- .map_err(|e| SourceError::Io(std::io::Error::new(std::io::ErrorKind::Other, e)))?;
6060+ .map_err(|e| SourceError::Io(std::io::Error::other(e)))?;
67616862 let output = result.map_err(|e| {
6963 if e.kind() == std::io::ErrorKind::NotFound {
···1009410195 async fn resolve(&self, url: &str, cache_dir: &Path) -> Result<String, SourceError> {
10296 if !url.starts_with("http://") && !url.starts_with("https://") {
103103- return Err(SourceError::Unsupported(format!("not an HTTP(S) URL: {url}")));
9797+ return Err(SourceError::Unsupported(format!(
9898+ "not an HTTP(S) URL: {url}"
9999+ )));
104100 }
105101106102 let cache_dir = cache_dir.to_path_buf();
···109105 std::process::Command::new("yt-dlp")
110106 .args([
111107 "-f",
112112- "b",
108108+ "b", // best single format (not bestvideo+bestaudio which needs merge)
113109 "-g",
114110 "--no-playlist",
115111 "--no-warnings",
···120116 .output()
121117 })
122118 .await
123123- .map_err(|e| SourceError::Io(std::io::Error::new(std::io::ErrorKind::Other, e)))?;
119119+ .map_err(|e| SourceError::Io(std::io::Error::other(e)))?;
124120125121 let output = result.map_err(|e| {
126122 if e.kind() == std::io::ErrorKind::NotFound {
···152148 Ok(stream_url)
153149 }
154150}
155155-156156-// ---------------------------------------------------------------------------
157157-// Tests
158158-// ---------------------------------------------------------------------------
159151160152#[cfg(test)]
161153mod tests {
+41-81
src/state.rs
···11-//! Pure playback state machine — deterministic, testable, IO-free.
11+//! Pure playback state machine: deterministic, testable, IO-free.
22//!
33//! Owns the in-memory queue, active track, and history. All transitions
44//! are pure functions returning [`Effect`]s that the caller executes.
55-//!
66-//! # Impure-pure-impure sandwich
77-//!
88-//! 1. **Gather** — caller fetches data (DB, wall clock)
99-//! 2. **Transition** — `PlaybackState::transition(event, now)` — pure
1010-//! 3. **Commit** — caller executes returned effects (DB writes, pipeline start)
1111-//!
1212-//! # Testing
1313-//!
1414-//! ```ignore
1515-//! let mut state = PlaybackState::default();
1616-//! let effects = state.transition(&Event::TrackQueued(my_track), 1_700_000_000_000);
1717-//! assert!(effects.contains(&Effect::StartPipeline(...)));
1818-//! ```
195206use crate::types::ActiveTrackInfo;
217···7763 PublishSnapshot,
7864}
79658080-/// Pure playback state — no IO handles, no DB connections.
8181-#[derive(Debug, Clone)]
6666+/// Pure playback state: no IO handles, no DB connections.
6767+#[derive(Debug, Clone, Default)]
8268pub(crate) struct PlaybackState {
8369 /// Currently playing track, if any.
8470 pub active: Option<ActiveTrackInfo>,
···8874 pub history: Vec<FinishedTrack>,
8975}
90769191-impl Default for PlaybackState {
9292- fn default() -> Self {
9393- Self {
9494- active: None,
9595- queue: Vec::new(),
9696- history: Vec::new(),
9797- }
9898- }
9999-}
100100-10177impl PlaybackState {
10278 /// Pure transition function.
10379 ///
104104- /// - `event` — what happened
105105- /// - `now` — current wall-clock epoch ms (gathered by caller for determinism)
106106- ///
10780 /// Returns effects for the caller to execute. Mutates `self` in place.
108108- /// Does NO I/O. Deterministic given the same inputs.
8181+ /// Does no I/O. Deterministic given the same inputs.
10982 pub fn transition(&mut self, event: &Event, now: i64) -> Vec<Effect> {
11083 match event {
11184 Event::TrackQueued(track) => self.handle_queued(track),
···12497 fn handle_queued(&mut self, track: &QueuedTrack) -> Vec<Effect> {
12598 self.queue.push(track.clone());
12699127127- // If nothing is playing, start immediately.
128128- if self.active.is_none() {
129129- self.advance(track.clone())
130130- } else {
131131- vec![Effect::PublishSnapshot]
100100+ if self.active.is_some() {
101101+ return vec![Effect::PublishSnapshot];
132102 }
103103+104104+ self.advance(track.clone())
133105 }
134106135107 /// User requested skip.
···148120 url: active.url,
149121 duration: active.duration,
150122 thumbnail: active.thumbnail,
151151- played_at: self.format_played_at(),
123123+ // RoomActor overrides via PersistFinished handler which sets
124124+ // the real played_at timestamp in the DB.
125125+ played_at: String::new(),
152126 },
153127 );
154128 }
···193167 url: active.url,
194168 duration: active.duration,
195169 thumbnail: active.thumbnail,
196196- played_at: self.format_played_at(),
170170+ // RoomActor overrides via PersistFinished handler which sets
171171+ // the real played_at timestamp in the DB.
172172+ played_at: String::new(),
197173 },
198174 );
199175 }
···210186211187 /// Asynchronous metadata resolution completed. Updates the matching item
212188 /// in the queue and/or the active track in-place. Does not change playback
213213- /// state — only updates display fields.
189189+ /// state, only updates display fields.
214190 fn handle_metadata_updated(
215191 &mut self,
216192 item_id: i64,
···240216 fn advance(&mut self, next: QueuedTrack) -> Vec<Effect> {
241217 self.queue.retain(|t| t.id != next.id);
242218243243- // Store active track with placeholder timestamp — RoomActor calls
219219+ // Store active track with placeholder timestamp. RoomActor calls
244220 // `resolve_started_at` with the real wall-clock time.
245221 let info = ActiveTrackInfo {
246222 id: next.id,
···266242 active.started_at_wall = real_started_at;
267243 }
268244 }
269269-270270- fn format_played_at(&self) -> String {
271271- // Simple ISO-like timestamp for frontend display.
272272- // This is a mock since we don't have Utc::now() in pure code.
273273- // RoomActor will override via PersistFinished handler.
274274- String::new()
275275- }
276245}
277246278247/// # Pure unit tests (no async, no DB, no IO)
···291260 }
292261 }
293262294294- // ── Effect ordering tests ─────────────────────────────────────────────
295295- // These verify the exact effect sequence returned by transition(). Order
296296- // matters: PersistStarted must precede StartPipeline (DB commit before
297297- // pipeline spawn), AbortPipeline must precede PersistFinished, etc.
298298-299263 #[test]
300300- fn idle_to_playing_effect_order() {
264264+ fn test_idle_to_playing_effect_order() {
301265 let mut s = PlaybackState::default();
302266 let effects = s.transition(&Event::TrackQueued(queued("A", 1)), 0);
303267···313277 }
314278315279 #[test]
316316- fn queue_second_while_playing_effect_order() {
280280+ fn test_queue_second_while_playing_effect_order() {
317281 let mut s = PlaybackState::default();
318282 s.transition(&Event::TrackQueued(queued("A", 1)), 0);
319283 let effects = s.transition(&Event::TrackQueued(queued("B", 2)), 0);
···321285 assert_eq!(
322286 effects,
323287 vec![Effect::PublishSnapshot],
324324- "playing → queue another: just notify — pipeline untouched"
288288+ "playing -> queue another: just notify, pipeline untouched"
325289 );
326290 }
327291328292 #[test]
329329- fn skip_with_next_effect_order() {
293293+ fn test_skip_with_next_effect_order() {
330294 let mut s = PlaybackState::default();
331295 s.transition(&Event::TrackQueued(queued("A", 1)), 0);
332296 s.transition(&Event::TrackQueued(queued("B", 2)), 0);
···354318 }
355319356320 #[test]
357357- fn skip_last_track_effect_order() {
321321+ fn test_skip_last_track_effect_order() {
358322 let mut s = PlaybackState::default();
359323 s.transition(&Event::TrackQueued(queued("A", 1)), 0);
360324 let effects = s.transition(&Event::Skip, 0);
···366330 Effect::PersistFinished(1),
367331 Effect::PublishSnapshot,
368332 ],
369369- "skip last: abort, finalize, notify — no StartPipeline"
333333+ "skip last: abort, finalize, notify, no StartPipeline"
370334 );
371335 }
372336373337 #[test]
374374- fn track_ended_with_next_effect_order() {
338338+ fn test_track_ended_with_next_effect_order() {
375339 let mut s = PlaybackState::default();
376340 s.transition(&Event::TrackQueued(queued("A", 1)), 0);
377341 s.transition(&Event::TrackQueued(queued("B", 2)), 0);
···380344 assert_eq!(
381345 &effects[..2],
382346 &[Effect::PersistFinished(1), Effect::PersistStarted(2),],
383383- "track ended (next): finalize A, persist B started_at — no AbortPipeline"
347347+ "track ended (next): finalize A, persist B started_at, no AbortPipeline"
384348 );
385349 assert!(
386350 effects
···395359 }
396360397361 #[test]
398398- fn track_ended_last_goes_idle_effect_order() {
362362+ fn test_track_ended_last_goes_idle_effect_order() {
399363 let mut s = PlaybackState::default();
400364 s.transition(&Event::TrackQueued(queued("A", 1)), 0);
401365 let effects = s.transition(&Event::TrackEnded { item_id: 1 }, 0);
···403367 assert_eq!(
404368 effects,
405369 vec![Effect::PersistFinished(1), Effect::PublishSnapshot,],
406406- "track ended last: finalize, notify — no StartPipeline, no AbortPipeline"
370370+ "track ended last: finalize, notify, no StartPipeline, no AbortPipeline"
407371 );
408372 }
409373410410- // ── State invariant tests ─────────────────────────────────────────────
411411-412374 #[test]
413413- fn idle_room_starts_playing_on_first_track() {
375375+ fn test_idle_room_starts_playing_on_first_track() {
414376 let mut s = PlaybackState::default();
415377 s.transition(&Event::TrackQueued(queued("A", 1)), 0);
416378···420382 }
421383422384 #[test]
423423- fn second_track_stays_in_queue() {
385385+ fn test_second_track_stays_in_queue() {
424386 let mut s = PlaybackState::default();
425387 s.transition(&Event::TrackQueued(queued("A", 1)), 0);
426388 s.transition(&Event::TrackQueued(queued("B", 2)), 0);
···431393 }
432394433395 #[test]
434434- fn skip_advances_to_next_track() {
396396+ fn test_skip_advances_to_next_track() {
435397 let mut s = PlaybackState::default();
436398 s.transition(&Event::TrackQueued(queued("A", 1)), 0);
437399 s.transition(&Event::TrackQueued(queued("B", 2)), 0);
···444406 }
445407446408 #[test]
447447- fn skip_on_idle_does_nothing() {
409409+ fn test_skip_on_idle_does_nothing() {
448410 let mut s = PlaybackState::default();
449411 let effects = s.transition(&Event::Skip, 0);
450412···453415 }
454416455417 #[test]
456456- fn skip_last_track_goes_idle() {
418418+ fn test_skip_last_track_goes_idle() {
457419 let mut s = PlaybackState::default();
458420 s.transition(&Event::TrackQueued(queued("A", 1)), 0);
459421 s.transition(&Event::Skip, 0);
···464426 }
465427466428 #[test]
467467- fn track_ended_advances_to_next() {
429429+ fn test_track_ended_advances_to_next() {
468430 let mut s = PlaybackState::default();
469431 s.transition(&Event::TrackQueued(queued("A", 1)), 0);
470432 s.transition(&Event::TrackQueued(queued("B", 2)), 0);
···476438 }
477439478440 #[test]
479479- fn track_ended_ignores_stale_id() {
441441+ fn test_track_ended_ignores_stale_id() {
480442 let mut s = PlaybackState::default();
481443 s.transition(&Event::TrackQueued(queued("A", 1)), 0);
482444···488450 }
489451490452 #[test]
491491- fn track_ended_last_track_goes_idle() {
453453+ fn test_track_ended_last_track_goes_idle() {
492454 let mut s = PlaybackState::default();
493455 s.transition(&Event::TrackQueued(queued("A", 1)), 0);
494456 s.transition(&Event::TrackEnded { item_id: 1 }, 0);
···498460 }
499461500462 #[test]
501501- fn resolve_started_at_updates_active_track() {
463463+ fn test_resolve_started_at_updates_active_track() {
502464 let mut s = PlaybackState::default();
503465 s.transition(&Event::TrackQueued(queued("A", 1)), 0);
504466···507469 }
508470509471 #[test]
510510- fn skip_then_track_ended_stale_is_ignored() {
472472+ fn test_skip_then_track_ended_stale_is_ignored() {
511473 let mut s = PlaybackState::default();
512474 s.transition(&Event::TrackQueued(queued("A", 1)), 0);
513475 s.transition(&Event::TrackQueued(queued("B", 2)), 0);
···520482 }
521483522484 #[test]
523523- fn resolve_started_at_noop_when_idle() {
485485+ fn test_resolve_started_at_noop_when_idle() {
524486 let mut s = PlaybackState::default();
525487 s.resolve_started_at(42);
526488 assert!(s.active.is_none());
527489 }
528490529529- // ── Metadata update tests ─────────────────────────────────────────
530530-531491 #[test]
532532- fn metadata_update_updates_queue_item() {
492492+ fn test_metadata_update_updates_queue_item() {
533493 let mut s = PlaybackState::default();
534494 s.transition(&Event::TrackQueued(queued("A", 1)), 0);
535495 s.transition(
···553513 }
554514555515 #[test]
556556- fn metadata_update_on_queued_item() {
516516+ fn test_metadata_update_on_queued_item() {
557517 let mut s = PlaybackState::default();
558518 s.transition(&Event::TrackQueued(queued("A", 1)), 0);
559519 s.transition(&Event::TrackQueued(queued("B", 2)), 0);
···574534 }
575535576536 #[test]
577577- fn metadata_update_on_unknown_id_is_noop() {
537537+ fn test_metadata_update_on_unknown_id_is_noop() {
578538 let mut s = PlaybackState::default();
579539 let effects = s.transition(
580540 &Event::MetadataUpdated {
···585545 },
586546 0,
587547 );
588588- // Still publishes snapshot even if nothing changed — clients may want
548548+ // Still publishes snapshot even if nothing changed: clients may want
589549 // to know the update was processed.
590550 assert_eq!(effects, vec![Effect::PublishSnapshot]);
591551 assert!(s.active.is_none());
+31-87
src/transport.rs
···4444 }
4545}
46464747+#[must_use]
4748pub(crate) fn decode_varint(buf: &[u8]) -> Option<(u64, usize)> {
4849 let first = *buf.first()?;
4950 match first >> 6 {
···8990 buf
9091}
91929393+#[must_use]
9294fn decode_string(buf: &[u8]) -> Option<(String, usize)> {
9395 let (len, mut offset) = decode_varint(buf)?;
9496 let end = offset.checked_add(len as usize)?;
···177179 Bytes::from(buf)
178180}
179181182182+#[must_use]
180183pub(crate) fn decode_message(buf: &[u8]) -> Option<(MoqMessage, usize)> {
181184 let (msg_type, mut offset) = decode_varint(buf)?;
182185 match msg_type {
···252255 format!("moqbox/room/{room_id}")
253256}
254257255255-// ---------------------------------------------------------------------------
256256-// Track publishers
257257-// ---------------------------------------------------------------------------
258258-259258/// Per-room MoQ track publishers, shared across all client sessions.
260259///
261260/// Cloning is cheap (all inner channels are `Arc`-based).
···272271273272impl TrackPublishers {
274273 pub(crate) fn new() -> Self {
275275- let (video, _) = broadcast::channel(256);
274274+ let (video, _) = broadcast::channel(256); // enough for bursty media objects without backpressure
276275 let (chat, _) = broadcast::channel(256);
277276 let (state, _) = watch::channel(StateValue {
278277 payload: Bytes::new(),
···345344}
346345347346/// Handle a WebSocket upgrade and run the MoQ session lifecycle.
348348-///
349349-/// 1. Sends `ANNOUNCE` for the room namespace.
350350-/// 2. Loops, reading MoQ messages from the client.
351351-/// 3. On `SUBSCRIBE`, subscribes to the corresponding broadcast receiver
352352-/// and spawns a forward task that relays objects to the WebSocket.
353353-/// 4. On `OBJECT` with track ID 2 (chat), forwards to the chat publisher.
354354-/// 5. On close, cancels all forward tasks.
355347pub(crate) async fn handle_ws_session(
356348 mut ws: WebSocket,
357349 room_id: RoomId,
···373365374366 let (mut ws_tx, mut ws_rx) = ws.split();
375367376376- // 1. Send ANNOUNCE for this room.
377368 let announce = encode_message(&MoqMessage::Announce {
378369 namespace: room_namespace(&room_id.0),
379370 });
···381372 return;
382373 }
383374384384- // Register this client with the room actor.
385375 let (resp_tx, resp_rx) = tokio::sync::oneshot::channel();
386376 let _ = cmd_tx
387377 .send(RoomCommand::RegisterClient { resp: resp_tx })
···391381 .ok()
392382 .and_then(|r| r.ok());
393383394394- // Channel that broadcast forward tasks use to send objects to the WS writer.
395395- // Unbounded to avoid deadlock between subscribe processing and the WS send loop.
396384 let (object_tx, mut object_rx) = mpsc::unbounded_channel::<Bytes>();
397385398398- // Handles of spawned forward tasks, cancelled on disconnect.
399386 let mut task_handles: Vec<tokio::task::JoinHandle<()>> = Vec::new();
400387401388 loop {
···423410 break;
424411 }
425412 None => break,
426426- // Ignore Ping / Pong / Text frames.
427413 _ => {}
428414 }
429415 }
430416431431- // Forward objects from broadcast receivers to the WebSocket.
432417 obj_bytes = object_rx.recv() => {
433418 match obj_bytes {
434419 Some(bytes) => {
···436421 break;
437422 }
438423 }
439439- // All senders dropped — no more objects will arrive.
440424 None => break,
441425 }
442426 }
443427 }
444428 }
445429446446- // Unregister this client with the room actor.
447430 let _ = cmd_tx.send(RoomCommand::UnregisterClient).await;
448431449449- // Cancel all forward tasks on disconnect.
450432 for handle in task_handles {
451433 handle.abort();
452434 }
···454436455437/// WebSocket session for video-only delivery.
456438///
457457-/// No MoQ protocol negotiation — the server auto-subscribes the client to
439439+/// No MoQ protocol negotiation: the server auto-subscribes the client to
458440/// the video broadcast. Messages are raw fMP4 bytes (no MoQ framing).
459441/// A separate WS-control connection handles chat, state, and MoQ handshake.
460460-pub(crate) async fn handle_video_session(
461461- ws: WebSocket,
462462- publishers: TrackPublishers,
463463-) {
442442+pub(crate) async fn handle_video_session(ws: WebSocket, publishers: TrackPublishers) {
464443 let (mut ws_tx, mut ws_rx) = ws.split();
465444466466- // Collect the cached init (or wait for the pipeline to produce one)
467467- // but don't send it yet — we need to subscribe to the broadcast first
468468- // so we can discard any init that was also published via broadcast,
469469- // which would otherwise arrive as a duplicate through the forward loop.
470445 let init = match publishers.get_init_segment() {
471446 Some(cached) => {
472447 tracing::debug!("video ws: using cached init segment");
···486461 }
487462 };
488463489489- // Subscribe to the broadcast before sending cached init so that
490490- // any init published via broadcast is caught by the catch-up loop
491491- // and discarded, preventing a duplicate on the forward path.
492464 let mut video_rx = publishers.video.subscribe();
493465 loop {
494466 match video_rx.try_recv() {
···499471 }
500472 }
501473502502- // Now send the cached init — the broadcast buffer has been drained
503503- // so the forward loop below won't re-deliver it.
504474 if let Some((_group_id, payload)) = init {
505475 if ws_tx.send(Message::Binary(payload)).await.is_err() {
506476 return;
507477 }
508478 }
509479510510- // Forward live video objects as raw fMP4 payloads.
511480 loop {
512481 tokio::select! {
513482 msg = ws_rx.next() => {
···533502 }
534503}
535504505505+#[allow(clippy::too_many_arguments)]
506506+// justified: all params are distinct concerns
536507async fn handle_moq_message(
537508 msg: MoqMessage,
538509 room_id: &RoomId,
···549520 let track_id = match TrackId::from_name(&track) {
550521 Some(id) => id,
551522 None => {
552552- // Unknown track — send SUBSCRIBE_RST.
553523 let rst = encode_message(&MoqMessage::SubscribeRst {
554524 namespace,
555525 track,
···560530 }
561531 };
562532563563- // Send SUBSCRIBE_OK first.
564533 let ok = encode_message(&MoqMessage::SubscribeOk {
565534 namespace: namespace.clone(),
566535 track: track.clone(),
···571540572541 match track_id {
573542 TrackId::Chat => {
574574- // Send recent chat history, then subscribe to live broadcast.
575543 if let Ok(messages) = registry.recent_chat(room_id, 50).await {
576544 for msg in messages.iter().rev() {
577545 let json = serde_json::json!({
···594562 }
595563 }
596564597597- // Subscribe to live chat broadcast.
598565 let rx = publishers.chat.subscribe();
599566 let tx = object_tx.clone();
600567 task_handles.push(tokio::spawn(async move {
···619586 }));
620587 }
621588 TrackId::State => {
622622- // State uses watch (latest-only). Send current value
623623- // immediately, then forward changes.
624589 let initial = publishers.state.borrow().clone();
625590 let obj = encode_message(&MoqMessage::Object {
626591 track_id: TrackId::State as u64,
···651616 }
652617 }
653618 }));
654654- }
655655- _ => {
656656- // Video and audio use a dedicated WebSocket endpoint.
657657- // Ignore subscribe requests on the control WS.
619619+ }
620620+ _ => {}
658621 }
659622 }
660660- }
661623662624 MoqMessage::Object {
663625 track_id,
···665627 object_id: _,
666628 payload,
667629 } => {
668668- // Incoming chat message from client: persist and rebroadcast.
669630 if track_id == TrackId::Chat as u64 {
670631 let json: serde_json::Value =
671632 serde_json::from_slice(&payload).unwrap_or(serde_json::Value::Null);
···683644 }
684645 }
685646686686- // Server-initiated messages received from client are unexpected.
687647 MoqMessage::Announce { .. }
688648 | MoqMessage::AnnounceOk { .. }
689649 | MoqMessage::SubscribeOk { .. }
···857817858818 publishers.publish_video(5, 99, Bytes::from("video-frame"));
859819860860- let obj = rx.recv().await.expect("should receive video object");
820820+ let obj = rx.recv().await.unwrap();
861821 assert_eq!(obj.track_id, TrackId::Video);
862822 assert_eq!(obj.payload, Bytes::from("video-frame"));
863823 }
···869829870830 publishers.publish_chat(0, Bytes::from("hello"));
871831872872- let obj = rx.recv().await.expect("should receive chat object");
832832+ let obj = rx.recv().await.unwrap();
873833 assert_eq!(obj.track_id, TrackId::Chat);
874834 assert_eq!(obj.object_id, 0);
875835 assert_eq!(obj.payload, Bytes::from("hello"));
···884844 publishers.publish_chat(0, Bytes::from("second"));
885845 publishers.publish_chat(0, Bytes::from("third"));
886846887887- let obj1 = rx.recv().await.expect("msg 1");
847847+ let obj1 = rx.recv().await.unwrap();
888848 assert_eq!(obj1.object_id, 0);
889849 assert_eq!(obj1.payload, Bytes::from("first"));
890850891891- let obj2 = rx.recv().await.expect("msg 2");
851851+ let obj2 = rx.recv().await.unwrap();
892852 assert_eq!(obj2.object_id, 1);
893853 assert_eq!(obj2.payload, Bytes::from("second"));
894854895895- let obj3 = rx.recv().await.expect("msg 3");
855855+ let obj3 = rx.recv().await.unwrap();
896856 assert_eq!(obj3.object_id, 2);
897857 assert_eq!(obj3.payload, Bytes::from("third"));
898858 }
···904864905865 publishers.publish_state(Bytes::from("playing"));
906866907907- // Watch receiver needs changed() to observe the update.
908908- rx.changed().await.expect("state should change");
867867+ rx.changed().await.unwrap();
909868 let val = rx.borrow().clone();
910869 assert_eq!(val.payload, Bytes::from("playing"));
911870 assert_eq!(val.seq, 1); // started at 0, now 1
···919878920879 publishers.publish_video(0, 0, Bytes::from("fanout"));
921880922922- let obj1 = rx1.recv().await.expect("subscriber 1");
923923- let obj2 = rx2.recv().await.expect("subscriber 2");
881881+ let obj1 = rx1.recv().await.unwrap();
882882+ let obj2 = rx2.recv().await.unwrap();
924883 assert_eq!(obj1.payload, Bytes::from("fanout"));
925884 assert_eq!(obj2.payload, Bytes::from("fanout"));
926885 }
927927-928928- // ── Init segment cache tests ───────────────────────────────────────────
929886930887 #[test]
931888 fn test_init_cache_and_retrieve() {
932889 let publishers = TrackPublishers::new();
933890934934- assert!(
935935- publishers.get_init_segment().is_none(),
936936- "init should be None before caching"
937937- );
891891+ assert!(publishers.get_init_segment().is_none());
938892939893 let payload = Bytes::from("ftyp-moov-data");
940894 publishers.cache_init_segment(payload.clone(), 42);
941895942896 let retrieved = publishers.get_init_segment();
943943- assert!(retrieved.is_some(), "init should be Some after caching");
897897+ assert!(retrieved.is_some());
944898 let (group_id, data) = retrieved.unwrap();
945945- assert_eq!(group_id, 42, "group_id should match what was cached");
946946- assert_eq!(data, payload, "payload should match what was cached");
899899+ assert_eq!(group_id, 42);
900900+ assert_eq!(data, payload);
947901 }
948902949903 #[tokio::test]
···951905 let publishers = TrackPublishers::new();
952906953907 let mut waiter = publishers.video_init_waiter();
954954- assert!(!*waiter.borrow(), "initial watch value should be false");
908908+ assert!(!*waiter.borrow());
955909956910 let pubs = publishers.clone();
957911 tokio::spawn(async move {
···961915962916 let _ = tokio::time::timeout(std::time::Duration::from_secs(2), waiter.changed())
963917 .await
964964- .expect("watch should fire within timeout when init is cached");
918918+ .expect("watch should fire within timeout");
965919966966- assert!(
967967- *waiter.borrow(),
968968- "watch value should be true after cache_init_segment"
969969- );
920920+ assert!(*waiter.borrow());
970921971922 let retrieved = publishers.get_init_segment();
972972- assert!(
973973- retrieved.is_some(),
974974- "init should be retrievable after watch fires"
975975- );
923923+ assert!(retrieved.is_some());
976924 let (group_id, data) = retrieved.unwrap();
977977- assert_eq!(group_id, 1, "group_id should match");
978978- assert_eq!(data, Bytes::from("init-payload"), "payload should match");
925925+ assert_eq!(group_id, 1);
926926+ assert_eq!(data, Bytes::from("init-payload"));
979927 }
980928981929 #[test]
···986934 publishers.cache_init_segment(Bytes::from("init-2"), 2);
987935988936 let retrieved = publishers.get_init_segment();
989989- assert!(retrieved.is_some(), "init should be cached");
937937+ assert!(retrieved.is_some());
990938 let (group_id, data) = retrieved.unwrap();
991991- assert_eq!(group_id, 2, "group_id should be from the latest cache call");
992992- assert_eq!(
993993- data,
994994- Bytes::from("init-2"),
995995- "payload should be from the latest cache call"
996996- );
939939+ assert_eq!(group_id, 2);
940940+ assert_eq!(data, Bytes::from("init-2"));
997941 }
998942}
+7-9
src/types.rs
···49495050/// Metadata for a single queued track.
5151///
5252-/// Source-agnostic — produced by any [`MediaSource`](crate::source::MediaSource)
5353-/// implementation (yt-dlp, direct URL detection, etc.).
5252+/// Source-agnostic: produced by any [`MediaSource`](crate::source::MediaSource)
5353+/// implementation.
5454#[derive(Debug, Clone, Serialize)]
5555pub(crate) struct TrackMeta {
5656 pub title: String,
···6868 pub(crate) title: String,
6969 pub(crate) url: String,
7070 pub(crate) duration: String,
7171+ pub(crate) thumbnail: Option<String>,
7172 pub(crate) started_at: i64,
7273}
7374···145146 pub(crate) seq: u64,
146147}
147148148148-/// All room operations — processed sequentially by the per-room actor.
149149+/// All room operations, processed sequentially by the per-room actor.
149150pub(crate) enum RoomCommand {
150150- /// Track with already-resolved metadata (legacy path, used by tests).
151151- #[allow(dead_code)]
152152- QueueTrack(TrackMeta),
153153- /// Raw URL — metadata extraction happens asynchronously in a spawned task.
151151+ /// Raw URL: metadata extraction happens asynchronously in a spawned task.
154152 QueueUrl(String),
155153 /// Async metadata extraction result.
156154 MetadataReady {
···188186 pub(crate) url: String,
189187 pub(crate) duration: String,
190188 pub(crate) thumbnail: Option<String>,
191191- /// Epoch ms when the track started — for client-side elapsed computation.
189189+ /// Epoch ms when the track started, for client-side elapsed computation.
192190 pub(crate) started_at_wall: i64,
193191}
194192195195-/// Public handle to a room — allows sending commands and reading publishers.
193193+/// Public handle to a room: allows sending commands and reading publishers.
196194#[derive(Clone)]
197195pub(crate) struct RoomHandle {
198196 pub(crate) cmd_tx: mpsc::Sender<RoomCommand>,