+3

Cargo.toml

Reviewed

··· 34 34 rustls = { version = "0.23", features = ["ring"] } 35 35 rcgen = "0.13" 36 36 37 37 + # ── Async ───────────────────────────────────────────────────────── 38 38 + async-trait = "0.1" 39 39 + 37 40 # ── Identifiers ──────────────────────────────────────────────────── 38 41 uuid = { version = "1", features = ["v4", "serde"] } 39 42

+32 -1

src/db.rs

Reviewed

··· 47 47 title TEXT NOT NULL, 48 48 duration TEXT NOT NULL DEFAULT '--:--', 49 49 thumbnail TEXT, 50 50 - extractor TEXT NOT NULL DEFAULT '', 50 50 + source TEXT NOT NULL DEFAULT '', 51 51 added_by TEXT, 52 52 position INTEGER NOT NULL, 53 53 added_at TEXT NOT NULL, ··· 84 84 sqlx::query("ALTER TABLE queue_items ADD COLUMN started_playing_at TEXT") 85 85 .execute(pool) 86 86 .await?; 87 87 + } 88 88 + 89 89 + // Migration: rename `extractor` column to `source` (schema cleanup). 90 90 + let has_source: bool = sqlx::query_scalar( 91 91 + "SELECT COUNT(*) FROM pragma_table_info('queue_items') WHERE name = 'source'", 92 92 + ) 93 93 + .fetch_one(pool) 94 94 + .await 95 95 + .map(|c: i64| c > 0) 96 96 + .unwrap_or(false); 97 97 + 98 98 + if !has_source { 99 99 + // For pre-refactor databases: rename extractor → source. 100 100 + let has_extractor: bool = sqlx::query_scalar( 101 101 + "SELECT COUNT(*) FROM pragma_table_info('queue_items') WHERE name = 'extractor'", 102 102 + ) 103 103 + .fetch_one(pool) 104 104 + .await 105 105 + .map(|c: i64| c > 0) 106 106 + .unwrap_or(false); 107 107 + 108 108 + if has_extractor { 109 109 + sqlx::query("ALTER TABLE queue_items RENAME COLUMN extractor TO source") 110 110 + .execute(pool) 111 111 + .await?; 112 112 + } else { 113 113 + // Safety net: neither column exists (shouldn't happen). 114 114 + sqlx::query("ALTER TABLE queue_items ADD COLUMN source TEXT NOT NULL DEFAULT ''") 115 115 + .execute(pool) 116 116 + .await?; 117 117 + } 87 118 } 88 119 89 120 Ok(())

+11 -2

src/main.rs

Reviewed

··· 5 5 6 6 mod db; 7 7 mod media; 8 8 + mod playback; 8 9 mod playlist; 9 10 mod room; 11 11 + mod source; 10 12 mod transport; 11 13 mod types; 12 14 mod web; ··· 14 16 use clap::Parser; 15 17 use std::net::SocketAddr; 16 18 use std::path::PathBuf; 19 19 + use std::sync::Arc; 17 20 use std::time::Duration; 18 21 use tracing_subscriber::EnvFilter; 19 22 ··· 55 58 db::run_migrations(&pool).await?; 56 59 tracing::info!("database ready at {}", args.db_path.display()); 57 60 58 58 - let rooms = room::Registry::new(pool.clone(), args.cache_dir.clone()); 61 61 + // ── Source registry ────────────────────────────────────────────── 62 62 + let mut sources = source::SourceRegistry::new(); 63 63 + sources.register(Box::new(source::ytdlp::YtdlpSource::new())); 64 64 + sources.register(Box::new(source::direct::DirectSource::new())); 65 65 + let sources = Arc::new(sources); 66 66 + 67 67 + let rooms = room::Registry::new(pool.clone(), args.cache_dir.clone(), sources.clone()); 59 68 60 69 // Idle room sweeper: every 60s, remove rooms idle for 600s (10 min). 61 70 let registry_clone = rooms.clone(); ··· 70 79 } 71 80 }); 72 81 73 73 - let app = web::router(rooms, pool, args.cache_dir); 82 82 + let app = web::router(rooms, pool, args.cache_dir, sources); 74 83 let listener = tokio::net::TcpListener::bind(args.http_addr).await?; 75 84 76 85 tracing::info!("HTTP server listening on {}", args.http_addr);

+99 -206

src/media.rs

Reviewed

··· 1 1 - //! Media extraction and transcoding pipeline. 1 1 + //! FFmpeg transcoding to fragmented MP4. 2 2 //! 3 3 - //! Wraps `yt-dlp` for metadata extraction and stream URL resolution, 4 4 - //! and `ffmpeg` for transcoding to raw fMP4 chunks on stdout. 5 5 - //! No fMP4 parsing — just reads stdout chunks and publishes as MoQ objects. 3 3 + //! Takes a playable stream URL, spawns `ffmpeg` to transcode to fMP4 at 4 4 + //! real-time speed (`-re`), reads fMP4 boxes from stdout as moof+mdat pairs, 5 5 + //! and publishes them over per-room broadcast channels. 6 6 + //! 7 7 + //! This module does **not** resolve URLs or extract metadata — that's the 8 8 + //! responsibility of [`crate::source`]. 6 9 7 7 - use std::path::Path; 8 8 - 9 9 - use bytes::Bytes; 10 10 - use serde::Deserialize; 11 10 use tokio::io::AsyncReadExt; 12 11 use tokio::process::Command; 13 12 14 14 - use crate::types::{TrackMeta, TrackPublishers}; 13 13 + use crate::transport::TrackPublishers; 15 14 16 16 - /// Errors that can occur during media pipeline operations. 15 15 + // --------------------------------------------------------------------------- 16 16 + // Error 17 17 + // --------------------------------------------------------------------------- 18 18 + 19 19 + /// Errors that can occur during FFmpeg transcoding. 17 20 #[derive(Debug, thiserror::Error)] 18 18 - pub(crate) enum MediaError { 19 19 - #[error("extraction failed: {0}")] 20 20 - Extraction(String), 21 21 + pub(crate) enum TranscodeError { 21 22 #[error("transcoding failed: {0}")] 22 23 Transcode(String), 23 24 #[error("io error: {0}")] ··· 26 27 Aborted, 27 28 } 28 29 29 29 - /// Raw yt-dlp `--dump-json` output fields we care about. 30 30 - #[derive(Debug, Deserialize)] 31 31 - #[serde(rename_all = "snake_case")] 32 32 - struct YtdlpMetadata { 33 33 - title: String, 34 34 - duration: Option<f64>, 35 35 - thumbnail: Option<String>, 36 36 - webpage_url: String, 37 37 - #[serde(default)] 38 38 - extractor: String, 39 39 - } 30 30 + // --------------------------------------------------------------------------- 31 31 + // Public API 32 32 + // --------------------------------------------------------------------------- 40 33 41 41 - /// Spawn FFmpeg pipeline and publish raw fMP4 chunks as MoQ objects. 34 34 + /// Transcode a playable stream URL to fMP4, publishing raw boxes as MoQ objects. 35 35 + /// 36 36 + /// Spawns `ffmpeg` with the stream URL as input, transcodes to fragmented MP4 37 37 + /// on stdout, and publishes each media segment (moof+mdat pair) as a single 38 38 + /// MoQ object. The init segment (ftyp+moov) is cached so late-joining clients 39 39 + /// can initialise their MediaSource. 42 40 /// 43 41 /// Returns `Ok(())` on normal EOF, `Err` on failure. 44 44 - pub(crate) async fn spawn_pipeline( 45 45 - url: &str, 46 46 - cache_dir: &Path, 42 42 + /// 43 43 + /// ## Abort behaviour 44 44 + /// 45 45 + /// When `abort` is signalled, the pipeline exits early with 46 46 + /// [`TranscodeError::Aborted`] and the ffmpeg process is killed. 47 47 + pub(crate) async fn transcode_to_fmp4( 48 48 + stream_url: &str, 47 49 publishers: TrackPublishers, 48 50 mut abort: tokio::sync::watch::Receiver<bool>, 49 51 group_id: u64, 50 50 - ) -> Result<(), MediaError> { 51 51 - tracing::debug!(url, "spawning ffmpeg pipeline"); 52 52 + ) -> Result<(), TranscodeError> { 53 53 + tracing::debug!(url = %stream_url, "spawning ffmpeg pipeline"); 52 54 53 53 - // 1. Get direct stream URL from yt-dlp (best single-file format). 54 54 - // Using -f b ensures a combined audio+video URL; -g alone returns 55 55 - // separate audio and video URLs which ffmpeg can't use with one -i. 56 56 - let output = Command::new("yt-dlp") 57 57 - .args([ 58 58 - "-f", 59 59 - "b", 60 60 - "-g", 61 61 - "--no-playlist", 62 62 - "--no-warnings", 63 63 - "--cache-dir", 64 64 - ]) 65 65 - .arg(cache_dir) 66 66 - .arg(url) 67 67 - .kill_on_drop(true) 68 68 - .output() 69 69 - .await 70 70 - .map_err(MediaError::Io)?; 71 71 - 72 72 - if !output.status.success() { 73 73 - let stderr = String::from_utf8_lossy(&output.stderr); 74 74 - return Err(MediaError::Extraction(stderr.trim().to_string())); 75 75 - } 76 76 - 77 77 - let stream_url = String::from_utf8(output.stdout) 78 78 - .map_err(|_| MediaError::Extraction("invalid UTF-8 from yt-dlp".into()))? 79 79 - .trim() 80 80 - .to_string(); 81 81 - 82 82 - if stream_url.is_empty() { 83 83 - return Err(MediaError::Extraction( 84 84 - "yt-dlp returned empty stream URL".into(), 85 85 - )); 86 86 - } 87 87 - // 2. Spawn FFmpeg transcoding to fMP4 on stdout. 88 88 - // -re throttles the input to native frame rate so output 89 89 - // stays in sync with wall-clock time. Without it FFmpeg 90 90 - // transcodes faster than real-time, making late-join 91 91 - // position calculations impossible. 55 55 + // Spawn FFmpeg transcoding to fMP4 on stdout. 56 56 + // -re throttles the input to native frame rate so output 57 57 + // stays in sync with wall-clock time. Without it FFmpeg 58 58 + // transcodes faster than real-time, making late-join 59 59 + // position calculations impossible. 92 60 let mut ffmpeg = Command::new("ffmpeg") 93 61 .args([ 94 62 "-re", 95 63 "-i", 96 96 - &stream_url, 64 64 + stream_url, 97 65 "-c:v", 98 66 "libx264", 99 67 "-profile:v", ··· 122 90 .stderr(std::process::Stdio::piped()) 123 91 .kill_on_drop(true) 124 92 .spawn() 125 125 - .map_err(|e| MediaError::Transcode(format!("failed to spawn ffmpeg: {e}")))?; 93 93 + .map_err(|e| TranscodeError::Transcode(format!("failed to spawn ffmpeg: {e}")))?; 126 94 127 95 tracing::debug!("ffmpeg pipeline started"); 128 96 129 97 let mut stdout = ffmpeg 130 98 .stdout 131 99 .take() 132 132 - .ok_or_else(|| MediaError::Transcode("no stdout from ffmpeg".into()))?; 100 100 + .ok_or_else(|| TranscodeError::Transcode("no stdout from ffmpeg".into()))?; 133 101 134 102 // Spawn background task to drain stderr (prevents pipe deadlock at 64KB). 135 103 let stderr_drain = ffmpeg.stderr.take().map(|stderr| { ··· 141 109 }) 142 110 }); 143 111 144 144 - // 3. Read and publish each fMP4 box as a MoQ object. 145 145 - // Box format: [4-byte big-endian size][4-byte type][payload] 112 112 + // 1. Read and publish each fMP4 box as a MoQ object. 146 113 // Publishing complete boxes ensures MSE can consume them directly. 147 114 read_and_publish_boxes(&mut stdout, &publishers, &mut abort, group_id).await?; 148 115 149 149 - // 4. Wait for FFmpeg to exit 116 116 + // 2. Wait for FFmpeg to exit 150 117 let status = ffmpeg 151 118 .wait() 152 119 .await 153 153 - .map_err(|e| MediaError::Transcode(format!("ffmpeg wait failed: {e}")))?; 120 120 + .map_err(|e| TranscodeError::Transcode(format!("ffmpeg wait failed: {e}")))?; 154 121 155 155 - // 5. Collect stderr output (background task already completed since 122 122 + // 3. Collect stderr output (background task already completed since 156 123 // ffmpeg closed stderr on exit). 157 124 let stderr_output = match stderr_drain { 158 125 Some(handle) => handle.await.unwrap_or_default(), ··· 169 136 Ok(()) 170 137 } 171 138 139 139 + // --------------------------------------------------------------------------- 140 140 + // fMP4 box parsing 141 141 + // --------------------------------------------------------------------------- 142 142 + 172 143 /// Read fMP4 boxes from FFmpeg stdout, buffer moof+mdat pairs, and publish 173 144 /// each complete segment as a single MoQ object. 174 174 - /// 175 175 - /// Box format: [4-byte big-endian size][4-byte type][payload] 176 145 /// 177 146 /// MSE requires moof and its following mdat to be appended as one buffer. 178 147 /// This function buffers moof, waits for the next mdat, concatenates them, 179 148 /// and publishes the pair as a single object. 180 180 - /// 181 181 - /// Returns Ok on EOF, Err on read failure. 182 149 async fn read_and_publish_boxes( 183 150 stdout: &mut (impl tokio::io::AsyncRead + Unpin), 184 151 publishers: &TrackPublishers, 185 152 abort: &mut tokio::sync::watch::Receiver<bool>, 186 153 group_id: u64, 187 187 - ) -> Result<(), MediaError> { 154 154 + ) -> Result<(), TranscodeError> { 188 155 // Check abort signal before starting (handles pre-signalled aborts). 189 156 if *abort.borrow() { 190 157 tracing::debug!("pipeline aborted before starting"); 191 191 - return Err(MediaError::Aborted); 158 158 + return Err(TranscodeError::Aborted); 192 159 } 193 160 194 161 let mut object_id: u64 = 0; ··· 199 166 // Read 8-byte box header 200 167 let mut header = [0u8; 8]; 201 168 tokio::select! { 202 202 - _ = abort.changed() => return Err(MediaError::Aborted), 169 169 + _ = abort.changed() => return Err(TranscodeError::Aborted), 203 170 result = stdout.read_exact(&mut header) => { 204 171 match result { 205 172 Err(ref e) if e.kind() == std::io::ErrorKind::UnexpectedEof => return Ok(()), 206 206 - Err(e) => return Err(MediaError::Io(e)), 173 173 + Err(e) => return Err(TranscodeError::Io(e)), 207 174 Ok(_) => {} 208 175 } 209 176 } ··· 214 181 215 182 match &box_type { 216 183 b"ftyp" => { 217 217 - // Buffer ftyp — wait for moov before publishing init segment. 218 184 init_data = box_data.to_vec(); 219 185 } 220 186 b"moov" => { 221 221 - // Concatenate with buffered ftyp and publish as one init segment. 222 187 if !init_data.is_empty() { 223 188 let mut init = std::mem::take(&mut init_data); 224 189 init.extend_from_slice(&box_data); 225 225 - publishers.cache_init_segment(Bytes::copy_from_slice(&init), group_id); 226 226 - // Init is NOT published to broadcast — the forward task sends 227 227 - // it explicitly from cache so late joiners can initialise. 190 190 + let payload = bytes::Bytes::copy_from_slice(&init); 191 191 + // Cache for late-joining clients (forward task sends from cache). 192 192 + publishers.cache_init_segment(payload.clone(), group_id); 193 193 + // Also publish to broadcast so existing clients on track change 194 194 + // (e.g. after skip) receive the new init and can initialise. 195 195 + publishers.publish_video(group_id, object_id, payload); 228 196 object_id += 1; 229 197 } else { 230 230 - publishers.publish_video(group_id, object_id, Bytes::from(box_data)); 198 198 + publishers.publish_video(group_id, object_id, bytes::Bytes::from(box_data)); 231 199 object_id += 1; 232 200 } 233 201 } ··· 238 206 if let Some(moof) = pending_moof.take() { 239 207 let mut combined = moof; 240 208 combined.extend_from_slice(&box_data); 241 241 - publishers.publish_video(group_id, object_id, Bytes::from(combined)); 209 209 + publishers.publish_video(group_id, object_id, bytes::Bytes::from(combined)); 242 210 object_id += 1; 243 211 } else { 244 244 - publishers.publish_video(group_id, object_id, Bytes::from(box_data)); 212 212 + publishers.publish_video(group_id, object_id, bytes::Bytes::from(box_data)); 245 213 object_id += 1; 246 214 } 247 215 } 248 216 _ => { 249 249 - publishers.publish_video(group_id, object_id, Bytes::from(box_data)); 217 217 + publishers.publish_video(group_id, object_id, bytes::Bytes::from(box_data)); 250 218 object_id += 1; 251 219 } 252 220 } ··· 254 222 } 255 223 256 224 /// Read the complete bytes of an ISOBMFF box given its 8-byte header. 257 257 - /// 258 258 - /// Returns the full box bytes (header + payload, including extended-size 259 259 - /// field if applicable), suitable for MSE SourceBuffer.appendBuffer(). 260 225 async fn read_box_payload<R: tokio::io::AsyncRead + Unpin>( 261 226 reader: &mut R, 262 227 header: &[u8; 8], 263 263 - ) -> Result<Vec<u8>, MediaError> { 228 228 + ) -> Result<Vec<u8>, TranscodeError> { 264 229 let size = u32::from_be_bytes([header[0], header[1], header[2], header[3]]); 265 230 266 231 if size == 0 { ··· 277 242 let mut ext = [0u8; 8]; 278 243 reader.read_exact(&mut ext).await?; 279 244 let total = u64::from_be_bytes(ext) as usize; 280 280 - let payload_len = total - 16; // header(8) + ext(8) 245 245 + let payload_len = total - 16; 281 246 let mut payload = vec![0u8; payload_len]; 282 247 reader.read_exact(&mut payload).await?; 283 248 let mut full = Vec::with_capacity(total); ··· 296 261 Ok(full) 297 262 } 298 263 299 299 - /// Run `yt-dlp --dump-json <url>` and parse the result. 300 300 - /// 301 301 - /// Returns a [`TrackMeta`] suitable for the room queue. 302 302 - /// 303 303 - /// # Errors 304 304 - /// 305 305 - /// Returns an error if yt-dlp is not installed, the URL is not 306 306 - /// extractable, or the process times out. 307 307 - pub(crate) async fn extract(url: &str, cache_dir: &Path) -> anyhow::Result<TrackMeta> { 308 308 - if !url.starts_with("http://") && !url.starts_with("https://") { 309 309 - anyhow::bail!("URL must start with http:// or https://"); 310 310 - } 264 264 + // --------------------------------------------------------------------------- 265 265 + // Formatting 266 266 + // --------------------------------------------------------------------------- 311 267 312 312 - let output = Command::new("yt-dlp") 313 313 - .args([ 314 314 - "--dump-json", 315 315 - "--no-playlist", 316 316 - "--no-warnings", 317 317 - "--cache-dir", 318 318 - ]) 319 319 - .arg(cache_dir) 320 320 - .arg(url) 321 321 - .kill_on_drop(true) 322 322 - .output() 323 323 - .await?; 324 324 - 325 325 - if !output.status.success() { 326 326 - let stderr = String::from_utf8_lossy(&output.stderr); 327 327 - anyhow::bail!("yt-dlp failed: {}", stderr.trim()); 328 328 - } 329 329 - 330 330 - let raw: YtdlpMetadata = serde_json::from_slice(&output.stdout)?; 331 331 - 332 332 - let duration = raw 333 333 - .duration 334 334 - .map(format_duration) 335 335 - .unwrap_or_else(|| "--:--".into()); 336 336 - 337 337 - Ok(TrackMeta { 338 338 - title: raw.title, 339 339 - duration, 340 340 - thumbnail: raw.thumbnail, 341 341 - url: raw.webpage_url, 342 342 - extractor: raw.extractor, 343 343 - }) 344 344 - } 345 345 - 346 346 - fn format_duration(total_secs: f64) -> String { 268 268 + /// Format a duration in seconds to a human-readable `M:SS` or `H:MM:SS` string. 269 269 + pub(crate) fn format_duration(total_secs: f64) -> String { 347 270 let total = total_secs as u64; 348 271 let hours = total / 3600; 349 272 let minutes = (total % 3600) / 60; ··· 356 279 } 357 280 } 358 281 282 282 + // --------------------------------------------------------------------------- 283 283 + // Tests 284 284 + // --------------------------------------------------------------------------- 285 285 + 359 286 #[cfg(test)] 360 287 mod tests { 361 288 use super::*; 362 362 - use std::path::Path; 363 289 364 290 #[test] 365 291 fn test_format_duration() { ··· 368 294 assert_eq!(format_duration(3661.0), "1:01:01"); 369 295 } 370 296 371 371 - #[test] 372 372 - fn test_extract_rejects_bad_urls() { 373 373 - let rt = tokio::runtime::Runtime::new().unwrap(); 374 374 - let result = rt.block_on(extract("javascript:alert(1)", Path::new("/tmp"))); 375 375 - assert!(result.is_err()); 376 376 - assert!(result.unwrap_err().to_string().contains("http")); 297 297 + fn make_box(box_type: &[u8; 4], payload: &[u8]) -> Vec<u8> { 298 298 + let size = 8 + payload.len() as u32; 299 299 + let mut buf = Vec::with_capacity(size as usize); 300 300 + buf.extend_from_slice(&size.to_be_bytes()); 301 301 + buf.extend_from_slice(box_type); 302 302 + buf.extend_from_slice(payload); 303 303 + buf 377 304 } 378 305 379 306 #[tokio::test] 380 307 async fn test_moof_mdat_combining() { 381 381 - fn make_box(box_type: &[u8; 4], payload: &[u8]) -> Vec<u8> { 382 382 - let size = 8 + payload.len() as u32; 383 383 - let mut buf = Vec::with_capacity(size as usize); 384 384 - buf.extend_from_slice(&size.to_be_bytes()); 385 385 - buf.extend_from_slice(box_type); 386 386 - buf.extend_from_slice(payload); 387 387 - buf 388 388 - } 389 389 - 390 390 - // Build a minimal fMP4 sequence: ftyp, moov, moof, mdat. 391 308 let ftyp = make_box(b"ftyp", b"iso5"); 392 309 let moov = make_box(b"moov", b"moovdata"); 393 310 let moof = make_box(b"moof", b"moofdata"); ··· 408 325 .await 409 326 .expect("read_and_publish_boxes should succeed on valid fMP4"); 410 327 411 411 - // Init segment is NOT published to broadcast — only cached. 412 412 - // Verify the cache contains ftyp+moov. 328 328 + // Init is cached (for late joiners) AND published to broadcast (for track changes). 413 329 let cached = publishers.get_init_segment(); 414 330 assert!(cached.is_some(), "init cache should contain ftyp+moov"); 415 331 let (cached_group, cached_data) = cached.unwrap(); ··· 418 334 expected_init.extend_from_slice(&moov); 419 335 assert_eq!(cached_data.to_vec(), expected_init); 420 336 421 421 - // Object 1: moof+mdat combined into one media segment 422 422 - // (object_id=0 was skipped — it's reserved for the forward task's cached init) 337 337 + // Object 0: init (ftyp+moov) published to broadcast. 338 338 + let init_obj = rx.recv().await.expect("should receive init via broadcast"); 339 339 + assert_eq!(init_obj.object_id, 0); 340 340 + let mut expected_init_broadcast = ftyp.clone(); 341 341 + expected_init_broadcast.extend_from_slice(&moov); 342 342 + assert_eq!(init_obj.payload.to_vec(), expected_init_broadcast); 343 343 + 344 344 + // Object 1: moof+mdat combined. 423 345 let obj1 = rx.recv().await.expect("should receive combined moof+mdat"); 424 346 assert_eq!(obj1.object_id, 1); 425 347 let mut expected_media = moof.clone(); ··· 427 349 assert_eq!(obj1.payload.to_vec(), expected_media); 428 350 } 429 351 430 430 - /// Verify init segment (ftyp+moov) is cached but NOT published to broadcast. 431 431 - /// Only the moof+mdat media segment should appear on the broadcast channel, 432 432 - /// with object_id=1 (object_id=0 is reserved for the forward task's cached init). 433 352 #[tokio::test] 434 434 - async fn test_init_not_published_to_broadcast() { 435 435 - fn make_box(box_type: &[u8; 4], payload: &[u8]) -> Vec<u8> { 436 436 - let size = 8 + payload.len() as u32; 437 437 - let mut buf = Vec::with_capacity(size as usize); 438 438 - buf.extend_from_slice(&size.to_be_bytes()); 439 439 - buf.extend_from_slice(box_type); 440 440 - buf.extend_from_slice(payload); 441 441 - buf 442 442 - } 443 443 - 353 353 + async fn test_init_broadcast_and_cache() { 444 354 let ftyp = make_box(b"ftyp", b"iso5"); 445 355 let moov = make_box(b"moov", b"moovdata"); 446 356 let moof = make_box(b"moof", b"moofdata"); ··· 461 371 .await 462 372 .expect("read_and_publish_boxes should succeed"); 463 373 464 464 - // Verify init is cached but NOT broadcast. 374 374 + // Init is both cached and broadcast. 465 375 let cached = publishers.get_init_segment(); 466 376 assert!(cached.is_some(), "init cache should contain ftyp+moov"); 467 467 - let (cached_group, cached_data) = cached.unwrap(); 468 468 - assert_eq!(cached_group, 0); 377 377 + 378 378 + // Object 0: init via broadcast. 379 379 + let init = rx.recv().await.expect("should receive init via broadcast"); 380 380 + assert_eq!(init.object_id, 0); 381 381 + assert_eq!(init.group_id, 0); 469 382 let mut expected_init = ftyp.clone(); 470 383 expected_init.extend_from_slice(&moov); 471 471 - assert_eq!(cached_data.to_vec(), expected_init); 384 384 + assert_eq!(init.payload.to_vec(), expected_init); 472 385 473 473 - // Only one broadcast message: moof+mdat at object_id=1. 386 386 + // Object 1: moof+mdat media segment. 474 387 let media = rx.recv().await.expect("should receive media segment"); 475 475 - assert_eq!( 476 476 - media.object_id, 1, 477 477 - "media segment should have object_id=1 (0 reserved for cached init)" 478 478 - ); 479 479 - assert_eq!(media.group_id, 0); 388 388 + assert_eq!(media.object_id, 1); 480 389 let mut expected_media = moof.clone(); 481 390 expected_media.extend_from_slice(&mdat); 482 391 assert_eq!(media.payload.to_vec(), expected_media); 483 483 - 484 484 - // Confirm no second broadcast message. 485 485 - match tokio::time::timeout(std::time::Duration::from_millis(100), rx.recv()).await { 486 486 - Err(tokio::time::error::Elapsed { .. }) => {} // expected: no more messages 487 487 - Ok(_) => panic!("should not receive a second broadcast message"), 488 488 - } 489 392 } 490 393 491 491 - /// Verify get_init_segment returns the correct group_id. 492 394 #[tokio::test] 493 395 async fn test_init_cache_has_group_id() { 494 494 - fn make_box(box_type: &[u8; 4], payload: &[u8]) -> Vec<u8> { 495 495 - let size = 8 + payload.len() as u32; 496 496 - let mut buf = Vec::with_capacity(size as usize); 497 497 - buf.extend_from_slice(&size.to_be_bytes()); 498 498 - buf.extend_from_slice(box_type); 499 499 - buf.extend_from_slice(payload); 500 500 - buf 501 501 - } 502 502 - 503 396 let ftyp = make_box(b"ftyp", b"iso5"); 504 397 let moov = make_box(b"moov", b"moovdata"); 505 398

