From da4905a2de0fc367e397df2ce074dc856d4e7909 Mon Sep 17 00:00:00 2001 From: karitham Date: Sun, 17 May 2026 23:27:34 +0200 Subject: [PATCH] formatting & cleanup --- Cargo.toml | 23 - README.md | 50 +- flake.nix | 7 +- rustfmt.toml | 1 + src/db.rs | 38 +- src/main.rs | 5 +- src/media.rs | 104 +- src/names.rs | 27 + src/playback.rs | 77 +- src/playlist.rs | 236 +--- src/room.rs | 789 +++++------- src/source/direct.rs | 65 +- src/source/mod.rs | 30 +- src/source/ytdlp.rs | 26 +- src/state.rs | 122 +- src/transport.rs | 118 +- src/types.rs | 16 +- src/util.rs | 5 + src/web.rs | 61 +- templates/index.html | 312 +++-- templates/room.html | 2882 +++++++++++++++++++++--------------------- treefmt.toml | 11 + 22 files changed, 2218 insertions(+), 2787 deletions(-) create mode 100644 rustfmt.toml create mode 100644 src/util.rs create mode 100644 treefmt.toml diff --git a/Cargo.toml b/Cargo.toml index 04d5639..e5d5e8c 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -6,41 +6,18 @@ description = "Ultra-low latency shared media rooms over MoQ" license = "MIT OR Apache-2.0" [dependencies] -# ── Async ────────────────────────────────────────────────────────── tokio = { version = "1", features = ["full"] } anyhow = "1" thiserror = "2" - -# ── HTTP / SSR ───────────────────────────────────────────────────── axum = { version = "0.8", features = ["ws"] } -tower = "0.5" -tower-http = { version = "0.6", features = ["cors", "trace", "fs", "compression-gzip"] } tracing = "0.1" tracing-subscriber = { version = "0.3", features = ["env-filter"] } serde = { version = "1", features = ["derive"] } serde_json = "1" - -# ── Templates ────────────────────────────────────────────────────── askama = "0.13" - -# ── CLI ──────────────────────────────────────────────────────────── clap = { version = "4", features = ["derive", "env"] } - -# ── WebSocket / MoQ Transport ─────────────────────────────────────── -tokio-tungstenite = "0.24" futures-util = "0.3" - -# ── TLS (reserved for future use) ───────────────────────────────────── -rustls = { version = "0.23", features = ["ring"] } -rcgen = "0.13" - -# ── Async ───────────────────────────────────────────────────────── async-trait = "0.1" - -# ── Identifiers ──────────────────────────────────────────────────── -uuid = { version = "1", features = ["v4", "serde"] } - -# ── Database ──────────────────────────────────────────────────────── sqlx = { version = "0.8", features = ["sqlite", "runtime-tokio", "chrono"] } chrono = { version = "0.4", features = ["serde"] } bytes = "1" diff --git a/README.md b/README.md index e6a7c29..4af3a34 100644 --- a/README.md +++ b/README.md @@ -7,16 +7,54 @@ Requires `ffmpeg` and `yt-dlp` on PATH. Run with `cargo run`, open `localhost:30 One Axum binary serving SSR pages, REST API, and a custom media streaming protocol over WebSocket on the same port. Per-room mpsc actor processes commands -(queue, skip, chat, register) sequentially — no locks. Room state persisted to -SQLite (WAL mode). +(queue, skip, chat, register, metadata) sequentially -- no locks. Room state +persisted to SQLite (WAL mode). + +### Async metadata extraction + +The ingest HTTP handler returns immediately after queuing a URL. Metadata +extraction (yt-dlp) runs in a spawned task within the actor. When extraction +completes, the result is sent back to the actor which updates the DB and pushes +the resolved metadata to clients via the state broadcast channel. + +```mermaid +sequenceDiagram + participant Client + participant HTTP as HTTP Handler + participant Actor as Room Actor + participant Extract as Extraction Task + + Client->>HTTP: POST /api/ingest + HTTP->>Actor: QueueUrl(url) + HTTP-->>Client: {ok: true} + Actor->>Actor: insert placeholder + Actor->>Extract: spawn extraction + Extract->>Actor: MetadataReady + Actor->>Client: PublishSnapshot via WS +``` + +### Playback pipeline yt-dlp resolves URLs to direct stream addresses. FFmpeg transcodes to fragmented MP4 at real-time speed (`-re`). fMP4 boxes are read from stdout as moof+mdat pairs and published over per-room broadcast channels. The video init segment -(ftyp+moov) is cached separately. Forward tasks send cached init first, then -catch up to the broadcast live edge before forwarding, so late joiners start at -the live position. +(ftyp+moov) is cached for late-joining clients. + +### State machine + +`PlaybackState` is pure -- all transitions are deterministic. The actor follows +the impure-pure-impure sandwich: + +1. **Gather** -- DB queries, wall clock +2. **Transition** -- `PlaybackState::transition(event, now)` -- returns `Effect`s +3. **Commit** -- actor executes effects + +Events: `TrackQueued`, `Skip`, `TrackEnded`, `MetadataUpdated`. +Effects: `AbortPipeline`, `StartPipeline`, `PersistStarted`, `PersistFinished`, `PublishSnapshot`. + +### Client delivery Clients use MSE with a single SourceBuffer in segments mode. A state track (watch channel) delivers current playback position, queue, and history as JSON. -Chat is persisted to SQLite, replayed on subscribe (last N), then live. +Queue items carry a `pending` flag while metadata is unresolved. Chat is +persisted to SQLite, replayed on subscribe, then live. diff --git a/flake.nix b/flake.nix index e37673d..f22a617 100644 --- a/flake.nix +++ b/flake.nix @@ -55,9 +55,8 @@ cargo-edit cargo-audit cargo-nextest - cargo-expand - cargo-insta - cargo-deny + treefmt + nixfmt yt-dlp ffmpeg openssl @@ -65,7 +64,7 @@ ]; }; - formatter = pkgs.nixfmt; + formatter = pkgs.treefmt; }; }; } diff --git a/rustfmt.toml b/rustfmt.toml new file mode 100644 index 0000000..f216078 --- /dev/null +++ b/rustfmt.toml @@ -0,0 +1 @@ +edition = "2024" diff --git a/src/db.rs b/src/db.rs index 4c14654..d9bcb68 100644 --- a/src/db.rs +++ b/src/db.rs @@ -9,7 +9,7 @@ use sqlx::{Pool, Sqlite}; pub(crate) async fn create_pool(path: &Path) -> Result, sqlx::Error> { let conn_str = format!("sqlite:{}?mode=rwc", path.to_string_lossy()); let pool = SqlitePoolOptions::new() - .max_connections(8) + .max_connections(8) // sqlite supports unlimited connections but 8 avoids contention .connect(&conn_str) .await?; @@ -26,7 +26,7 @@ pub(crate) async fn create_pool(path: &Path) -> Result, sqlx::Error Ok(pool) } -/// Run all migrations (idempotent — uses `CREATE TABLE IF NOT EXISTS`). +/// Run all migrations. Idempotent: uses `CREATE TABLE IF NOT EXISTS`. pub(crate) async fn run_migrations(pool: &Pool) -> Result<(), sqlx::Error> { sqlx::query( "CREATE TABLE IF NOT EXISTS rooms ( @@ -118,13 +118,12 @@ pub(crate) async fn run_migrations(pool: &Pool) -> Result<(), sqlx::Erro } // Migration: add name column to rooms. - let has_name: bool = sqlx::query_scalar( - "SELECT COUNT(*) FROM pragma_table_info('rooms') WHERE name = 'name'", - ) - .fetch_one(pool) - .await - .map(|c: i64| c > 0) - .unwrap_or(false); + let has_name: bool = + sqlx::query_scalar("SELECT COUNT(*) FROM pragma_table_info('rooms') WHERE name = 'name'") + .fetch_one(pool) + .await + .map(|c: i64| c > 0) + .unwrap_or(false); if !has_name { sqlx::query("ALTER TABLE rooms ADD COLUMN name TEXT NOT NULL DEFAULT ''") @@ -135,12 +134,6 @@ pub(crate) async fn run_migrations(pool: &Pool) -> Result<(), sqlx::Erro Ok(()) } -fn now_iso() -> String { - chrono::Utc::now() - .format("%Y-%m-%dT%H:%M:%S%.3fZ") - .to_string() -} - pub(crate) struct NewChatMessage { pub(crate) room_id: String, pub(crate) user_name: String, @@ -164,7 +157,7 @@ pub(crate) async fn room_insert( name: &str, persistent: bool, ) -> Result<(), sqlx::Error> { - let now = now_iso(); + let now = crate::util::now_iso(); sqlx::query( "INSERT INTO rooms (id, name, created_at, updated_at, persistent) VALUES (?1, ?2, ?3, ?4, ?5)", ) @@ -241,6 +234,7 @@ pub(crate) async fn chat_recent( #[cfg(test)] mod tests { use super::*; + use crate::util; async fn setup() -> Pool { let pool = SqlitePoolOptions::new() @@ -267,8 +261,12 @@ mod tests { #[tokio::test] async fn test_room_crud() { let pool = setup().await; - room_insert(&pool, "test-room", "test-name", false).await.unwrap(); + room_insert(&pool, "test-room", "test-name", false) + .await + .unwrap(); assert!(room_exists(&pool, "test-room").await.unwrap()); + let name = get_room_name(&pool, "test-room").await; + assert_eq!(name, "test-name"); room_delete(&pool, "test-room").await.unwrap(); assert!(!room_exists(&pool, "test-room").await.unwrap()); } @@ -276,7 +274,9 @@ mod tests { #[tokio::test] async fn test_chat_crud() { let pool = setup().await; - room_insert(&pool, "test-room", "test-name", false).await.unwrap(); + room_insert(&pool, "test-room", "test-name", false) + .await + .unwrap(); for i in 0..5 { let msg = NewChatMessage { @@ -284,7 +284,7 @@ mod tests { user_name: "alice".into(), content: format!("message {i}"), msg_type: "message".into(), - created_at: now_iso(), + created_at: util::now_iso(), }; chat_insert(&pool, &msg).await.unwrap(); } diff --git a/src/main.rs b/src/main.rs index da1abba..3eb2934 100644 --- a/src/main.rs +++ b/src/main.rs @@ -1,4 +1,4 @@ -//! moqbox — Ultra-low latency shared media rooms over MoQ. +//! moqbox: Ultra-low latency shared media rooms over MoQ. #![deny(rust_2018_idioms, unsafe_code)] #![cfg_attr(not(test), deny(clippy::unwrap_used))] @@ -13,6 +13,7 @@ mod source; mod state; mod transport; mod types; +mod util; mod web; use clap::Parser; @@ -46,7 +47,6 @@ async fn main() -> anyhow::Result<()> { let args = Args::parse(); - // Ensure cache directory exists. tokio::fs::create_dir_all(&args.cache_dir).await?; // Ensure database parent directory exists. @@ -60,7 +60,6 @@ async fn main() -> anyhow::Result<()> { db::run_migrations(&pool).await?; tracing::info!("database ready at {}", args.db_path.display()); - // ── Source registry ────────────────────────────────────────────── let mut sources = source::SourceRegistry::new(); sources.register(Box::new(source::ytdlp::YtdlpSource::new())); sources.register(Box::new(source::direct::DirectSource::new())); diff --git a/src/media.rs b/src/media.rs index aa54888..bee52c4 100644 --- a/src/media.rs +++ b/src/media.rs @@ -4,7 +4,7 @@ //! real-time speed (`-re`), reads fMP4 boxes from stdout as moof+mdat pairs, //! and publishes them over per-room broadcast channels. //! -//! This module does **not** resolve URLs or extract metadata — that's the +//! This module does **not** resolve URLs or extract metadata: that is the //! responsibility of [`crate::source`]. use tokio::io::AsyncReadExt; @@ -12,10 +12,6 @@ use tokio::process::Command; use crate::transport::TrackPublishers; -// --------------------------------------------------------------------------- -// Error -// --------------------------------------------------------------------------- - /// Errors that can occur during FFmpeg transcoding. #[derive(Debug, thiserror::Error)] pub(crate) enum TranscodeError { @@ -27,10 +23,6 @@ pub(crate) enum TranscodeError { Aborted, } -// --------------------------------------------------------------------------- -// Public API -// --------------------------------------------------------------------------- - /// Transcode a playable stream URL to fMP4, publishing raw boxes as MoQ objects. /// /// Spawns `ffmpeg` with the stream URL as input, transcodes to fragmented MP4 @@ -40,8 +32,6 @@ pub(crate) enum TranscodeError { /// /// Returns `Ok(())` on normal EOF, `Err` on failure. /// -/// ## Abort behaviour -/// /// When `abort` is signalled, the pipeline exits early with /// [`TranscodeError::Aborted`] and the ffmpeg process is killed. pub(crate) async fn transcode_to_fmp4( @@ -113,7 +103,6 @@ pub(crate) async fn transcode_to_fmp4( // Publishing complete boxes ensures MSE can consume them directly. read_and_publish_boxes(&mut stdout, &publishers, &mut abort, group_id).await?; - // 2. Wait for FFmpeg to exit let status = ffmpeg .wait() .await @@ -128,7 +117,7 @@ pub(crate) async fn transcode_to_fmp4( if !status.success() { tracing::warn!("ffmpeg exited with status {status}; stderr: {stderr_output}"); - // Still return Ok — EOF is EOF even if ffmpeg reported an error at the end + // Still return Ok: EOF is EOF even if ffmpeg reported an error at the end } tracing::info!(status = %status, stderr_len = %stderr_output.len(), "ffmpeg pipeline finished"); @@ -136,10 +125,6 @@ pub(crate) async fn transcode_to_fmp4( Ok(()) } -// --------------------------------------------------------------------------- -// fMP4 box parsing -// --------------------------------------------------------------------------- - /// Read fMP4 boxes from FFmpeg stdout, buffer moof+mdat pairs, and publish /// each complete segment as a single MoQ object. /// @@ -163,7 +148,6 @@ async fn read_and_publish_boxes( let mut pending_moof: Option> = None; loop { - // Read 8-byte box header let mut header = [0u8; 8]; tokio::select! { _ = abort.changed() => return Err(TranscodeError::Aborted), @@ -229,7 +213,7 @@ async fn read_box_payload( let size = u32::from_be_bytes([header[0], header[1], header[2], header[3]]); if size == 0 { - // Last box — read all remaining data. + // Last box: read all remaining data. let mut remaining = Vec::new(); reader.read_to_end(&mut remaining).await?; let mut full = header.to_vec(); @@ -261,10 +245,6 @@ async fn read_box_payload( Ok(full) } -// --------------------------------------------------------------------------- -// Formatting -// --------------------------------------------------------------------------- - /// Format a duration in seconds to a human-readable `M:SS` or `H:MM:SS` string. pub(crate) fn format_duration(total_secs: f64) -> String { let total = total_secs as u64; @@ -279,10 +259,6 @@ pub(crate) fn format_duration(total_secs: f64) -> String { } } -// --------------------------------------------------------------------------- -// Tests -// --------------------------------------------------------------------------- - #[cfg(test)] mod tests { use super::*; @@ -323,26 +299,23 @@ mod tests { let mut slice: &[u8] = &input; read_and_publish_boxes(&mut slice, &publishers, &mut abort_rx, 0) .await - .expect("read_and_publish_boxes should succeed on valid fMP4"); + .unwrap(); - // Init is cached (for late joiners) AND published to broadcast (for track changes). let cached = publishers.get_init_segment(); - assert!(cached.is_some(), "init cache should contain ftyp+moov"); + assert!(cached.is_some()); let (cached_group, cached_data) = cached.unwrap(); assert_eq!(cached_group, 0); let mut expected_init = ftyp.clone(); expected_init.extend_from_slice(&moov); assert_eq!(cached_data.to_vec(), expected_init); - // Object 0: init (ftyp+moov) published to broadcast. - let init_obj = rx.recv().await.expect("should receive init via broadcast"); + let init_obj = rx.recv().await.unwrap(); assert_eq!(init_obj.object_id, 0); let mut expected_init_broadcast = ftyp.clone(); expected_init_broadcast.extend_from_slice(&moov); assert_eq!(init_obj.payload.to_vec(), expected_init_broadcast); - // Object 1: moof+mdat combined. - let obj1 = rx.recv().await.expect("should receive combined moof+mdat"); + let obj1 = rx.recv().await.unwrap(); assert_eq!(obj1.object_id, 1); let mut expected_media = moof.clone(); expected_media.extend_from_slice(&mdat); @@ -350,70 +323,45 @@ mod tests { } #[tokio::test] - async fn test_init_broadcast_and_cache() { + async fn test_init_cache_has_group_id() { let ftyp = make_box(b"ftyp", b"iso5"); let moov = make_box(b"moov", b"moovdata"); - let moof = make_box(b"moof", b"moofdata"); - let mdat = make_box(b"mdat", b"datadata"); let mut input = Vec::new(); input.extend_from_slice(&ftyp); input.extend_from_slice(&moov); - input.extend_from_slice(&moof); - input.extend_from_slice(&mdat); let publishers = TrackPublishers::new(); - let mut rx = publishers.video.subscribe(); let (_tx, mut abort_rx) = tokio::sync::watch::channel(false); let mut slice: &[u8] = &input; - read_and_publish_boxes(&mut slice, &publishers, &mut abort_rx, 0) + read_and_publish_boxes(&mut slice, &publishers, &mut abort_rx, 42) .await - .expect("read_and_publish_boxes should succeed"); + .unwrap(); - // Init is both cached and broadcast. let cached = publishers.get_init_segment(); - assert!(cached.is_some(), "init cache should contain ftyp+moov"); - - // Object 0: init via broadcast. - let init = rx.recv().await.expect("should receive init via broadcast"); - assert_eq!(init.object_id, 0); - assert_eq!(init.group_id, 0); - let mut expected_init = ftyp.clone(); - expected_init.extend_from_slice(&moov); - assert_eq!(init.payload.to_vec(), expected_init); - - // Object 1: moof+mdat media segment. - let media = rx.recv().await.expect("should receive media segment"); - assert_eq!(media.object_id, 1); - let mut expected_media = moof.clone(); - expected_media.extend_from_slice(&mdat); - assert_eq!(media.payload.to_vec(), expected_media); + assert!(cached.is_some()); + let (group_id, _data) = cached.unwrap(); + assert_eq!(group_id, 42); } #[tokio::test] - async fn test_init_cache_has_group_id() { - let ftyp = make_box(b"ftyp", b"iso5"); - let moov = make_box(b"moov", b"moovdata"); - - let mut input = Vec::new(); - input.extend_from_slice(&ftyp); - input.extend_from_slice(&moov); - + async fn test_empty_input_no_crash() { let publishers = TrackPublishers::new(); let (_tx, mut abort_rx) = tokio::sync::watch::channel(false); - - let mut slice: &[u8] = &input; - read_and_publish_boxes(&mut slice, &publishers, &mut abort_rx, 42) + let mut empty: &[u8] = &[]; + read_and_publish_boxes(&mut empty, &publishers, &mut abort_rx, 0) .await - .expect("read_and_publish_boxes should succeed"); + .unwrap(); + } - let cached = publishers.get_init_segment(); - assert!(cached.is_some(), "init should be cached"); - let (group_id, _data) = cached.unwrap(); - assert_eq!( - group_id, 42, - "group_id in cache should match the input group_id" - ); + #[tokio::test] + async fn test_partial_box_returns_error() { + let publishers = TrackPublishers::new(); + let (_tx, mut abort_rx) = tokio::sync::watch::channel(false); + let mut partial: &[u8] = &[0, 0, 0, 100, b'f', b't', b'y', b'p']; + read_and_publish_boxes(&mut partial, &publishers, &mut abort_rx, 0) + .await + .unwrap_err(); } } diff --git a/src/names.rs b/src/names.rs index e059704..6e0d1e4 100644 --- a/src/names.rs +++ b/src/names.rs @@ -337,3 +337,30 @@ const NOUNS: &[&str] = &[ "yacht", "zebra", ]; + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn test_generate_room_name_is_deterministic() { + let name1 = generate_room_name("12345"); + let name2 = generate_room_name("12345"); + assert_eq!(name1, name2); + } + + #[test] + fn test_generate_room_name_differs_for_diff_ids() { + let name1 = generate_room_name("11111"); + let name2 = generate_room_name("22222"); + assert_ne!(name1, name2); + } + + #[test] + fn test_generate_room_name_has_hyphen() { + let name = generate_room_name("99999"); + assert!(name.contains('-'), "room name should be adjective-noun"); + let parts: Vec<&str> = name.split('-').collect(); + assert_eq!(parts.len(), 2); + } +} diff --git a/src/playback.rs b/src/playback.rs index 1d5eefc..d0ba15d 100644 --- a/src/playback.rs +++ b/src/playback.rs @@ -1,17 +1,8 @@ //! Playback lifecycle management. //! //! The [`Player`] struct owns the abort handle for the currently-running -//! transcoding pipeline. It does **not** track what's playing — that's the +//! transcoding pipeline. It does **not** track what is playing: that is the //! [`PlaybackState`](crate::state::PlaybackState) machine's job. -//! -//! # Lifecycle -//! -//! 1. **Start**: [`Player::start`] receives a fully-formed [`ActiveTrackInfo`], -//! resolves the URL via the source registry, and spawns ffmpeg transcoding. -//! 2. **End**: The spawned task sends [`RoomCommand::TrackEnded`] when ffmpeg -//! finishes or errors. The room actor calls [`Player::on_track_ended`] to -//! acknowledge. -//! 3. **Skip**: [`Player::abort`] signals the transcoding task to stop. use std::path::PathBuf; use std::sync::Arc; @@ -26,7 +17,7 @@ use crate::types::{ActiveTrackInfo, RoomCommand}; /// Manages the abort handle for the currently-running transcoding pipeline. /// -/// The player is **not** responsible for queue or state management — that +/// The player is **not** responsible for queue or state management: that /// belongs to the [`PlaybackState`](crate::state::PlaybackState) machine in the /// room actor. pub(crate) struct Player { @@ -69,7 +60,6 @@ impl Player { let pipeline_tx = self.pipeline_tx.clone(); tokio::spawn(async move { - // 1. Resolve URL to a playable stream. let stream_url = match source_registry.resolve(&url, &cache_dir).await { Ok(url) => url, Err(e) => { @@ -79,7 +69,6 @@ impl Player { } }; - // 2. Transcode to fMP4 via ffmpeg. let group_id = track_id as u64; let result = media::transcode_to_fmp4(&stream_url, publishers, abort_rx, group_id).await; @@ -109,6 +98,7 @@ impl Player { /// /// Returns `true` if `item_id` matched the currently-tracked pipeline, /// `false` if it was a stale event from an older pipeline. + #[must_use] pub(crate) fn on_track_ended(&mut self, item_id: i64) -> bool { if self.track_id == Some(item_id) { self.track_id = None; @@ -126,20 +116,12 @@ async fn send_track_ended(tx: &mpsc::Sender, item_id: i64) { } } -// --------------------------------------------------------------------------- -// Tests -// --------------------------------------------------------------------------- - #[cfg(test)] mod tests { use super::*; - use crate::db; - use crate::playlist::Playlist; - use crate::types::TrackMeta; - use sqlx::sqlite::SqlitePoolOptions; - use sqlx::Pool; use std::path::Path; use std::sync::Arc; + use std::sync::atomic::{AtomicI64, Ordering}; struct MockPlayerSource; @@ -149,7 +131,7 @@ mod tests { &self, url: &str, _cache_dir: &Path, - ) -> Result { + ) -> Result { Err(crate::source::SourceError::Unsupported(url.into())) } @@ -168,31 +150,9 @@ mod tests { Arc::new(reg) } - async fn test_pool() -> Pool { - let pool = SqlitePoolOptions::new() - .max_connections(1) - .connect(":memory:") - .await - .unwrap(); - db::run_migrations(&pool).await.unwrap(); - db::room_insert(&pool, "test-room", "test-name", false) - .await - .unwrap(); - pool - } - - async fn make_queue_item(pool: &Pool) -> i64 { - let pl = Playlist::new(pool.clone(), "test-room"); - pl.push(&TrackMeta { - title: "Test".into(), - duration: "3:45".into(), - thumbnail: None, - url: "https://example.com/track".into(), - source: crate::types::SourceKind::Direct, - }) - .await - .unwrap() - .id + fn make_item_id() -> i64 { + static NEXT_ID: AtomicI64 = AtomicI64::new(1); + NEXT_ID.fetch_add(1, Ordering::Relaxed) } fn test_info(id: i64) -> ActiveTrackInfo { @@ -208,8 +168,7 @@ mod tests { #[tokio::test] async fn test_player_start_then_abort() { - let pool = test_pool().await; - let item_id = make_queue_item(&pool).await; + let item_id = make_item_id(); let (tx, _rx) = mpsc::channel(256); let mut player = Player::new(test_registry(), PathBuf::from("/tmp"), tx); @@ -217,35 +176,33 @@ mod tests { player.abort(); - // on_track_ended with a stale id should return false. assert!(!player.on_track_ended(99999)); } #[tokio::test] async fn test_player_abort_idempotent() { - let pool = test_pool().await; - let item_id = make_queue_item(&pool).await; + let item_id = make_item_id(); let (tx, _rx) = mpsc::channel(256); let mut player = Player::new(test_registry(), PathBuf::from("/tmp"), tx); player.start(&test_info(item_id), TrackPublishers::new()); player.abort(); - player.abort(); // second abort is a no-op - // Did not panic = pass + player.abort(); + assert!(!player.on_track_ended(99999)); } #[tokio::test] async fn test_player_abort_when_idle() { let (tx, _rx) = mpsc::channel(256); let mut player = Player::new(test_registry(), PathBuf::from("/tmp"), tx); - player.abort(); // no-op + player.abort(); + assert!(!player.on_track_ended(99999)); } #[tokio::test] async fn test_player_on_track_ended_matches_current() { - let pool = test_pool().await; - let item_id = make_queue_item(&pool).await; + let item_id = make_item_id(); let (tx, _rx) = mpsc::channel(256); let mut player = Player::new(test_registry(), PathBuf::from("/tmp"), tx); @@ -256,14 +213,12 @@ mod tests { #[tokio::test] async fn test_player_on_track_ended_ignores_old_id() { - let pool = test_pool().await; - let item_id = make_queue_item(&pool).await; + let item_id = make_item_id(); let (tx, _rx) = mpsc::channel(256); let mut player = Player::new(test_registry(), PathBuf::from("/tmp"), tx); player.start(&test_info(item_id), TrackPublishers::new()); - // Stale id should be ignored. assert!(!player.on_track_ended(99999)); } } diff --git a/src/playlist.rs b/src/playlist.rs index b925215..06c28cf 100644 --- a/src/playlist.rs +++ b/src/playlist.rs @@ -1,18 +1,9 @@ -//! Ordered track list for a room — upcoming queue and played history. -//! -//! Wraps SQLite queue operations with room-scoped queries, providing -//! push, pop-next, upcoming-list, and history-list operations. +//! Ordered track list for a room: upcoming queue and played history. use sqlx::{Pool, Sqlite}; use crate::types::TrackMeta; -fn now_iso() -> String { - chrono::Utc::now() - .format("%Y-%m-%dT%H:%M:%S%.3fZ") - .to_string() -} - /// A single queue item, returned from all Playlist operations. #[derive(Debug, Clone, sqlx::FromRow, serde::Serialize)] pub(crate) struct QueueItem { @@ -62,7 +53,7 @@ impl Playlist { .bind(meta.source.to_string()) .bind(None::) // added_by .bind(position) - .bind(now_iso()) + .bind(crate::util::now_iso()) .execute(&self.pool) .await?; @@ -79,44 +70,6 @@ impl Playlist { Ok(item) } - /// Atomically fetch and mark the first unplayed item as playing. - /// - /// Returns the item with `started_playing_at` set, or `None` when the - /// queue is empty. Unlike the old `pop_next`, this does NOT set - /// `played = 1` — that happens later via `mark_finished`. - #[allow(dead_code)] - pub(crate) async fn pop_next(&self) -> Result, sqlx::Error> { - let mut tx = self.pool.begin().await?; - - let item = sqlx::query_as::<_, QueueItem>( - "SELECT id, url, title, duration, thumbnail, source, position, played, played_at, started_playing_at - FROM queue_items WHERE room_id = ? AND played = 0 AND started_playing_at IS NULL - ORDER BY position ASC LIMIT 1", - ) - .bind(&self.room_id) - .fetch_optional(&mut *tx) - .await?; - - let result = if let Some(item) = item { - let now = now_iso(); - sqlx::query("UPDATE queue_items SET started_playing_at = ? WHERE id = ?") - .bind(&now) - .bind(item.id) - .execute(&mut *tx) - .await?; - - Some(QueueItem { - started_playing_at: Some(now), - ..item - }) - } else { - None - }; - - tx.commit().await?; - Ok(result) - } - /// List queued items (not playing, not finished) ordered by position ASC. pub(crate) async fn upcoming(&self) -> Result, sqlx::Error> { let items = sqlx::query_as::<_, QueueItem>( @@ -155,7 +108,7 @@ impl Playlist { /// Mark a currently-playing item as finished (moves it from "playing" to "played"). pub(crate) async fn mark_finished(&self, id: i64) -> Result<(), sqlx::Error> { - let now = now_iso(); + let now = crate::util::now_iso(); sqlx::query( "UPDATE queue_items SET played = 1, played_at = ?, started_playing_at = NULL WHERE id = ?", ) @@ -204,8 +157,9 @@ mod tests { .await .unwrap(); db::run_migrations(&pool).await.unwrap(); - // Create a room row so foreign-key constraints are satisfied. - db::room_insert(&pool, "test-room", "test-name", false).await.unwrap(); + db::room_insert(&pool, "test-room", "test-name", false) + .await + .unwrap(); Playlist::new(pool, "test-room") } @@ -240,69 +194,6 @@ mod tests { assert!(!upcoming[1].played); } - #[tokio::test] - async fn test_pop_next() { - let pl = test_playlist().await; - - pl.push(&track("First")).await.unwrap(); - pl.push(&track("Second")).await.unwrap(); - - let popped = pl.pop_next().await.unwrap().expect("should pop first"); - assert_eq!(popped.title, "First"); - assert!(!popped.played, "pop_next should NOT mark as played"); - assert!( - popped.played_at.is_none(), - "pop_next should NOT set played_at" - ); - assert!( - popped.started_playing_at.is_some(), - "pop_next should set started_playing_at" - ); - - // Upcoming should exclude items that have started playing. - let upcoming = pl.upcoming().await.unwrap(); - assert_eq!(upcoming.len(), 1); - assert_eq!(upcoming[0].title, "Second"); - } - - #[tokio::test] - async fn test_history() { - let pl = test_playlist().await; - - pl.push(&track("A")).await.unwrap(); - pl.push(&track("B")).await.unwrap(); - pl.push(&track("C")).await.unwrap(); - - let a = pl.pop_next().await.unwrap().unwrap(); - let b = pl.pop_next().await.unwrap().unwrap(); - - // Items only appear in history after mark_finished. - let history = pl.history(10).await.unwrap(); - assert_eq!( - history.len(), - 0, - "items should not appear in history until mark_finished" - ); - - pl.mark_finished(a.id).await.unwrap(); - pl.mark_finished(b.id).await.unwrap(); - - let history = pl.history(10).await.unwrap(); - assert_eq!(history.len(), 2); - // Most recently played first. - assert_eq!(history[0].title, "B"); - assert_eq!(history[1].title, "A"); - assert!(history[0].played); - assert!(history[0].played_at.is_some()); - } - - #[tokio::test] - async fn test_pop_empty() { - let pl = test_playlist().await; - let popped = pl.pop_next().await.unwrap(); - assert!(popped.is_none()); - } - #[tokio::test] async fn test_push_positions() { let pl = test_playlist().await; @@ -319,82 +210,86 @@ mod tests { #[tokio::test] async fn test_mark_finished_non_existent_id() { let pl = test_playlist().await; - // Should not panic or error. - let result = pl.mark_finished(99999).await; - assert!( - result.is_ok(), - "mark_finished on non-existent id should be Ok" - ); + pl.mark_finished(99999).await.unwrap(); } #[tokio::test] - async fn test_pop_next_returns_none_when_all_played() { + async fn test_upcoming_excludes_playing_items() { let pl = test_playlist().await; - pl.push(&track("A")).await.unwrap(); - pl.pop_next().await.unwrap(); - let next = pl.pop_next().await.unwrap(); - assert!( - next.is_none(), - "pop_next should return None when queue is empty" - ); + let a = pl.push(&track("A")).await.unwrap(); + pl.push(&track("B")).await.unwrap(); + + let now = chrono::Utc::now() + .format("%Y-%m-%dT%H:%M:%S%.3fZ") + .to_string(); + pl.mark_started(a.id, &now).await.unwrap(); + + let upcoming = pl.upcoming().await.unwrap(); + assert_eq!(upcoming.len(), 1); + assert_eq!(upcoming[0].title, "B"); } #[tokio::test] - async fn test_upcoming_excludes_playing_items() { + async fn test_history() { let pl = test_playlist().await; - pl.push(&track("A")).await.unwrap(); + + let a = pl.push(&track("A")).await.unwrap(); pl.push(&track("B")).await.unwrap(); + let c = pl.push(&track("C")).await.unwrap(); - // Pop A (marks as playing, not finished). - let a = pl.pop_next().await.unwrap().unwrap(); - assert_eq!(a.title, "A"); - assert!(!a.played, "popped item should not be marked played"); - assert!( - a.started_playing_at.is_some(), - "popped item should have started_playing_at" - ); + let now_a = chrono::Utc::now() + .format("%Y-%m-%dT%H:%M:%S%.3fZ") + .to_string(); + pl.mark_started(a.id, &now_a).await.unwrap(); + pl.mark_finished(a.id).await.unwrap(); - // Upcoming should only have B. - let upcoming = pl.upcoming().await.unwrap(); - assert_eq!(upcoming.len(), 1, "upcoming should exclude playing item"); - assert_eq!(upcoming[0].title, "B"); + tokio::time::sleep(std::time::Duration::from_millis(5)).await; + let now_c = chrono::Utc::now() + .format("%Y-%m-%dT%H:%M:%S%.3fZ") + .to_string(); + pl.mark_started(c.id, &now_c).await.unwrap(); + pl.mark_finished(c.id).await.unwrap(); + + let history = pl.history(10).await.unwrap(); + assert_eq!(history.len(), 2); + assert_eq!(history[0].title, "C"); + assert_eq!(history[1].title, "A"); + assert!(history[0].played); + assert!(history[0].played_at.is_some()); } #[tokio::test] async fn test_history_only_after_mark_finished() { let pl = test_playlist().await; - pl.push(&track("A")).await.unwrap(); - let a = pl.pop_next().await.unwrap().unwrap(); + let a = pl.push(&track("A")).await.unwrap(); + + let now = chrono::Utc::now() + .format("%Y-%m-%dT%H:%M:%S%.3fZ") + .to_string(); + pl.mark_started(a.id, &now).await.unwrap(); - // A is playing — should NOT be in history. let history = pl.history(10).await.unwrap(); - assert!( - history.is_empty(), - "playing items should not appear in history" - ); + assert!(history.is_empty()); - // Mark as finished. pl.mark_finished(a.id).await.unwrap(); - // Now A should be in history. let history = pl.history(10).await.unwrap(); assert_eq!(history.len(), 1); assert_eq!(history[0].title, "A"); assert!(history[0].played); - assert!( - history[0].played_at.is_some(), - "finished items should have played_at" - ); + assert!(history[0].played_at.is_some()); } #[tokio::test] async fn test_push_after_mark_finished() { let pl = test_playlist().await; - pl.push(&track("A")).await.unwrap(); - let a = pl.pop_next().await.unwrap().unwrap(); + let a = pl.push(&track("A")).await.unwrap(); + let now = chrono::Utc::now() + .format("%Y-%m-%dT%H:%M:%S%.3fZ") + .to_string(); + pl.mark_started(a.id, &now).await.unwrap(); pl.mark_finished(a.id).await.unwrap(); - // Push another track pl.push(&track("B")).await.unwrap(); let upcoming = pl.upcoming().await.unwrap(); @@ -409,19 +304,28 @@ mod tests { #[tokio::test] async fn test_history_order_multiple_finished() { let pl = test_playlist().await; - pl.push(&track("A")).await.unwrap(); - pl.push(&track("B")).await.unwrap(); - pl.push(&track("C")).await.unwrap(); + let a = pl.push(&track("A")).await.unwrap(); + let b = pl.push(&track("B")).await.unwrap(); + let c = pl.push(&track("C")).await.unwrap(); - let a = pl.pop_next().await.unwrap().unwrap(); - tokio::time::sleep(std::time::Duration::from_millis(5)).await; + let now_a = chrono::Utc::now() + .format("%Y-%m-%dT%H:%M:%S%.3fZ") + .to_string(); + pl.mark_started(a.id, &now_a).await.unwrap(); pl.mark_finished(a.id).await.unwrap(); - let b = pl.pop_next().await.unwrap().unwrap(); tokio::time::sleep(std::time::Duration::from_millis(5)).await; + let now_b = chrono::Utc::now() + .format("%Y-%m-%dT%H:%M:%S%.3fZ") + .to_string(); + pl.mark_started(b.id, &now_b).await.unwrap(); pl.mark_finished(b.id).await.unwrap(); - let c = pl.pop_next().await.unwrap().unwrap(); + tokio::time::sleep(std::time::Duration::from_millis(5)).await; + let now_c = chrono::Utc::now() + .format("%Y-%m-%dT%H:%M:%S%.3fZ") + .to_string(); + pl.mark_started(c.id, &now_c).await.unwrap(); pl.mark_finished(c.id).await.unwrap(); let history = pl.history(10).await.unwrap(); diff --git a/src/room.rs b/src/room.rs index 9e7d5e5..3c62a55 100644 --- a/src/room.rs +++ b/src/room.rs @@ -4,12 +4,11 @@ //! SQLite, and holds ephemeral playback state in memory via a pure //! [`PlaybackState`] machine. //! -//! All mutations flow through a per-room mpsc event-loop actor, -//! following the impure-pure-impure sandwich: +//! All mutations flow through a per-room mpsc event-loop actor: //! -//! 1. **Gather** — fetch data (DB, wall clock) -//! 2. **Transition** — `PlaybackState::transition()` (pure) -//! 3. **Commit** — execute returned [`Effect`]s +//! 1. Gather: fetch data (DB, wall clock) +//! 2. Transition: `PlaybackState::transition()` (pure) +//! 3. Commit: execute returned [`Effect`]s use std::collections::HashMap; use std::path::PathBuf; @@ -32,17 +31,16 @@ use crate::source::SourceRegistry; use crate::state::{self, Effect, Event, PlaybackState}; use crate::transport::TrackPublishers; use crate::types::{ - ChatMessage, HistoryEntry, QueueSummary, RoomCommand, RoomHandle, RoomId, TrackMeta, - TrackState, + ChatMessage, HistoryEntry, QueueSummary, RoomCommand, RoomHandle, RoomId, TrackMeta, TrackState, }; /// Distributed unique ID generator (Discord-style snowflake). /// /// Bit layout: -/// 1 bit — unused (always 0) -/// 41 bits — timestamp in milliseconds since custom epoch -/// 10 bits — node ID (0–1023) -/// 12 bits — sequence number (0–4095) +/// 1 bit unused (always 0) +/// 41 bits timestamp in milliseconds since custom epoch +/// 10 bits node ID (0-1023) +/// 12 bits sequence number (0-4095) /// /// Thread-safety is provided by a single `AtomicU64` that packs /// `(timestamp_ms << 12) | sequence`, allowing lock-free CAS updates. @@ -51,7 +49,7 @@ pub(crate) struct SnowflakeIdGen { state: AtomicU64, // (timestamp_ms << 12) | sequence (12 bits) } -const CUSTOM_EPOCH_MS: u64 = 173_568_960_0000; // 2025-01-01T00:00:00Z +const CUSTOM_EPOCH_MS: u64 = 1_735_689_600_000; // 2025-01-01T00:00:00Z impl SnowflakeIdGen { pub(crate) fn new(node_id: u16) -> Self { @@ -67,6 +65,7 @@ impl SnowflakeIdGen { /// Thread-safe. Under extreme load (≥4096 IDs in the same /// millisecond) the generator yields to the tokio runtime until /// the next millisecond tick. + #[must_use] pub(crate) async fn next_id(&self) -> u64 { loop { let now = SystemTime::now() @@ -89,8 +88,6 @@ impl SnowflakeIdGen { { return (ts << 22) | ((self.node_id as u64) << 12); } - // CAS lost — another thread already advanced or incremented. - // Retry with fresh state. continue; } @@ -105,7 +102,6 @@ impl SnowflakeIdGen { { return (last_ts << 22) | ((self.node_id as u64) << 12) | seq as u64; } - // CAS lost — another thread claimed this seq first; retry. continue; } @@ -124,7 +120,7 @@ struct RoomActor { pool: Pool, source_registry: Arc, cache_dir: PathBuf, - /// Clone of the command sender — used by spawned extraction tasks to + /// Clone of the command sender, used by spawned extraction tasks to /// deliver MetadataReady / MetadataFailed back to the actor. cmd_tx: mpsc::Sender, @@ -168,7 +164,7 @@ impl Registry { } /// Create a new room with a snowflake ID, persist to SQLite. - pub(crate) async fn create(&self) -> RoomRef { + pub(crate) async fn create(&self) -> RoomId { let id = RoomId(self.id_gen.next_id().await.to_string()); let room_name = generate_room_name(&id.0); let now = Utc::now(); @@ -178,11 +174,7 @@ impl Registry { let (tx, rx) = mpsc::channel(256); let (last_active_tx, last_active_rx) = watch::channel(now); - let player = Player::new( - self.sources.clone(), - self.cache_dir.clone(), - tx.clone(), - ); + let player = Player::new(self.sources.clone(), self.cache_dir.clone(), tx.clone()); let actor = RoomActor { rx, room_id: id.clone(), @@ -209,10 +201,7 @@ impl Registry { }, ); - RoomRef { - id, - _registry: self.clone(), - } + id } /// Get the room handle if the room is active. @@ -221,6 +210,7 @@ impl Registry { } /// Check whether a room exists in SQLite. + #[must_use] pub(crate) async fn exists(&self, id: &RoomId) -> bool { db::room_exists(&self.pool, &id.0).await.unwrap_or(false) } @@ -236,41 +226,22 @@ impl Registry { duration: i.duration, thumbnail: i.thumbnail, url: i.url, - source: i - .source - .parse() - .unwrap_or_else(|s: String| { - tracing::warn!(source = %s, "unknown source kind in queue_items"); - crate::types::SourceKind::Direct - }), + source: i.source.parse().unwrap_or_else(|s: String| { + tracing::warn!(source = %s, "unknown source kind in queue_items"); + crate::types::SourceKind::Direct + }), }) .collect() } - /// Push a track onto a room's queue (persisted to SQLite via Playlist). - #[allow(dead_code)] // only used by tests - pub(crate) async fn push(&self, id: &RoomId, meta: TrackMeta) { - if let Some(handle) = self.handle(id).await { - let _ = handle.cmd_tx.send(RoomCommand::QueueTrack(meta)).await; - } - } - /// Push a raw URL onto a room's queue. Metadata extraction (yt-dlp) runs - /// asynchronously in a spawned task — the caller returns immediately. + /// asynchronously in a spawned task. The caller returns immediately. pub(crate) async fn push_url(&self, id: &RoomId, url: String) { - let t0 = std::time::Instant::now(); let handle = match self.handle(id).await { Some(h) => h, None => return, }; - let t1 = t0.elapsed(); let _ = handle.cmd_tx.send(RoomCommand::QueueUrl(url)).await; - let t2 = t0.elapsed(); - tracing::info!( - room = %id, - "push_url timing: handle={:.0?} send={:.0?} total={:.0?}", - t1, t2 - t1, t2 - ); } /// Remove a room (from SQLite and in-memory state). @@ -278,7 +249,7 @@ impl Registry { pub(crate) async fn remove(&self, id: &RoomId) { if let Some(handle) = self.handle(id).await { let _ = tokio::time::timeout( - Duration::from_secs(1), + Duration::from_secs(1), // generous 1s window for graceful shutdown handle.cmd_tx.send(RoomCommand::Shutdown), ) .await; @@ -397,7 +368,7 @@ impl Registry { for id in &idle_rooms { if let Some(handle) = self.handle(id).await { if tokio::time::timeout( - Duration::from_secs(1), + Duration::from_secs(1), // generous 1s window for graceful shutdown handle.cmd_tx.send(RoomCommand::Shutdown), ) .await @@ -416,12 +387,6 @@ impl Registry { } } -/// An owned handle to a room, returned by [`Registry::create`]. -pub(crate) struct RoomRef { - pub id: RoomId, - _registry: Registry, -} - impl RoomActor { /// Run the event loop. Returns when rx is closed or Shutdown received. async fn run(mut self) { @@ -431,7 +396,6 @@ impl RoomActor { break; } } - // On shutdown, abort any active pipeline. self.player.abort(); tracing::info!(room = %self.room_id, "room actor shut down"); } @@ -439,47 +403,15 @@ impl RoomActor { /// Process a single command. /// /// Returns `true` if the actor should shut down. + #[must_use] async fn process(&mut self, cmd: RoomCommand) -> bool { match cmd { - RoomCommand::QueueTrack(meta) => { - tracing::debug!(room = %self.room_id, title = %meta.title, "track queued"); - - // ── Gather ────────────────────────────────────────── - let playlist = Playlist::new(self.pool.clone(), &self.room_id.0); - let item = match playlist.push(&meta).await { - Ok(item) => item, - Err(e) => { - tracing::warn!("failed to persist queue item: {e}"); - return false; - } - }; - let now = Utc::now().timestamp_millis(); - - // ── Transition (pure) ─────────────────────────────── - let effects = self.state.transition( - &Event::TrackQueued(state::QueuedTrack { - id: item.id, - title: item.title, - url: item.url, - duration: item.duration, - thumbnail: item.thumbnail, - pending: false, - }), - now, - ); - - // ── Commit ────────────────────────────────────────── - self.execute_effects(effects).await; - } - RoomCommand::QueueUrl(url) => { - let t0 = std::time::Instant::now(); tracing::debug!(room = %self.room_id, %url, "track url queued"); - // ── Gather ────────────────────────────────────────── let playlist = Playlist::new(self.pool.clone(), &self.room_id.0); let placeholder = TrackMeta { - title: "Loading…".into(), + title: "Loading...".into(), duration: "--:--".into(), thumbnail: None, url: url.clone(), @@ -492,10 +424,8 @@ impl RoomActor { return false; } }; - let t1 = t0.elapsed(); let now = Utc::now().timestamp_millis(); - // ── Transition (pure) ─────────────────────────────── let effects = self.state.transition( &Event::TrackQueued(state::QueuedTrack { id: item.id, @@ -507,13 +437,9 @@ impl RoomActor { }), now, ); - let t2 = t0.elapsed(); - // ── Commit ────────────────────────────────────────── self.execute_effects(effects).await; - let t3 = t0.elapsed(); - // ── Spawn async extraction (non-blocking) ─────────── let item_id = item.id; let cmd_tx = self.cmd_tx.clone(); let sources = self.source_registry.clone(); @@ -535,27 +461,18 @@ impl RoomActor { } } }); - - tracing::info!( - room = %self.room_id, - "QueueUrl actor timing: db_push={:.0?} transition={:.0?} effects={:.0?} total={:.0?}", - t1, t2 - t1, t3 - t2, t3 - ); } RoomCommand::MetadataReady { item_id, meta } => { tracing::debug!(room = %self.room_id, item_id, title = %meta.title, "metadata ready"); - // ── Gather ────────────────────────────────────────── let now = Utc::now().timestamp_millis(); - // ── DB update (synchronous commit, actor-safe) ────── let playlist = Playlist::new(self.pool.clone(), &self.room_id.0); if let Err(e) = playlist.update_metadata(item_id, &meta).await { tracing::warn!(item_id, error = %e, "failed to persist metadata update"); } - // ── Transition (pure) ─────────────────────────────── let effects = self.state.transition( &Event::MetadataUpdated { item_id, @@ -566,19 +483,17 @@ impl RoomActor { now, ); - // ── Commit ────────────────────────────────────────── self.execute_effects(effects).await; } RoomCommand::MetadataFailed { item_id, error } => { tracing::debug!(room = %self.room_id, item_id, %error, "metadata extraction failed"); - // ── Transition (pure) — set error state in title ──── let now = Utc::now().timestamp_millis(); let effects = self.state.transition( &Event::MetadataUpdated { item_id, - title: format!("Failed to load — {error}"), + title: format!("Failed to load: {error}"), duration: "--:--".into(), thumbnail: None, }, @@ -590,31 +505,24 @@ impl RoomActor { RoomCommand::Skip => { tracing::info!(room = %self.room_id, "skip requested"); - // ── Gather ────────────────────────────────────────── let now = Utc::now().timestamp_millis(); - // ── Transition (pure) ─────────────────────────────── let effects = self.state.transition(&Event::Skip, now); - // ── Commit ────────────────────────────────────────── self.execute_effects(effects).await; } RoomCommand::TrackEnded { item_id } => { tracing::debug!(room = %self.room_id, item_id, "track ended"); - // Gate on player first — stale events are silently dropped. if !self.player.on_track_ended(item_id) { return false; } - // ── Gather ────────────────────────────────────────── let now = Utc::now().timestamp_millis(); - // ── Transition (pure) ─────────────────────────────── let effects = self.state.transition(&Event::TrackEnded { item_id }, now); - // ── Commit ────────────────────────────────────────── self.execute_effects(effects).await; } @@ -677,11 +585,9 @@ impl RoomActor { } Effect::PersistStarted(id) => { - let t0 = std::time::Instant::now(); let now = Utc::now().format("%Y-%m-%dT%H:%M:%S%.3fZ").to_string(); let playlist = Playlist::new(self.pool.clone(), &self.room_id.0); let _ = playlist.mark_started(id, &now).await; - tracing::debug!(room = %self.room_id, id, "PersistStarted: {:.0?}", t0.elapsed()); } Effect::StartPipeline(track) => { @@ -712,13 +618,14 @@ impl RoomActor { /// Publish the current state snapshot to all connected clients. /// - /// Reads from in-memory [`PlaybackState`] — no SQLite queries. + /// Reads from in-memory [`PlaybackState`], no SQLite queries. async fn publish_state_snapshot(&self) { let current = self.state.active.as_ref().map(|t| TrackState { id: t.id, title: t.title.clone(), url: t.url.clone(), duration: t.duration.clone(), + thumbnail: t.thumbnail.clone(), started_at: t.started_at_wall, }); @@ -774,17 +681,35 @@ mod tests { use super::*; use sqlx::sqlite::SqlitePoolOptions; - async fn test_pool() -> Pool { + async fn poll_until(check: F, timeout: Duration, label: &str) + where + F: Fn() -> Fut, + Fut: std::future::Future, + { + let start = std::time::Instant::now(); + loop { + if check().await { + return; + } + if start.elapsed() > timeout { + panic!("timed out: {label}"); + } + tokio::time::sleep(Duration::from_millis(10)).await; + } + } + + async fn test_registry_and_room() -> (Registry, RoomId) { let pool = SqlitePoolOptions::new() .max_connections(1) .connect(":memory:") .await .unwrap(); db::run_migrations(&pool).await.unwrap(); - pool + let reg = Registry::new(pool, PathBuf::from("/tmp/moqbox"), test_registry()); + let room_id = reg.create().await; + (reg, room_id) } - /// A mock source that handles all URLs. struct PanaceaSource; #[async_trait::async_trait] @@ -794,8 +719,9 @@ mod tests { url: &str, _cache_dir: &std::path::Path, ) -> Result { + let title = url.rsplit('/').next().unwrap_or("test"); Ok(TrackMeta { - title: "test".into(), + title: title.into(), duration: "3:45".into(), thumbnail: None, url: url.into(), @@ -844,64 +770,42 @@ mod tests { #[tokio::test] async fn test_create_and_exists() { - let pool = test_pool().await; - let reg = Registry::new(pool, PathBuf::from("/tmp/moqbox"), test_registry()); - let room = reg.create().await; - assert!(reg.exists(&room.id).await); + let (reg, room_id) = test_registry_and_room().await; + assert!(reg.exists(&room_id).await); } #[tokio::test] async fn test_push_and_queue() { - let pool = test_pool().await; - let reg = Registry::new(pool, PathBuf::from("/tmp/moqbox"), test_registry()); - let room = reg.create().await; - - let meta_a = TrackMeta { - title: "Track A".into(), - duration: "3:45".into(), - thumbnail: None, - url: "https://example.com/a".into(), - source: crate::types::SourceKind::Direct, - }; - let meta_b = TrackMeta { - title: "Track B".into(), - duration: "4:20".into(), - thumbnail: None, - url: "https://example.com/b".into(), - source: crate::types::SourceKind::Direct, - }; - - // Push first track — it gets popped and starts playing immediately. - reg.push(&room.id, meta_a.clone()).await; - // Push second track — stays in the upcoming queue. - reg.push(&room.id, meta_b.clone()).await; + let (reg, room_id) = test_registry_and_room().await; - // Give the actor time to process both commands. - tokio::time::sleep(Duration::from_millis(100)).await; + reg.push_url(&room_id, "https://example.com/a".into()).await; + reg.push_url(&room_id, "https://example.com/b".into()).await; - let queue = reg.queue(&room.id).await; - assert_eq!(queue.len(), 1, "only unplayed track should appear in queue"); - assert_eq!(queue[0].title, "Track B"); + poll_until( + || async { + let q = reg.queue(&room_id).await; + q.len() == 1 && q[0].title == "b" + }, + Duration::from_secs(2), + "queue should have 1 item with title 'b'", + ) + .await; } #[tokio::test] async fn test_remove() { - let pool = test_pool().await; - let reg = Registry::new(pool, PathBuf::from("/tmp/moqbox"), test_registry()); - let room = reg.create().await; + let (reg, room_id) = test_registry_and_room().await; - reg.remove(&room.id).await; - assert!(!reg.exists(&room.id).await); + reg.remove(&room_id).await; + assert!(!reg.exists(&room_id).await); } #[tokio::test] async fn test_send_chat() { - let pool = test_pool().await; - let reg = Registry::new(pool, PathBuf::from("/tmp/moqbox"), test_registry()); - let room = reg.create().await; + let (reg, room_id) = test_registry_and_room().await; let msg = reg - .send_chat(&room.id, "alice", "hello world", "message") + .send_chat(&room_id, "alice", "hello world", "message") .await .unwrap(); assert_eq!(msg.user_name, "alice"); @@ -912,23 +816,25 @@ mod tests { #[tokio::test] async fn test_send_chat_persistence() { - let pool = test_pool().await; - let reg = Registry::new(pool, PathBuf::from("/tmp/moqbox"), test_registry()); - let room = reg.create().await; + let (reg, room_id) = test_registry_and_room().await; let _ = reg - .send_chat(&room.id, "alice", "msg1", "message") + .send_chat(&room_id, "alice", "msg1", "message") .await .unwrap(); let _ = reg - .send_chat(&room.id, "bob", "msg2", "message") + .send_chat(&room_id, "bob", "msg2", "message") .await .unwrap(); - tokio::time::sleep(Duration::from_millis(100)).await; + poll_until( + || async { reg.recent_chat(&room_id, 10).await.unwrap().len() == 2 }, + Duration::from_secs(2), + "recent_chat should have 2 messages", + ) + .await; - let recent = reg.recent_chat(&room.id, 10).await.unwrap(); - assert_eq!(recent.len(), 2); + let recent = reg.recent_chat(&room_id, 10).await.unwrap(); assert_eq!(recent[0].user_name, "bob"); assert_eq!(recent[0].content, "msg2"); assert_eq!(recent[1].user_name, "alice"); @@ -937,510 +843,387 @@ mod tests { #[tokio::test] async fn test_send_chat_empty_content() { - let pool = test_pool().await; - let reg = Registry::new(pool, PathBuf::from("/tmp/moqbox"), test_registry()); - let room = reg.create().await; + let (reg, room_id) = test_registry_and_room().await; - let result = reg.send_chat(&room.id, "alice", "", "message").await; - assert!(result.is_err()); + let result = reg.send_chat(&room_id, "alice", "", "message").await; + result.unwrap_err(); } #[tokio::test] async fn test_send_chat_long_content() { - let pool = test_pool().await; - let reg = Registry::new(pool, PathBuf::from("/tmp/moqbox"), test_registry()); - let room = reg.create().await; + let (reg, room_id) = test_registry_and_room().await; let long = "a".repeat(2001); - let result = reg.send_chat(&room.id, "alice", &long, "message").await; - assert!(result.is_err()); + let result = reg.send_chat(&room_id, "alice", &long, "message").await; + result.unwrap_err(); } #[tokio::test] async fn test_send_chat_long_username() { - let pool = test_pool().await; - let reg = Registry::new(pool, PathBuf::from("/tmp/moqbox"), test_registry()); - let room = reg.create().await; + let (reg, room_id) = test_registry_and_room().await; let long = "a".repeat(33); - let result = reg.send_chat(&room.id, &long, "hello", "message").await; - assert!(result.is_err()); + let result = reg.send_chat(&room_id, &long, "hello", "message").await; + result.unwrap_err(); } #[tokio::test] async fn test_register_client() { - let pool = test_pool().await; - let reg = Registry::new(pool, PathBuf::from("/tmp/moqbox"), test_registry()); - let room = reg.create().await; + let (reg, room_id) = test_registry_and_room().await; - let id1 = reg.register_client(&room.id).await; + let id1 = reg.register_client(&room_id).await; assert_eq!(id1, Some(1)); - let id2 = reg.register_client(&room.id).await; + let id2 = reg.register_client(&room_id).await; assert_eq!(id2, Some(2)); } #[tokio::test] async fn test_register_client_nonexistent_room() { - let pool = test_pool().await; - let reg = Registry::new(pool, PathBuf::from("/tmp/moqbox"), test_registry()); + let (reg, _) = test_registry_and_room().await; let fake_id = RoomId("nonexistent".into()); assert_eq!(reg.register_client(&fake_id).await, None); } #[tokio::test] async fn test_unregister_client() { - let pool = test_pool().await; - let reg = Registry::new(pool, PathBuf::from("/tmp/moqbox"), test_registry()); - let room = reg.create().await; + let (reg, room_id) = test_registry_and_room().await; - reg.register_client(&room.id).await; - reg.unregister_client(&room.id).await; + reg.register_client(&room_id).await; + reg.unregister_client(&room_id).await; - let id = reg.register_client(&room.id).await; + let id = reg.register_client(&room_id).await; assert_eq!(id, Some(2)); } #[tokio::test] async fn test_skip_when_idle_does_not_panic() { - let pool = test_pool().await; - let reg = Registry::new(pool, PathBuf::from("/tmp/moqbox"), test_registry()); - let room = reg.create().await; - reg.skip(&room.id).await; + let (reg, room_id) = test_registry_and_room().await; + reg.skip(&room_id).await; } #[tokio::test] async fn test_double_skip_does_not_deadlock() { - let pool = test_pool().await; - let reg = Registry::new(pool, PathBuf::from("/tmp/moqbox"), test_registry()); - let room = reg.create().await; - - let meta = TrackMeta { - title: "A".into(), - duration: "3:45".into(), - thumbnail: None, - url: "https://example.com/a".into(), - source: crate::types::SourceKind::Direct, - }; - reg.push(&room.id, meta).await; - tokio::time::sleep(Duration::from_millis(100)).await; + let (reg, room_id) = test_registry_and_room().await; + + reg.push_url(&room_id, "https://example.com/a".into()).await; + poll_until( + || async { reg.queue(&room_id).await.is_empty() }, + Duration::from_secs(2), + "queue should be empty after push starts playing", + ) + .await; let r1 = reg.clone(); - let rid1 = room.id.clone(); + let rid1 = room_id.clone(); let h1 = tokio::spawn(async move { r1.skip(&rid1).await }); let r2 = reg.clone(); - let rid2 = room.id.clone(); + let rid2 = room_id.clone(); let h2 = tokio::spawn(async move { r2.skip(&rid2).await }); let _ = tokio::join!(h1, h2); } #[tokio::test] async fn test_send_chat_returns_id_zero() { - let pool = test_pool().await; - let reg = Registry::new(pool, PathBuf::from("/tmp/moqbox"), test_registry()); - let room = reg.create().await; + let (reg, room_id) = test_registry_and_room().await; let msg = reg - .send_chat(&room.id, "alice", "hello", "message") + .send_chat(&room_id, "alice", "hello", "message") .await .unwrap(); - assert_eq!(msg.id, 0, "send_chat should return id 0 in actor mode"); + assert_eq!(msg.id, 0); } #[tokio::test] async fn test_sweep_idle_removes_inactive_rooms() { - let pool = test_pool().await; - let reg = Registry::new(pool, PathBuf::from("/tmp/moqbox"), test_registry()); - let room = reg.create().await; + let (reg, room_id) = test_registry_and_room().await; tokio::time::sleep(Duration::from_millis(1)).await; let swept = reg.sweep_idle(Duration::from_secs(0)).await; - assert!(swept.contains(&room.id), "room should be swept immediately"); - assert!( - !reg.exists(&room.id).await, - "room should no longer exist after sweep" - ); + assert!(swept.contains(&room_id)); + assert!(!reg.exists(&room_id).await); } #[tokio::test] async fn test_sweep_idle_preserves_active_rooms() { - let pool = test_pool().await; - let reg = Registry::new(pool, PathBuf::from("/tmp/moqbox"), test_registry()); - let room = reg.create().await; + let (reg, room_id) = test_registry_and_room().await; - reg.register_client(&room.id).await; - tokio::time::sleep(Duration::from_millis(50)).await; + reg.register_client(&room_id).await; let swept = reg.sweep_idle(Duration::from_millis(100)).await; - assert!(!swept.contains(&room.id), "active room should not be swept"); + assert!(!swept.contains(&room_id)); } #[tokio::test] async fn test_push_starts_playback_when_idle() { - let pool = test_pool().await; - let reg = Registry::new(pool, PathBuf::from("/tmp/moqbox"), test_registry()); - let room = reg.create().await; - - let meta = TrackMeta { - title: "Test".into(), - duration: "3:45".into(), - thumbnail: None, - url: "https://example.com/t".into(), - source: crate::types::SourceKind::Direct, - }; - reg.push(&room.id, meta).await; - - tokio::time::sleep(Duration::from_millis(100)).await; + let (reg, room_id) = test_registry_and_room().await; - let queue = reg.queue(&room.id).await; - assert!( - queue.is_empty(), - "track should be popped when playback starts" - ); + reg.push_url(&room_id, "https://example.com/t".into()).await; + poll_until( + || async { reg.queue(&room_id).await.is_empty() }, + Duration::from_secs(2), + "queue should be empty (item started playing)", + ) + .await; } #[tokio::test] async fn test_push_appends_when_already_playing() { - let pool = test_pool().await; - let reg = Registry::new(pool, PathBuf::from("/tmp/moqbox"), test_registry()); - let room = reg.create().await; + let (reg, room_id) = test_registry_and_room().await; - reg.push( - &room.id, - TrackMeta { - title: "A".into(), - duration: "3:45".into(), - thumbnail: None, - url: "https://example.com/a".into(), - source: crate::types::SourceKind::Direct, - }, + reg.push_url(&room_id, "https://example.com/a".into()).await; + poll_until( + || async { reg.queue(&room_id).await.is_empty() }, + Duration::from_secs(2), + "first push should have started playing", ) .await; - tokio::time::sleep(Duration::from_millis(100)).await; - reg.push( - &room.id, - TrackMeta { - title: "B".into(), - duration: "3:45".into(), - thumbnail: None, - url: "https://example.com/b".into(), - source: crate::types::SourceKind::Direct, - }, + reg.push_url(&room_id, "https://example.com/b".into()).await; + poll_until( + || async { reg.queue(&room_id).await.len() == 1 }, + Duration::from_secs(2), + "second push should be queued", ) .await; - tokio::time::sleep(Duration::from_millis(100)).await; - let queue = reg.queue(&room.id).await; - assert_eq!(queue.len(), 1, "second track should be queued"); - assert_eq!(queue[0].title, "B"); + let queue = reg.queue(&room_id).await; + assert_eq!(queue[0].title, "b"); - // History should be empty (first track is still "playing", not finished). - let playlist = crate::playlist::Playlist::new(reg.pool.clone(), &room.id.0); + let playlist = crate::playlist::Playlist::new(reg.pool.clone(), &room_id.0); let history = playlist.history(10).await.unwrap(); - assert!( - history.is_empty(), - "no history yet — first track is still playing" - ); + assert!(history.is_empty()); } #[tokio::test] async fn test_skip_when_queue_empty_after_track_ends() { - let pool = test_pool().await; - let reg = Registry::new(pool, PathBuf::from("/tmp/moqbox"), test_registry()); - let room = reg.create().await; + let (reg, room_id) = test_registry_and_room().await; - reg.push( - &room.id, - TrackMeta { - title: "A".into(), - duration: "3:45".into(), - thumbnail: None, - url: "https://example.com/a".into(), - source: crate::types::SourceKind::Direct, - }, + reg.push_url(&room_id, "https://example.com/a".into()).await; + poll_until( + || async { reg.queue(&room_id).await.is_empty() }, + Duration::from_secs(2), + "first push should have started playing", ) .await; - tokio::time::sleep(Duration::from_millis(100)).await; - reg.skip(&room.id).await; - tokio::time::sleep(Duration::from_millis(100)).await; + reg.skip(&room_id).await; + poll_until( + || async { + let playlist = crate::playlist::Playlist::new(reg.pool.clone(), &room_id.0); + playlist.history(10).await.unwrap().len() == 1 + }, + Duration::from_secs(2), + "skip should move track to history", + ) + .await; - let queue = reg.queue(&room.id).await; - assert!( - queue.is_empty(), - "queue should be empty after skip with no next track" - ); + let queue = reg.queue(&room_id).await; + assert!(queue.is_empty()); } #[tokio::test] async fn test_double_skip_does_not_corrupt_state() { - let pool = test_pool().await; - let reg = Registry::new(pool, PathBuf::from("/tmp/moqbox"), test_registry()); - let room = reg.create().await; + let (reg, room_id) = test_registry_and_room().await; - reg.push( - &room.id, - TrackMeta { - title: "A".into(), - duration: "3:45".into(), - thumbnail: None, - url: "https://example.com/a".into(), - source: crate::types::SourceKind::Direct, - }, + reg.push_url(&room_id, "https://example.com/a".into()).await; + poll_until( + || async { reg.queue(&room_id).await.is_empty() }, + Duration::from_secs(2), + "first push started playing", ) .await; - tokio::time::sleep(Duration::from_millis(100)).await; - reg.push( - &room.id, - TrackMeta { - title: "B".into(), - duration: "3:45".into(), - thumbnail: None, - url: "https://example.com/b".into(), - source: crate::types::SourceKind::Direct, - }, + reg.push_url(&room_id, "https://example.com/b".into()).await; + poll_until( + || async { reg.queue(&room_id).await.len() == 1 }, + Duration::from_secs(2), + "second push queued", ) .await; - tokio::time::sleep(Duration::from_millis(50)).await; - reg.skip(&room.id).await; - reg.skip(&room.id).await; - tokio::time::sleep(Duration::from_millis(100)).await; + reg.skip(&room_id).await; + reg.skip(&room_id).await; + poll_until( + || async { + let playlist = crate::playlist::Playlist::new(reg.pool.clone(), &room_id.0); + let h = playlist.history(10).await.unwrap(); + !h.is_empty() && h[0].title == "b" + }, + Duration::from_secs(2), + "history should have 'b' as most recent", + ) + .await; - let playlist = crate::playlist::Playlist::new(reg.pool.clone(), &room.id.0); + let playlist = crate::playlist::Playlist::new(reg.pool.clone(), &room_id.0); let history = playlist.history(10).await.unwrap(); - assert!( - !history.is_empty(), - "at least one track should be in history after double skip" - ); - assert_eq!(history[0].title, "B", "last skipped track should be B"); + assert!(!history.is_empty()); + assert_eq!(history[0].title, "b"); } #[tokio::test] - async fn test_register_client_overflow() { - let pool = test_pool().await; - let reg = Registry::new(pool, PathBuf::from("/tmp/moqbox"), test_registry()); - let room = reg.create().await; + async fn test_register_client_runs_three_times() { + let (reg, room_id) = test_registry_and_room().await; - for _ in 0..100 { - let id = reg.register_client(&room.id).await; - assert!(id.is_some(), "register_client should return Some(id)"); + for _ in 0..3 { + let id = reg.register_client(&room_id).await; + assert!(id.is_some()); } - tokio::time::sleep(Duration::from_millis(50)).await; - let exists = reg.exists(&room.id).await; - assert!(exists, "room should still exist"); + assert!(reg.exists(&room_id).await); } #[tokio::test] async fn test_remove_while_playing_does_not_panic() { - let pool = test_pool().await; - let reg = Registry::new(pool, PathBuf::from("/tmp/moqbox"), test_registry()); - let room = reg.create().await; + let (reg, room_id) = test_registry_and_room().await; - reg.push( - &room.id, - TrackMeta { - title: "A".into(), - duration: "3:45".into(), - thumbnail: None, - url: "https://example.com/a".into(), - source: crate::types::SourceKind::Direct, - }, + reg.push_url(&room_id, "https://example.com/a".into()).await; + poll_until( + || async { reg.queue(&room_id).await.is_empty() }, + Duration::from_secs(2), + "push should have started playing", ) .await; - tokio::time::sleep(Duration::from_millis(100)).await; - - reg.remove(&room.id).await; - tokio::time::sleep(Duration::from_millis(100)).await; - assert!( - !reg.exists(&room.id).await, - "room should not exist after remove" - ); + reg.remove(&room_id).await; + assert!(!reg.exists(&room_id).await); } #[tokio::test] async fn test_sweep_idle_keeps_active_room() { - let pool = test_pool().await; - let reg = Registry::new(pool, PathBuf::from("/tmp/moqbox"), test_registry()); - let room = reg.create().await; + let (reg, room_id) = test_registry_and_room().await; - reg.register_client(&room.id).await; + reg.register_client(&room_id).await; let swept = reg.sweep_idle(Duration::from_millis(10_000)).await; - assert!( - !swept.contains(&room.id), - "active room should not be swept while clients are connected" - ); + assert!(!swept.contains(&room_id)); } #[tokio::test] async fn test_client_count_tracking() { - let pool = test_pool().await; - let reg = Registry::new(pool, PathBuf::from("/tmp/moqbox"), test_registry()); - let room = reg.create().await; + let (reg, room_id) = test_registry_and_room().await; - let id1 = reg.register_client(&room.id).await; - let id2 = reg.register_client(&room.id).await; - let id3 = reg.register_client(&room.id).await; - assert!(id1.is_some() && id2.is_some() && id3.is_some()); + let id1 = reg.register_client(&room_id).await; + let id2 = reg.register_client(&room_id).await; + let id3 = reg.register_client(&room_id).await; + assert!(id1.is_some(), "id1 should be Some"); + assert!(id2.is_some(), "id2 should be Some"); + assert!(id3.is_some(), "id3 should be Some"); - reg.unregister_client(&room.id).await; - tokio::time::sleep(Duration::from_millis(50)).await; + reg.unregister_client(&room_id).await; - let id4 = reg.register_client(&room.id).await; - assert!(id4.is_some(), "new client should get an id"); - assert_ne!(id4, id1, "ids should be unique"); + let id4 = reg.register_client(&room_id).await; + assert!(id4.is_some()); + assert_ne!(id4, id1); } #[tokio::test] async fn test_multiple_rooms_dont_interfere() { - let pool = test_pool().await; - let reg = Registry::new(pool, PathBuf::from("/tmp/moqbox"), test_registry()); - let room_a = reg.create().await; + let (reg, room_a) = test_registry_and_room().await; let room_b = reg.create().await; - assert_ne!(room_a.id, room_b.id, "room IDs should be unique"); - assert!(reg.exists(&room_a.id).await); - assert!(reg.exists(&room_b.id).await); + assert_ne!(room_a, room_b); + assert!(reg.exists(&room_a).await); + assert!(reg.exists(&room_b).await); - reg.push( - &room_a.id, - TrackMeta { - title: "A".into(), - duration: "1:00".into(), - thumbnail: None, - url: "https://example.com/a".into(), - source: crate::types::SourceKind::Direct, - }, + reg.push_url(&room_a, "https://example.com/a".into()).await; + reg.push_url(&room_b, "https://example.com/b".into()).await; + + poll_until( + || async { reg.queue(&room_a).await.is_empty() }, + Duration::from_secs(2), + "room_a queue empty", ) .await; - reg.push( - &room_b.id, - TrackMeta { - title: "B".into(), - duration: "2:00".into(), - thumbnail: None, - url: "https://example.com/b".into(), - source: crate::types::SourceKind::Direct, - }, + poll_until( + || async { reg.queue(&room_b).await.is_empty() }, + Duration::from_secs(2), + "room_b queue empty", ) .await; - tokio::time::sleep(Duration::from_millis(100)).await; - - let queue_a = reg.queue(&room_a.id).await; - let queue_b = reg.queue(&room_b.id).await; - assert!( - queue_a.is_empty(), - "room A should be empty after first track starts" - ); - assert!( - queue_b.is_empty(), - "room B should be empty after first track starts" - ); - } - - #[tokio::test] - async fn test_skip_on_empty_queue_no_crash() { - let pool = test_pool().await; - let reg = Registry::new(pool, PathBuf::from("/tmp/moqbox"), test_registry()); - let room = reg.create().await; - reg.skip(&room.id).await; - tokio::time::sleep(Duration::from_millis(50)).await; } #[tokio::test] async fn test_push_after_skip_works() { - let pool = test_pool().await; - let reg = Registry::new(pool, PathBuf::from("/tmp/moqbox"), test_registry()); - let room = reg.create().await; + let (reg, room_id) = test_registry_and_room().await; - reg.push( - &room.id, - TrackMeta { - title: "A".into(), - duration: "3:45".into(), - thumbnail: None, - url: "https://a".into(), - source: crate::types::SourceKind::Direct, - }, + reg.push_url(&room_id, "https://a".into()).await; + poll_until( + || async { reg.queue(&room_id).await.is_empty() }, + Duration::from_secs(2), + "first push started playing", ) .await; - tokio::time::sleep(Duration::from_millis(100)).await; - reg.skip(&room.id).await; - tokio::time::sleep(Duration::from_millis(100)).await; - - reg.push( - &room.id, - TrackMeta { - title: "B".into(), - duration: "3:45".into(), - thumbnail: None, - url: "https://b".into(), - source: crate::types::SourceKind::Direct, + + reg.skip(&room_id).await; + poll_until( + || async { + let playlist = crate::playlist::Playlist::new(reg.pool.clone(), &room_id.0); + !playlist.history(10).await.unwrap().is_empty() }, + Duration::from_secs(2), + "skip should move track to history", ) .await; - tokio::time::sleep(Duration::from_millis(100)).await; - let queue = reg.queue(&room.id).await; - assert!(queue.is_empty(), "new track should be playing"); + reg.push_url(&room_id, "https://b".into()).await; + poll_until( + || async { reg.queue(&room_id).await.is_empty() }, + Duration::from_secs(2), + "second push should start playing immediately (idle after skip)", + ) + .await; - let playlist = crate::playlist::Playlist::new(reg.pool.clone(), &room.id.0); + let playlist = crate::playlist::Playlist::new(reg.pool.clone(), &room_id.0); let history = playlist.history(10).await.unwrap(); - assert!(!history.is_empty(), "skipped track should be in history"); + assert!(!history.is_empty()); } #[tokio::test] async fn test_client_count_does_not_underflow() { - let pool = test_pool().await; - let reg = Registry::new(pool, PathBuf::from("/tmp/moqbox"), test_registry()); - let room = reg.create().await; + let (reg, room_id) = test_registry_and_room().await; - reg.unregister_client(&room.id).await; - tokio::time::sleep(Duration::from_millis(50)).await; + reg.unregister_client(&room_id).await; - assert!(reg.exists(&room.id).await); + let re_registered = reg.register_client(&room_id).await; + assert!( + re_registered.is_some(), + "room should accept new clients after underflow" + ); } #[tokio::test] async fn test_multiple_skips_sequential() { - let pool = test_pool().await; - let reg = Registry::new(pool, PathBuf::from("/tmp/moqbox"), test_registry()); - let room = reg.create().await; - - for title in ["A", "B", "C"] { - reg.push( - &room.id, - TrackMeta { - title: title.into(), - duration: "1:00".into(), - thumbnail: None, - url: "https://x".into(), - source: crate::types::SourceKind::Direct, - }, - ) - .await; + let (reg, room_id) = test_registry_and_room().await; + + for url in ["https://x/a", "https://x/b", "https://x/c"] { + reg.push_url(&room_id, url.into()).await; } - tokio::time::sleep(Duration::from_millis(100)).await; + poll_until( + || async { reg.queue(&room_id).await.len() == 2 }, + Duration::from_secs(2), + "3 pushes → 1 active, 2 queued", + ) + .await; - let mut queue = reg.queue(&room.id).await; - assert_eq!(queue.len(), 2, "B and C should be queued"); + reg.skip(&room_id).await; + poll_until( + || async { reg.queue(&room_id).await.len() == 1 }, + Duration::from_secs(2), + "skip → 1 item left in queue", + ) + .await; - reg.skip(&room.id).await; - tokio::time::sleep(Duration::from_millis(100)).await; - queue = reg.queue(&room.id).await; - assert_eq!(queue.len(), 1, "only C should remain after skipping A"); - assert_eq!(queue[0].title, "C"); + let queue = reg.queue(&room_id).await; + assert_eq!(queue[0].title, "c"); - reg.skip(&room.id).await; - tokio::time::sleep(Duration::from_millis(100)).await; - queue = reg.queue(&room.id).await; - assert!(queue.is_empty(), "nothing should remain after skipping B"); + reg.skip(&room_id).await; + poll_until( + || async { reg.queue(&room_id).await.is_empty() }, + Duration::from_secs(2), + "skip → queue empty", + ) + .await; - reg.skip(&room.id).await; - tokio::time::sleep(Duration::from_millis(100)).await; + reg.skip(&room_id).await; } } diff --git a/src/source/direct.rs b/src/source/direct.rs index f135b5e..842bf2c 100644 --- a/src/source/direct.rs +++ b/src/source/direct.rs @@ -2,7 +2,7 @@ //! //! Handles URLs that point directly to media files (`.mp4`, `.webm`, `.mp3`, //! etc.) without going through yt-dlp. Metadata is extracted heuristically -//! from the URL path — no network requests are made during extraction. +//! from the URL path. No network requests are made during extraction. use std::path::Path; @@ -11,14 +11,10 @@ use crate::types::TrackMeta; /// Media file extensions that DirectSource can handle. const MEDIA_EXTENSIONS: &[&str] = &[ - "mp4", "webm", "mkv", "avi", "mov", "m4v", - "mp3", "ogg", "flac", "wav", "m4a", "aac", "opus", "wma", + "mp4", "webm", "mkv", "avi", "mov", "m4v", "mp3", "ogg", "flac", "wav", "m4a", "aac", "opus", + "wma", ]; -// --------------------------------------------------------------------------- -// Source -// --------------------------------------------------------------------------- - /// A media source that handles direct media file URLs. /// /// Detects media files by URL path extension. If the URL has a recognised @@ -29,8 +25,8 @@ const MEDIA_EXTENSIONS: &[&str] = &[ /// replaces separators with spaces) /// - **resolve** returns the URL unchanged /// -/// This source is intended as a fallback after [`YtdlpSource`](super::ytdlp::YtdlpSource) -/// — it only matches URLs with media extensions whose filenames look like +/// This source is intended as a fallback after [`YtdlpSource`](super::ytdlp::YtdlpSource). +/// It only matches URLs with media extensions whose filenames look like /// media files. pub(crate) struct DirectSource; @@ -44,7 +40,9 @@ impl DirectSource { impl MediaSource for DirectSource { async fn extract(&self, url: &str, _cache_dir: &Path) -> Result { let Some(ext) = extension(url) else { - return Err(SourceError::Unsupported(format!("no file extension in URL: {url}"))); + return Err(SourceError::Unsupported(format!( + "no file extension in URL: {url}" + ))); }; if !MEDIA_EXTENSIONS.contains(&ext.as_str()) { @@ -66,18 +64,15 @@ impl MediaSource for DirectSource { async fn resolve(&self, url: &str, _cache_dir: &Path) -> Result { if !url.starts_with("http://") && !url.starts_with("https://") { - return Err(SourceError::Unsupported(format!("not an HTTP(S) URL: {url}"))); + return Err(SourceError::Unsupported(format!( + "not an HTTP(S) URL: {url}" + ))); } - // Direct URLs are already playable as-is. Ok(url.to_string()) } } -// --------------------------------------------------------------------------- -// Helpers -// --------------------------------------------------------------------------- - /// Extract the file extension (without dot) from a URL path, if any. fn extension(url: &str) -> Option { let path = url.split('?').next().unwrap_or(url); @@ -123,22 +118,23 @@ fn extract_title(url: &str, ext: &str) -> String { .collect::>() .join(" "); - if result.is_empty() { "Untitled".into() } else { result } + if result.is_empty() { + "Untitled".into() + } else { + result + } } -// --------------------------------------------------------------------------- -// Tests -// --------------------------------------------------------------------------- - #[cfg(test)] mod tests { use super::*; - // -- Extension detection ------------------------------------------------- - #[test] fn test_extension_mp4() { - assert_eq!(extension("https://example.com/video.mp4"), Some("mp4".into())); + assert_eq!( + extension("https://example.com/video.mp4"), + Some("mp4".into()) + ); } #[test] @@ -161,14 +157,18 @@ mod tests { #[test] fn test_extension_uppercase() { - assert_eq!(extension("https://example.com/video.MP4"), Some("mp4".into())); + assert_eq!( + extension("https://example.com/video.MP4"), + Some("mp4".into()) + ); } - // -- Title extraction ---------------------------------------------------- - #[test] fn test_title_basic() { - assert_eq!(extract_title("https://example.com/my-video.mp4", "mp4"), "my video"); + assert_eq!( + extract_title("https://example.com/my-video.mp4", "mp4"), + "my video" + ); } #[test] @@ -212,14 +212,9 @@ mod tests { #[test] fn test_title_multi_space_collapse() { - assert_eq!( - extract_title("https://example.com/a---b.mp4", "mp4"), - "a b" - ); + assert_eq!(extract_title("https://example.com/a---b.mp4", "mp4"), "a b"); } - // -- MediaSource implementation ------------------------------------------ - #[tokio::test] async fn test_direct_source_accepts_mp4() { let src = DirectSource::new(); @@ -270,6 +265,6 @@ mod tests { assert!(matches!(result, Err(SourceError::Unsupported(_)))); } - // Integration test with yt-dlp is in src/tests/ — this file focuses on + // Integration test with yt-dlp is in src/tests/. This file focuses on // DirectSource unit tests without network dependencies. } diff --git a/src/source/mod.rs b/src/source/mod.rs index c1e3ced..20599b9 100644 --- a/src/source/mod.rs +++ b/src/source/mod.rs @@ -1,6 +1,6 @@ //! Pluggable media source resolution. //! -//! Defines [`MediaSource`] — the seam between URL-based track selection and +//! Defines [`MediaSource`]: the seam between URL-based track selection and //! playable stream production. Different implementations handle different URL //! patterns (yt-dlp, direct media URLs, etc.). //! @@ -8,7 +8,7 @@ //! //! 1. Implement [`MediaSource`] for your source type. //! 2. Register it with [`SourceRegistry::register`]. -//! 3. Sources are tried in registration order — first match wins. +//! 3. Sources are tried in registration order: first match wins. pub(crate) mod direct; pub(crate) mod ytdlp; @@ -17,10 +17,6 @@ use std::path::Path; use crate::types::TrackMeta; -// --------------------------------------------------------------------------- -// Error -// --------------------------------------------------------------------------- - /// Errors from media source resolution. #[derive(Debug, thiserror::Error)] pub(crate) enum SourceError { @@ -41,10 +37,6 @@ pub(crate) enum SourceError { Io(#[from] std::io::Error), } -// --------------------------------------------------------------------------- -// Trait -// --------------------------------------------------------------------------- - /// A media source knows how to extract metadata from a URL and resolve it to a /// playable stream URL. #[async_trait::async_trait] @@ -52,21 +44,17 @@ pub(crate) trait MediaSource: Send + Sync { /// Extract metadata from `url`. /// /// Returns [`SourceError::Unsupported`] if this source cannot handle the - /// URL — the registry will try the next source. Any other error is + /// URL. The registry will try the next source. Any other error is /// terminal for this URL. async fn extract(&self, url: &str, cache_dir: &Path) -> Result; /// Resolve `url` to a playable stream URL. /// /// Returns [`SourceError::Unsupported`] if this source cannot handle the - /// URL — the registry will try the next source. + /// URL. The registry will try the next source. async fn resolve(&self, url: &str, cache_dir: &Path) -> Result; } -// --------------------------------------------------------------------------- -// Registry -// --------------------------------------------------------------------------- - /// A registry of media sources tried in registration order. /// /// When a URL arrives, each source is asked via @@ -127,11 +115,7 @@ impl SourceRegistry { } /// Resolve a URL to a playable stream URL by trying each registered source. - pub(crate) async fn resolve( - &self, - url: &str, - cache_dir: &Path, - ) -> Result { + pub(crate) async fn resolve(&self, url: &str, cache_dir: &Path) -> Result { let mut unsupported_hint = String::new(); for source in &self.sources { match source.resolve(url, cache_dir).await { @@ -149,10 +133,6 @@ impl SourceRegistry { } } -// --------------------------------------------------------------------------- -// Tests -// --------------------------------------------------------------------------- - #[cfg(test)] mod tests { use super::*; diff --git a/src/source/ytdlp.rs b/src/source/ytdlp.rs index 51c599a..5cb76e0 100644 --- a/src/source/ytdlp.rs +++ b/src/source/ytdlp.rs @@ -12,10 +12,6 @@ use crate::media; use crate::source::{MediaSource, SourceError}; use crate::types::TrackMeta; -// --------------------------------------------------------------------------- -// yt-dlp JSON output fields -// --------------------------------------------------------------------------- - #[derive(Debug, Deserialize)] #[serde(rename_all = "snake_case")] struct YtdlpMetadata { @@ -25,10 +21,6 @@ struct YtdlpMetadata { webpage_url: String, } -// --------------------------------------------------------------------------- -// Source -// --------------------------------------------------------------------------- - /// A media source backed by yt-dlp. /// /// Handles URLs from YouTube, SoundCloud, Bandcamp, and hundreds of other @@ -45,7 +37,9 @@ impl YtdlpSource { impl MediaSource for YtdlpSource { async fn extract(&self, url: &str, cache_dir: &Path) -> Result { if !url.starts_with("http://") && !url.starts_with("https://") { - return Err(SourceError::Unsupported(format!("not an HTTP(S) URL: {url}"))); + return Err(SourceError::Unsupported(format!( + "not an HTTP(S) URL: {url}" + ))); } let cache_dir = cache_dir.to_path_buf(); @@ -63,7 +57,7 @@ impl MediaSource for YtdlpSource { .output() }) .await - .map_err(|e| SourceError::Io(std::io::Error::new(std::io::ErrorKind::Other, e)))?; + .map_err(|e| SourceError::Io(std::io::Error::other(e)))?; let output = result.map_err(|e| { if e.kind() == std::io::ErrorKind::NotFound { @@ -100,7 +94,9 @@ impl MediaSource for YtdlpSource { async fn resolve(&self, url: &str, cache_dir: &Path) -> Result { if !url.starts_with("http://") && !url.starts_with("https://") { - return Err(SourceError::Unsupported(format!("not an HTTP(S) URL: {url}"))); + return Err(SourceError::Unsupported(format!( + "not an HTTP(S) URL: {url}" + ))); } let cache_dir = cache_dir.to_path_buf(); @@ -109,7 +105,7 @@ impl MediaSource for YtdlpSource { std::process::Command::new("yt-dlp") .args([ "-f", - "b", + "b", // best single format (not bestvideo+bestaudio which needs merge) "-g", "--no-playlist", "--no-warnings", @@ -120,7 +116,7 @@ impl MediaSource for YtdlpSource { .output() }) .await - .map_err(|e| SourceError::Io(std::io::Error::new(std::io::ErrorKind::Other, e)))?; + .map_err(|e| SourceError::Io(std::io::Error::other(e)))?; let output = result.map_err(|e| { if e.kind() == std::io::ErrorKind::NotFound { @@ -153,10 +149,6 @@ impl MediaSource for YtdlpSource { } } -// --------------------------------------------------------------------------- -// Tests -// --------------------------------------------------------------------------- - #[cfg(test)] mod tests { use super::*; diff --git a/src/state.rs b/src/state.rs index ee02ffe..9aa6490 100644 --- a/src/state.rs +++ b/src/state.rs @@ -1,21 +1,7 @@ -//! Pure playback state machine — deterministic, testable, IO-free. +//! Pure playback state machine: deterministic, testable, IO-free. //! //! Owns the in-memory queue, active track, and history. All transitions //! are pure functions returning [`Effect`]s that the caller executes. -//! -//! # Impure-pure-impure sandwich -//! -//! 1. **Gather** — caller fetches data (DB, wall clock) -//! 2. **Transition** — `PlaybackState::transition(event, now)` — pure -//! 3. **Commit** — caller executes returned effects (DB writes, pipeline start) -//! -//! # Testing -//! -//! ```ignore -//! let mut state = PlaybackState::default(); -//! let effects = state.transition(&Event::TrackQueued(my_track), 1_700_000_000_000); -//! assert!(effects.contains(&Effect::StartPipeline(...))); -//! ``` use crate::types::ActiveTrackInfo; @@ -77,8 +63,8 @@ pub(crate) enum Effect { PublishSnapshot, } -/// Pure playback state — no IO handles, no DB connections. -#[derive(Debug, Clone)] +/// Pure playback state: no IO handles, no DB connections. +#[derive(Debug, Clone, Default)] pub(crate) struct PlaybackState { /// Currently playing track, if any. pub active: Option, @@ -88,24 +74,11 @@ pub(crate) struct PlaybackState { pub history: Vec, } -impl Default for PlaybackState { - fn default() -> Self { - Self { - active: None, - queue: Vec::new(), - history: Vec::new(), - } - } -} - impl PlaybackState { /// Pure transition function. /// - /// - `event` — what happened - /// - `now` — current wall-clock epoch ms (gathered by caller for determinism) - /// /// Returns effects for the caller to execute. Mutates `self` in place. - /// Does NO I/O. Deterministic given the same inputs. + /// Does no I/O. Deterministic given the same inputs. pub fn transition(&mut self, event: &Event, now: i64) -> Vec { match event { Event::TrackQueued(track) => self.handle_queued(track), @@ -124,12 +97,11 @@ impl PlaybackState { fn handle_queued(&mut self, track: &QueuedTrack) -> Vec { self.queue.push(track.clone()); - // If nothing is playing, start immediately. - if self.active.is_none() { - self.advance(track.clone()) - } else { - vec![Effect::PublishSnapshot] + if self.active.is_some() { + return vec![Effect::PublishSnapshot]; } + + self.advance(track.clone()) } /// User requested skip. @@ -148,7 +120,9 @@ impl PlaybackState { url: active.url, duration: active.duration, thumbnail: active.thumbnail, - played_at: self.format_played_at(), + // RoomActor overrides via PersistFinished handler which sets + // the real played_at timestamp in the DB. + played_at: String::new(), }, ); } @@ -193,7 +167,9 @@ impl PlaybackState { url: active.url, duration: active.duration, thumbnail: active.thumbnail, - played_at: self.format_played_at(), + // RoomActor overrides via PersistFinished handler which sets + // the real played_at timestamp in the DB. + played_at: String::new(), }, ); } @@ -210,7 +186,7 @@ impl PlaybackState { /// Asynchronous metadata resolution completed. Updates the matching item /// in the queue and/or the active track in-place. Does not change playback - /// state — only updates display fields. + /// state, only updates display fields. fn handle_metadata_updated( &mut self, item_id: i64, @@ -240,7 +216,7 @@ impl PlaybackState { fn advance(&mut self, next: QueuedTrack) -> Vec { self.queue.retain(|t| t.id != next.id); - // Store active track with placeholder timestamp — RoomActor calls + // Store active track with placeholder timestamp. RoomActor calls // `resolve_started_at` with the real wall-clock time. let info = ActiveTrackInfo { id: next.id, @@ -266,13 +242,6 @@ impl PlaybackState { active.started_at_wall = real_started_at; } } - - fn format_played_at(&self) -> String { - // Simple ISO-like timestamp for frontend display. - // This is a mock since we don't have Utc::now() in pure code. - // RoomActor will override via PersistFinished handler. - String::new() - } } /// # Pure unit tests (no async, no DB, no IO) @@ -291,13 +260,8 @@ mod tests { } } - // ── Effect ordering tests ───────────────────────────────────────────── - // These verify the exact effect sequence returned by transition(). Order - // matters: PersistStarted must precede StartPipeline (DB commit before - // pipeline spawn), AbortPipeline must precede PersistFinished, etc. - #[test] - fn idle_to_playing_effect_order() { + fn test_idle_to_playing_effect_order() { let mut s = PlaybackState::default(); let effects = s.transition(&Event::TrackQueued(queued("A", 1)), 0); @@ -313,7 +277,7 @@ mod tests { } #[test] - fn queue_second_while_playing_effect_order() { + fn test_queue_second_while_playing_effect_order() { let mut s = PlaybackState::default(); s.transition(&Event::TrackQueued(queued("A", 1)), 0); let effects = s.transition(&Event::TrackQueued(queued("B", 2)), 0); @@ -321,12 +285,12 @@ mod tests { assert_eq!( effects, vec![Effect::PublishSnapshot], - "playing → queue another: just notify — pipeline untouched" + "playing -> queue another: just notify, pipeline untouched" ); } #[test] - fn skip_with_next_effect_order() { + fn test_skip_with_next_effect_order() { let mut s = PlaybackState::default(); s.transition(&Event::TrackQueued(queued("A", 1)), 0); s.transition(&Event::TrackQueued(queued("B", 2)), 0); @@ -354,7 +318,7 @@ mod tests { } #[test] - fn skip_last_track_effect_order() { + fn test_skip_last_track_effect_order() { let mut s = PlaybackState::default(); s.transition(&Event::TrackQueued(queued("A", 1)), 0); let effects = s.transition(&Event::Skip, 0); @@ -366,12 +330,12 @@ mod tests { Effect::PersistFinished(1), Effect::PublishSnapshot, ], - "skip last: abort, finalize, notify — no StartPipeline" + "skip last: abort, finalize, notify, no StartPipeline" ); } #[test] - fn track_ended_with_next_effect_order() { + fn test_track_ended_with_next_effect_order() { let mut s = PlaybackState::default(); s.transition(&Event::TrackQueued(queued("A", 1)), 0); s.transition(&Event::TrackQueued(queued("B", 2)), 0); @@ -380,7 +344,7 @@ mod tests { assert_eq!( &effects[..2], &[Effect::PersistFinished(1), Effect::PersistStarted(2),], - "track ended (next): finalize A, persist B started_at — no AbortPipeline" + "track ended (next): finalize A, persist B started_at, no AbortPipeline" ); assert!( effects @@ -395,7 +359,7 @@ mod tests { } #[test] - fn track_ended_last_goes_idle_effect_order() { + fn test_track_ended_last_goes_idle_effect_order() { let mut s = PlaybackState::default(); s.transition(&Event::TrackQueued(queued("A", 1)), 0); let effects = s.transition(&Event::TrackEnded { item_id: 1 }, 0); @@ -403,14 +367,12 @@ mod tests { assert_eq!( effects, vec![Effect::PersistFinished(1), Effect::PublishSnapshot,], - "track ended last: finalize, notify — no StartPipeline, no AbortPipeline" + "track ended last: finalize, notify, no StartPipeline, no AbortPipeline" ); } - // ── State invariant tests ───────────────────────────────────────────── - #[test] - fn idle_room_starts_playing_on_first_track() { + fn test_idle_room_starts_playing_on_first_track() { let mut s = PlaybackState::default(); s.transition(&Event::TrackQueued(queued("A", 1)), 0); @@ -420,7 +382,7 @@ mod tests { } #[test] - fn second_track_stays_in_queue() { + fn test_second_track_stays_in_queue() { let mut s = PlaybackState::default(); s.transition(&Event::TrackQueued(queued("A", 1)), 0); s.transition(&Event::TrackQueued(queued("B", 2)), 0); @@ -431,7 +393,7 @@ mod tests { } #[test] - fn skip_advances_to_next_track() { + fn test_skip_advances_to_next_track() { let mut s = PlaybackState::default(); s.transition(&Event::TrackQueued(queued("A", 1)), 0); s.transition(&Event::TrackQueued(queued("B", 2)), 0); @@ -444,7 +406,7 @@ mod tests { } #[test] - fn skip_on_idle_does_nothing() { + fn test_skip_on_idle_does_nothing() { let mut s = PlaybackState::default(); let effects = s.transition(&Event::Skip, 0); @@ -453,7 +415,7 @@ mod tests { } #[test] - fn skip_last_track_goes_idle() { + fn test_skip_last_track_goes_idle() { let mut s = PlaybackState::default(); s.transition(&Event::TrackQueued(queued("A", 1)), 0); s.transition(&Event::Skip, 0); @@ -464,7 +426,7 @@ mod tests { } #[test] - fn track_ended_advances_to_next() { + fn test_track_ended_advances_to_next() { let mut s = PlaybackState::default(); s.transition(&Event::TrackQueued(queued("A", 1)), 0); s.transition(&Event::TrackQueued(queued("B", 2)), 0); @@ -476,7 +438,7 @@ mod tests { } #[test] - fn track_ended_ignores_stale_id() { + fn test_track_ended_ignores_stale_id() { let mut s = PlaybackState::default(); s.transition(&Event::TrackQueued(queued("A", 1)), 0); @@ -488,7 +450,7 @@ mod tests { } #[test] - fn track_ended_last_track_goes_idle() { + fn test_track_ended_last_track_goes_idle() { let mut s = PlaybackState::default(); s.transition(&Event::TrackQueued(queued("A", 1)), 0); s.transition(&Event::TrackEnded { item_id: 1 }, 0); @@ -498,7 +460,7 @@ mod tests { } #[test] - fn resolve_started_at_updates_active_track() { + fn test_resolve_started_at_updates_active_track() { let mut s = PlaybackState::default(); s.transition(&Event::TrackQueued(queued("A", 1)), 0); @@ -507,7 +469,7 @@ mod tests { } #[test] - fn skip_then_track_ended_stale_is_ignored() { + fn test_skip_then_track_ended_stale_is_ignored() { let mut s = PlaybackState::default(); s.transition(&Event::TrackQueued(queued("A", 1)), 0); s.transition(&Event::TrackQueued(queued("B", 2)), 0); @@ -520,16 +482,14 @@ mod tests { } #[test] - fn resolve_started_at_noop_when_idle() { + fn test_resolve_started_at_noop_when_idle() { let mut s = PlaybackState::default(); s.resolve_started_at(42); assert!(s.active.is_none()); } - // ── Metadata update tests ───────────────────────────────────────── - #[test] - fn metadata_update_updates_queue_item() { + fn test_metadata_update_updates_queue_item() { let mut s = PlaybackState::default(); s.transition(&Event::TrackQueued(queued("A", 1)), 0); s.transition( @@ -553,7 +513,7 @@ mod tests { } #[test] - fn metadata_update_on_queued_item() { + fn test_metadata_update_on_queued_item() { let mut s = PlaybackState::default(); s.transition(&Event::TrackQueued(queued("A", 1)), 0); s.transition(&Event::TrackQueued(queued("B", 2)), 0); @@ -574,7 +534,7 @@ mod tests { } #[test] - fn metadata_update_on_unknown_id_is_noop() { + fn test_metadata_update_on_unknown_id_is_noop() { let mut s = PlaybackState::default(); let effects = s.transition( &Event::MetadataUpdated { @@ -585,7 +545,7 @@ mod tests { }, 0, ); - // Still publishes snapshot even if nothing changed — clients may want + // Still publishes snapshot even if nothing changed: clients may want // to know the update was processed. assert_eq!(effects, vec![Effect::PublishSnapshot]); assert!(s.active.is_none()); diff --git a/src/transport.rs b/src/transport.rs index 7366ad9..b340a9e 100644 --- a/src/transport.rs +++ b/src/transport.rs @@ -44,6 +44,7 @@ pub(crate) fn encode_varint(value: u64) -> Vec { } } +#[must_use] pub(crate) fn decode_varint(buf: &[u8]) -> Option<(u64, usize)> { let first = *buf.first()?; match first >> 6 { @@ -89,6 +90,7 @@ fn encode_string(s: &str) -> Vec { buf } +#[must_use] fn decode_string(buf: &[u8]) -> Option<(String, usize)> { let (len, mut offset) = decode_varint(buf)?; let end = offset.checked_add(len as usize)?; @@ -177,6 +179,7 @@ pub(crate) fn encode_message(msg: &MoqMessage) -> Bytes { Bytes::from(buf) } +#[must_use] pub(crate) fn decode_message(buf: &[u8]) -> Option<(MoqMessage, usize)> { let (msg_type, mut offset) = decode_varint(buf)?; match msg_type { @@ -252,10 +255,6 @@ pub(crate) fn room_namespace(room_id: &str) -> String { format!("moqbox/room/{room_id}") } -// --------------------------------------------------------------------------- -// Track publishers -// --------------------------------------------------------------------------- - /// Per-room MoQ track publishers, shared across all client sessions. /// /// Cloning is cheap (all inner channels are `Arc`-based). @@ -272,7 +271,7 @@ pub(crate) struct TrackPublishers { impl TrackPublishers { pub(crate) fn new() -> Self { - let (video, _) = broadcast::channel(256); + let (video, _) = broadcast::channel(256); // enough for bursty media objects without backpressure let (chat, _) = broadcast::channel(256); let (state, _) = watch::channel(StateValue { payload: Bytes::new(), @@ -345,13 +344,6 @@ impl TrackPublishers { } /// Handle a WebSocket upgrade and run the MoQ session lifecycle. -/// -/// 1. Sends `ANNOUNCE` for the room namespace. -/// 2. Loops, reading MoQ messages from the client. -/// 3. On `SUBSCRIBE`, subscribes to the corresponding broadcast receiver -/// and spawns a forward task that relays objects to the WebSocket. -/// 4. On `OBJECT` with track ID 2 (chat), forwards to the chat publisher. -/// 5. On close, cancels all forward tasks. pub(crate) async fn handle_ws_session( mut ws: WebSocket, room_id: RoomId, @@ -373,7 +365,6 @@ pub(crate) async fn handle_ws_session( let (mut ws_tx, mut ws_rx) = ws.split(); - // 1. Send ANNOUNCE for this room. let announce = encode_message(&MoqMessage::Announce { namespace: room_namespace(&room_id.0), }); @@ -381,7 +372,6 @@ pub(crate) async fn handle_ws_session( return; } - // Register this client with the room actor. let (resp_tx, resp_rx) = tokio::sync::oneshot::channel(); let _ = cmd_tx .send(RoomCommand::RegisterClient { resp: resp_tx }) @@ -391,11 +381,8 @@ pub(crate) async fn handle_ws_session( .ok() .and_then(|r| r.ok()); - // Channel that broadcast forward tasks use to send objects to the WS writer. - // Unbounded to avoid deadlock between subscribe processing and the WS send loop. let (object_tx, mut object_rx) = mpsc::unbounded_channel::(); - // Handles of spawned forward tasks, cancelled on disconnect. let mut task_handles: Vec> = Vec::new(); loop { @@ -423,12 +410,10 @@ pub(crate) async fn handle_ws_session( break; } None => break, - // Ignore Ping / Pong / Text frames. _ => {} } } - // Forward objects from broadcast receivers to the WebSocket. obj_bytes = object_rx.recv() => { match obj_bytes { Some(bytes) => { @@ -436,17 +421,14 @@ pub(crate) async fn handle_ws_session( break; } } - // All senders dropped — no more objects will arrive. None => break, } } } } - // Unregister this client with the room actor. let _ = cmd_tx.send(RoomCommand::UnregisterClient).await; - // Cancel all forward tasks on disconnect. for handle in task_handles { handle.abort(); } @@ -454,19 +436,12 @@ pub(crate) async fn handle_ws_session( /// WebSocket session for video-only delivery. /// -/// No MoQ protocol negotiation — the server auto-subscribes the client to +/// No MoQ protocol negotiation: the server auto-subscribes the client to /// the video broadcast. Messages are raw fMP4 bytes (no MoQ framing). /// A separate WS-control connection handles chat, state, and MoQ handshake. -pub(crate) async fn handle_video_session( - ws: WebSocket, - publishers: TrackPublishers, -) { +pub(crate) async fn handle_video_session(ws: WebSocket, publishers: TrackPublishers) { let (mut ws_tx, mut ws_rx) = ws.split(); - // Collect the cached init (or wait for the pipeline to produce one) - // but don't send it yet — we need to subscribe to the broadcast first - // so we can discard any init that was also published via broadcast, - // which would otherwise arrive as a duplicate through the forward loop. let init = match publishers.get_init_segment() { Some(cached) => { tracing::debug!("video ws: using cached init segment"); @@ -486,9 +461,6 @@ pub(crate) async fn handle_video_session( } }; - // Subscribe to the broadcast before sending cached init so that - // any init published via broadcast is caught by the catch-up loop - // and discarded, preventing a duplicate on the forward path. let mut video_rx = publishers.video.subscribe(); loop { match video_rx.try_recv() { @@ -499,15 +471,12 @@ pub(crate) async fn handle_video_session( } } - // Now send the cached init — the broadcast buffer has been drained - // so the forward loop below won't re-deliver it. if let Some((_group_id, payload)) = init { if ws_tx.send(Message::Binary(payload)).await.is_err() { return; } } - // Forward live video objects as raw fMP4 payloads. loop { tokio::select! { msg = ws_rx.next() => { @@ -533,6 +502,8 @@ pub(crate) async fn handle_video_session( } } +#[allow(clippy::too_many_arguments)] +// justified: all params are distinct concerns async fn handle_moq_message( msg: MoqMessage, room_id: &RoomId, @@ -549,7 +520,6 @@ async fn handle_moq_message( let track_id = match TrackId::from_name(&track) { Some(id) => id, None => { - // Unknown track — send SUBSCRIBE_RST. let rst = encode_message(&MoqMessage::SubscribeRst { namespace, track, @@ -560,7 +530,6 @@ async fn handle_moq_message( } }; - // Send SUBSCRIBE_OK first. let ok = encode_message(&MoqMessage::SubscribeOk { namespace: namespace.clone(), track: track.clone(), @@ -571,7 +540,6 @@ async fn handle_moq_message( match track_id { TrackId::Chat => { - // Send recent chat history, then subscribe to live broadcast. if let Ok(messages) = registry.recent_chat(room_id, 50).await { for msg in messages.iter().rev() { let json = serde_json::json!({ @@ -594,7 +562,6 @@ async fn handle_moq_message( } } - // Subscribe to live chat broadcast. let rx = publishers.chat.subscribe(); let tx = object_tx.clone(); task_handles.push(tokio::spawn(async move { @@ -619,8 +586,6 @@ async fn handle_moq_message( })); } TrackId::State => { - // State uses watch (latest-only). Send current value - // immediately, then forward changes. let initial = publishers.state.borrow().clone(); let obj = encode_message(&MoqMessage::Object { track_id: TrackId::State as u64, @@ -651,13 +616,10 @@ async fn handle_moq_message( } } })); - } - _ => { - // Video and audio use a dedicated WebSocket endpoint. - // Ignore subscribe requests on the control WS. + } + _ => {} } } - } MoqMessage::Object { track_id, @@ -665,7 +627,6 @@ async fn handle_moq_message( object_id: _, payload, } => { - // Incoming chat message from client: persist and rebroadcast. if track_id == TrackId::Chat as u64 { let json: serde_json::Value = serde_json::from_slice(&payload).unwrap_or(serde_json::Value::Null); @@ -683,7 +644,6 @@ async fn handle_moq_message( } } - // Server-initiated messages received from client are unexpected. MoqMessage::Announce { .. } | MoqMessage::AnnounceOk { .. } | MoqMessage::SubscribeOk { .. } @@ -857,7 +817,7 @@ mod tests { publishers.publish_video(5, 99, Bytes::from("video-frame")); - let obj = rx.recv().await.expect("should receive video object"); + let obj = rx.recv().await.unwrap(); assert_eq!(obj.track_id, TrackId::Video); assert_eq!(obj.payload, Bytes::from("video-frame")); } @@ -869,7 +829,7 @@ mod tests { publishers.publish_chat(0, Bytes::from("hello")); - let obj = rx.recv().await.expect("should receive chat object"); + let obj = rx.recv().await.unwrap(); assert_eq!(obj.track_id, TrackId::Chat); assert_eq!(obj.object_id, 0); assert_eq!(obj.payload, Bytes::from("hello")); @@ -884,15 +844,15 @@ mod tests { publishers.publish_chat(0, Bytes::from("second")); publishers.publish_chat(0, Bytes::from("third")); - let obj1 = rx.recv().await.expect("msg 1"); + let obj1 = rx.recv().await.unwrap(); assert_eq!(obj1.object_id, 0); assert_eq!(obj1.payload, Bytes::from("first")); - let obj2 = rx.recv().await.expect("msg 2"); + let obj2 = rx.recv().await.unwrap(); assert_eq!(obj2.object_id, 1); assert_eq!(obj2.payload, Bytes::from("second")); - let obj3 = rx.recv().await.expect("msg 3"); + let obj3 = rx.recv().await.unwrap(); assert_eq!(obj3.object_id, 2); assert_eq!(obj3.payload, Bytes::from("third")); } @@ -904,8 +864,7 @@ mod tests { publishers.publish_state(Bytes::from("playing")); - // Watch receiver needs changed() to observe the update. - rx.changed().await.expect("state should change"); + rx.changed().await.unwrap(); let val = rx.borrow().clone(); assert_eq!(val.payload, Bytes::from("playing")); assert_eq!(val.seq, 1); // started at 0, now 1 @@ -919,31 +878,26 @@ mod tests { publishers.publish_video(0, 0, Bytes::from("fanout")); - let obj1 = rx1.recv().await.expect("subscriber 1"); - let obj2 = rx2.recv().await.expect("subscriber 2"); + let obj1 = rx1.recv().await.unwrap(); + let obj2 = rx2.recv().await.unwrap(); assert_eq!(obj1.payload, Bytes::from("fanout")); assert_eq!(obj2.payload, Bytes::from("fanout")); } - // ── Init segment cache tests ─────────────────────────────────────────── - #[test] fn test_init_cache_and_retrieve() { let publishers = TrackPublishers::new(); - assert!( - publishers.get_init_segment().is_none(), - "init should be None before caching" - ); + assert!(publishers.get_init_segment().is_none()); let payload = Bytes::from("ftyp-moov-data"); publishers.cache_init_segment(payload.clone(), 42); let retrieved = publishers.get_init_segment(); - assert!(retrieved.is_some(), "init should be Some after caching"); + assert!(retrieved.is_some()); let (group_id, data) = retrieved.unwrap(); - assert_eq!(group_id, 42, "group_id should match what was cached"); - assert_eq!(data, payload, "payload should match what was cached"); + assert_eq!(group_id, 42); + assert_eq!(data, payload); } #[tokio::test] @@ -951,7 +905,7 @@ mod tests { let publishers = TrackPublishers::new(); let mut waiter = publishers.video_init_waiter(); - assert!(!*waiter.borrow(), "initial watch value should be false"); + assert!(!*waiter.borrow()); let pubs = publishers.clone(); tokio::spawn(async move { @@ -961,21 +915,15 @@ mod tests { let _ = tokio::time::timeout(std::time::Duration::from_secs(2), waiter.changed()) .await - .expect("watch should fire within timeout when init is cached"); + .expect("watch should fire within timeout"); - assert!( - *waiter.borrow(), - "watch value should be true after cache_init_segment" - ); + assert!(*waiter.borrow()); let retrieved = publishers.get_init_segment(); - assert!( - retrieved.is_some(), - "init should be retrievable after watch fires" - ); + assert!(retrieved.is_some()); let (group_id, data) = retrieved.unwrap(); - assert_eq!(group_id, 1, "group_id should match"); - assert_eq!(data, Bytes::from("init-payload"), "payload should match"); + assert_eq!(group_id, 1); + assert_eq!(data, Bytes::from("init-payload")); } #[test] @@ -986,13 +934,9 @@ mod tests { publishers.cache_init_segment(Bytes::from("init-2"), 2); let retrieved = publishers.get_init_segment(); - assert!(retrieved.is_some(), "init should be cached"); + assert!(retrieved.is_some()); let (group_id, data) = retrieved.unwrap(); - assert_eq!(group_id, 2, "group_id should be from the latest cache call"); - assert_eq!( - data, - Bytes::from("init-2"), - "payload should be from the latest cache call" - ); + assert_eq!(group_id, 2); + assert_eq!(data, Bytes::from("init-2")); } } diff --git a/src/types.rs b/src/types.rs index ca18f29..b3acd05 100644 --- a/src/types.rs +++ b/src/types.rs @@ -49,8 +49,8 @@ impl std::str::FromStr for SourceKind { /// Metadata for a single queued track. /// -/// Source-agnostic — produced by any [`MediaSource`](crate::source::MediaSource) -/// implementation (yt-dlp, direct URL detection, etc.). +/// Source-agnostic: produced by any [`MediaSource`](crate::source::MediaSource) +/// implementation. #[derive(Debug, Clone, Serialize)] pub(crate) struct TrackMeta { pub title: String, @@ -68,6 +68,7 @@ pub(crate) struct TrackState { pub(crate) title: String, pub(crate) url: String, pub(crate) duration: String, + pub(crate) thumbnail: Option, pub(crate) started_at: i64, } @@ -145,12 +146,9 @@ pub(crate) struct StateValue { pub(crate) seq: u64, } -/// All room operations — processed sequentially by the per-room actor. +/// All room operations, processed sequentially by the per-room actor. pub(crate) enum RoomCommand { - /// Track with already-resolved metadata (legacy path, used by tests). - #[allow(dead_code)] - QueueTrack(TrackMeta), - /// Raw URL — metadata extraction happens asynchronously in a spawned task. + /// Raw URL: metadata extraction happens asynchronously in a spawned task. QueueUrl(String), /// Async metadata extraction result. MetadataReady { @@ -188,11 +186,11 @@ pub(crate) struct ActiveTrackInfo { pub(crate) url: String, pub(crate) duration: String, pub(crate) thumbnail: Option, - /// Epoch ms when the track started — for client-side elapsed computation. + /// Epoch ms when the track started, for client-side elapsed computation. pub(crate) started_at_wall: i64, } -/// Public handle to a room — allows sending commands and reading publishers. +/// Public handle to a room: allows sending commands and reading publishers. #[derive(Clone)] pub(crate) struct RoomHandle { pub(crate) cmd_tx: mpsc::Sender, diff --git a/src/util.rs b/src/util.rs new file mode 100644 index 0000000..6fad486 --- /dev/null +++ b/src/util.rs @@ -0,0 +1,5 @@ +pub(crate) fn now_iso() -> String { + chrono::Utc::now() + .format("%Y-%m-%dT%H:%M:%S%.3fZ") + .to_string() +} diff --git a/src/web.rs b/src/web.rs index 84eda2b..3341c51 100644 --- a/src/web.rs +++ b/src/web.rs @@ -1,4 +1,4 @@ -//! SSR frontend — Axum router, handlers, and templates. +//! SSR frontend: Axum router, handlers, and templates. use std::collections::HashMap; use std::net::SocketAddr; @@ -59,19 +59,15 @@ pub(crate) struct AppState { pub rate_limiter: RateLimiter, } -pub(crate) fn router( - rooms: room::Registry, - pool: Pool, -) -> Router { +pub(crate) fn router(rooms: room::Registry, pool: Pool) -> Router { let rate_limiter = RateLimiter::new(); - // Periodic cleanup for rate limiter entries (every 30s). - let rl = rate_limiter.clone(); + let cleanup_limiter = rate_limiter.clone(); tokio::spawn(async move { let mut interval = tokio::time::interval(std::time::Duration::from_secs(30)); loop { interval.tick().await; - rl.cleanup().await; + cleanup_limiter.cleanup().await; } }); @@ -123,7 +119,6 @@ async fn room_view( State(state): State, Path(id): Path, ) -> Result, StatusCode> { - // Validate room ID format (snowflake digits only). if !id.chars().all(|c| c.is_ascii_digit()) { return Err(StatusCode::BAD_REQUEST); } @@ -131,7 +126,6 @@ async fn room_view( let queue = state.rooms.queue(&room_id).await; let queue_empty = queue.is_empty(); - // Fetch history via Playlist. let history = { let playlist = crate::playlist::Playlist::new(state.pool.clone(), &id); playlist @@ -144,13 +138,10 @@ async fn room_view( duration: i.duration, thumbnail: i.thumbnail, url: i.url, - source: i - .source - .parse() - .unwrap_or_else(|s: String| { - tracing::warn!(source = %s, "unknown source kind in queue_items"); - crate::types::SourceKind::Direct - }), + source: i.source.parse().unwrap_or_else(|s: String| { + tracing::warn!(source = %s, "unknown source kind in queue_items"); + crate::types::SourceKind::Direct + }), }) .collect::>() }; @@ -172,7 +163,7 @@ async fn room_view( async fn create_room(State(state): State) -> Result { let room = state.rooms.create().await; - Ok(Json(serde_json::json!({ "room_id": room.id.0 }))) + Ok(Json(serde_json::json!({ "room_id": room.0 }))) } #[derive(serde::Deserialize)] @@ -190,44 +181,35 @@ async fn ingest_url( ConnectInfo(addr): ConnectInfo, axum::Json(req): axum::Json, ) -> Result, axum::response::Response> { - // Input validation — URL max 2048 characters. if req.url.len() > 2048 { + // 2048 is the practical URL max across browsers return Err(err_response(StatusCode::BAD_REQUEST, "URL too long")); } - // Input validation — room ID must be digits (snowflake format). if !req.room_id.chars().all(|c| c.is_ascii_digit()) { return Err(err_response(StatusCode::BAD_REQUEST, "invalid room ID")); } - // Rate limiting — max 1 request per 5 seconds per IP. if let Err(retry_after) = state.rate_limiter.check(addr).await { let mut res = err_response(StatusCode::TOO_MANY_REQUESTS, "rate limited"); res.headers_mut().insert( axum::http::header::RETRY_AFTER, - retry_after.as_secs().to_string().parse().expect("valid retry-after seconds"), + retry_after + .as_secs() + .to_string() + .parse() + .expect("valid retry-after seconds"), ); return Err(res); } let room_id = RoomId(req.room_id); - let t0 = std::time::Instant::now(); - - // Validate the room exists. if !state.rooms.exists(&room_id).await { return Err(err_response(StatusCode::NOT_FOUND, "room not found")); } - let t1 = t0.elapsed(); - // Push URL to room — metadata extraction runs async, no client wait. state.rooms.push_url(&room_id, req.url).await; - let t2 = t0.elapsed(); - - tracing::info!( - "ingest_url timing: exists={:.0?} push_url={:.0?} total={:.0?}", - t1, t2 - t1, t2 - ); Ok(Json(serde_json::json!({ "ok": true }))) } @@ -243,7 +225,6 @@ async fn skip_handler( ) -> Result, (StatusCode, Json)> { let room_id = RoomId(req.room_id); - // Validate the room exists. if !state.rooms.exists(&room_id).await { return Err(( StatusCode::NOT_FOUND, @@ -256,23 +237,18 @@ async fn skip_handler( Ok(Json(serde_json::json!({ "ok": true }))) } -/// WebSocket upgrade endpoint: `/ws/{room_id}?name={user_name}`. -/// -/// Validates the room exists, then upgrades to the MoQ session handler. async fn ws_handler( ws: WebSocketUpgrade, Path(room_id): Path, Query(params): Query>, State(state): State, ) -> Result { - // Validate room ID format (snowflake digits only). if !room_id.chars().all(|c| c.is_ascii_digit()) { return Err((StatusCode::BAD_REQUEST, "invalid room ID")); } let rid = RoomId(room_id); - // Reject unknown rooms before upgrading. if !state.rooms.exists(&rid).await { return Err((StatusCode::NOT_FOUND, "room not found")); } @@ -288,8 +264,8 @@ async fn ws_handler( .cloned() .unwrap_or_else(|| "anonymous".into()); - // Validate username. if user_name.is_empty() || user_name.len() > 32 { + // 32 chars is enough for display names return Err((StatusCode::BAD_REQUEST, "invalid username")); } @@ -306,11 +282,6 @@ async fn ws_handler( })) } -/// WebSocket upgrade endpoint for video-only delivery: `/ws/{room_id}/video`. -/// -/// No MoQ protocol — server auto-subscribes to the video broadcast and -/// forwards raw fMP4 payloads. The client identifies init segments by -/// checking for the ftyp box in the data. async fn video_ws_handler( ws: WebSocketUpgrade, Path(room_id): Path, diff --git a/templates/index.html b/templates/index.html index b8c5f61..d7ef34b 100644 --- a/templates/index.html +++ b/templates/index.html @@ -1,165 +1,165 @@ - - + + moqbox - - + + + >