+299

src/playback.rs

Reviewed

··· 1 1 + //! Playback lifecycle management. 2 2 + //! 3 3 + //! The [`Player`] struct owns the lifecycle of the currently-playing track: 4 4 + //! starting a new track (resolving its stream URL and transcoding to fMP4), 5 5 + //! aborting playback, and detecting when a track has ended. 6 6 + //! 7 7 + //! This separates playback concerns from room orchestration — the room actor 8 8 + //! tells the player *what* to play, not *how* to play it. 9 9 + 10 10 + use std::path::PathBuf; 11 11 + use std::sync::Arc; 12 12 + use std::time::Duration; 13 13 + 14 14 + use tokio::sync::{mpsc, watch}; 15 15 + 16 16 + use crate::media; 17 17 + use crate::playlist::QueueItem; 18 18 + use crate::source::SourceRegistry; 19 19 + use crate::transport::TrackPublishers; 20 20 + use crate::types::{ActiveTrackInfo, RoomCommand}; 21 21 + 22 22 + /// Manages the lifecycle of the currently-playing track. 23 23 + /// 24 24 + /// # Lifecycle 25 25 + /// 26 26 + /// 1. **Start**: [`Player::start`] resolves the URL via the source registry, 27 27 + /// spawns ffmpeg transcoding, and stores an abort handle. 28 28 + /// 2. **End**: The spawned task sends [`RoomCommand::TrackEnded`] when ffmpeg 29 29 + /// finishes or errors. The room actor calls [`Player::on_track_ended`] to 30 30 + /// acknowledge. 31 31 + /// 3. **Skip**: [`Player::abort`] signals the transcoding task to stop. 32 32 + /// 33 33 + /// The player is **not** responsible for queue management — that belongs to 34 34 + /// the room actor. 35 35 + pub(crate) struct Player { 36 36 + source_registry: Arc<SourceRegistry>, 37 37 + cache_dir: PathBuf, 38 38 + /// Channel to send [`RoomCommand::TrackEnded`] back to the room actor. 39 39 + pipeline_tx: mpsc::Sender<RoomCommand>, 40 40 + active_track: Option<ActiveTrackInfo>, 41 41 + abort_tx: Option<watch::Sender<bool>>, 42 42 + } 43 43 + 44 44 + impl Player { 45 45 + pub(crate) fn new( 46 46 + source_registry: Arc<SourceRegistry>, 47 47 + cache_dir: PathBuf, 48 48 + pipeline_tx: mpsc::Sender<RoomCommand>, 49 49 + ) -> Self { 50 50 + Self { 51 51 + source_registry, 52 52 + cache_dir, 53 53 + pipeline_tx, 54 54 + active_track: None, 55 55 + abort_tx: None, 56 56 + } 57 57 + } 58 58 + 59 59 + /// Start playing a queue item. 60 60 + /// 61 61 + /// Resolves its URL to a playable stream via the source registry, spawns 62 62 + /// ffmpeg transcoding to fMP4, and returns the active track info for state 63 63 + /// publishing. 64 64 + /// 65 65 + /// If a track is already playing, it is **not** aborted — call [`abort`] 66 66 + /// first. 67 67 + /// 68 68 + /// [`abort`]: Player::abort 69 69 + pub(crate) fn start( 70 70 + &mut self, 71 71 + item: &QueueItem, 72 72 + publishers: TrackPublishers, 73 73 + ) -> ActiveTrackInfo { 74 74 + let (abort_tx, abort_rx) = watch::channel(false); 75 75 + let info = ActiveTrackInfo::from_item(item); 76 76 + let track_id = item.id; 77 77 + 78 78 + let source_registry = self.source_registry.clone(); 79 79 + let url = item.url.clone(); 80 80 + let cache_dir = self.cache_dir.clone(); 81 81 + let pipeline_tx = self.pipeline_tx.clone(); 82 82 + 83 83 + tokio::spawn(async move { 84 84 + // 1. Resolve URL to a playable stream. 85 85 + let stream_url = match source_registry.resolve(&url, &cache_dir).await { 86 86 + Ok(url) => url, 87 87 + Err(e) => { 88 88 + tracing::error!(%url, error = %e, "player: failed to resolve stream"); 89 89 + send_track_ended(&pipeline_tx, track_id).await; 90 90 + return; 91 91 + } 92 92 + }; 93 93 + 94 94 + // 2. Transcode to fMP4 via ffmpeg. 95 95 + let group_id = track_id as u64; 96 96 + let result = media::transcode_to_fmp4(&stream_url, publishers, abort_rx, group_id) 97 97 + .await; 98 98 + 99 99 + if let Err(e) = &result { 100 100 + tracing::warn!(%url, error = %e, "player: transcoding finished with error"); 101 101 + } 102 102 + 103 103 + // Small delay to give Skip time to clear state first. 104 104 + tokio::time::sleep(Duration::from_millis(200)).await; 105 105 + send_track_ended(&pipeline_tx, track_id).await; 106 106 + }); 107 107 + 108 108 + self.active_track = Some(info.clone()); 109 109 + self.abort_tx = Some(abort_tx); 110 110 + 111 111 + info 112 112 + } 113 113 + 114 114 + /// Abort the currently playing track. 115 115 + /// 116 116 + /// Returns the DB id of the aborted track, or `None` if nothing was playing. 117 117 + pub(crate) fn abort(&mut self) -> Option<i64> { 118 118 + let id = self.active_track.take().map(|t| t.id); 119 119 + if let Some(abort) = self.abort_tx.take() { 120 120 + let _ = abort.send(true); 121 121 + } 122 122 + id 123 123 + } 124 124 + 125 125 + /// Acknowledge that a track has ended. 126 126 + /// 127 127 + /// Returns the track's DB id if `item_id` matches the currently active 128 128 + /// track, or `None` if it doesn't match (e.g. it was already skipped). 129 129 + pub(crate) fn on_track_ended(&mut self, item_id: i64) -> Option<i64> { 130 130 + let is_current = self 131 131 + .active_track 132 132 + .as_ref() 133 133 + .map(|t| t.id == item_id) 134 134 + .unwrap_or(false); 135 135 + if is_current { 136 136 + self.active_track = None; 137 137 + self.abort_tx = None; 138 138 + Some(item_id) 139 139 + } else { 140 140 + None 141 141 + } 142 142 + } 143 143 + 144 144 + /// Is a track currently playing? 145 145 + pub(crate) fn is_playing(&self) -> bool { 146 146 + self.active_track.is_some() 147 147 + } 148 148 + 149 149 + /// Get a snapshot of the currently playing track, if any. 150 150 + pub(crate) fn active_track(&self) -> Option<ActiveTrackInfo> { 151 151 + self.active_track.clone() 152 152 + } 153 153 + } 154 154 + 155 155 + /// Send [`RoomCommand::TrackEnded`] with best-effort logging on failure. 156 156 + async fn send_track_ended(tx: &mpsc::Sender<RoomCommand>, item_id: i64) { 157 157 + if let Err(e) = tx.send(RoomCommand::TrackEnded { item_id }).await { 158 158 + tracing::warn!(item_id, error = %e, "player: failed to send TrackEnded"); 159 159 + } 160 160 + } 161 161 + 162 162 + // --------------------------------------------------------------------------- 163 163 + // Tests 164 164 + // --------------------------------------------------------------------------- 165 165 + 166 166 + #[cfg(test)] 167 167 + mod tests { 168 168 + use super::*; 169 169 + use crate::db; 170 170 + use crate::playlist::Playlist; 171 171 + use crate::types::TrackMeta; 172 172 + use sqlx::sqlite::SqlitePoolOptions; 173 173 + use sqlx::Pool; 174 174 + use std::path::Path; 175 175 + use std::sync::Arc; 176 176 + 177 177 + /// A mock source that resolves all URLs to a fake stream. 178 178 + struct MockPlayerSource; 179 179 + 180 180 + #[async_trait::async_trait] 181 181 + impl crate::source::MediaSource for MockPlayerSource { 182 182 + async fn extract( 183 183 + &self, 184 184 + url: &str, 185 185 + _cache_dir: &Path, 186 186 + ) -> Result<TrackMeta, crate::source::SourceError> { 187 187 + Err(crate::source::SourceError::Unsupported(url.into())) 188 188 + } 189 189 + 190 190 + async fn resolve( 191 191 + &self, 192 192 + url: &str, 193 193 + _cache_dir: &Path, 194 194 + ) -> Result<String, crate::source::SourceError> { 195 195 + // Fake resolve: return the URL with /stream appended. 196 196 + Ok(format!("{url}/stream")) 197 197 + } 198 198 + } 199 199 + 200 200 + fn test_registry() -> Arc<SourceRegistry> { 201 201 + let mut reg = SourceRegistry::new(); 202 202 + reg.register(Box::new(MockPlayerSource)); 203 203 + Arc::new(reg) 204 204 + } 205 205 + 206 206 + async fn test_pool() -> Pool<sqlx::Sqlite> { 207 207 + let pool = SqlitePoolOptions::new() 208 208 + .max_connections(1) 209 209 + .connect(":memory:") 210 210 + .await 211 211 + .unwrap(); 212 212 + db::run_migrations(&pool).await.unwrap(); 213 213 + db::room_insert(&pool, "test-room", false) 214 214 + .await 215 215 + .unwrap(); 216 216 + pool 217 217 + } 218 218 + 219 219 + async fn make_queue_item(pool: &Pool<sqlx::Sqlite>) -> QueueItem { 220 220 + let pl = Playlist::new(pool.clone(), "test-room"); 221 221 + pl.push(&TrackMeta { 222 222 + title: "Test".into(), 223 223 + duration: "3:45".into(), 224 224 + thumbnail: None, 225 225 + url: "https://example.com/track".into(), 226 226 + source: crate::types::SourceKind::Direct, 227 227 + }) 228 228 + .await 229 229 + .unwrap() 230 230 + } 231 231 + 232 232 + #[tokio::test] 233 233 + async fn test_player_start_sets_active_track() { 234 234 + let pool = test_pool().await; 235 235 + let item = make_queue_item(&pool).await; 236 236 + let (tx, _rx) = mpsc::channel(256); 237 237 + 238 238 + let mut player = Player::new(test_registry(), PathBuf::from("/tmp"), tx); 239 239 + assert!(!player.is_playing()); 240 240 + 241 241 + let info = player.start(&item, TrackPublishers::new()); 242 242 + assert!(player.is_playing()); 243 243 + assert_eq!(info.title, "Test"); 244 244 + assert_eq!(info.id, item.id); 245 245 + } 246 246 + 247 247 + #[tokio::test] 248 248 + async fn test_player_abort_clears_active_track() { 249 249 + let pool = test_pool().await; 250 250 + let item = make_queue_item(&pool).await; 251 251 + let (tx, _rx) = mpsc::channel(256); 252 252 + 253 253 + let mut player = Player::new(test_registry(), PathBuf::from("/tmp"), tx); 254 254 + player.start(&item, TrackPublishers::new()); 255 255 + assert!(player.is_playing()); 256 256 + 257 257 + let aborted_id = player.abort(); 258 258 + assert_eq!(aborted_id, Some(item.id)); 259 259 + assert!(!player.is_playing()); 260 260 + assert!(player.active_track().is_none()); 261 261 + } 262 262 + 263 263 + #[tokio::test] 264 264 + async fn test_player_abort_when_idle_returns_none() { 265 265 + let (tx, _rx) = mpsc::channel(256); 266 266 + let mut player = Player::new(test_registry(), PathBuf::from("/tmp"), tx); 267 267 + assert!(player.abort().is_none()); 268 268 + } 269 269 + 270 270 + #[tokio::test] 271 271 + async fn test_player_on_track_ended_matches_current() { 272 272 + let pool = test_pool().await; 273 273 + let item = make_queue_item(&pool).await; 274 274 + let (tx, _rx) = mpsc::channel(256); 275 275 + 276 276 + let mut player = Player::new(test_registry(), PathBuf::from("/tmp"), tx); 277 277 + player.start(&item, TrackPublishers::new()); 278 278 + 279 279 + let result = player.on_track_ended(item.id); 280 280 + assert_eq!(result, Some(item.id)); 281 281 + assert!(!player.is_playing()); 282 282 + } 283 283 + 284 284 + #[tokio::test] 285 285 + async fn test_player_on_track_ended_ignores_old_id() { 286 286 + let pool = test_pool().await; 287 287 + let item = make_queue_item(&pool).await; 288 288 + let (tx, _rx) = mpsc::channel(256); 289 289 + 290 290 + let mut player = Player::new(test_registry(), PathBuf::from("/tmp"), tx); 291 291 + player.start(&item, TrackPublishers::new()); 292 292 + assert!(player.is_playing()); 293 293 + 294 294 + // An old/stale item_id should be ignored. 295 295 + let result = player.on_track_ended(99999); 296 296 + assert!(result.is_none()); 297 297 + assert!(player.is_playing()); 298 298 + } 299 299 + }

+8 -8

src/playlist.rs

Reviewed

··· 21 21 pub(crate) title: String, 22 22 pub(crate) duration: String, 23 23 pub(crate) thumbnail: Option<String>, 24 24 - pub(crate) extractor: String, 24 24 + pub(crate) source: String, 25 25 pub(crate) position: i32, 26 26 pub(crate) played: bool, 27 27 pub(crate) played_at: Option<String>, ··· 51 51 let position = count as i32; 52 52 53 53 let result = sqlx::query( 54 54 - "INSERT INTO queue_items (room_id, url, title, duration, thumbnail, extractor, added_by, position, added_at, played) 54 54 + "INSERT INTO queue_items (room_id, url, title, duration, thumbnail, source, added_by, position, added_at, played) 55 55 VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, 0)", 56 56 ) 57 57 .bind(&self.room_id) ··· 59 59 .bind(&meta.title) 60 60 .bind(&meta.duration) 61 61 .bind(&meta.thumbnail) 62 62 - .bind(&meta.extractor) 62 62 + .bind(meta.source.to_string()) 63 63 .bind(None::<String>) // added_by 64 64 .bind(position) 65 65 .bind(now_iso()) ··· 69 69 let id = result.last_insert_rowid(); 70 70 71 71 let item = sqlx::query_as::<_, QueueItem>( 72 72 - "SELECT id, url, title, duration, thumbnail, extractor, position, played, played_at, started_playing_at 72 72 + "SELECT id, url, title, duration, thumbnail, source, position, played, played_at, started_playing_at 73 73 FROM queue_items WHERE id = ?", 74 74 ) 75 75 .bind(id) ··· 88 88 let mut tx = self.pool.begin().await?; 89 89 90 90 let item = sqlx::query_as::<_, QueueItem>( 91 91 - "SELECT id, url, title, duration, thumbnail, extractor, position, played, played_at, started_playing_at 91 91 + "SELECT id, url, title, duration, thumbnail, source, position, played, played_at, started_playing_at 92 92 FROM queue_items WHERE room_id = ? AND played = 0 AND started_playing_at IS NULL 93 93 ORDER BY position ASC LIMIT 1", 94 94 ) ··· 119 119 /// List queued items (not playing, not finished) ordered by position ASC. 120 120 pub(crate) async fn upcoming(&self) -> Result<Vec<QueueItem>, sqlx::Error> { 121 121 let items = sqlx::query_as::<_, QueueItem>( 122 122 - "SELECT id, url, title, duration, thumbnail, extractor, position, played, played_at, started_playing_at 122 122 + "SELECT id, url, title, duration, thumbnail, source, position, played, played_at, started_playing_at 123 123 FROM queue_items WHERE room_id = ? AND played = 0 AND started_playing_at IS NULL 124 124 ORDER BY position ASC", 125 125 ) ··· 132 132 /// List played items ordered by played_at DESC (newest first), limited to `limit`. 133 133 pub(crate) async fn history(&self, limit: i64) -> Result<Vec<QueueItem>, sqlx::Error> { 134 134 let items = sqlx::query_as::<_, QueueItem>( 135 135 - "SELECT id, url, title, duration, thumbnail, extractor, position, played, played_at, started_playing_at 135 135 + "SELECT id, url, title, duration, thumbnail, source, position, played, played_at, started_playing_at 136 136 FROM queue_items WHERE room_id = ? AND played = 1 ORDER BY played_at DESC, id DESC LIMIT ?", 137 137 ) 138 138 .bind(&self.room_id) ··· 181 181 duration: "3:45".into(), 182 182 thumbnail: None, 183 183 url: "https://example.com/track".into(), 184 184 - extractor: "test".into(), 184 184 + source: crate::types::SourceKind::Direct, 185 185 } 186 186 } 187 187

+111 -123

src/room.rs

Reviewed