moqbox

- Ultra-low latency shared media rooms over MoQ.
+ Ultra-low latency shared media rooms over MoQ.
Paste a link from YouTube, SoundCloud, or Bandcamp to start.

@@ -172,22 +172,20 @@ placeholder="Enter room code..." maxlength="64" onkeydown="if(event.key==='Enter'){window.location='/room/'+this.value}" - /> + >
- moqbox v{{ version }} + moqbox v{{ version }}
diff --git a/templates/room.html b/templates/room.html index 519c47a..53c380e 100644 --- a/templates/room.html +++ b/templates/room.html @@ -1,747 +1,764 @@ - - + + {{ room_name }} — moqbox - - + + + > @@ -800,12 +819,7 @@ {{ room_id }}
- 0 viewers @@ -817,18 +831,11 @@
🔇 - +
- +
@@ -854,8 +861,7 @@
🎵
Queue is empty
- Paste a link from YouTube, SoundCloud, or Bandcamp above to get - started. + Paste a link from YouTube, SoundCloud, or Bandcamp above to get started.
{% else %} @@ -863,7 +869,7 @@ {% for item in queue %}
  • {% if let Some(thumb) = item.thumbnail %} - + {% else %}
    {% endif %} @@ -887,15 +893,9 @@ type="text" id="url-input" placeholder="Paste YouTube / SoundCloud / Bandcamp URL..." - /> - - + +
    • @@ -906,17 +906,13 @@ {% for item in history %}
  • {% if let Some(thumb) = item.thumbnail %} - + {% else %}
    {% endif %}
    {{ item.title }}
    {{ item.duration }}
    - source
    • @@ -927,727 +923,677 @@
    -
    - disconnected -
    +
    disconnected
    - +
    diff --git a/treefmt.toml b/treefmt.toml new file mode 100644 index 0000000..8f50c5b --- /dev/null +++ b/treefmt.toml @@ -0,0 +1,11 @@ +excludes = ["flake.lock", ".envrc"] + +[formatter.nix] +command = "nixfmt" +includes = ["*.nix"] + +[formatter.rust] +command = "rustfmt" +includes = ["*.rs"] + + -- knotmirror 0.0.1