··· 21 21 use tokio::sync::{RwLock, mpsc, watch}; 22 22 23 23 use crate::db; 24 24 - use crate::media; 25 25 - use crate::playlist::{Playlist, QueueItem}; 24 24 + use crate::playback::Player; 25 25 + use crate::playlist::Playlist; 26 26 + use crate::source::SourceRegistry; 27 27 + use crate::transport::TrackPublishers; 26 28 use crate::types::{ 27 29 ChatMessage, HistoryEntry, QueueSummary, RoomCommand, RoomHandle, RoomId, TrackMeta, 28 28 - TrackPublishers, TrackState, 30 30 + TrackState, 29 31 }; 30 32 31 33 /// Distributed unique ID generator (Discord-style snowflake). ··· 107 109 } 108 110 } 109 111 110 110 - /// Ephemeral info about the active track, stored in-memory only. 111 111 - struct ActiveTrackInfo { 112 112 - id: i64, 113 113 - title: String, 114 114 - url: String, 115 115 - duration: String, 116 116 - /// Epoch ms when the track started — for client-side elapsed computation. 117 117 - started_at_wall: i64, 118 118 - } 119 119 - 120 120 - impl ActiveTrackInfo { 121 121 - fn from_item(item: &QueueItem) -> Self { 122 122 - Self { 123 123 - id: item.id, 124 124 - title: item.title.clone(), 125 125 - url: item.url.clone(), 126 126 - duration: item.duration.clone(), 127 127 - started_at_wall: chrono::Utc::now().timestamp_millis(), 128 128 - } 129 129 - } 130 130 - } 131 131 - 132 112 /// Per-room event-loop actor. Owns all mutable state. 133 113 struct RoomActor { 134 114 rx: mpsc::Receiver<RoomCommand>, 135 135 - /// Clone of tx passed to pipeline tasks so they can send TrackEnded back. 136 136 - pipeline_tx: mpsc::Sender<RoomCommand>, 137 115 room_id: RoomId, 138 116 publishers: TrackPublishers, 139 117 pool: Pool<Sqlite>, 140 140 - cache_dir: PathBuf, 141 118 142 119 // Mutable state (owned by this actor, no locks needed) 143 143 - active_track: Option<ActiveTrackInfo>, 144 144 - pipeline_abort: Option<watch::Sender<bool>>, 120 120 + player: Player, 145 121 client_count: u64, 146 122 next_user_id: u64, 147 123 last_active_tx: watch::Sender<DateTime<Utc>>, ··· 153 129 pool: Pool<Sqlite>, 154 130 id_gen: Arc<SnowflakeIdGen>, 155 131 cache_dir: PathBuf, 132 132 + sources: Arc<SourceRegistry>, 156 133 inner: Arc<RwLock<Inner>>, 157 134 } 158 135 ··· 161 138 } 162 139 163 140 impl Registry { 164 164 - pub(crate) fn new(pool: Pool<Sqlite>, cache_dir: PathBuf) -> Self { 141 141 + pub(crate) fn new( 142 142 + pool: Pool<Sqlite>, 143 143 + cache_dir: PathBuf, 144 144 + sources: Arc<SourceRegistry>, 145 145 + ) -> Self { 165 146 Self { 166 147 pool, 167 148 id_gen: Arc::new(SnowflakeIdGen::new(1)), ··· 169 150 rooms: HashMap::new(), 170 151 })), 171 152 cache_dir, 153 153 + sources, 172 154 } 173 155 } 174 156 ··· 182 164 let (tx, rx) = mpsc::channel(256); 183 165 let (last_active_tx, last_active_rx) = watch::channel(now); 184 166 167 167 + let player = Player::new( 168 168 + self.sources.clone(), 169 169 + self.cache_dir.clone(), 170 170 + tx.clone(), 171 171 + ); 185 172 let actor = RoomActor { 186 173 rx, 187 187 - pipeline_tx: tx.clone(), 188 174 room_id: id.clone(), 189 175 publishers: publishers.clone(), 190 176 pool: self.pool.clone(), 191 191 - cache_dir: self.cache_dir.clone(), 192 192 - active_track: None, 193 193 - pipeline_abort: None, 177 177 + player, 194 178 client_count: 0, 195 179 next_user_id: 1, 196 180 last_active_tx, ··· 233 217 duration: i.duration, 234 218 thumbnail: i.thumbnail, 235 219 url: i.url, 236 236 - extractor: i.extractor, 220 220 + source: i 221 221 + .source 222 222 + .parse() 223 223 + .unwrap_or_else(|s: String| { 224 224 + tracing::warn!(source = %s, "unknown source kind in queue_items"); 225 225 + crate::types::SourceKind::Direct 226 226 + }), 237 227 }) 238 228 .collect() 239 229 } ··· 416 406 } 417 407 } 418 408 // On shutdown, abort any active pipeline. 419 419 - if let Some(abort) = self.pipeline_abort.take() { 420 420 - let _ = abort.send(true); 421 421 - } 409 409 + self.player.abort(); 422 410 tracing::info!(room = %self.room_id, "room actor shut down"); 423 411 } 424 412 ··· 430 418 if let Err(e) = playlist.push(&meta).await { 431 419 tracing::warn!("failed to persist queue item: {e}"); 432 420 } 433 433 - if self.active_track.is_none() { 421 421 + if !self.player.is_playing() { 434 422 self.start_next().await; 435 423 } 436 424 self.publish_state_snapshot().await; 437 425 } 438 426 RoomCommand::Skip => { 439 427 tracing::info!(room = %self.room_id, "skip requested"); 440 440 - let finished_id = self.active_track.take().map(|t| t.id); 441 441 - if let Some(abort) = self.pipeline_abort.take() { 442 442 - let _ = abort.send(true); 443 443 - } 428 428 + let finished_id = self.player.abort(); 444 429 if let Some(id) = finished_id { 445 430 let playlist = Playlist::new(self.pool.clone(), &self.room_id.0); 446 431 let _ = playlist.mark_finished(id).await; ··· 450 435 } 451 436 RoomCommand::TrackEnded { item_id } => { 452 437 tracing::debug!(room = %self.room_id, item_id, "track ended"); 453 453 - let is_current = self 454 454 - .active_track 455 455 - .as_ref() 456 456 - .map(|t| t.id == item_id) 457 457 - .unwrap_or(false); 458 458 - if is_current { 459 459 - self.active_track = None; 460 460 - self.pipeline_abort = None; 438 438 + if self.player.on_track_ended(item_id).is_some() { 461 439 let playlist = Playlist::new(self.pool.clone(), &self.room_id.0); 462 440 let _ = playlist.mark_finished(item_id).await; 463 441 self.start_next().await; ··· 505 483 } 506 484 RoomCommand::Shutdown => { 507 485 tracing::info!(room = %self.room_id, "actor shutdown requested"); 508 508 - if let Some(abort) = self.pipeline_abort.take() { 509 509 - let _ = abort.send(true); 510 510 - } 486 486 + self.player.abort(); 511 487 return true; 512 488 } 513 489 } ··· 525 501 } 526 502 }; 527 503 tracing::info!(room = %self.room_id, title = %item.title, item_id = item.id, "starting track"); 528 528 - let (abort_tx, abort_rx) = watch::channel(false); 529 529 - let info = ActiveTrackInfo::from_item(&item); 530 530 - self.active_track = Some(info); 531 531 - self.pipeline_abort = Some(abort_tx); 532 532 - 533 533 - let url = item.url.clone(); 534 534 - let cache = self.cache_dir.clone(); 535 535 - let publishers = self.publishers.clone(); 536 536 - let pipeline_tx = self.pipeline_tx.clone(); 537 537 - let track_item_id = item.id; 538 538 - let group_id = item.id as u64; 539 539 - 540 540 - tokio::spawn(async move { 541 541 - let _result = media::spawn_pipeline(&url, &cache, publishers, abort_rx, group_id).await; 542 542 - // Small delay before sending TrackEnded (avoids tight loop, 543 543 - // gives Skip time to clear state first). 544 544 - tokio::time::sleep(Duration::from_millis(200)).await; 545 545 - let _ = pipeline_tx 546 546 - .send(RoomCommand::TrackEnded { 547 547 - item_id: track_item_id, 548 548 - }) 549 549 - .await; 550 550 - }); 504 504 + self.player.start(&item, self.publishers.clone()); 551 505 } 552 506 553 507 async fn publish_state_snapshot(&self) { 554 554 - let current = self.active_track.as_ref().map(|t| TrackState { 508 508 + let current = self.player.active_track().map(|t| TrackState { 555 509 id: t.id, 556 510 title: t.title.clone(), 557 511 url: t.url.clone(), ··· 615 569 pool 616 570 } 617 571 572 572 + /// A mock source that handles all URLs (for tests that need playback). 573 573 + struct PanaceaSource; 574 574 + 575 575 + #[async_trait::async_trait] 576 576 + impl crate::source::MediaSource for PanaceaSource { 577 577 + async fn extract( 578 578 + &self, 579 579 + url: &str, 580 580 + _cache_dir: &std::path::Path, 581 581 + ) -> Result<TrackMeta, crate::source::SourceError> { 582 582 + Ok(TrackMeta { 583 583 + title: "test".into(), 584 584 + duration: "3:45".into(), 585 585 + thumbnail: None, 586 586 + url: url.into(), 587 587 + source: crate::types::SourceKind::Direct, 588 588 + }) 589 589 + } 590 590 + 591 591 + async fn resolve( 592 592 + &self, 593 593 + url: &str, 594 594 + _cache_dir: &std::path::Path, 595 595 + ) -> Result<String, crate::source::SourceError> { 596 596 + Ok(format!("{url}/stream")) 597 597 + } 598 598 + } 599 599 + 600 600 + fn test_registry() -> Arc<SourceRegistry> { 601 601 + let mut reg = SourceRegistry::new(); 602 602 + reg.register(Box::new(PanaceaSource)); 603 603 + Arc::new(reg) 604 604 + } 605 605 + 618 606 #[tokio::test] 619 607 async fn test_snowflake_uniqueness() { 620 608 let gen_id = Arc::new(SnowflakeIdGen::new(1)); ··· 642 630 #[tokio::test] 643 631 async fn test_create_and_exists() { 644 632 let pool = test_pool().await; 645 645 - let reg = Registry::new(pool, PathBuf::from("/tmp/moqbox")); 633 633 + let reg = Registry::new(pool, PathBuf::from("/tmp/moqbox"), test_registry()); 646 634 let room = reg.create().await; 647 635 assert!(reg.exists(&room.id).await); 648 636 } ··· 650 638 #[tokio::test] 651 639 async fn test_push_and_queue() { 652 640 let pool = test_pool().await; 653 653 - let reg = Registry::new(pool, PathBuf::from("/tmp/moqbox")); 641 641 + let reg = Registry::new(pool, PathBuf::from("/tmp/moqbox"), test_registry()); 654 642 let room = reg.create().await; 655 643 656 644 let meta_a = TrackMeta { ··· 658 646 duration: "3:45".into(), 659 647 thumbnail: None, 660 648 url: "https://example.com/a".into(), 661 661 - extractor: "test".into(), 649 649 + source: crate::types::SourceKind::Direct, 662 650 }; 663 651 let meta_b = TrackMeta { 664 652 title: "Track B".into(), 665 653 duration: "4:20".into(), 666 654 thumbnail: None, 667 655 url: "https://example.com/b".into(), 668 668 - extractor: "test".into(), 656 656 + source: crate::types::SourceKind::Direct, 669 657 }; 670 658 671 659 // Push first track — it gets popped and starts playing immediately. ··· 684 672 #[tokio::test] 685 673 async fn test_remove() { 686 674 let pool = test_pool().await; 687 687 - let reg = Registry::new(pool, PathBuf::from("/tmp/moqbox")); 675 675 + let reg = Registry::new(pool, PathBuf::from("/tmp/moqbox"), test_registry()); 688 676 let room = reg.create().await; 689 677 690 678 reg.remove(&room.id).await; ··· 694 682 #[tokio::test] 695 683 async fn test_send_chat() { 696 684 let pool = test_pool().await; 697 697 - let reg = Registry::new(pool, PathBuf::from("/tmp/moqbox")); 685 685 + let reg = Registry::new(pool, PathBuf::from("/tmp/moqbox"), test_registry()); 698 686 let room = reg.create().await; 699 687 700 688 let msg = reg ··· 710 698 #[tokio::test] 711 699 async fn test_send_chat_persistence() { 712 700 let pool = test_pool().await; 713 713 - let reg = Registry::new(pool, PathBuf::from("/tmp/moqbox")); 701 701 + let reg = Registry::new(pool, PathBuf::from("/tmp/moqbox"), test_registry()); 714 702 let room = reg.create().await; 715 703 716 704 let _ = reg ··· 737 725 #[tokio::test] 738 726 async fn test_send_chat_empty_content() { 739 727 let pool = test_pool().await; 740 740 - let reg = Registry::new(pool, PathBuf::from("/tmp/moqbox")); 728 728 + let reg = Registry::new(pool, PathBuf::from("/tmp/moqbox"), test_registry()); 741 729 let room = reg.create().await; 742 730 743 731 let result = reg.send_chat(&room.id, "alice", "", "message").await; ··· 747 735 #[tokio::test] 748 736 async fn test_send_chat_long_content() { 749 737 let pool = test_pool().await; 750 750 - let reg = Registry::new(pool, PathBuf::from("/tmp/moqbox")); 738 738 + let reg = Registry::new(pool, PathBuf::from("/tmp/moqbox"), test_registry()); 751 739 let room = reg.create().await; 752 740 753 741 let long = "a".repeat(2001); ··· 758 746 #[tokio::test] 759 747 async fn test_send_chat_long_username() { 760 748 let pool = test_pool().await; 761 761 - let reg = Registry::new(pool, PathBuf::from("/tmp/moqbox")); 749 749 + let reg = Registry::new(pool, PathBuf::from("/tmp/moqbox"), test_registry()); 762 750 let room = reg.create().await; 763 751 764 752 let long = "a".repeat(33); ··· 769 757 #[tokio::test] 770 758 async fn test_register_client() { 771 759 let pool = test_pool().await; 772 772 - let reg = Registry::new(pool, PathBuf::from("/tmp/moqbox")); 760 760 + let reg = Registry::new(pool, PathBuf::from("/tmp/moqbox"), test_registry()); 773 761 let room = reg.create().await; 774 762 775 763 let id1 = reg.register_client(&room.id).await; ··· 782 770 #[tokio::test] 783 771 async fn test_register_client_nonexistent_room() { 784 772 let pool = test_pool().await; 785 785 - let reg = Registry::new(pool, PathBuf::from("/tmp/moqbox")); 773 773 + let reg = Registry::new(pool, PathBuf::from("/tmp/moqbox"), test_registry()); 786 774 let fake_id = RoomId("nonexistent".into()); 787 775 assert_eq!(reg.register_client(&fake_id).await, None); 788 776 } ··· 790 778 #[tokio::test] 791 779 async fn test_unregister_client() { 792 780 let pool = test_pool().await; 793 793 - let reg = Registry::new(pool, PathBuf::from("/tmp/moqbox")); 781 781 + let reg = Registry::new(pool, PathBuf::from("/tmp/moqbox"), test_registry()); 794 782 let room = reg.create().await; 795 783 796 784 // Register and unregister — subsequent register should still work. ··· 804 792 #[tokio::test] 805 793 async fn test_skip_when_idle_does_not_panic() { 806 794 let pool = test_pool().await; 807 807 - let reg = Registry::new(pool, PathBuf::from("/tmp/moqbox")); 795 795 + let reg = Registry::new(pool, PathBuf::from("/tmp/moqbox"), test_registry()); 808 796 let room = reg.create().await; 809 797 // Skip on an empty, idle room should not panic or deadlock. 810 798 reg.skip(&room.id).await; ··· 814 802 #[tokio::test] 815 803 async fn test_double_skip_does_not_deadlock() { 816 804 let pool = test_pool().await; 817 817 - let reg = Registry::new(pool, PathBuf::from("/tmp/moqbox")); 805 805 + let reg = Registry::new(pool, PathBuf::from("/tmp/moqbox"), test_registry()); 818 806 let room = reg.create().await; 819 807 820 808 let meta = TrackMeta { ··· 822 810 duration: "3:45".into(), 823 811 thumbnail: None, 824 812 url: "https://example.com/a".into(), 825 825 - extractor: "test".into(), 813 813 + source: crate::types::SourceKind::Direct, 826 814 }; 827 815 reg.push(&room.id, meta).await; 828 816 tokio::time::sleep(Duration::from_millis(100)).await; ··· 840 828 #[tokio::test] 841 829 async fn test_send_chat_returns_id_zero() { 842 830 let pool = test_pool().await; 843 843 - let reg = Registry::new(pool, PathBuf::from("/tmp/moqbox")); 831 831 + let reg = Registry::new(pool, PathBuf::from("/tmp/moqbox"), test_registry()); 844 832 let room = reg.create().await; 845 833 let msg = reg 846 834 .send_chat(&room.id, "alice", "hello", "message") ··· 853 841 #[tokio::test] 854 842 async fn test_sweep_idle_removes_inactive_rooms() { 855 843 let pool = test_pool().await; 856 856 - let reg = Registry::new(pool, PathBuf::from("/tmp/moqbox")); 844 844 + let reg = Registry::new(pool, PathBuf::from("/tmp/moqbox"), test_registry()); 857 845 let room = reg.create().await; 858 846 859 847 // Tiny sleep so the room has a non-zero idle duration. ··· 871 859 #[tokio::test] 872 860 async fn test_sweep_idle_preserves_active_rooms() { 873 861 let pool = test_pool().await; 874 874 - let reg = Registry::new(pool, PathBuf::from("/tmp/moqbox")); 862 862 + let reg = Registry::new(pool, PathBuf::from("/tmp/moqbox"), test_registry()); 875 863 let room = reg.create().await; 876 864 877 865 // Register a client (triggers state snapshot, updates last_active). ··· 885 873 #[tokio::test] 886 874 async fn test_push_starts_playback_when_idle() { 887 875 let pool = test_pool().await; 888 888 - let reg = Registry::new(pool, PathBuf::from("/tmp/moqbox")); 876 876 + let reg = Registry::new(pool, PathBuf::from("/tmp/moqbox"), test_registry()); 889 877 let room = reg.create().await; 890 878 891 879 let meta = TrackMeta { ··· 893 881 duration: "3:45".into(), 894 882 thumbnail: None, 895 883 url: "https://example.com/t".into(), 896 896 - extractor: "test".into(), 884 884 + source: crate::types::SourceKind::Direct, 897 885 }; 898 886 reg.push(&room.id, meta).await; 899 887 ··· 911 899 #[tokio::test] 912 900 async fn test_push_appends_when_already_playing() { 913 901 let pool = test_pool().await; 914 914 - let reg = Registry::new(pool, PathBuf::from("/tmp/moqbox")); 902 902 + let reg = Registry::new(pool, PathBuf::from("/tmp/moqbox"), test_registry()); 915 903 let room = reg.create().await; 916 904 917 905 // First track starts playing immediately. ··· 922 910 duration: "3:45".into(), 923 911 thumbnail: None, 924 912 url: "https://example.com/a".into(), 925 925 - extractor: "test".into(), 913 913 + source: crate::types::SourceKind::Direct, 926 914 }, 927 915 ) 928 916 .await; ··· 936 924 duration: "3:45".into(), 937 925 thumbnail: None, 938 926 url: "https://example.com/b".into(), 939 939 - extractor: "test".into(), 927 927 + source: crate::types::SourceKind::Direct, 940 928 }, 941 929 ) 942 930 .await; ··· 958 946 #[tokio::test] 959 947 async fn test_skip_when_queue_empty_after_track_ends() { 960 948 let pool = test_pool().await; 961 961 - let reg = Registry::new(pool, PathBuf::from("/tmp/moqbox")); 949 949 + let reg = Registry::new(pool, PathBuf::from("/tmp/moqbox"), test_registry()); 962 950 let room = reg.create().await; 963 951 964 952 // Push a track, let it start playing, then skip. ··· 969 957 duration: "3:45".into(), 970 958 thumbnail: None, 971 959 url: "https://example.com/a".into(), 972 972 - extractor: "test".into(), 960 960 + source: crate::types::SourceKind::Direct, 973 961 }, 974 962 ) 975 963 .await; ··· 989 977 #[tokio::test] 990 978 async fn test_double_skip_does_not_corrupt_state() { 991 979 let pool = test_pool().await; 992 992 - let reg = Registry::new(pool, PathBuf::from("/tmp/moqbox")); 980 980 + let reg = Registry::new(pool, PathBuf::from("/tmp/moqbox"), test_registry()); 993 981 let room = reg.create().await; 994 982 995 983 // Push two tracks and play the first. ··· 1000 988 duration: "3:45".into(), 1001 989 thumbnail: None, 1002 990 url: "https://example.com/a".into(), 1003 1003 - extractor: "test".into(), 991 991 + source: crate::types::SourceKind::Direct, 1004 992 }, 1005 993 ) 1006 994 .await; ··· 1014 1002 duration: "3:45".into(), 1015 1003 thumbnail: None, 1016 1004 url: "https://example.com/b".into(), 1017 1017 - extractor: "test".into(), 1005 1005 + source: crate::types::SourceKind::Direct, 1018 1006 }, 1019 1007 ) 1020 1008 .await; ··· 1040 1028 #[tokio::test] 1041 1029 async fn test_register_client_overflow() { 1042 1030 let pool = test_pool().await; 1043 1043 - let reg = Registry::new(pool, PathBuf::from("/tmp/moqbox")); 1031 1031 + let reg = Registry::new(pool, PathBuf::from("/tmp/moqbox"), test_registry()); 1044 1032 let room = reg.create().await; 1045 1033 1046 1034 // Register many clients — overflow should not panic. ··· 1058 1046 #[tokio::test] 1059 1047 async fn test_remove_while_playing_does_not_panic() { 1060 1048 let pool = test_pool().await; 1061 1061 - let reg = Registry::new(pool, PathBuf::from("/tmp/moqbox")); 1049 1049 + let reg = Registry::new(pool, PathBuf::from("/tmp/moqbox"), test_registry()); 1062 1050 let room = reg.create().await; 1063 1051 1064 1052 reg.push( ··· 1068 1056 duration: "3:45".into(), 1069 1057 thumbnail: None, 1070 1058 url: "https://example.com/a".into(), 1071 1071 - extractor: "test".into(), 1059 1059 + source: crate::types::SourceKind::Direct, 1072 1060 }, 1073 1061 ) 1074 1062 .await; ··· 1087 1075 #[tokio::test] 1088 1076 async fn test_sweep_idle_keeps_active_room() { 1089 1077 let pool = test_pool().await; 1090 1090 - let reg = Registry::new(pool, PathBuf::from("/tmp/moqbox")); 1078 1078 + let reg = Registry::new(pool, PathBuf::from("/tmp/moqbox"), test_registry()); 1091 1079 let room = reg.create().await; 1092 1080 1093 1081 // Register a client (updates last_active via actor while we await). ··· 1105 1093 #[tokio::test] 1106 1094 async fn test_client_count_tracking() { 1107 1095 let pool = test_pool().await; 1108 1108 - let reg = Registry::new(pool, PathBuf::from("/tmp/moqbox")); 1096 1096 + let reg = Registry::new(pool, PathBuf::from("/tmp/moqbox"), test_registry()); 1109 1097 let room = reg.create().await; 1110 1098 1111 1099 // Register 3 clients. ··· 1127 1115 #[tokio::test] 1128 1116 async fn test_multiple_rooms_dont_interfere() { 1129 1117 let pool = test_pool().await; 1130 1130 - let reg = Registry::new(pool, PathBuf::from("/tmp/moqbox")); 1118 1118 + let reg = Registry::new(pool, PathBuf::from("/tmp/moqbox"), test_registry()); 1131 1119 let room_a = reg.create().await; 1132 1120 let room_b = reg.create().await; 1133 1121 ··· 1142 1130 duration: "1:00".into(), 1143 1131 thumbnail: None, 1144 1132 url: "https://example.com/a".into(), 1145 1145 - extractor: "test".into(), 1133 1133 + source: crate::types::SourceKind::Direct, 1146 1134 }, 1147 1135 ) 1148 1136 .await; ··· 1153 1141 duration: "2:00".into(), 1154 1142 thumbnail: None, 1155 1143 url: "https://example.com/b".into(), 1156 1156 - extractor: "test".into(), 1144 1144 + source: crate::types::SourceKind::Direct, 1157 1145 }, 1158 1146 ) 1159 1147 .await; ··· 1175 1163 #[tokio::test] 1176 1164 async fn test_skip_on_empty_queue_no_crash() { 1177 1165 let pool = test_pool().await; 1178 1178 - let reg = Registry::new(pool, PathBuf::from("/tmp/moqbox")); 1166 1166 + let reg = Registry::new(pool, PathBuf::from("/tmp/moqbox"), test_registry()); 1179 1167 let room = reg.create().await; 1180 1168 // Skip with no tracks ever queued. 1181 1169 reg.skip(&room.id).await; ··· 1186 1174 #[tokio::test] 1187 1175 async fn test_push_after_skip_works() { 1188 1176 let pool = test_pool().await; 1189 1189 - let reg = Registry::new(pool, PathBuf::from("/tmp/moqbox")); 1177 1177 + let reg = Registry::new(pool, PathBuf::from("/tmp/moqbox"), test_registry()); 1190 1178 let room = reg.create().await; 1191 1179 1192 1180 // Push and skip. ··· 1197 1185 duration: "3:45".into(), 1198 1186 thumbnail: None, 1199 1187 url: "https://a".into(), 1200 1200 - extractor: "test".into(), 1188 1188 + source: crate::types::SourceKind::Direct, 1201 1189 }, 1202 1190 ) 1203 1191 .await; ··· 1213 1201 duration: "3:45".into(), 1214 1202 thumbnail: None, 1215 1203 url: "https://b".into(), 1216 1216 - extractor: "test".into(), 1204 1204 + source: crate::types::SourceKind::Direct, 1217 1205 }, 1218 1206 ) 1219 1207 .await; ··· 1232 1220 #[tokio::test] 1233 1221 async fn test_client_count_does_not_underflow() { 1234 1222 let pool = test_pool().await; 1235 1235 - let reg = Registry::new(pool, PathBuf::from("/tmp/moqbox")); 1223 1223 + let reg = Registry::new(pool, PathBuf::from("/tmp/moqbox"), test_registry()); 1236 1224 let room = reg.create().await; 1237 1225 1238 1226 // Unregister with no clients registered — should not underflow. ··· 1246 1234 #[tokio::test] 1247 1235 async fn test_multiple_skips_sequential() { 1248 1236 let pool = test_pool().await; 1249 1249 - let reg = Registry::new(pool, PathBuf::from("/tmp/moqbox")); 1237 1237 + let reg = Registry::new(pool, PathBuf::from("/tmp/moqbox"), test_registry()); 1250 1238 let room = reg.create().await; 1251 1239 1252 1240 // Push 3 tracks, skip through all. ··· 1258 1246 duration: "1:00".into(), 1259 1247 thumbnail: None, 1260 1248 url: "https://x".into(), 1261 1261 - extractor: "test".into(), 1249 1249 + source: crate::types::SourceKind::Direct, 1262 1250 }, 1263 1251 ) 1264 1252 .await;

+172 -3

src/transport.rs

Reviewed

··· 3 3 //! Provides varint encoding/decoding, MoQ message types, track publishers 4 4 //! with broadcast channels, and a WebSocket session handler. 5 5 6 6 + use std::sync::atomic::{AtomicU64, Ordering}; 7 7 + use std::sync::{Arc, Mutex}; 8 8 + use std::time::Duration; 9 9 + 6 10 use axum::extract::ws::{CloseFrame, Message, WebSocket}; 7 11 use bytes::Bytes; 8 12 use futures_util::{SinkExt, StreamExt}; 9 9 - use std::time::Duration; 10 10 - use tokio::sync::{broadcast, mpsc}; 13 13 + use tokio::sync::{broadcast, mpsc, watch}; 11 14 12 15 use crate::room; 13 13 - use crate::types::{RoomCommand, RoomId, TrackId, TrackPublishers}; 16 16 + use crate::types::{MoqObject, RoomCommand, RoomId, StateValue, TrackId}; 14 17 15 18 pub(crate) fn encode_varint(value: u64) -> Vec<u8> { 16 19 if value < 64 { ··· 247 250 248 251 pub(crate) fn room_namespace(room_id: &str) -> String { 249 252 format!("moqbox/room/{room_id}") 253 253 + } 254 254 + 255 255 + // --------------------------------------------------------------------------- 256 256 + // Track publishers 257 257 + // --------------------------------------------------------------------------- 258 258 + 259 259 + /// Per-room MoQ track publishers, shared across all client sessions. 260 260 + /// 261 261 + /// Cloning is cheap (all inner channels are `Arc`-based). 262 262 + #[derive(Clone)] 263 263 + pub(crate) struct TrackPublishers { 264 264 + pub(crate) audio: broadcast::Sender<MoqObject>, 265 265 + pub(crate) video: broadcast::Sender<MoqObject>, 266 266 + pub(crate) chat: broadcast::Sender<MoqObject>, 267 267 + pub(crate) state: watch::Sender<StateValue>, 268 268 + /// Cached fMP4 init segment (ftyp+moov boxes) for late-joining clients. 269 269 + video_init: Arc<Mutex<Option<(u64, Bytes)>>>, // (group_id, payload) 270 270 + video_init_watch: watch::Sender<bool>, 271 271 + chat_seq: Arc<AtomicU64>, 272 272 + } 273 273 + 274 274 + impl TrackPublishers { 275 275 + pub(crate) fn new() -> Self { 276 276 + let (audio, _) = broadcast::channel(256); 277 277 + let (video, _) = broadcast::channel(256); 278 278 + let (chat, _) = broadcast::channel(256); 279 279 + let (state, _) = watch::channel(StateValue { 280 280 + payload: Bytes::new(), 281 281 + seq: 0, 282 282 + }); 283 283 + let (video_init_watch, _) = watch::channel(false); 284 284 + TrackPublishers { 285 285 + audio, 286 286 + video, 287 287 + chat, 288 288 + state, 289 289 + video_init: Arc::new(Mutex::new(None)), 290 290 + video_init_watch, 291 291 + chat_seq: Arc::new(AtomicU64::new(0)), 292 292 + } 293 293 + } 294 294 + 295 295 + pub(crate) fn publish_video(&self, group_id: u64, object_id: u64, payload: Bytes) { 296 296 + let _ = self.video.send(MoqObject { 297 297 + track_id: TrackId::Video, 298 298 + group_id, 299 299 + object_id, 300 300 + payload, 301 301 + }); 302 302 + } 303 303 + 304 304 + /// Publish a chat object with an auto-incrementing object_id. 305 305 + pub(crate) fn publish_chat(&self, group_id: u64, payload: Bytes) { 306 306 + let object_id = self.chat_seq.fetch_add(1, Ordering::Relaxed); 307 307 + let _ = self.chat.send(MoqObject { 308 308 + track_id: TrackId::Chat, 309 309 + group_id, 310 310 + object_id, 311 311 + payload, 312 312 + }); 313 313 + } 314 314 + 315 315 + /// Publish a state update (latest-only, atomically increments seq). 316 316 + pub(crate) fn publish_state(&self, payload: Bytes) { 317 317 + let prev = self.state.borrow().clone(); 318 318 + let new = StateValue { 319 319 + payload, 320 320 + seq: prev.seq + 1, 321 321 + }; 322 322 + let _ = self.state.send(new); 323 323 + } 324 324 + 325 325 + /// Cache the fMP4 init segment (ftyp+moov boxes) so late-joining clients 326 326 + /// can initialise their MediaSource before receiving live fragments. 327 327 + pub(crate) fn cache_init_segment(&self, payload: Bytes, group_id: u64) { 328 328 + tracing::debug!(group_id, size = payload.len(), "caching init segment"); 329 329 + if let Ok(mut guard) = self.video_init.lock() { 330 330 + *guard = Some((group_id, payload.clone())); 331 331 + } 332 332 + self.video_init_watch.send_replace(true); 333 333 + } 334 334 + 335 335 + pub(crate) fn get_init_segment(&self) -> Option<(u64, Bytes)> { 336 336 + let result = self.video_init.lock().ok().and_then(|g| g.clone()); 337 337 + if let Some((group_id, ref data)) = result { 338 338 + tracing::debug!(group_id, size = data.len(), "retrieved cached init segment"); 339 339 + } else { 340 340 + tracing::debug!("no cached init segment found"); 341 341 + } 342 342 + result 343 343 + } 344 344 + 345 345 + pub(crate) fn video_init_waiter(&self) -> watch::Receiver<bool> { 346 346 + self.video_init_watch.subscribe() 347 347 + } 250 348 } 251 349 252 350 /// Handle a WebSocket upgrade and run the MoQ session lifecycle. ··· 866 964 let obj2 = rx2.recv().await.expect("subscriber 2"); 867 965 assert_eq!(obj1.payload, Bytes::from("fanout")); 868 966 assert_eq!(obj2.payload, Bytes::from("fanout")); 967 967 + } 968 968 + 969 969 + // ── Init segment cache tests ─────────────────────────────────────────── 970 970 + 971 971 + #[test] 972 972 + fn test_init_cache_and_retrieve() { 973 973 + let publishers = TrackPublishers::new(); 974 974 + 975 975 + assert!( 976 976 + publishers.get_init_segment().is_none(), 977 977 + "init should be None before caching" 978 978 + ); 979 979 + 980 980 + let payload = Bytes::from("ftyp-moov-data"); 981 981 + publishers.cache_init_segment(payload.clone(), 42); 982 982 + 983 983 + let retrieved = publishers.get_init_segment(); 984 984 + assert!(retrieved.is_some(), "init should be Some after caching"); 985 985 + let (group_id, data) = retrieved.unwrap(); 986 986 + assert_eq!(group_id, 42, "group_id should match what was cached"); 987 987 + assert_eq!(data, payload, "payload should match what was cached"); 988 988 + } 989 989 + 990 990 + #[tokio::test] 991 991 + async fn test_init_watch_signals() { 992 992 + let publishers = TrackPublishers::new(); 993 993 + 994 994 + let mut waiter = publishers.video_init_waiter(); 995 995 + assert!(!*waiter.borrow(), "initial watch value should be false"); 996 996 + 997 997 + let pubs = publishers.clone(); 998 998 + tokio::spawn(async move { 999 999 + tokio::time::sleep(std::time::Duration::from_millis(50)).await; 1000 1000 + pubs.cache_init_segment(Bytes::from("init-payload"), 1); 1001 1001 + }); 1002 1002 + 1003 1003 + let _ = tokio::time::timeout(std::time::Duration::from_secs(2), waiter.changed()) 1004 1004 + .await 1005 1005 + .expect("watch should fire within timeout when init is cached"); 1006 1006 + 1007 1007 + assert!( 1008 1008 + *waiter.borrow(), 1009 1009 + "watch value should be true after cache_init_segment" 1010 1010 + ); 1011 1011 + 1012 1012 + let retrieved = publishers.get_init_segment(); 1013 1013 + assert!( 1014 1014 + retrieved.is_some(), 1015 1015 + "init should be retrievable after watch fires" 1016 1016 + ); 1017 1017 + let (group_id, data) = retrieved.unwrap(); 1018 1018 + assert_eq!(group_id, 1, "group_id should match"); 1019 1019 + assert_eq!(data, Bytes::from("init-payload"), "payload should match"); 1020 1020 + } 1021 1021 + 1022 1022 + #[test] 1023 1023 + fn test_cache_overwrite_on_new_group() { 1024 1024 + let publishers = TrackPublishers::new(); 1025 1025 + 1026 1026 + publishers.cache_init_segment(Bytes::from("init-1"), 1); 1027 1027 + publishers.cache_init_segment(Bytes::from("init-2"), 2); 1028 1028 + 1029 1029 + let retrieved = publishers.get_init_segment(); 1030 1030 + assert!(retrieved.is_some(), "init should be cached"); 1031 1031 + let (group_id, data) = retrieved.unwrap(); 1032 1032 + assert_eq!(group_id, 2, "group_id should be from the latest cache call"); 1033 1033 + assert_eq!( 1034 1034 + data, 1035 1035 + Bytes::from("init-2"), 1036 1036 + "payload should be from the latest cache call" 1037 1037 + ); 869 1038 } 870 1039 }

+60 -183

src/types.rs

Reviewed

··· 1 1 //! Shared types used across moqbox modules. 2 2 - //! 3 3 - //! Consolidates room state types, MoQ transport types, and track 4 4 - //! publishing infrastructure into a single module to avoid circular 5 5 - //! dependencies and keep the module graph acyclic. 6 2 7 3 use std::fmt; 8 8 - use std::sync::atomic::{AtomicU64, Ordering}; 9 9 - use std::sync::{Arc, Mutex}; 10 4 11 5 use bytes::Bytes; 12 6 use chrono::{DateTime, Utc}; 13 7 use serde::Serialize; 14 14 - use tokio::sync::{broadcast, mpsc, oneshot, watch}; 8 8 + use tokio::sync::{mpsc, oneshot, watch}; 9 9 + 10 10 + pub(crate) use crate::transport::TrackPublishers; 15 11 16 12 /// Opaque room identifier, human-readable and URL-safe. 17 13 #[derive(Debug, Clone, Hash, Eq, PartialEq, Serialize)] ··· 23 19 } 24 20 } 25 21 26 26 - /// Metadata for a single queued track, extracted by yt-dlp. 22 22 + /// Identifies which media source resolved a track. 23 23 + #[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize)] 24 24 + #[serde(rename_all = "snake_case")] 25 25 + pub(crate) enum SourceKind { 26 26 + Ytdlp, 27 27 + Direct, 28 28 + } 29 29 + 30 30 + impl std::fmt::Display for SourceKind { 31 31 + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { 32 32 + match self { 33 33 + SourceKind::Ytdlp => write!(f, "ytdlp"), 34 34 + SourceKind::Direct => write!(f, "direct"), 35 35 + } 36 36 + } 37 37 + } 38 38 + 39 39 + impl std::str::FromStr for SourceKind { 40 40 + type Err = String; 41 41 + fn from_str(s: &str) -> Result<Self, Self::Err> { 42 42 + match s { 43 43 + "ytdlp" => Ok(SourceKind::Ytdlp), 44 44 + "direct" => Ok(SourceKind::Direct), 45 45 + _ => Err(format!("unknown source kind: {s}")), 46 46 + } 47 47 + } 48 48 + } 49 49 + 50 50 + /// Metadata for a single queued track. 51 51 + /// 52 52 + /// Source-agnostic — produced by any [`MediaSource`](crate::source::MediaSource) 53 53 + /// implementation (yt-dlp, direct URL detection, etc.). 27 54 #[derive(Debug, Clone, Serialize)] 28 55 pub(crate) struct TrackMeta { 29 56 pub title: String, 30 57 pub duration: String, 31 58 pub thumbnail: Option<String>, 32 59 pub url: String, 33 33 - pub extractor: String, 60 60 + /// How this track was resolved (yt-dlp, direct URL, etc.). 61 61 + pub source: SourceKind, 34 62 } 35 63 36 64 /// Snapshot of the currently playing track for state publishing. ··· 110 138 pub(crate) seq: u64, 111 139 } 112 140 113 113 - /// Per-room MoQ track publishers, shared across all client sessions. 114 114 - /// 115 115 - /// Cloning is cheap (all inner channels are `Arc`-based). 116 116 - #[derive(Clone)] 117 117 - pub(crate) struct TrackPublishers { 118 118 - pub(crate) audio: broadcast::Sender<MoqObject>, 119 119 - pub(crate) video: broadcast::Sender<MoqObject>, 120 120 - pub(crate) chat: broadcast::Sender<MoqObject>, 121 121 - pub(crate) state: watch::Sender<StateValue>, 122 122 - /// Cached fMP4 init segment (ftyp+moov boxes) for late-joining clients. 123 123 - video_init: Arc<Mutex<Option<(u64, Bytes)>>>, // (group_id, payload) 124 124 - video_init_watch: watch::Sender<bool>, 125 125 - chat_seq: Arc<AtomicU64>, 126 126 - } 127 127 - 128 128 - impl TrackPublishers { 129 129 - pub(crate) fn new() -> Self { 130 130 - let (audio, _) = broadcast::channel(256); 131 131 - let (video, _) = broadcast::channel(256); 132 132 - let (chat, _) = broadcast::channel(256); 133 133 - let (state, _) = watch::channel(StateValue { 134 134 - payload: Bytes::new(), 135 135 - seq: 0, 136 136 - }); 137 137 - let (video_init_watch, _) = watch::channel(false); 138 138 - TrackPublishers { 139 139 - audio, 140 140 - video, 141 141 - chat, 142 142 - state, 143 143 - video_init: Arc::new(Mutex::new(None)), 144 144 - video_init_watch, 145 145 - chat_seq: Arc::new(AtomicU64::new(0)), 146 146 - } 147 147 - } 148 148 - 149 149 - pub(crate) fn publish_video(&self, group_id: u64, object_id: u64, payload: Bytes) { 150 150 - let _ = self.video.send(MoqObject { 151 151 - track_id: TrackId::Video, 152 152 - group_id, 153 153 - object_id, 154 154 - payload, 155 155 - }); 156 156 - } 157 157 - 158 158 - /// Publish a chat object with an auto-incrementing object_id. 159 159 - pub(crate) fn publish_chat(&self, group_id: u64, payload: Bytes) { 160 160 - let object_id = self.chat_seq.fetch_add(1, Ordering::Relaxed); 161 161 - let _ = self.chat.send(MoqObject { 162 162 - track_id: TrackId::Chat, 163 163 - group_id, 164 164 - object_id, 165 165 - payload, 166 166 - }); 167 167 - } 168 168 - 169 169 - /// Publish a state update (latest-only, atomically increments seq). 170 170 - pub(crate) fn publish_state(&self, payload: Bytes) { 171 171 - let prev = self.state.borrow().clone(); 172 172 - let new = StateValue { 173 173 - payload, 174 174 - seq: prev.seq + 1, 175 175 - }; 176 176 - let _ = self.state.send(new); 177 177 - } 178 178 - 179 179 - /// Cache the fMP4 init segment (ftyp+moov boxes) so late-joining clients 180 180 - /// can initialise their MediaSource before receiving live fragments. 181 181 - pub(crate) fn cache_init_segment(&self, payload: Bytes, group_id: u64) { 182 182 - tracing::debug!(group_id, size = payload.len(), "caching init segment"); 183 183 - if let Ok(mut guard) = self.video_init.lock() { 184 184 - *guard = Some((group_id, payload.clone())); 185 185 - } 186 186 - self.video_init_watch.send_replace(true); 187 187 - } 188 188 - 189 189 - pub(crate) fn get_init_segment(&self) -> Option<(u64, Bytes)> { 190 190 - let result = self.video_init.lock().ok().and_then(|g| g.clone()); 191 191 - if let Some((group_id, ref data)) = result { 192 192 - tracing::debug!(group_id, size = data.len(), "retrieved cached init segment"); 193 193 - } else { 194 194 - tracing::debug!("no cached init segment found"); 195 195 - } 196 196 - result 197 197 - } 198 198 - 199 199 - pub(crate) fn video_init_waiter(&self) -> watch::Receiver<bool> { 200 200 - self.video_init_watch.subscribe() 201 201 - } 202 202 - } 203 203 - 204 141 /// All room operations — processed sequentially by the per-room actor. 205 142 pub(crate) enum RoomCommand { 206 143 QueueTrack(TrackMeta), ··· 222 159 Shutdown, 223 160 } 224 161 162 162 + /// Ephemeral info about the active track, stored in-memory only. 163 163 + #[derive(Debug, Clone)] 164 164 + pub(crate) struct ActiveTrackInfo { 165 165 + pub(crate) id: i64, 166 166 + pub(crate) title: String, 167 167 + pub(crate) url: String, 168 168 + pub(crate) duration: String, 169 169 + /// Epoch ms when the track started — for client-side elapsed computation. 170 170 + pub(crate) started_at_wall: i64, 171 171 + } 172 172 + 173 173 + impl ActiveTrackInfo { 174 174 + pub(crate) fn from_item(item: &crate::playlist::QueueItem) -> Self { 175 175 + Self { 176 176 + id: item.id, 177 177 + title: item.title.clone(), 178 178 + url: item.url.clone(), 179 179 + duration: item.duration.clone(), 180 180 + started_at_wall: chrono::Utc::now().timestamp_millis(), 181 181 + } 182 182 + } 183 183 + } 184 184 + 225 185 /// Public handle to a room — allows sending commands and reading publishers. 226 186 #[derive(Clone)] 227 187 pub(crate) struct RoomHandle { ··· 233 193 #[cfg(test)] 234 194 mod tests { 235 195 use super::*; 236 236 - use std::time::Duration; 237 196 238 197 #[test] 239 198 fn test_room_id_display() { ··· 249 208 assert_eq!(TrackId::from_name("state"), Some(TrackId::State)); 250 209 assert_eq!(TrackId::from_name("invalid"), None); 251 210 assert_eq!(TrackId::from_name(""), None); 252 252 - } 253 253 - 254 254 - /// Test 1: Verify cache_init_segment stores and get_init_segment retrieves, 255 255 - /// including correct group_id matching. 256 256 - #[test] 257 257 - fn test_init_cache_and_retrieve() { 258 258 - let publishers = TrackPublishers::new(); 259 259 - 260 260 - // Should be None initially 261 261 - assert!( 262 262 - publishers.get_init_segment().is_none(), 263 263 - "init should be None before caching" 264 264 - ); 265 265 - 266 266 - // Cache with group_id=42 267 267 - let payload = Bytes::from("ftyp-moov-data"); 268 268 - publishers.cache_init_segment(payload.clone(), 42); 269 269 - 270 270 - // Retrieve and verify 271 271 - let retrieved = publishers.get_init_segment(); 272 272 - assert!(retrieved.is_some(), "init should be Some after caching"); 273 273 - let (group_id, data) = retrieved.unwrap(); 274 274 - assert_eq!(group_id, 42, "group_id should match what was cached"); 275 275 - assert_eq!(data, payload, "payload should match what was cached"); 276 276 - } 277 277 - 278 278 - /// Test 2: Verify that the video_init_watch fires when cache_init_segment is called, 279 279 - /// so a task waiting on it can proceed. 280 280 - #[tokio::test] 281 281 - async fn test_init_watch_signals() { 282 282 - let publishers = TrackPublishers::new(); 283 283 - 284 284 - // Get a waiter before the init is cached 285 285 - let mut waiter = publishers.video_init_waiter(); 286 286 - assert!(!*waiter.borrow(), "initial watch value should be false"); 287 287 - 288 288 - // Spawn a task that caches init after a short delay 289 289 - let pubs = publishers.clone(); 290 290 - tokio::spawn(async move { 291 291 - tokio::time::sleep(Duration::from_millis(50)).await; 292 292 - pubs.cache_init_segment(Bytes::from("init-payload"), 1); 293 293 - }); 294 294 - 295 295 - // Wait for the watch to fire (with timeout) 296 296 - let _ = tokio::time::timeout(Duration::from_secs(2), waiter.changed()) 297 297 - .await 298 298 - .expect("watch should fire within timeout when init is cached"); 299 299 - 300 300 - // After the watch fires, the value should be true 301 301 - assert!( 302 302 - *waiter.borrow(), 303 303 - "watch value should be true after cache_init_segment" 304 304 - ); 305 305 - 306 306 - // And the init should be retrievable 307 307 - let retrieved = publishers.get_init_segment(); 308 308 - assert!( 309 309 - retrieved.is_some(), 310 310 - "init should be retrievable after watch fires" 311 311 - ); 312 312 - let (group_id, data) = retrieved.unwrap(); 313 313 - assert_eq!(group_id, 1, "group_id should match"); 314 314 - assert_eq!(data, Bytes::from("init-payload"), "payload should match"); 315 315 - } 316 316 - 317 317 - /// Verify caching a new init segment for a new group overwrites the old one. 318 318 - #[test] 319 319 - fn test_cache_overwrite_on_new_group() { 320 320 - let publishers = TrackPublishers::new(); 321 321 - 322 322 - publishers.cache_init_segment(Bytes::from("init-1"), 1); 323 323 - publishers.cache_init_segment(Bytes::from("init-2"), 2); 324 324 - 325 325 - let retrieved = publishers.get_init_segment(); 326 326 - assert!(retrieved.is_some(), "init should be cached"); 327 327 - let (group_id, data) = retrieved.unwrap(); 328 328 - assert_eq!(group_id, 2, "group_id should be from the latest cache call"); 329 329 - assert_eq!( 330 330 - data, 331 331 - Bytes::from("init-2"), 332 332 - "payload should be from the latest cache call" 333 333 - ); 334 211 } 335 212 }

+27 -15

src/web.rs

Reviewed

··· 16 16 use sqlx::{Pool, Sqlite}; 17 17 use tokio::sync::Mutex; 18 18 19 19 - use crate::media; 20 19 use crate::room; 20 20 + use crate::source::SourceRegistry; 21 21 use crate::types::{RoomId, TrackMeta}; 22 22 23 23 /// Per-IP rate limiter for POST /api/ingest (max 1 req/5s per IP). ··· 58 58 pub rooms: room::Registry, 59 59 pub pool: Pool<Sqlite>, 60 60 pub cache_dir: PathBuf, 61 61 + pub sources: Arc<SourceRegistry>, 61 62 pub rate_limiter: RateLimiter, 62 63 } 63 64 64 64 - pub(crate) fn router(rooms: room::Registry, pool: Pool<Sqlite>, cache_dir: PathBuf) -> Router { 65 65 + pub(crate) fn router( 66 66 + rooms: room::Registry, 67 67 + pool: Pool<Sqlite>, 68 68 + cache_dir: PathBuf, 69 69 + sources: Arc<SourceRegistry>, 70 70 + ) -> Router { 65 71 let rate_limiter = RateLimiter::new(); 66 72 67 73 // Periodic cleanup for rate limiter entries (every 30s). ··· 78 84 rooms, 79 85 pool, 80 86 cache_dir, 87 87 + sources, 81 88 rate_limiter, 82 89 }; 83 90 ··· 141 148 duration: i.duration, 142 149 thumbnail: i.thumbnail, 143 150 url: i.url, 144 144 - extractor: i.extractor, 151 151 + source: i 152 152 + .source 153 153 + .parse() 154 154 + .unwrap_or_else(|s: String| { 155 155 + tracing::warn!(source = %s, "unknown source kind in queue_items"); 156 156 + crate::types::SourceKind::Direct 157 157 + }), 145 158 }) 146 159 .collect::<Vec<_>>() 147 160 }; ··· 196 209 } 197 210 198 211 // Rate limiting — max 1 request per 5 seconds per IP. 199 199 - match state.rate_limiter.check(addr).await { 200 200 - Err(retry_after) => { 201 201 - let mut res = err_response(StatusCode::TOO_MANY_REQUESTS, "rate limited"); 202 202 - res.headers_mut().insert( 203 203 - axum::http::header::RETRY_AFTER, 204 204 - retry_after.as_secs().to_string().parse().unwrap(), 205 205 - ); 206 206 - return Err(res); 207 207 - } 208 208 - Ok(()) => {} 212 212 + if let Err(retry_after) = state.rate_limiter.check(addr).await { 213 213 + let mut res = err_response(StatusCode::TOO_MANY_REQUESTS, "rate limited"); 214 214 + res.headers_mut().insert( 215 215 + axum::http::header::RETRY_AFTER, 216 216 + retry_after.as_secs().to_string().parse().expect("valid retry-after seconds"), 217 217 + ); 218 218 + return Err(res); 209 219 } 210 220 211 221 let room_id = RoomId(req.room_id); ··· 215 225 return Err(err_response(StatusCode::NOT_FOUND, "room not found")); 216 226 } 217 227 218 218 - // Run yt-dlp extraction. 219 219 - let meta = media::extract(&req.url, &state.cache_dir) 228 228 + // Extract metadata via the source registry. 229 229 + let meta = state 230 230 + .sources 231 231 + .extract(&req.url, &state.cache_dir) 220 232 .await 221 233 .map_err(|e| err_response(StatusCode::BAD_REQUEST, &e.to_string()))?; 222 234

+115 -127

templates/room.html

Reviewed

··· 193 193 button:active { 194 194 transform: scale(0.96); 195 195 } 196 196 + button:disabled { 197 197 + opacity: 0.5; 198 198 + cursor: not-allowed; 199 199 + transform: none; 200 200 + filter: none; 201 201 + } 196 202 button.btn-primary { 197 203 background: var(--green); 198 204 color: var(--crust); ··· 274 280 color: var(--blue); 275 281 text-decoration: none; 276 282 font-size: 0.8rem; 277 277 - padding: 2px 8px; 283 283 + padding: 6px 10px; 278 284 border-radius: var(--radius-sm); 279 285 background: color-mix(in srgb, var(--blue) 15%, transparent); 280 286 transition: background 150ms ease-out; 281 287 flex-shrink: 0; 288 288 + min-height: 40px; 289 289 + display: inline-flex; 290 290 + align-items: center; 282 291 } 283 292 .source-link:hover { 284 293 background: color-mix(in srgb, var(--blue) 25%, transparent); 294 294 + } 295 295 + 296 296 + .history-item { 297 297 + opacity: 0.7; 298 298 + } 299 299 + 300 300 + /* Connection status */ 301 301 + .conn-status { 302 302 + font-size: 0.75rem; 303 303 + color: var(--overlay0); 304 304 + margin-bottom: 0.5rem; 305 305 + text-align: right; 306 306 + transition: color 300ms ease-out; 307 307 + } 308 308 + .conn-status.connected { 309 309 + color: var(--green); 310 310 + } 311 311 + .conn-status.disconnected { 312 312 + color: var(--red); 285 313 } 286 314 287 315 /* History header */ ··· 295 323 } 296 324 297 325 /* Empty state */ 298 298 - .empty { 299 299 - color: var(--overlay0); 326 326 + .empty-state { 327 327 + display: flex; 328 328 + flex-direction: column; 329 329 + align-items: center; 330 330 + justify-content: center; 331 331 + gap: 0.75rem; 332 332 + padding: 3rem 1.5rem; 300 333 text-align: center; 301 301 - padding: 3rem 1rem; 302 302 - font-size: 0.9375rem; 334 334 + color: var(--overlay1); 335 335 + font-size: 0.875rem; 336 336 + line-height: 1.5; 337 337 + background: var(--surface0); 338 338 + border-radius: var(--radius-lg); 339 339 + border: 1px dashed var(--surface2); 340 340 + margin-bottom: 0.5rem; 341 341 + } 342 342 + .empty-state .icon { 343 343 + font-size: 1.75rem; 344 344 + line-height: 1; 345 345 + opacity: 0.5; 346 346 + } 347 347 + .empty-state .label { 348 348 + color: var(--subtext0); 349 349 + font-weight: 500; 350 350 + } 351 351 + .empty-state .hint { 352 352 + color: var(--overlay0); 353 353 + font-size: 0.8125rem; 354 354 + max-width: 24em; 355 355 + } 356 356 + 357 357 + /* Now-playing empty */ 358 358 + #now-playing.empty { 359 359 + color: var(--overlay0); 360 360 + font-size: 0.875rem; 361 361 + font-style: italic; 303 362 } 304 363 305 364 /* Chat */ ··· 477 536 478 537 <div id="queue-section" style="min-height: 3rem"> 479 538 {% if queue_empty %} 480 480 - <div class="empty"> 481 481 - Queue is empty. Paste a link above to get started. 539 539 + <div class="empty-state"> 540 540 + <div class="icon">&#x1F3B5;</div> 541 541 + <div class="label">Queue is empty</div> 542 542 + <div class="hint">Paste a link from YouTube, SoundCloud, or Bandcamp above to get started.</div> 482 543 </div> 483 544 {% else %} 484 545 <ul class="queue"> ··· 509 570 <div class="history-header">Played</div> 510 571 <ul class="queue"> 511 572 {% for item in history %} 512 512 - <li style="opacity: 0.7"> 573 573 + <li class="history-item"> 513 574 {% if let Some(thumb) = item.thumbnail %} 514 575 <img src="{{ thumb }}" class="thumb-img" alt="" /> 515 576 {% else %} ··· 532 593 </div> 533 594 534 595 <div class="chat-col"> 596 596 + <div id="conn-status" class="conn-status disconnected">disconnected</div> 535 597 <div id="chat-messages" class="chat-messages"></div> 536 598 <div class="chat-input-row"> 537 599 <input ··· 611 673 return [str, o + len]; 612 674 } 613 675 676 676 + // ── Debug logging (opt-in via ?debug or localStorage) ──────────── 677 677 + const DEBUG = new URLSearchParams(location.search).has("debug") || 678 678 + localStorage.getItem("moqbox_debug"); 679 679 + function debug(...args) { if (DEBUG) console.log(...args); } 680 680 + 614 681 // ── Chat WebSocket client ─────────────────────────────────────── 615 682 const roomId = "{{ room_id }}"; 616 683 let userName = sessionStorage.getItem("moqbox_user"); ··· 632 699 let seekInitDone = false; // true after first seek on join or track change 633 700 634 701 function initVideo() { 635 635 - console.log( 636 636 - "initVideo() called, readyState:", 637 637 - mediaSource ? mediaSource.readyState : "null", 638 638 - ); 702 702 + // Guard: don't create a second player if already initialized. 703 703 + if (document.getElementById("video-player")) return; 704 704 + debug("initVideo called"); 639 705 const container = 640 706 document.getElementById("video-container") || 641 707 document.querySelector(".main-col"); ··· 644 710 video.controls = true; 645 711 video.autoplay = true; 646 712 video.muted = true; 647 647 - video.style.cssText = 648 648 - "width:100%;max-height:85vh;object-fit:contain;background:#000;display:block;"; 649 713 container.insertBefore(video, container.firstChild); 650 714 mediaSource = new MediaSource(); 651 651 - console.log( 652 652 - "initVideo: created MediaSource, readyState:", 653 653 - mediaSource.readyState, 654 654 - ); 655 715 mediaSource.addEventListener("sourceopen", () => { 656 656 - console.log( 657 657 - "MediaSource sourceopen fired, readyState:", 658 658 - mediaSource.readyState, 659 659 - ); 716 716 + debug("MediaSource sourceopen"); 660 717 try { 661 718 sourceBuffer = mediaSource.addSourceBuffer( 662 719 'video/mp4; codecs="avc1.4d0028,mp4a.40.2"', 663 720 ); 664 721 sourceBuffer.mode = "segments"; 665 665 - console.log( 666 666 - "SourceBuffer created, mode:", 667 667 - sourceBuffer.mode, 668 668 - "updating:", 669 669 - sourceBuffer.updating, 670 670 - ); 671 722 sourceBuffer.addEventListener("updateend", () => { 672 672 - console.log( 673 673 - "SourceBuffer updateend, pendingBoxes.length:", 674 674 - pendingBoxes.length, 675 675 - ); 676 723 if (pendingBoxes.length > 0) { 677 724 const next = pendingBoxes.shift(); 678 725 try { ··· 681 728 console.warn("SourceBuffer append failed:", e); 682 729 } 683 730 } 684 684 - // Late-join seek: after first media segment is appended, 685 685 - // seek to the start of buffered data to get out of the 686 686 - // t=0 stall. The periodic sync in updateNowPlaying handles 687 687 - // drift correction to the live position. 688 731 if (!seekInitDone) { 689 732 const v = document.getElementById("video-player"); 690 733 if (v && v.buffered.length > 0) { 691 734 seekInitDone = true; 692 735 v.currentTime = v.buffered.start(0); 693 693 - console.log( 694 694 - "Initial seek to:", 695 695 - v.currentTime, 696 696 - "buffered:", 697 697 - v.buffered.start(0), 698 698 - "-", 699 699 - v.buffered.end(0), 700 700 - ); 701 736 } 702 737 } 703 738 }); 704 704 - // Flush any boxes that arrived before sourceopen. 705 705 - console.log( 706 706 - "sourceopen flush: pendingBoxes.length before flush:", 707 707 - pendingBoxes.length, 708 708 - ); 739 739 + // Flush boxes that arrived before sourceopen. 709 740 while (pendingBoxes.length > 0) { 710 741 const box = pendingBoxes.shift(); 711 742 if (!sourceBuffer.updating) { ··· 729 760 730 761 function appendVideo(data) { 731 762 if (sourceBuffer && !sourceBuffer.updating) { 732 732 - console.log( 733 733 - "appendVideo: appending to sourceBuffer, size:", 734 734 - data.length, 735 735 - ); 736 763 sourceBuffer.appendBuffer(data); 737 764 } else { 738 738 - // Buffer if sourceBuffer is still null or busy updating. 739 739 - console.log( 740 740 - "appendVideo: buffering to pendingBoxes, size:", 741 741 - data.length, 742 742 - "sourceBuffer:", 743 743 - !!sourceBuffer, 744 744 - "updating:", 745 745 - sourceBuffer ? sourceBuffer.updating : "N/A", 746 746 - ); 747 765 pendingBoxes.push(data); 748 766 } 749 767 } ··· 775 793 ); 776 794 } 777 795 796 796 + function updateConnStatus(connected) { 797 797 + const el = document.getElementById("conn-status"); 798 798 + if (!el) return; 799 799 + el.className = "conn-status " + (connected ? "connected" : "disconnected"); 800 800 + el.textContent = connected ? "connected" : "disconnected"; 801 801 + } 802 802 + 778 803 function connect() { 779 804 ws = new WebSocket( 780 805 proto + ··· 788 813 ws.binaryType = "arraybuffer"; 789 814 790 815 ws.onopen = () => { 791 791 - console.log( 792 792 - "WS onopen: resetting awaitingInit=true, clearing pendingBoxes", 793 793 - ); 816 816 + debug("WS connected"); 817 817 + updateConnStatus(true); 794 818 reconnectDelay = 1000; 795 795 - // Full reset on reconnect — ensures init is re-sent and accepted. 796 819 awaitingInit = true; 797 820 pendingBoxes.length = 0; 798 821 currentTrackId = null; ··· 811 834 const [objectId, o4] = decVarint(buf, o3); 812 835 if (trackId === 1) { 813 836 const videoData = new Uint8Array(buf.slice(o4)); 814 814 - console.log( 815 815 - "Video obj received, size:", 816 816 - videoData.length, 817 817 - "awaitingInit:", 818 818 - awaitingInit, 819 819 - "pendingBoxes:", 820 820 - pendingBoxes.length, 821 821 - ); 822 837 823 838 if (awaitingInit) { 824 839 const isInit = ··· 830 845 videoData[7], 831 846 ) === "ftyp"; 832 847 833 833 - console.log( 834 834 - "awaitingInit check: isInit:", 835 835 - isInit, 836 836 - "first 4 bytes:", 837 837 - String.fromCharCode( 838 838 - videoData[4], 839 839 - videoData[5], 840 840 - videoData[6], 841 841 - videoData[7], 842 842 - ), 843 843 - ); 844 844 - 845 848 if (isInit) { 846 846 - console.log( 847 847 - "INIT RECEIVED — setting awaitingInit=false, clearing pendingBoxes", 848 848 - ); 849 849 + debug("init segment received"); 849 850 awaitingInit = false; 850 851 pendingBoxes.length = 0; 851 852 if (sourceBuffer && !sourceBuffer.updating) { ··· 860 861 } catch (_) {} 861 862 } 862 863 } else { 863 863 - // Buffer until init arrives. 864 864 - console.log( 865 865 - "Buffering non-init data, pendingBoxes now:", 866 866 - pendingBoxes.length + 1, 867 867 - ); 868 864 pendingBoxes.push(videoData); 869 865 off = buf.length; 870 866 return; ··· 891 887 off = buf.length; 892 888 } 893 889 } else if (msgType === 0x02) { 894 894 - // SUBSCRIBE_OK — just advance past it. 895 890 const [ns, o2] = decString(buf, o1); 896 891 const [tr, o3] = decString(buf, o2); 897 892 off = o3; 898 893 } else if (msgType === 0x06) { 899 899 - // SUBSCRIBE_RST — advance past it. 900 894 const [ns, o2] = decString(buf, o1); 901 895 const [tr, o3] = decString(buf, o2); 902 896 const [reason, o4] = decString(buf, o3); ··· 908 902 }; 909 903 910 904 ws.onclose = () => { 911 911 - console.log("WS closed, reconnecting in " + reconnectDelay + "ms"); 905 905 + updateConnStatus(false); 912 906 setTimeout(() => { 913 907 reconnectDelay = Math.min(reconnectDelay * 2, 30000); 914 908 connect(); ··· 972 966 // Detect track change — set awaitingInit so the next video object 973 967 // is treated as the new init segment. 974 968 if (currentTrackId !== null && currentTrackId !== trackId) { 975 975 - console.log( 976 976 - "Track change detected:", 977 977 - currentTrackId, 978 978 - "->", 979 979 - trackId, 980 980 - "setting awaitingInit=true", 981 981 - ); 969 969 + debug("track change", currentTrackId, "->", trackId); 982 970 awaitingInit = true; 983 971 pendingBoxes.length = 0; 984 972 seekInitDone = false; 985 973 } 986 986 - console.log( 987 987 - "updateNowPlaying: trackId=", 988 988 - trackId, 989 989 - "currentTrackId=", 990 990 - currentTrackId, 991 991 - "awaitingInit=", 992 992 - awaitingInit, 993 993 - ); 994 974 currentTrackId = trackId; 995 975 996 976 const startedAt = state.current_track.started_at; ··· 1038 1018 } else { 1039 1019 currentTrackId = null; 1040 1020 el.innerHTML = 1041 1041 - '<span style="color:#6c7086">Waiting for tracks...</span>'; 1021 1021 + '<div class="empty-state" style="border-style:none;background:transparent;padding:2rem 1rem"><div class="icon">&#x25D4;</div><div class="label">Waiting for tracks</div><div class="hint">Paste a link above or wait for the next track to start.</div></div>'; 1042 1022 } 1043 1023 1044 1024 // Update queue section from state. ··· 1076 1056 qSection.innerHTML = html; 1077 1057 } else if (state.current_track) { 1078 1058 qSection.innerHTML = 1079 1079 - '<div class="empty">No upcoming tracks. Add another link to the queue.</div>'; 1059 1059 + '<div class="empty-state"><div class="icon">&#x25B6;</div><div class="label">No upcoming tracks</div><div class="hint">Add another link to keep the queue going.</div></div>'; 1080 1060 } else { 1081 1061 qSection.innerHTML = 1082 1082 - '<div class="empty">Queue is empty. Paste a link above to get started.</div>'; 1062 1062 + '<div class="empty-state"><div class="icon">&#x1F3B5;</div><div class="label">Queue is empty</div><div class="hint">Paste a link from YouTube, SoundCloud, or Bandcamp above to get started.</div></div>'; 1083 1063 } 1084 1064 1085 1065 if (hSection && state.history && state.history.length > 0) { ··· 1115 1095 function formatTime(secs) { 1116 1096 const m = Math.floor(secs / 60); 1117 1097 const s = Math.floor(secs % 60); 1118 1118 - return m + ":" + (s < 10 ? "0" : "") + s; 1098 1098 + return m + ":" + String(s).padStart(2, "0"); 1119 1099 } 1120 1100 1121 1101 function esc(s) { ··· 1158 1138 // ── Ingest helper ─────────────────────────────────────────────── 1159 1139 async function ingest() { 1160 1140 const input = document.getElementById("url-input"); 1161 1161 - const resp = await fetch("/api/ingest", { 1162 1162 - method: "POST", 1163 1163 - headers: { "Content-Type": "application/json" }, 1164 1164 - body: JSON.stringify({ url: input.value, room_id: roomId }), 1165 1165 - }); 1166 1166 - if (resp.ok) { 1167 1167 - input.value = ""; 1168 1168 - // Queue will update via state track broadcast — no reload needed. 1169 1169 - } else { 1170 1170 - const err = await resp.json(); 1171 1171 - alert(err.error || "ingest failed"); 1141 1141 + const btn = document.querySelector(".input-row .btn-primary"); 1142 1142 + if (!input.value.trim()) return; 1143 1143 + btn.disabled = true; 1144 1144 + btn.textContent = "Adding…"; 1145 1145 + try { 1146 1146 + const resp = await fetch("/api/ingest", { 1147 1147 + method: "POST", 1148 1148 + headers: { "Content-Type": "application/json" }, 1149 1149 + body: JSON.stringify({ url: input.value, room_id: roomId }), 1150 1150 + }); 1151 1151 + if (resp.ok) { 1152 1152 + input.value = ""; 1153 1153 + } else { 1154 1154 + const err = await resp.json(); 1155 1155 + alert(err.error || "ingest failed"); 1156 1156 + } 1157 1157 + } finally { 1158 1158 + btn.disabled = false; 1159 1159 + btn.textContent = "Add"; 1172 1160 } 1173 1161 } 1174 1162

+275

src/source/direct.rs

Reviewed

··· 1 1 + //! Direct URL media source implementation. 2 2 + //! 3 3 + //! Handles URLs that point directly to media files (`.mp4`, `.webm`, `.mp3`, 4 4 + //! etc.) without going through yt-dlp. Metadata is extracted heuristically 5 5 + //! from the URL path — no network requests are made during extraction. 6 6 + 7 7 + use std::path::Path; 8 8 + 9 9 + use crate::source::{MediaSource, SourceError}; 10 10 + use crate::types::TrackMeta; 11 11 + 12 12 + /// Media file extensions that DirectSource can handle. 13 13 + const MEDIA_EXTENSIONS: &[&str] = &[ 14 14 + "mp4", "webm", "mkv", "avi", "mov", "m4v", 15 15 + "mp3", "ogg", "flac", "wav", "m4a", "aac", "opus", "wma", 16 16 + ]; 17 17 + 18 18 + // --------------------------------------------------------------------------- 19 19 + // Source 20 20 + // --------------------------------------------------------------------------- 21 21 + 22 22 + /// A media source that handles direct media file URLs. 23 23 + /// 24 24 + /// Detects media files by URL path extension. If the URL has a recognised 25 25 + /// media file extension (`mp4`, `webm`, `mp3`, etc.), it is treated as a 26 26 + /// direct-play URL: 27 27 + /// 28 28 + /// - **extract** derives the title from the filename (strips extension, 29 29 + /// replaces separators with spaces) 30 30 + /// - **resolve** returns the URL unchanged 31 31 + /// 32 32 + /// This source is intended as a fallback after [`YtdlpSource`](super::ytdlp::YtdlpSource) 33 33 + /// — it only matches URLs with media extensions whose filenames look like 34 34 + /// media files. 35 35 + pub(crate) struct DirectSource; 36 36 + 37 37 + impl DirectSource { 38 38 + pub(crate) fn new() -> Self { 39 39 + Self 40 40 + } 41 41 + } 42 42 + 43 43 + #[async_trait::async_trait] 44 44 + impl MediaSource for DirectSource { 45 45 + async fn extract(&self, url: &str, _cache_dir: &Path) -> Result<TrackMeta, SourceError> { 46 46 + let Some(ext) = extension(url) else { 47 47 + return Err(SourceError::Unsupported(format!("no file extension in URL: {url}"))); 48 48 + }; 49 49 + 50 50 + if !MEDIA_EXTENSIONS.contains(&ext.as_str()) { 51 51 + return Err(SourceError::Unsupported(format!( 52 52 + "unrecognised media extension '.{ext}' in URL", 53 53 + ))); 54 54 + } 55 55 + 56 56 + let title = extract_title(url, &ext); 57 57 + 58 58 + Ok(TrackMeta { 59 59 + title, 60 60 + duration: "--:--".into(), 61 61 + thumbnail: None, 62 62 + url: url.to_string(), 63 63 + source: crate::types::SourceKind::Direct, 64 64 + }) 65 65 + } 66 66 + 67 67 + async fn resolve(&self, url: &str, _cache_dir: &Path) -> Result<String, SourceError> { 68 68 + if !url.starts_with("http://") && !url.starts_with("https://") { 69 69 + return Err(SourceError::Unsupported(format!("not an HTTP(S) URL: {url}"))); 70 70 + } 71 71 + 72 72 + // Direct URLs are already playable as-is. 73 73 + Ok(url.to_string()) 74 74 + } 75 75 + } 76 76 + 77 77 + // --------------------------------------------------------------------------- 78 78 + // Helpers 79 79 + // --------------------------------------------------------------------------- 80 80 + 81 81 + /// Extract the file extension (without dot) from a URL path, if any. 82 82 + fn extension(url: &str) -> Option<String> { 83 83 + let path = url.split('?').next().unwrap_or(url); 84 84 + let path = path.split('#').next().unwrap_or(path); 85 85 + let dot = path.rfind('.')?; 86 86 + 87 87 + // Extension must be after the last path segment. 88 88 + let after_dot = &path[dot + 1..]; 89 89 + if after_dot.is_empty() || after_dot.contains('/') { 90 90 + return None; 91 91 + } 92 92 + 93 93 + Some(after_dot.to_ascii_lowercase()) 94 94 + } 95 95 + 96 96 + /// Derive a human-readable title from a URL path. 97 97 + /// 98 98 + /// Strips the extension, takes the last path segment, and replaces common 99 99 + /// URL separators (`-`, `_`, `+`, `.`) with spaces. 100 100 + fn extract_title(url: &str, ext: &str) -> String { 101 101 + let path = url.split('?').next().unwrap_or(url); 102 102 + let path = path.split('#').next().unwrap_or(path); 103 103 + 104 104 + // Take the last non-empty path segment. 105 105 + let filename = path 106 106 + .rsplit('/') 107 107 + .find(|s| !s.is_empty()) 108 108 + .unwrap_or("untitled"); 109 109 + 110 110 + // Strip the extension from the filename. 111 111 + let dot_len = ext.len() + 1; // +1 for the dot 112 112 + let stem = if filename.len() >= dot_len && filename.ends_with(&format!(".{ext}")) { 113 113 + &filename[..filename.len() - dot_len] 114 114 + } else { 115 115 + filename 116 116 + }; 117 117 + 118 118 + // Replace URL separators with spaces, then collapse whitespace. 119 119 + let separators = ['-', '_', '+', '.']; 120 120 + let result: String = stem 121 121 + .replace(&separators[..], " ") 122 122 + .split_whitespace() 123 123 + .collect::<Vec<_>>() 124 124 + .join(" "); 125 125 + 126 126 + if result.is_empty() { "Untitled".into() } else { result } 127 127 + } 128 128 + 129 129 + // --------------------------------------------------------------------------- 130 130 + // Tests 131 131 + // --------------------------------------------------------------------------- 132 132 + 133 133 + #[cfg(test)] 134 134 + mod tests { 135 135 + use super::*; 136 136 + 137 137 + // -- Extension detection ------------------------------------------------- 138 138 + 139 139 + #[test] 140 140 + fn test_extension_mp4() { 141 141 + assert_eq!(extension("https://example.com/video.mp4"), Some("mp4".into())); 142 142 + } 143 143 + 144 144 + #[test] 145 145 + fn test_extension_with_query() { 146 146 + assert_eq!( 147 147 + extension("https://example.com/v.mp4?t=30"), 148 148 + Some("mp4".into()) 149 149 + ); 150 150 + } 151 151 + 152 152 + #[test] 153 153 + fn test_extension_no_ext() { 154 154 + assert_eq!(extension("https://example.com/video"), None); 155 155 + } 156 156 + 157 157 + #[test] 158 158 + fn test_extension_empty_path() { 159 159 + assert_eq!(extension("https://example.com/"), None); 160 160 + } 161 161 + 162 162 + #[test] 163 163 + fn test_extension_uppercase() { 164 164 + assert_eq!(extension("https://example.com/video.MP4"), Some("mp4".into())); 165 165 + } 166 166 + 167 167 + // -- Title extraction ---------------------------------------------------- 168 168 + 169 169 + #[test] 170 170 + fn test_title_basic() { 171 171 + assert_eq!(extract_title("https://example.com/my-video.mp4", "mp4"), "my video"); 172 172 + } 173 173 + 174 174 + #[test] 175 175 + fn test_title_with_underscores() { 176 176 + assert_eq!( 177 177 + extract_title("https://example.com/great_song.mp3", "mp3"), 178 178 + "great song" 179 179 + ); 180 180 + } 181 181 + 182 182 + #[test] 183 183 + fn test_title_with_dots() { 184 184 + assert_eq!( 185 185 + extract_title("https://cdn.example.com/file.name.with.dots.mp4", "mp4"), 186 186 + "file name with dots" 187 187 + ); 188 188 + } 189 189 + 190 190 + #[test] 191 191 + fn test_title_no_stem() { 192 192 + assert_eq!(extract_title("https://example.com/.mp4", "mp4"), "Untitled"); 193 193 + } 194 194 + 195 195 + #[test] 196 196 + fn test_title_deep_path() { 197 197 + assert_eq!( 198 198 + extract_title("https://cdn.example.com/path/to/some/deep-file.mp4", "mp4"), 199 199 + "deep file" 200 200 + ); 201 201 + } 202 202 + 203 203 + #[test] 204 204 + fn test_title_no_extension_match() { 205 205 + // If the filename doesn't actually end with the given extension 206 206 + // (shouldn't happen in practice), use the whole filename. 207 207 + assert_eq!( 208 208 + extract_title("https://example.com/video.mkv", "mp4"), 209 209 + "video mkv" 210 210 + ); 211 211 + } 212 212 + 213 213 + #[test] 214 214 + fn test_title_multi_space_collapse() { 215 215 + assert_eq!( 216 216 + extract_title("https://example.com/a---b.mp4", "mp4"), 217 217 + "a b" 218 218 + ); 219 219 + } 220 220 + 221 221 + // -- MediaSource implementation ------------------------------------------ 222 222 + 223 223 + #[tokio::test] 224 224 + async fn test_direct_source_accepts_mp4() { 225 225 + let src = DirectSource::new(); 226 226 + let meta = src 227 227 + .extract("https://example.com/video.mp4", Path::new("/tmp")) 228 228 + .await 229 229 + .expect("DirectSource should accept .mp4 URLs"); 230 230 + assert_eq!(meta.title, "video"); 231 231 + assert_eq!(meta.source, crate::types::SourceKind::Direct); 232 232 + assert_eq!(meta.duration, "--:--"); 233 233 + assert!(meta.thumbnail.is_none()); 234 234 + } 235 235 + 236 236 + #[tokio::test] 237 237 + async fn test_direct_source_rejects_no_extension() { 238 238 + let src = DirectSource::new(); 239 239 + let result = src 240 240 + .extract("https://example.com/video", Path::new("/tmp")) 241 241 + .await; 242 242 + assert!(matches!(result, Err(SourceError::Unsupported(_)))); 243 243 + } 244 244 + 245 245 + #[tokio::test] 246 246 + async fn test_direct_source_rejects_unknown_extension() { 247 247 + let src = DirectSource::new(); 248 248 + let result = src 249 249 + .extract("https://example.com/file.pdf", Path::new("/tmp")) 250 250 + .await; 251 251 + assert!(matches!(result, Err(SourceError::Unsupported(_)))); 252 252 + } 253 253 + 254 254 + #[tokio::test] 255 255 + async fn test_direct_source_resolve_returns_url() { 256 256 + let src = DirectSource::new(); 257 257 + let stream = src 258 258 + .resolve("https://example.com/video.mp4", Path::new("/tmp")) 259 259 + .await 260 260 + .expect("DirectSource should resolve .mp4 URLs"); 261 261 + assert_eq!(stream, "https://example.com/video.mp4"); 262 262 + } 263 263 + 264 264 + #[tokio::test] 265 265 + async fn test_direct_source_resolve_rejects_non_http() { 266 266 + let src = DirectSource::new(); 267 267 + let result = src 268 268 + .resolve("ftp://example.com/video.mp4", Path::new("/tmp")) 269 269 + .await; 270 270 + assert!(matches!(result, Err(SourceError::Unsupported(_)))); 271 271 + } 272 272 + 273 273 + // Integration test with yt-dlp is in src/tests/ — this file focuses on 274 274 + // DirectSource unit tests without network dependencies. 275 275 + }

+275

src/source/mod.rs

Reviewed

··· 1 1 + //! Pluggable media source resolution. 2 2 + //! 3 3 + //! Defines [`MediaSource`] — the seam between URL-based track selection and 4 4 + //! playable stream production. Different implementations handle different URL 5 5 + //! patterns (yt-dlp, direct media URLs, etc.). 6 6 + //! 7 7 + //! # Adding a new source 8 8 + //! 9 9 + //! 1. Implement [`MediaSource`] for your source type. 10 10 + //! 2. Register it with [`SourceRegistry::register`]. 11 11 + //! 3. Sources are tried in registration order — first match wins. 12 12 + 13 13 + pub(crate) mod direct; 14 14 + pub(crate) mod ytdlp; 15 15 + 16 16 + use std::path::Path; 17 17 + 18 18 + use crate::types::TrackMeta; 19 19 + 20 20 + // --------------------------------------------------------------------------- 21 21 + // Error 22 22 + // --------------------------------------------------------------------------- 23 23 + 24 24 + /// Errors from media source resolution. 25 25 + #[derive(Debug, thiserror::Error)] 26 26 + pub(crate) enum SourceError { 27 27 + /// The source cannot handle this URL (try the next source in the registry). 28 28 + #[error("unsupported URL: {0}")] 29 29 + Unsupported(String), 30 30 + 31 31 + /// The source recognised the URL but couldn't extract metadata. 32 32 + #[error("extraction failed: {0}")] 33 33 + Extraction(String), 34 34 + 35 35 + /// The source recognised the URL but couldn't resolve a playable stream. 36 36 + #[error("resolution failed: {0}")] 37 37 + Resolution(String), 38 38 + 39 39 + /// I/O error during source operations. 40 40 + #[error("io error: {0}")] 41 41 + Io(#[from] std::io::Error), 42 42 + } 43 43 + 44 44 + // --------------------------------------------------------------------------- 45 45 + // Trait 46 46 + // --------------------------------------------------------------------------- 47 47 + 48 48 + /// A media source knows how to extract metadata from a URL and resolve it to a 49 49 + /// playable stream URL. 50 50 + #[async_trait::async_trait] 51 51 + pub(crate) trait MediaSource: Send + Sync { 52 52 + /// Extract metadata from `url`. 53 53 + /// 54 54 + /// Returns [`SourceError::Unsupported`] if this source cannot handle the 55 55 + /// URL — the registry will try the next source. Any other error is 56 56 + /// terminal for this URL. 57 57 + async fn extract(&self, url: &str, cache_dir: &Path) -> Result<TrackMeta, SourceError>; 58 58 + 59 59 + /// Resolve `url` to a playable stream URL. 60 60 + /// 61 61 + /// Returns [`SourceError::Unsupported`] if this source cannot handle the 62 62 + /// URL — the registry will try the next source. 63 63 + async fn resolve(&self, url: &str, cache_dir: &Path) -> Result<String, SourceError>; 64 64 + } 65 65 + 66 66 + // --------------------------------------------------------------------------- 67 67 + // Registry 68 68 + // --------------------------------------------------------------------------- 69 69 + 70 70 + /// A registry of media sources tried in registration order. 71 71 + /// 72 72 + /// When a URL arrives, each source is asked via 73 73 + /// [`MediaSource::extract`] / [`MediaSource::resolve`]. The first source 74 74 + /// that does **not** return [`SourceError::Unsupported`] wins. 75 75 + /// 76 76 + /// # Example 77 77 + /// 78 78 + /// ```ignore 79 79 + /// let mut registry = SourceRegistry::new(); 80 80 + /// registry.register(YtdlpSource::new()); 81 81 + /// registry.register(DirectSource::new()); 82 82 + /// 83 83 + /// let meta = registry.extract("https://youtu.be/...", &cache_dir).await?; 84 84 + /// ``` 85 85 + pub(crate) struct SourceRegistry { 86 86 + sources: Vec<Box<dyn MediaSource>>, 87 87 + } 88 88 + 89 89 + impl Default for SourceRegistry { 90 90 + fn default() -> Self { 91 91 + Self::new() 92 92 + } 93 93 + } 94 94 + 95 95 + impl SourceRegistry { 96 96 + pub(crate) fn new() -> Self { 97 97 + Self { 98 98 + sources: Vec::new(), 99 99 + } 100 100 + } 101 101 + 102 102 + /// Register a source. Sources are tried in registration order. 103 103 + pub(crate) fn register(&mut self, source: Box<dyn MediaSource>) { 104 104 + self.sources.push(source); 105 105 + } 106 106 + 107 107 + /// Extract metadata from a URL by trying each registered source in order. 108 108 + pub(crate) async fn extract( 109 109 + &self, 110 110 + url: &str, 111 111 + cache_dir: &Path, 112 112 + ) -> Result<TrackMeta, SourceError> { 113 113 + let mut unsupported_hint = String::new(); 114 114 + for source in &self.sources { 115 115 + match source.extract(url, cache_dir).await { 116 116 + Ok(meta) => return Ok(meta), 117 117 + Err(SourceError::Unsupported(hint)) => { 118 118 + unsupported_hint = hint; 119 119 + continue; 120 120 + } 121 121 + Err(e) => return Err(e), 122 122 + } 123 123 + } 124 124 + Err(SourceError::Unsupported(format!( 125 125 + "no registered source could handle URL: {unsupported_hint}" 126 126 + ))) 127 127 + } 128 128 + 129 129 + /// Resolve a URL to a playable stream URL by trying each registered source. 130 130 + pub(crate) async fn resolve( 131 131 + &self, 132 132 + url: &str, 133 133 + cache_dir: &Path, 134 134 + ) -> Result<String, SourceError> { 135 135 + let mut unsupported_hint = String::new(); 136 136 + for source in &self.sources { 137 137 + match source.resolve(url, cache_dir).await { 138 138 + Ok(stream) => return Ok(stream), 139 139 + Err(SourceError::Unsupported(hint)) => { 140 140 + unsupported_hint = hint; 141 141 + continue; 142 142 + } 143 143 + Err(e) => return Err(e), 144 144 + } 145 145 + } 146 146 + Err(SourceError::Unsupported(format!( 147 147 + "no registered source could resolve URL: {unsupported_hint}" 148 148 + ))) 149 149 + } 150 150 + } 151 151 + 152 152 + // --------------------------------------------------------------------------- 153 153 + // Tests 154 154 + // --------------------------------------------------------------------------- 155 155 + 156 156 + #[cfg(test)] 157 157 + mod tests { 158 158 + use super::*; 159 159 + 160 160 + /// A mock source that handles URLs starting with "mock://". 161 161 + struct MockSource; 162 162 + 163 163 + #[async_trait::async_trait] 164 164 + impl MediaSource for MockSource { 165 165 + async fn extract(&self, url: &str, _cache_dir: &Path) -> Result<TrackMeta, SourceError> { 166 166 + if url.starts_with("mock://") { 167 167 + Ok(TrackMeta { 168 168 + title: "Mock Track".into(), 169 169 + duration: "3:00".into(), 170 170 + thumbnail: None, 171 171 + url: url.into(), 172 172 + source: crate::types::SourceKind::Ytdlp, 173 173 + }) 174 174 + } else { 175 175 + Err(SourceError::Unsupported(url.into())) 176 176 + } 177 177 + } 178 178 + 179 179 + async fn resolve(&self, url: &str, _cache_dir: &Path) -> Result<String, SourceError> { 180 180 + if url.starts_with("mock://") { 181 181 + Ok(format!("{url}/stream")) 182 182 + } else { 183 183 + Err(SourceError::Unsupported(url.into())) 184 184 + } 185 185 + } 186 186 + } 187 187 + 188 188 + /// A mock source that only matches a specific domain. 189 189 + struct SpecificSource(&'static str); 190 190 + 191 191 + #[async_trait::async_trait] 192 192 + impl MediaSource for SpecificSource { 193 193 + async fn extract(&self, url: &str, _cache_dir: &Path) -> Result<TrackMeta, SourceError> { 194 194 + if url.contains(self.0) { 195 195 + Ok(TrackMeta { 196 196 + title: format!("From {}", self.0), 197 197 + duration: "4:20".into(), 198 198 + thumbnail: None, 199 199 + url: url.into(), 200 200 + source: crate::types::SourceKind::Direct, 201 201 + }) 202 202 + } else { 203 203 + Err(SourceError::Unsupported(url.into())) 204 204 + } 205 205 + } 206 206 + 207 207 + async fn resolve(&self, url: &str, _cache_dir: &Path) -> Result<String, SourceError> { 208 208 + if url.contains(self.0) { 209 209 + Ok(url.into()) 210 210 + } else { 211 211 + Err(SourceError::Unsupported(url.into())) 212 212 + } 213 213 + } 214 214 + } 215 215 + 216 216 + #[tokio::test] 217 217 + async fn test_registry_tries_sources_in_order() { 218 218 + let mut reg = SourceRegistry::new(); 219 219 + reg.register(Box::new(MockSource)); 220 220 + 221 221 + let meta = reg 222 222 + .extract("mock://example.com/track", Path::new("/tmp")) 223 223 + .await 224 224 + .expect("mock source should handle mock://"); 225 225 + assert_eq!(meta.title, "Mock Track"); 226 226 + } 227 227 + 228 228 + #[tokio::test] 229 229 + async fn test_registry_unsupported_url() { 230 230 + let mut reg = SourceRegistry::new(); 231 231 + reg.register(Box::new(MockSource)); 232 232 + 233 233 + let result = reg.extract("unknown://url", Path::new("/tmp")).await; 234 234 + assert!(matches!(result, Err(SourceError::Unsupported(_)))); 235 235 + } 236 236 + 237 237 + #[tokio::test] 238 238 + async fn test_registry_resolve_mock_url() { 239 239 + let mut reg = SourceRegistry::new(); 240 240 + reg.register(Box::new(MockSource)); 241 241 + 242 242 + let stream = reg 243 243 + .resolve("mock://example.com/track", Path::new("/tmp")) 244 244 + .await 245 245 + .expect("mock source should resolve mock://"); 246 246 + assert_eq!(stream, "mock://example.com/track/stream"); 247 247 + } 248 248 + 249 249 + #[tokio::test] 250 250 + async fn test_registry_falls_through_sources() { 251 251 + let mut reg = SourceRegistry::new(); 252 252 + reg.register(Box::new(SpecificSource("alpha"))); 253 253 + reg.register(Box::new(SpecificSource("beta"))); 254 254 + 255 255 + let meta = reg 256 256 + .extract("https://beta.com/video", Path::new("/tmp")) 257 257 + .await 258 258 + .expect("second source should handle beta URLs"); 259 259 + assert_eq!(meta.title, "From beta"); 260 260 + 261 261 + // First source should still work for its URLs. 262 262 + let meta = reg 263 263 + .extract("https://alpha.com/video", Path::new("/tmp")) 264 264 + .await 265 265 + .expect("first source should handle alpha URLs"); 266 266 + assert_eq!(meta.title, "From alpha"); 267 267 + } 268 268 + 269 269 + #[tokio::test] 270 270 + async fn test_registry_empty() { 271 271 + let reg = SourceRegistry::new(); 272 272 + let result = reg.extract("anything", Path::new("/tmp")).await; 273 273 + assert!(matches!(result, Err(SourceError::Unsupported(_)))); 274 274 + } 275 275 + }

+199

src/source/ytdlp.rs

Reviewed

··· 1 1 + //! yt-dlp based media source implementation. 2 2 + //! 3 3 + //! Handles URLs that yt-dlp recognises (YouTube, SoundCloud, Bandcamp, etc.): 4 4 + //! - `extract` calls `yt-dlp --dump-json` for metadata 5 5 + //! - `resolve` calls `yt-dlp -f b -g` for a direct stream URL 6 6 + 7 7 + use std::path::Path; 8 8 + 9 9 + use serde::Deserialize; 10 10 + use tokio::process::Command; 11 11 + 12 12 + use crate::media; 13 13 + use crate::source::{MediaSource, SourceError}; 14 14 + use crate::types::TrackMeta; 15 15 + 16 16 + // --------------------------------------------------------------------------- 17 17 + // yt-dlp JSON output fields 18 18 + // --------------------------------------------------------------------------- 19 19 + 20 20 + #[derive(Debug, Deserialize)] 21 21 + #[serde(rename_all = "snake_case")] 22 22 + struct YtdlpMetadata { 23 23 + title: String, 24 24 + duration: Option<f64>, 25 25 + thumbnail: Option<String>, 26 26 + webpage_url: String, 27 27 + } 28 28 + 29 29 + // --------------------------------------------------------------------------- 30 30 + // Source 31 31 + // --------------------------------------------------------------------------- 32 32 + 33 33 + /// A media source backed by yt-dlp. 34 34 + /// 35 35 + /// Handles URLs from YouTube, SoundCloud, Bandcamp, and hundreds of other 36 36 + /// sites supported by yt-dlp. Requires `yt-dlp` on `$PATH`. 37 37 + pub(crate) struct YtdlpSource; 38 38 + 39 39 + impl YtdlpSource { 40 40 + pub(crate) fn new() -> Self { 41 41 + Self 42 42 + } 43 43 + } 44 44 + 45 45 + #[async_trait::async_trait] 46 46 + impl MediaSource for YtdlpSource { 47 47 + async fn extract(&self, url: &str, cache_dir: &Path) -> Result<TrackMeta, SourceError> { 48 48 + if !url.starts_with("http://") && !url.starts_with("https://") { 49 49 + return Err(SourceError::Unsupported(format!("not an HTTP(S) URL: {url}"))); 50 50 + } 51 51 + 52 52 + let output = Command::new("yt-dlp") 53 53 + .args([ 54 54 + "--dump-json", 55 55 + "--no-playlist", 56 56 + "--no-warnings", 57 57 + "--cache-dir", 58 58 + ]) 59 59 + .arg(cache_dir) 60 60 + .arg(url) 61 61 + .kill_on_drop(true) 62 62 + .output() 63 63 + .await 64 64 + .map_err(|e| { 65 65 + if e.kind() == std::io::ErrorKind::NotFound { 66 66 + SourceError::Unsupported("yt-dlp not found on PATH".into()) 67 67 + } else { 68 68 + SourceError::Io(e) 69 69 + } 70 70 + })?; 71 71 + 72 72 + if !output.status.success() { 73 73 + let stderr = String::from_utf8_lossy(&output.stderr); 74 74 + if stderr.contains("Unsupported URL") { 75 75 + return Err(SourceError::Unsupported(stderr.trim().to_string())); 76 76 + } 77 77 + return Err(SourceError::Extraction(stderr.trim().to_string())); 78 78 + } 79 79 + 80 80 + let raw: YtdlpMetadata = serde_json::from_slice(&output.stdout) 81 81 + .map_err(|e| SourceError::Extraction(format!("invalid yt-dlp JSON: {e}")))?; 82 82 + 83 83 + let duration = raw 84 84 + .duration 85 85 + .map(media::format_duration) 86 86 + .unwrap_or_else(|| "--:--".into()); 87 87 + 88 88 + Ok(TrackMeta { 89 89 + title: raw.title, 90 90 + duration, 91 91 + thumbnail: raw.thumbnail, 92 92 + url: raw.webpage_url, 93 93 + source: crate::types::SourceKind::Ytdlp, 94 94 + }) 95 95 + } 96 96 + 97 97 + async fn resolve(&self, url: &str, cache_dir: &Path) -> Result<String, SourceError> { 98 98 + if !url.starts_with("http://") && !url.starts_with("https://") { 99 99 + return Err(SourceError::Unsupported(format!("not an HTTP(S) URL: {url}"))); 100 100 + } 101 101 + 102 102 + let output = Command::new("yt-dlp") 103 103 + .args([ 104 104 + "-f", 105 105 + "b", 106 106 + "-g", 107 107 + "--no-playlist", 108 108 + "--no-warnings", 109 109 + "--cache-dir", 110 110 + ]) 111 111 + .arg(cache_dir) 112 112 + .arg(url) 113 113 + .kill_on_drop(true) 114 114 + .output() 115 115 + .await 116 116 + .map_err(|e| { 117 117 + if e.kind() == std::io::ErrorKind::NotFound { 118 118 + SourceError::Unsupported("yt-dlp not found on PATH".into()) 119 119 + } else { 120 120 + SourceError::Io(e) 121 121 + } 122 122 + })?; 123 123 + 124 124 + if !output.status.success() { 125 125 + let stderr = String::from_utf8_lossy(&output.stderr); 126 126 + if stderr.contains("Unsupported URL") { 127 127 + return Err(SourceError::Unsupported(stderr.trim().to_string())); 128 128 + } 129 129 + return Err(SourceError::Resolution(stderr.trim().to_string())); 130 130 + } 131 131 + 132 132 + let stream_url = String::from_utf8(output.stdout) 133 133 + .map_err(|_| SourceError::Resolution("invalid UTF-8 from yt-dlp".into()))? 134 134 + .trim() 135 135 + .to_string(); 136 136 + 137 137 + if stream_url.is_empty() { 138 138 + return Err(SourceError::Resolution( 139 139 + "yt-dlp returned empty stream URL".into(), 140 140 + )); 141 141 + } 142 142 + 143 143 + Ok(stream_url) 144 144 + } 145 145 + } 146 146 + 147 147 + // --------------------------------------------------------------------------- 148 148 + // Tests 149 149 + // --------------------------------------------------------------------------- 150 150 + 151 151 + #[cfg(test)] 152 152 + mod tests { 153 153 + use super::*; 154 154 + use std::path::Path; 155 155 + 156 156 + #[tokio::test] 157 157 + async fn test_extract_rejects_non_http() { 158 158 + let source = YtdlpSource::new(); 159 159 + let result = source 160 160 + .extract("javascript:alert(1)", Path::new("/tmp")) 161 161 + .await; 162 162 + assert!( 163 163 + matches!(result, Err(SourceError::Unsupported(_))), 164 164 + "non-HTTP URLs should be rejected as unsupported" 165 165 + ); 166 166 + } 167 167 + 168 168 + #[tokio::test] 169 169 + async fn test_extract_rejects_empty_url() { 170 170 + let source = YtdlpSource::new(); 171 171 + let result = source.extract("", Path::new("/tmp")).await; 172 172 + assert!( 173 173 + matches!(result, Err(SourceError::Unsupported(_))), 174 174 + "empty URLs should be rejected as unsupported" 175 175 + ); 176 176 + } 177 177 + 178 178 + #[tokio::test] 179 179 + async fn test_resolve_rejects_non_http() { 180 180 + let source = YtdlpSource::new(); 181 181 + let result = source 182 182 + .resolve("ftp://example.com/video.mp4", Path::new("/tmp")) 183 183 + .await; 184 184 + assert!( 185 185 + matches!(result, Err(SourceError::Unsupported(_))), 186 186 + "non-HTTP URLs should be rejected as unsupported" 187 187 + ); 188 188 + } 189 189 + 190 190 + #[tokio::test] 191 191 + async fn test_resolve_rejects_empty_url() { 192 192 + let source = YtdlpSource::new(); 193 193 + let result = source.resolve("", Path::new("/tmp")).await; 194 194 + assert!( 195 195 + matches!(result, Err(SourceError::Unsupported(_))), 196 196 + "empty URLs should be rejected as unsupported" 197 197 + ); 198 198 + } 199 199 + }