+1 -1

src/main.rs

Reviewed

··· 81 81 } 82 82 }); 83 83 84 84 - let app = web::router(rooms, pool, args.cache_dir, sources); 84 84 + let app = web::router(rooms, pool); 85 85 let listener = tokio::net::TcpListener::bind(args.http_addr).await?; 86 86 87 87 tracing::info!("HTTP server listening on {}", args.http_addr);

+23

src/playlist.rs

Reviewed

··· 165 165 .await?; 166 166 Ok(()) 167 167 } 168 168 + 169 169 + /// Update metadata fields after async extraction completes. 170 170 + /// 171 171 + /// Updates title, duration, thumbnail, and source for the given item. 172 172 + /// Used when yt-dlp extraction finishes after the item was already 173 173 + /// queued with placeholder values. 174 174 + pub(crate) async fn update_metadata( 175 175 + &self, 176 176 + id: i64, 177 177 + meta: &TrackMeta, 178 178 + ) -> Result<(), sqlx::Error> { 179 179 + sqlx::query( 180 180 + "UPDATE queue_items SET title = ?, duration = ?, thumbnail = ?, source = ? WHERE id = ?", 181 181 + ) 182 182 + .bind(&meta.title) 183 183 + .bind(&meta.duration) 184 184 + .bind(&meta.thumbnail) 185 185 + .bind(meta.source.to_string()) 186 186 + .bind(id) 187 187 + .execute(&self.pool) 188 188 + .await?; 189 189 + Ok(()) 190 190 + } 168 191 } 169 192 170 193 #[cfg(test)]

+147 -1

src/room.rs

Reviewed

··· 122 122 room_name: String, 123 123 publishers: TrackPublishers, 124 124 pool: Pool<Sqlite>, 125 125 + source_registry: Arc<SourceRegistry>, 126 126 + cache_dir: PathBuf, 127 127 + /// Clone of the command sender — used by spawned extraction tasks to 128 128 + /// deliver MetadataReady / MetadataFailed back to the actor. 129 129 + cmd_tx: mpsc::Sender<RoomCommand>, 125 130 126 131 // Mutable state (owned by this actor, no locks needed) 127 132 state: PlaybackState, ··· 184 189 room_name: room_name.clone(), 185 190 publishers: publishers.clone(), 186 191 pool: self.pool.clone(), 192 192 + source_registry: self.sources.clone(), 193 193 + cache_dir: self.cache_dir.clone(), 194 194 + cmd_tx: tx.clone(), 187 195 state: PlaybackState::default(), 188 196 player, 189 197 client_count: 0, ··· 240 248 } 241 249 242 250 /// Push a track onto a room's queue (persisted to SQLite via Playlist). 251 251 + #[allow(dead_code)] // only used by tests 243 252 pub(crate) async fn push(&self, id: &RoomId, meta: TrackMeta) { 244 253 if let Some(handle) = self.handle(id).await { 245 254 let _ = handle.cmd_tx.send(RoomCommand::QueueTrack(meta)).await; 246 255 } 256 256 + } 257 257 + 258 258 + /// Push a raw URL onto a room's queue. Metadata extraction (yt-dlp) runs 259 259 + /// asynchronously in a spawned task — the caller returns immediately. 260 260 + pub(crate) async fn push_url(&self, id: &RoomId, url: String) { 261 261 + let t0 = std::time::Instant::now(); 262 262 + let handle = match self.handle(id).await { 263 263 + Some(h) => h, 264 264 + None => return, 265 265 + }; 266 266 + let t1 = t0.elapsed(); 267 267 + let _ = handle.cmd_tx.send(RoomCommand::QueueUrl(url)).await; 268 268 + let t2 = t0.elapsed(); 269 269 + tracing::info!( 270 270 + room = %id, 271 271 + "push_url timing: handle={:.0?} send={:.0?} total={:.0?}", 272 272 + t1, t2 - t1, t2 273 273 + ); 247 274 } 248 275 249 276 /// Remove a room (from SQLite and in-memory state). ··· 436 463 url: item.url, 437 464 duration: item.duration, 438 465 thumbnail: item.thumbnail, 466 466 + pending: false, 439 467 }), 440 468 now, 441 469 ); ··· 444 472 self.execute_effects(effects).await; 445 473 } 446 474 475 475 + RoomCommand::QueueUrl(url) => { 476 476 + let t0 = std::time::Instant::now(); 477 477 + tracing::debug!(room = %self.room_id, %url, "track url queued"); 478 478 + 479 479 + // ── Gather ────────────────────────────────────────── 480 480 + let playlist = Playlist::new(self.pool.clone(), &self.room_id.0); 481 481 + let placeholder = TrackMeta { 482 482 + title: "Loading…".into(), 483 483 + duration: "--:--".into(), 484 484 + thumbnail: None, 485 485 + url: url.clone(), 486 486 + source: crate::types::SourceKind::Direct, 487 487 + }; 488 488 + let item = match playlist.push(&placeholder).await { 489 489 + Ok(item) => item, 490 490 + Err(e) => { 491 491 + tracing::warn!("failed to persist queue item: {e}"); 492 492 + return false; 493 493 + } 494 494 + }; 495 495 + let t1 = t0.elapsed(); 496 496 + let now = Utc::now().timestamp_millis(); 497 497 + 498 498 + // ── Transition (pure) ─────────────────────────────── 499 499 + let effects = self.state.transition( 500 500 + &Event::TrackQueued(state::QueuedTrack { 501 501 + id: item.id, 502 502 + title: placeholder.title, 503 503 + url: placeholder.url, 504 504 + duration: placeholder.duration, 505 505 + thumbnail: placeholder.thumbnail, 506 506 + pending: true, 507 507 + }), 508 508 + now, 509 509 + ); 510 510 + let t2 = t0.elapsed(); 511 511 + 512 512 + // ── Commit ────────────────────────────────────────── 513 513 + self.execute_effects(effects).await; 514 514 + let t3 = t0.elapsed(); 515 515 + 516 516 + // ── Spawn async extraction (non-blocking) ─────────── 517 517 + let item_id = item.id; 518 518 + let cmd_tx = self.cmd_tx.clone(); 519 519 + let sources = self.source_registry.clone(); 520 520 + let cache_dir = self.cache_dir.clone(); 521 521 + tokio::spawn(async move { 522 522 + match sources.extract(&url, &cache_dir).await { 523 523 + Ok(meta) => { 524 524 + let _ = cmd_tx 525 525 + .send(RoomCommand::MetadataReady { item_id, meta }) 526 526 + .await; 527 527 + } 528 528 + Err(e) => { 529 529 + let _ = cmd_tx 530 530 + .send(RoomCommand::MetadataFailed { 531 531 + item_id, 532 532 + error: e.to_string(), 533 533 + }) 534 534 + .await; 535 535 + } 536 536 + } 537 537 + }); 538 538 + 539 539 + tracing::info!( 540 540 + room = %self.room_id, 541 541 + "QueueUrl actor timing: db_push={:.0?} transition={:.0?} effects={:.0?} total={:.0?}", 542 542 + t1, t2 - t1, t3 - t2, t3 543 543 + ); 544 544 + } 545 545 + 546 546 + RoomCommand::MetadataReady { item_id, meta } => { 547 547 + tracing::debug!(room = %self.room_id, item_id, title = %meta.title, "metadata ready"); 548 548 + 549 549 + // ── Gather ────────────────────────────────────────── 550 550 + let now = Utc::now().timestamp_millis(); 551 551 + 552 552 + // ── DB update (synchronous commit, actor-safe) ────── 553 553 + let playlist = Playlist::new(self.pool.clone(), &self.room_id.0); 554 554 + if let Err(e) = playlist.update_metadata(item_id, &meta).await { 555 555 + tracing::warn!(item_id, error = %e, "failed to persist metadata update"); 556 556 + } 557 557 + 558 558 + // ── Transition (pure) ─────────────────────────────── 559 559 + let effects = self.state.transition( 560 560 + &Event::MetadataUpdated { 561 561 + item_id, 562 562 + title: meta.title, 563 563 + duration: meta.duration, 564 564 + thumbnail: meta.thumbnail, 565 565 + }, 566 566 + now, 567 567 + ); 568 568 + 569 569 + // ── Commit ────────────────────────────────────────── 570 570 + self.execute_effects(effects).await; 571 571 + } 572 572 + 573 573 + RoomCommand::MetadataFailed { item_id, error } => { 574 574 + tracing::debug!(room = %self.room_id, item_id, %error, "metadata extraction failed"); 575 575 + 576 576 + // ── Transition (pure) — set error state in title ──── 577 577 + let now = Utc::now().timestamp_millis(); 578 578 + let effects = self.state.transition( 579 579 + &Event::MetadataUpdated { 580 580 + item_id, 581 581 + title: format!("Failed to load — {error}"), 582 582 + duration: "--:--".into(), 583 583 + thumbnail: None, 584 584 + }, 585 585 + now, 586 586 + ); 587 587 + self.execute_effects(effects).await; 588 588 + } 589 589 + 447 590 RoomCommand::Skip => { 448 591 tracing::info!(room = %self.room_id, "skip requested"); 449 592 ··· 534 677 } 535 678 536 679 Effect::PersistStarted(id) => { 680 680 + let t0 = std::time::Instant::now(); 537 681 let now = Utc::now().format("%Y-%m-%dT%H:%M:%S%.3fZ").to_string(); 538 682 let playlist = Playlist::new(self.pool.clone(), &self.room_id.0); 539 683 let _ = playlist.mark_started(id, &now).await; 684 684 + tracing::debug!(room = %self.room_id, id, "PersistStarted: {:.0?}", t0.elapsed()); 540 685 } 541 686 542 687 Effect::StartPipeline(track) => { ··· 587 732 url: q.url.clone(), 588 733 duration: q.duration.clone(), 589 734 thumbnail: q.thumbnail.clone(), 735 735 + pending: q.pending, 590 736 }) 591 737 .collect(); 592 738 ··· 608 754 609 755 tracing::trace!( 610 756 room = %self.room_id, queue_len = %queue.len(), history_len = %history.len(), 611 611 - clients = %client_count, "state snapshot published" 757 757 + clients = %client_count, "publish_state_snapshot", 612 758 ); 613 759 614 760 let snapshot = serde_json::json!({

+50 -41

src/source/ytdlp.rs

Reviewed

··· 7 7 use std::path::Path; 8 8 9 9 use serde::Deserialize; 10 10 - use tokio::process::Command; 11 10 12 11 use crate::media; 13 12 use crate::source::{MediaSource, SourceError}; ··· 49 48 return Err(SourceError::Unsupported(format!("not an HTTP(S) URL: {url}"))); 50 49 } 51 50 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 - })?; 51 51 + let cache_dir = cache_dir.to_path_buf(); 52 52 + let url = url.to_string(); 53 53 + let result = tokio::task::spawn_blocking(move || { 54 54 + std::process::Command::new("yt-dlp") 55 55 + .args([ 56 56 + "--dump-json", 57 57 + "--no-playlist", 58 58 + "--no-warnings", 59 59 + "--cache-dir", 60 60 + ]) 61 61 + .arg(&cache_dir) 62 62 + .arg(&url) 63 63 + .output() 64 64 + }) 65 65 + .await 66 66 + .map_err(|e| SourceError::Io(std::io::Error::new(std::io::ErrorKind::Other, e)))?; 67 67 + 68 68 + let output = result.map_err(|e| { 69 69 + if e.kind() == std::io::ErrorKind::NotFound { 70 70 + SourceError::Unsupported("yt-dlp not found on PATH".into()) 71 71 + } else { 72 72 + SourceError::Io(e) 73 73 + } 74 74 + })?; 71 75 72 76 if !output.status.success() { 73 77 let stderr = String::from_utf8_lossy(&output.stderr); ··· 99 103 return Err(SourceError::Unsupported(format!("not an HTTP(S) URL: {url}"))); 100 104 } 101 105 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 - })?; 106 106 + let cache_dir = cache_dir.to_path_buf(); 107 107 + let url = url.to_string(); 108 108 + let result = tokio::task::spawn_blocking(move || { 109 109 + std::process::Command::new("yt-dlp") 110 110 + .args([ 111 111 + "-f", 112 112 + "b", 113 113 + "-g", 114 114 + "--no-playlist", 115 115 + "--no-warnings", 116 116 + "--cache-dir", 117 117 + ]) 118 118 + .arg(&cache_dir) 119 119 + .arg(&url) 120 120 + .output() 121 121 + }) 122 122 + .await 123 123 + .map_err(|e| SourceError::Io(std::io::Error::new(std::io::ErrorKind::Other, e)))?; 124 124 + 125 125 + let output = result.map_err(|e| { 126 126 + if e.kind() == std::io::ErrorKind::NotFound { 127 127 + SourceError::Unsupported("yt-dlp not found on PATH".into()) 128 128 + } else { 129 129 + SourceError::Io(e) 130 130 + } 131 131 + })?; 123 132 124 133 if !output.status.success() { 125 134 let stderr = String::from_utf8_lossy(&output.stderr);

+232 -41

src/state.rs

Reviewed

··· 27 27 pub url: String, 28 28 pub duration: String, 29 29 pub thumbnail: Option<String>, 30 30 + /// True while metadata is being extracted asynchronously. 31 31 + pub pending: bool, 30 32 } 31 33 32 34 /// A finished track in history. ··· 49 51 Skip, 50 52 /// The transcoding pipeline finished (naturally or via abort). 51 53 TrackEnded { item_id: i64 }, 54 54 + /// Asynchronous metadata extraction completed for a queued/playing track. 55 55 + MetadataUpdated { 56 56 + item_id: i64, 57 57 + title: String, 58 58 + duration: String, 59 59 + thumbnail: Option<String>, 60 60 + }, 52 61 } 53 62 54 63 /// Side effects to execute after a transition. ··· 102 111 Event::TrackQueued(track) => self.handle_queued(track), 103 112 Event::Skip => self.handle_skip(now), 104 113 Event::TrackEnded { item_id } => self.handle_track_ended(*item_id, now), 114 114 + Event::MetadataUpdated { 115 115 + item_id, 116 116 + title, 117 117 + duration, 118 118 + thumbnail, 119 119 + } => self.handle_metadata_updated(*item_id, title, duration, thumbnail), 105 120 } 106 121 } 107 122 ··· 193 208 effects 194 209 } 195 210 211 211 + /// Asynchronous metadata resolution completed. Updates the matching item 212 212 + /// in the queue and/or the active track in-place. Does not change playback 213 213 + /// state — only updates display fields. 214 214 + fn handle_metadata_updated( 215 215 + &mut self, 216 216 + item_id: i64, 217 217 + title: &str, 218 218 + duration: &str, 219 219 + thumbnail: &Option<String>, 220 220 + ) -> Vec<Effect> { 221 221 + if let Some(item) = self.queue.iter_mut().find(|t| t.id == item_id) { 222 222 + item.title = title.to_string(); 223 223 + item.duration = duration.to_string(); 224 224 + item.thumbnail.clone_from(thumbnail); 225 225 + item.pending = false; 226 226 + } 227 227 + if let Some(ref mut active) = self.active { 228 228 + if active.id == item_id { 229 229 + active.title = title.to_string(); 230 230 + active.duration = duration.to_string(); 231 231 + active.thumbnail.clone_from(thumbnail); 232 232 + } 233 233 + } 234 234 + vec![Effect::PublishSnapshot] 235 235 + } 236 236 + 196 237 /// Pop the first track from the queue and start playing it. 197 238 /// 198 239 /// Returns effects for starting the pipeline and publishing state. ··· 246 287 url: format!("https://example.com/{title}"), 247 288 duration: "3:45".into(), 248 289 thumbnail: None, 290 290 + pending: false, 249 291 } 250 292 } 251 293 294 294 + // ── Effect ordering tests ───────────────────────────────────────────── 295 295 + // These verify the exact effect sequence returned by transition(). Order 296 296 + // matters: PersistStarted must precede StartPipeline (DB commit before 297 297 + // pipeline spawn), AbortPipeline must precede PersistFinished, etc. 298 298 + 299 299 + #[test] 300 300 + fn idle_to_playing_effect_order() { 301 301 + let mut s = PlaybackState::default(); 302 302 + let effects = s.transition(&Event::TrackQueued(queued("A", 1)), 0); 303 303 + 304 304 + assert_eq!( 305 305 + effects, 306 306 + vec![ 307 307 + Effect::PersistStarted(1), 308 308 + Effect::StartPipeline(queued("A", 1)), 309 309 + Effect::PublishSnapshot, 310 310 + ], 311 311 + "idle → first track: persist started_at, spawn pipeline, notify clients" 312 312 + ); 313 313 + } 314 314 + 315 315 + #[test] 316 316 + fn queue_second_while_playing_effect_order() { 317 317 + let mut s = PlaybackState::default(); 318 318 + s.transition(&Event::TrackQueued(queued("A", 1)), 0); 319 319 + let effects = s.transition(&Event::TrackQueued(queued("B", 2)), 0); 320 320 + 321 321 + assert_eq!( 322 322 + effects, 323 323 + vec![Effect::PublishSnapshot], 324 324 + "playing → queue another: just notify — pipeline untouched" 325 325 + ); 326 326 + } 327 327 + 328 328 + #[test] 329 329 + fn skip_with_next_effect_order() { 330 330 + let mut s = PlaybackState::default(); 331 331 + s.transition(&Event::TrackQueued(queued("A", 1)), 0); 332 332 + s.transition(&Event::TrackQueued(queued("B", 2)), 0); 333 333 + let effects = s.transition(&Event::Skip, 1_700_000_000_001); 334 334 + 335 335 + assert_eq!( 336 336 + &effects[..3], 337 337 + &[ 338 338 + Effect::AbortPipeline, 339 339 + Effect::PersistFinished(1), 340 340 + Effect::PersistStarted(2), 341 341 + ], 342 342 + "skip with next: abort current, finalize A, persist B started_at" 343 343 + ); 344 344 + assert!( 345 345 + effects 346 346 + .iter() 347 347 + .any(|e| matches!(e, Effect::StartPipeline(t) if t.title == "B")), 348 348 + "skip with next: must include StartPipeline for B" 349 349 + ); 350 350 + assert!( 351 351 + effects.contains(&Effect::PublishSnapshot), 352 352 + "skip with next: must notify clients" 353 353 + ); 354 354 + } 355 355 + 356 356 + #[test] 357 357 + fn skip_last_track_effect_order() { 358 358 + let mut s = PlaybackState::default(); 359 359 + s.transition(&Event::TrackQueued(queued("A", 1)), 0); 360 360 + let effects = s.transition(&Event::Skip, 0); 361 361 + 362 362 + assert_eq!( 363 363 + effects, 364 364 + vec![ 365 365 + Effect::AbortPipeline, 366 366 + Effect::PersistFinished(1), 367 367 + Effect::PublishSnapshot, 368 368 + ], 369 369 + "skip last: abort, finalize, notify — no StartPipeline" 370 370 + ); 371 371 + } 372 372 + 373 373 + #[test] 374 374 + fn track_ended_with_next_effect_order() { 375 375 + let mut s = PlaybackState::default(); 376 376 + s.transition(&Event::TrackQueued(queued("A", 1)), 0); 377 377 + s.transition(&Event::TrackQueued(queued("B", 2)), 0); 378 378 + let effects = s.transition(&Event::TrackEnded { item_id: 1 }, 0); 379 379 + 380 380 + assert_eq!( 381 381 + &effects[..2], 382 382 + &[Effect::PersistFinished(1), Effect::PersistStarted(2),], 383 383 + "track ended (next): finalize A, persist B started_at — no AbortPipeline" 384 384 + ); 385 385 + assert!( 386 386 + effects 387 387 + .iter() 388 388 + .any(|e| matches!(e, Effect::StartPipeline(t) if t.title == "B")), 389 389 + "track ended (next): must start B" 390 390 + ); 391 391 + assert!( 392 392 + effects.contains(&Effect::PublishSnapshot), 393 393 + "track ended (next): must notify" 394 394 + ); 395 395 + } 396 396 + 397 397 + #[test] 398 398 + fn track_ended_last_goes_idle_effect_order() { 399 399 + let mut s = PlaybackState::default(); 400 400 + s.transition(&Event::TrackQueued(queued("A", 1)), 0); 401 401 + let effects = s.transition(&Event::TrackEnded { item_id: 1 }, 0); 402 402 + 403 403 + assert_eq!( 404 404 + effects, 405 405 + vec![Effect::PersistFinished(1), Effect::PublishSnapshot,], 406 406 + "track ended last: finalize, notify — no StartPipeline, no AbortPipeline" 407 407 + ); 408 408 + } 409 409 + 410 410 + // ── State invariant tests ───────────────────────────────────────────── 411 411 + 252 412 #[test] 253 413 fn idle_room_starts_playing_on_first_track() { 254 414 let mut s = PlaybackState::default(); 255 255 - let effects = s.transition(&Event::TrackQueued(queued("A", 1)), 0); 415 415 + s.transition(&Event::TrackQueued(queued("A", 1)), 0); 256 416 257 417 assert!(s.active.is_some()); 258 418 assert_eq!(s.active.as_ref().unwrap().title, "A"); 259 419 assert!(s.queue.is_empty()); 260 260 - assert!(effects.contains(&Effect::PublishSnapshot)); 261 261 - assert!(effects 262 262 - .iter() 263 263 - .any(|e| matches!(e, Effect::StartPipeline(_)))); 264 420 } 265 421 266 422 #[test] 267 423 fn second_track_stays_in_queue() { 268 424 let mut s = PlaybackState::default(); 269 425 s.transition(&Event::TrackQueued(queued("A", 1)), 0); 270 270 - let effects = s.transition(&Event::TrackQueued(queued("B", 2)), 0); 426 426 + s.transition(&Event::TrackQueued(queued("B", 2)), 0); 271 427 272 428 assert_eq!(s.queue.len(), 1); 273 429 assert_eq!(s.queue[0].title, "B"); 274 430 assert_eq!(s.active.as_ref().unwrap().title, "A"); 275 275 - assert_eq!(effects, vec![Effect::PublishSnapshot]); 276 431 } 277 432 278 433 #[test] ··· 280 435 let mut s = PlaybackState::default(); 281 436 s.transition(&Event::TrackQueued(queued("A", 1)), 0); 282 437 s.transition(&Event::TrackQueued(queued("B", 2)), 0); 283 283 - let effects = s.transition(&Event::Skip, 1_700_000_000_001); 438 438 + s.transition(&Event::Skip, 1_700_000_000_001); 284 439 285 285 - // Should have advanced to B. 286 440 assert_eq!(s.active.as_ref().unwrap().title, "B"); 287 441 assert!(s.queue.is_empty()); 288 288 - 289 289 - // A should be in history. 290 442 assert_eq!(s.history.len(), 1); 291 443 assert_eq!(s.history[0].title, "A"); 292 292 - 293 293 - // Effects: abort prev, persist finished, start B, publish snapshot. 294 294 - assert!(effects.contains(&Effect::AbortPipeline)); 295 295 - assert!(effects.contains(&Effect::PersistFinished(1))); 296 296 - assert!(effects 297 297 - .iter() 298 298 - .any(|e| matches!(e, Effect::StartPipeline(t) if t.title == "B"))); 299 299 - assert!(effects.contains(&Effect::PublishSnapshot)); 300 444 } 301 445 302 446 #[test] ··· 312 456 fn skip_last_track_goes_idle() { 313 457 let mut s = PlaybackState::default(); 314 458 s.transition(&Event::TrackQueued(queued("A", 1)), 0); 315 315 - let effects = s.transition(&Event::Skip, 0); 459 459 + s.transition(&Event::Skip, 0); 316 460 317 461 assert!(s.active.is_none()); 318 462 assert_eq!(s.history.len(), 1); 319 463 assert_eq!(s.history[0].title, "A"); 320 320 - assert!(effects.contains(&Effect::AbortPipeline)); 321 321 - assert!(effects.contains(&Effect::PersistFinished(1))); 322 322 - assert!(effects.contains(&Effect::PublishSnapshot)); 323 323 - // No StartPipeline since queue is empty. 324 324 - assert!(!effects 325 325 - .iter() 326 326 - .any(|e| matches!(e, Effect::StartPipeline(_)))); 327 464 } 328 465 329 466 #[test] ··· 331 468 let mut s = PlaybackState::default(); 332 469 s.transition(&Event::TrackQueued(queued("A", 1)), 0); 333 470 s.transition(&Event::TrackQueued(queued("B", 2)), 0); 334 334 - 335 335 - let effects = s.transition(&Event::TrackEnded { item_id: 1 }, 0); 471 471 + s.transition(&Event::TrackEnded { item_id: 1 }, 0); 336 472 337 473 assert_eq!(s.active.as_ref().unwrap().title, "B"); 338 474 assert_eq!(s.history.len(), 1); 339 475 assert_eq!(s.history[0].title, "A"); 340 340 - assert!(effects.contains(&Effect::PersistFinished(1))); 341 341 - assert!(effects.contains(&Effect::PublishSnapshot)); 342 476 } 343 477 344 478 #[test] ··· 346 480 let mut s = PlaybackState::default(); 347 481 s.transition(&Event::TrackQueued(queued("A", 1)), 0); 348 482 349 349 - // Stale TrackEnded for a different id. 350 483 let effects = s.transition(&Event::TrackEnded { item_id: 999 }, 0); 351 484 352 485 assert!(s.active.is_some()); ··· 358 491 fn track_ended_last_track_goes_idle() { 359 492 let mut s = PlaybackState::default(); 360 493 s.transition(&Event::TrackQueued(queued("A", 1)), 0); 361 361 - 362 362 - let effects = s.transition(&Event::TrackEnded { item_id: 1 }, 0); 494 494 + s.transition(&Event::TrackEnded { item_id: 1 }, 0); 363 495 364 496 assert!(s.active.is_none()); 365 497 assert_eq!(s.history.len(), 1); 366 366 - assert!(effects.contains(&Effect::PublishSnapshot)); 367 367 - assert!(!effects 368 368 - .iter() 369 369 - .any(|e| matches!(e, Effect::StartPipeline(_)))); 370 498 } 371 499 372 500 #[test] ··· 383 511 let mut s = PlaybackState::default(); 384 512 s.transition(&Event::TrackQueued(queued("A", 1)), 0); 385 513 s.transition(&Event::TrackQueued(queued("B", 2)), 0); 386 386 - 387 387 - // Skip A → B starts playing. 388 514 s.transition(&Event::Skip, 0); 389 515 assert_eq!(s.active.as_ref().unwrap().title, "B"); 390 516 391 391 - // Stale TrackEnded for A arrives — should be ignored. 392 517 let effects = s.transition(&Event::TrackEnded { item_id: 1 }, 0); 393 518 assert!(effects.is_empty()); 394 519 assert_eq!(s.active.as_ref().unwrap().title, "B"); ··· 399 524 let mut s = PlaybackState::default(); 400 525 s.resolve_started_at(42); 401 526 assert!(s.active.is_none()); 527 527 + } 528 528 + 529 529 + // ── Metadata update tests ───────────────────────────────────────── 530 530 + 531 531 + #[test] 532 532 + fn metadata_update_updates_queue_item() { 533 533 + let mut s = PlaybackState::default(); 534 534 + s.transition(&Event::TrackQueued(queued("A", 1)), 0); 535 535 + s.transition( 536 536 + &Event::MetadataUpdated { 537 537 + item_id: 1, 538 538 + title: "Track A (Real)".into(), 539 539 + duration: "4:20".into(), 540 540 + thumbnail: Some("https://img.example/a.jpg".into()), 541 541 + }, 542 542 + 0, 543 543 + ); 544 544 + 545 545 + // Queue should have empty (track was advanced to active), but active 546 546 + // should have updated metadata. 547 547 + assert_eq!(s.active.as_ref().unwrap().title, "Track A (Real)"); 548 548 + assert_eq!(s.active.as_ref().unwrap().duration, "4:20"); 549 549 + assert_eq!( 550 550 + s.active.as_ref().unwrap().thumbnail, 551 551 + Some("https://img.example/a.jpg".into()) 552 552 + ); 553 553 + } 554 554 + 555 555 + #[test] 556 556 + fn metadata_update_on_queued_item() { 557 557 + let mut s = PlaybackState::default(); 558 558 + s.transition(&Event::TrackQueued(queued("A", 1)), 0); 559 559 + s.transition(&Event::TrackQueued(queued("B", 2)), 0); 560 560 + 561 561 + // Update the queued item (B, id=2). 562 562 + let effects = s.transition( 563 563 + &Event::MetadataUpdated { 564 564 + item_id: 2, 565 565 + title: "Track B (Real)".into(), 566 566 + duration: "5:55".into(), 567 567 + thumbnail: None, 568 568 + }, 569 569 + 0, 570 570 + ); 571 571 + assert_eq!(effects, vec![Effect::PublishSnapshot]); 572 572 + assert_eq!(s.queue[0].title, "Track B (Real)"); 573 573 + assert!(!s.queue[0].pending); 574 574 + } 575 575 + 576 576 + #[test] 577 577 + fn metadata_update_on_unknown_id_is_noop() { 578 578 + let mut s = PlaybackState::default(); 579 579 + let effects = s.transition( 580 580 + &Event::MetadataUpdated { 581 581 + item_id: 999, 582 582 + title: "Ghost".into(), 583 583 + duration: "0:00".into(), 584 584 + thumbnail: None, 585 585 + }, 586 586 + 0, 587 587 + ); 588 588 + // Still publishes snapshot even if nothing changed — clients may want 589 589 + // to know the update was processed. 590 590 + assert_eq!(effects, vec![Effect::PublishSnapshot]); 591 591 + assert!(s.active.is_none()); 592 592 + assert!(s.queue.is_empty()); 402 593 } 403 594 }

+86 -127

src/transport.rs

Reviewed

··· 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 264 pub(crate) video: broadcast::Sender<MoqObject>, 266 265 pub(crate) chat: broadcast::Sender<MoqObject>, 267 266 pub(crate) state: watch::Sender<StateValue>, ··· 273 272 274 273 impl TrackPublishers { 275 274 pub(crate) fn new() -> Self { 276 276 - let (audio, _) = broadcast::channel(256); 277 275 let (video, _) = broadcast::channel(256); 278 276 let (chat, _) = broadcast::channel(256); 279 277 let (state, _) = watch::channel(StateValue { ··· 282 280 }); 283 281 let (video_init_watch, _) = watch::channel(false); 284 282 TrackPublishers { 285 285 - audio, 286 283 video, 287 284 chat, 288 285 state, ··· 455 452 } 456 453 } 457 454 455 455 + /// WebSocket session for video-only delivery. 456 456 + /// 457 457 + /// No MoQ protocol negotiation — the server auto-subscribes the client to 458 458 + /// the video broadcast. Messages are raw fMP4 bytes (no MoQ framing). 459 459 + /// A separate WS-control connection handles chat, state, and MoQ handshake. 460 460 + pub(crate) async fn handle_video_session( 461 461 + ws: WebSocket, 462 462 + publishers: TrackPublishers, 463 463 + ) { 464 464 + let (mut ws_tx, mut ws_rx) = ws.split(); 465 465 + 466 466 + // Collect the cached init (or wait for the pipeline to produce one) 467 467 + // but don't send it yet — we need to subscribe to the broadcast first 468 468 + // so we can discard any init that was also published via broadcast, 469 469 + // which would otherwise arrive as a duplicate through the forward loop. 470 470 + let init = match publishers.get_init_segment() { 471 471 + Some(cached) => { 472 472 + tracing::debug!("video ws: using cached init segment"); 473 473 + Some(cached) 474 474 + } 475 475 + None => { 476 476 + tracing::debug!("video ws: no cached init, waiting for watch signal"); 477 477 + let mut init_rx = publishers.video_init_waiter(); 478 478 + let _ = init_rx.changed().await; 479 479 + let result = publishers.get_init_segment(); 480 480 + if result.is_some() { 481 481 + tracing::debug!("video ws: received init segment via watch"); 482 482 + } else { 483 483 + tracing::warn!("video ws: watch fired but no init available"); 484 484 + } 485 485 + result 486 486 + } 487 487 + }; 488 488 + 489 489 + // Subscribe to the broadcast before sending cached init so that 490 490 + // any init published via broadcast is caught by the catch-up loop 491 491 + // and discarded, preventing a duplicate on the forward path. 492 492 + let mut video_rx = publishers.video.subscribe(); 493 493 + loop { 494 494 + match video_rx.try_recv() { 495 495 + Ok(_) => continue, 496 496 + Err(broadcast::error::TryRecvError::Empty) => break, 497 497 + Err(broadcast::error::TryRecvError::Closed) => return, 498 498 + Err(broadcast::error::TryRecvError::Lagged(_)) => break, 499 499 + } 500 500 + } 501 501 + 502 502 + // Now send the cached init — the broadcast buffer has been drained 503 503 + // so the forward loop below won't re-deliver it. 504 504 + if let Some((_group_id, payload)) = init { 505 505 + if ws_tx.send(Message::Binary(payload)).await.is_err() { 506 506 + return; 507 507 + } 508 508 + } 509 509 + 510 510 + // Forward live video objects as raw fMP4 payloads. 511 511 + loop { 512 512 + tokio::select! { 513 513 + msg = ws_rx.next() => { 514 514 + match msg { 515 515 + Some(Ok(Message::Close(_))) | None => break, 516 516 + _ => continue, 517 517 + } 518 518 + } 519 519 + result = video_rx.recv() => { 520 520 + match result { 521 521 + Ok(obj) => { 522 522 + if ws_tx.send(Message::Binary(obj.payload)).await.is_err() { 523 523 + break; 524 524 + } 525 525 + } 526 526 + Err(broadcast::error::RecvError::Lagged(n)) => { 527 527 + tracing::debug!("video ws: broadcast lagged by {n}"); 528 528 + } 529 529 + Err(broadcast::error::RecvError::Closed) => break, 530 530 + } 531 531 + } 532 532 + } 533 533 + } 534 534 + } 535 535 + 458 536 async fn handle_moq_message( 459 537 msg: MoqMessage, 460 538 room_id: &RoomId, ··· 573 651 } 574 652 } 575 653 })); 576 576 - } 577 577 - TrackId::Video => { 578 578 - // Send cached init (or wait for it) and forward live broadcast 579 579 - // from the same spawned task to guarantee ordering for late joiners. 580 580 - let init_rx = publishers.video_init_waiter(); 581 581 - let cached = publishers.get_init_segment(); // snapshot before spawn 582 582 - let tx = object_tx.clone(); 583 583 - let publishers_for_task = publishers.clone(); 584 584 - let rx = publishers.video.subscribe(); 585 585 - task_handles.push(tokio::spawn(async move { 586 586 - // 1. Ensure init segment is available (wait if pipeline 587 587 - // hasn't produced it yet). 588 588 - let init = if let Some(cached) = cached { 589 589 - tracing::debug!("forward task: using cached init segment"); 590 590 - Some(cached) 591 591 - } else { 592 592 - tracing::debug!( 593 593 - "forward task: no cached init, waiting for watch signal" 594 594 - ); 595 595 - let mut init_rx = init_rx; 596 596 - let _ = init_rx.changed().await; 597 597 - let result = publishers_for_task.get_init_segment(); 598 598 - if result.is_some() { 599 599 - tracing::debug!("forward task: received init segment via watch"); 600 600 - } else { 601 601 - tracing::warn!("forward task: watch fired but no init available"); 602 602 - } 603 603 - result 604 604 - }; 605 605 - 606 606 - if let Some((group_id, init)) = init { 607 607 - tracing::debug!( 608 608 - group_id, 609 609 - size = init.len(), 610 610 - "forward task: sending init segment" 611 611 - ); 612 612 - let obj = encode_message(&MoqMessage::Object { 613 613 - track_id: TrackId::Video as u64, 614 614 - group_id, 615 615 - object_id: 0, 616 616 - payload: init, 617 617 - }); 618 618 - if tx.send(obj).is_err() { 619 619 - return; 620 620 - } 621 621 - } else { 622 622 - // No init and never got one — nothing to play. 623 623 - tracing::warn!("forward task: no init segment available, aborting"); 624 624 - return; 625 625 - } 626 626 - 627 627 - // 2. Catch up to live edge: discard any messages that were 628 628 - // already in the broadcast buffer before subscription. 629 629 - // Without this, a late subscriber would replay all prior 630 630 - // segments from the oldest in the buffer. 631 631 - let mut rx = rx; 632 632 - loop { 633 633 - match rx.try_recv() { 634 634 - Ok(_) => continue, 635 635 - Err(broadcast::error::TryRecvError::Empty) => break, 636 636 - Err(broadcast::error::TryRecvError::Closed) => return, 637 637 - Err(broadcast::error::TryRecvError::Lagged(_)) => break, 638 638 - } 639 639 - } 640 640 - 641 641 - // 3. Forward live broadcast messages. 642 642 - tracing::debug!("forward task: entering broadcast loop"); 643 643 - loop { 644 644 - match rx.recv().await { 645 645 - Ok(obj) => { 646 646 - let msg = encode_message(&MoqMessage::Object { 647 647 - track_id: obj.track_id as u64, 648 648 - group_id: obj.group_id, 649 649 - object_id: obj.object_id, 650 650 - payload: obj.payload, 651 651 - }); 652 652 - if tx.send(msg).is_err() { 653 653 - break; 654 654 - } 655 655 - } 656 656 - Err(broadcast::error::RecvError::Lagged(n)) => { 657 657 - tracing::debug!("forward task: video broadcast lagged by {n}"); 658 658 - } 659 659 - Err(broadcast::error::RecvError::Closed) => break, 660 660 - } 661 661 - } 662 662 - })); 663 663 - } 664 664 - TrackId::Audio => { 665 665 - let rx = publishers.audio.subscribe(); 666 666 - let tx = object_tx.clone(); 667 667 - task_handles.push(tokio::spawn(async move { 668 668 - let mut rx = rx; 669 669 - // Catch up to live edge — discard prior messages. 670 670 - loop { 671 671 - match rx.try_recv() { 672 672 - Ok(_) => continue, 673 673 - Err(broadcast::error::TryRecvError::Empty) => break, 674 674 - Err(broadcast::error::TryRecvError::Closed) => return, 675 675 - Err(broadcast::error::TryRecvError::Lagged(_)) => break, 676 676 - } 677 677 - } 678 678 - loop { 679 679 - match rx.recv().await { 680 680 - Ok(obj) => { 681 681 - let msg = encode_message(&MoqMessage::Object { 682 682 - track_id: obj.track_id as u64, 683 683 - group_id: obj.group_id, 684 684 - object_id: obj.object_id, 685 685 - payload: obj.payload, 686 686 - }); 687 687 - if tx.send(msg).is_err() { 688 688 - break; 689 689 - } 690 690 - } 691 691 - Err(broadcast::error::RecvError::Lagged(_)) => { 692 692 - // Live media — skip missed objects. 693 693 - continue; 694 694 - } 695 695 - Err(broadcast::error::RecvError::Closed) => break, 696 696 - } 697 697 - } 698 698 - })); 699 699 - } 654 654 + } 655 655 + _ => { 656 656 + // Video and audio use a dedicated WebSocket endpoint. 657 657 + // Ignore subscribe requests on the control WS. 700 658 } 701 659 } 660 660 + } 702 661 703 662 MoqMessage::Object { 704 663 track_id,

+21

src/types.rs

Reviewed

··· 79 79 pub(crate) url: String, 80 80 pub(crate) duration: String, 81 81 pub(crate) thumbnail: Option<String>, 82 82 + /// True while metadata is being extracted asynchronously. 83 83 + #[serde(skip_serializing_if = "is_false")] 84 84 + pub(crate) pending: bool, 85 85 + } 86 86 + 87 87 + fn is_false(b: &bool) -> bool { 88 88 + !*b 82 89 } 83 90 84 91 #[derive(Debug, Clone, Serialize)] ··· 140 147 141 148 /// All room operations — processed sequentially by the per-room actor. 142 149 pub(crate) enum RoomCommand { 150 150 + /// Track with already-resolved metadata (legacy path, used by tests). 151 151 + #[allow(dead_code)] 143 152 QueueTrack(TrackMeta), 153 153 + /// Raw URL — metadata extraction happens asynchronously in a spawned task. 154 154 + QueueUrl(String), 155 155 + /// Async metadata extraction result. 156 156 + MetadataReady { 157 157 + item_id: i64, 158 158 + meta: TrackMeta, 159 159 + }, 160 160 + /// Async metadata extraction failed. 161 161 + MetadataFailed { 162 162 + item_id: i64, 163 163 + error: String, 164 164 + }, 144 165 Skip, 145 166 /// Sent by the pipeline task when it finishes (or errors, or is aborted). 146 167 TrackEnded {

+44 -29

src/web.rs

Reviewed

··· 2 2 3 3 use std::collections::HashMap; 4 4 use std::net::SocketAddr; 5 5 - use std::path::PathBuf; 6 5 use std::sync::Arc; 7 6 8 7 use askama::Template; ··· 18 17 19 18 use crate::db; 20 19 use crate::room; 21 21 - use crate::source::SourceRegistry; 22 20 use crate::types::{RoomId, TrackMeta}; 23 21 24 22 /// Per-IP rate limiter for POST /api/ingest (max 1 req/5s per IP). ··· 58 56 pub(crate) struct AppState { 59 57 pub rooms: room::Registry, 60 58 pub pool: Pool<Sqlite>, 61 61 - pub cache_dir: PathBuf, 62 62 - pub sources: Arc<SourceRegistry>, 63 59 pub rate_limiter: RateLimiter, 64 60 } 65 61 66 62 pub(crate) fn router( 67 63 rooms: room::Registry, 68 64 pool: Pool<Sqlite>, 69 69 - cache_dir: PathBuf, 70 70 - sources: Arc<SourceRegistry>, 71 65 ) -> Router { 72 66 let rate_limiter = RateLimiter::new(); 73 67 ··· 84 78 let state = AppState { 85 79 rooms, 86 80 pool, 87 87 - cache_dir, 88 88 - sources, 89 81 rate_limiter, 90 82 }; 91 83 ··· 96 88 .route("/api/ingest", post(ingest_url)) 97 89 .route("/api/skip", post(skip_handler)) 98 90 .route("/ws/{room_id}", get(ws_handler)) 91 91 + .route("/ws/{room_id}/video", get(video_ws_handler)) 99 92 .with_state(state) 100 93 } 101 94 ··· 188 181 room_id: String, 189 182 } 190 183 191 191 - #[derive(serde::Serialize)] 192 192 - struct IngestResponse { 193 193 - title: String, 194 194 - duration: String, 195 195 - thumbnail: Option<String>, 196 196 - } 197 197 - 198 184 fn err_response(status: StatusCode, msg: &str) -> axum::response::Response { 199 185 (status, Json(serde_json::json!({ "error": msg }))).into_response() 200 186 } ··· 203 189 State(state): State<AppState>, 204 190 ConnectInfo(addr): ConnectInfo<SocketAddr>, 205 191 axum::Json(req): axum::Json<IngestRequest>, 206 206 - ) -> Result<Json<IngestResponse>, axum::response::Response> { 192 192 + ) -> Result<Json<serde_json::Value>, axum::response::Response> { 207 193 // Input validation — URL max 2048 characters. 208 194 if req.url.len() > 2048 { 209 195 return Err(err_response(StatusCode::BAD_REQUEST, "URL too long")); ··· 226 212 227 213 let room_id = RoomId(req.room_id); 228 214 215 215 + let t0 = std::time::Instant::now(); 216 216 + 229 217 // Validate the room exists. 230 218 if !state.rooms.exists(&room_id).await { 231 219 return Err(err_response(StatusCode::NOT_FOUND, "room not found")); 232 220 } 221 221 + let t1 = t0.elapsed(); 233 222 234 234 - // Extract metadata via the source registry. 235 235 - let meta = state 236 236 - .sources 237 237 - .extract(&req.url, &state.cache_dir) 238 238 - .await 239 239 - .map_err(|e| err_response(StatusCode::BAD_REQUEST, &e.to_string()))?; 223 223 + // Push URL to room — metadata extraction runs async, no client wait. 224 224 + state.rooms.push_url(&room_id, req.url).await; 225 225 + let t2 = t0.elapsed(); 240 226 241 241 - // Push to room queue. 242 242 - state.rooms.push(&room_id, meta.clone()).await; 227 227 + tracing::info!( 228 228 + "ingest_url timing: exists={:.0?} push_url={:.0?} total={:.0?}", 229 229 + t1, t2 - t1, t2 230 230 + ); 243 231 244 244 - Ok(Json(IngestResponse { 245 245 - title: meta.title, 246 246 - duration: meta.duration, 247 247 - thumbnail: meta.thumbnail, 248 248 - })) 232 232 + Ok(Json(serde_json::json!({ "ok": true }))) 249 233 } 250 234 251 235 #[derive(serde::Deserialize)] ··· 321 305 .await; 322 306 })) 323 307 } 308 308 + 309 309 + /// WebSocket upgrade endpoint for video-only delivery: `/ws/{room_id}/video`. 310 310 + /// 311 311 + /// No MoQ protocol — server auto-subscribes to the video broadcast and 312 312 + /// forwards raw fMP4 payloads. The client identifies init segments by 313 313 + /// checking for the ftyp box in the data. 314 314 + async fn video_ws_handler( 315 315 + ws: WebSocketUpgrade, 316 316 + Path(room_id): Path<String>, 317 317 + State(state): State<AppState>, 318 318 + ) -> Result<impl IntoResponse, (StatusCode, &'static str)> { 319 319 + if !room_id.chars().all(|c| c.is_ascii_digit()) { 320 320 + return Err((StatusCode::BAD_REQUEST, "invalid room ID")); 321 321 + } 322 322 + 323 323 + let rid = RoomId(room_id); 324 324 + 325 325 + if !state.rooms.exists(&rid).await { 326 326 + return Err((StatusCode::NOT_FOUND, "room not found")); 327 327 + } 328 328 + 329 329 + let handle = state 330 330 + .rooms 331 331 + .handle(&rid) 332 332 + .await 333 333 + .ok_or((StatusCode::NOT_FOUND, "room not found"))?; 334 334 + 335 335 + Ok(ws.on_upgrade(move |socket| async move { 336 336 + crate::transport::handle_video_session(socket, handle.publishers).await; 337 337 + })) 338 338 + }

+142 -100

templates/room.html

Reviewed

··· 435 435 transition: 436 436 background 150ms ease-out, 437 437 box-shadow 150ms ease-out; 438 438 - animation: queueFadeIn 0.3s ease-out both; 438 438 + animation: queueFadeIn 0.12s ease-out both; 439 439 } 440 440 .queue li:hover { 441 441 background: var(--surface1); ··· 666 666 box-shadow: var(--shadow-card-hover); 667 667 } 668 668 669 669 + /* Skeleton loading state for pending queue items */ 670 670 + @keyframes shimmer { 671 671 + 0% { 672 672 + background-position: -200% 0; 673 673 + } 674 674 + 100% { 675 675 + background-position: 200% 0; 676 676 + } 677 677 + } 678 678 + .queue li.pending { 679 679 + position: relative; 680 680 + } 681 681 + .queue li.pending .title, 682 682 + .queue li.pending .duration { 683 683 + display: inline-block; 684 684 + border-radius: var(--radius-sm); 685 685 + background: linear-gradient( 686 686 + 90deg, 687 687 + var(--surface1) 25%, 688 688 + var(--surface2) 37%, 689 689 + var(--surface1) 63% 690 690 + ); 691 691 + background-size: 200% 100%; 692 692 + animation: shimmer 1.5s ease-in-out infinite; 693 693 + color: transparent !important; 694 694 + user-select: none; 695 695 + } 696 696 + .queue li.pending .title { 697 697 + min-width: 16ch; 698 698 + max-width: 24ch; 699 699 + } 700 700 + .queue li.pending .duration { 701 701 + min-width: 5ch; 702 702 + } 703 703 + .queue li.pending .source-link { 704 704 + visibility: hidden; 705 705 + } 706 706 + 669 707 /* Stagger animation */ 670 708 @keyframes queueFadeIn { 671 709 from { 672 672 - opacity: 0; 673 673 - transform: translateY(8px); 710 710 + opacity: 0.3; 711 711 + transform: translateY(4px); 674 712 } 675 713 to { 676 714 opacity: 1; 677 715 transform: translateY(0); 678 716 } 679 717 } 680 680 - .queue li:nth-child(1) { 681 681 - animation-delay: 0.02s; 682 682 - } 683 683 - .queue li:nth-child(2) { 684 684 - animation-delay: 0.04s; 685 685 - } 686 686 - .queue li:nth-child(3) { 687 687 - animation-delay: 0.06s; 688 688 - } 689 689 - .queue li:nth-child(4) { 690 690 - animation-delay: 0.08s; 691 691 - } 692 692 - .queue li:nth-child(5) { 693 693 - animation-delay: 0.1s; 694 694 - } 695 695 - .queue li:nth-child(6) { 696 696 - animation-delay: 0.12s; 697 697 - } 698 698 - .queue li:nth-child(7) { 699 699 - animation-delay: 0.14s; 700 700 - } 701 701 - .queue li:nth-child(8) { 702 702 - animation-delay: 0.16s; 703 703 - } 704 704 - .queue li:nth-child(9) { 705 705 - animation-delay: 0.18s; 706 706 - } 707 707 - .queue li:nth-child(10) { 708 708 - animation-delay: 0.2s; 709 709 - } 710 710 - .queue li:nth-child(11) { 711 711 - animation-delay: 0.22s; 712 712 - } 713 713 - .queue li:nth-child(12) { 714 714 - animation-delay: 0.24s; 715 715 - } 718 718 + .queue li:nth-child(1) { animation-delay: 0s; } 719 719 + .queue li:nth-child(2) { animation-delay: 0.01s; } 720 720 + .queue li:nth-child(3) { animation-delay: 0.02s; } 721 721 + .queue li:nth-child(4) { animation-delay: 0.03s; } 722 722 + .queue li:nth-child(5) { animation-delay: 0.04s; } 723 723 + .queue li:nth-child(6) { animation-delay: 0.05s; } 724 724 + .queue li:nth-child(n+7) { animation-delay: 0.05s; } 716 725 717 726 /* Mobile responsive */ 718 727 @media (max-width: 768px) { ··· 1115 1124 1116 1125 function appendVideo(data) { 1117 1126 if (sourceBuffer && !sourceBuffer.updating) { 1118 1118 - sourceBuffer.appendBuffer(data); 1127 1127 + try { 1128 1128 + sourceBuffer.appendBuffer(data); 1129 1129 + } catch (e) { 1130 1130 + console.warn("SourceBuffer append failed:", e); 1131 1131 + } 1119 1132 } else { 1120 1133 pendingBoxes.push(data); 1121 1134 } ··· 1128 1141 encVarint(0x01), 1129 1142 encString("moqbox/room/" + roomId + "/chat"), 1130 1143 encString("chat"), 1131 1131 - ]), 1132 1132 - ); 1133 1133 - // Subscribe to video track for MediaSource playback. 1134 1134 - ws.send( 1135 1135 - concat([ 1136 1136 - encVarint(0x01), 1137 1137 - encString("moqbox/room/" + roomId + "/video"), 1138 1138 - encString("video"), 1139 1144 ]), 1140 1145 ); 1141 1146 // Subscribe to state track for now-playing updates. ··· 1172 1177 debug("WS connected"); 1173 1178 updateConnStatus(true); 1174 1179 reconnectDelay = 1000; 1175 1175 - awaitingInit = true; 1176 1176 - pendingBoxes.length = 0; 1177 1180 currentTrackId = null; 1178 1178 - seekInitDone = false; 1179 1181 subscribeTracks(); 1180 1182 }; 1181 1183 ··· 1188 1190 const [trackId, o2] = decVarint(buf, o1); 1189 1191 const [groupId, o3] = decVarint(buf, o2); 1190 1192 const [objectId, o4] = decVarint(buf, o3); 1191 1191 - if (trackId === 1) { 1192 1192 - const videoData = new Uint8Array(buf.slice(o4)); 1193 1193 - 1194 1194 - if (awaitingInit) { 1195 1195 - const isInit = 1196 1196 - videoData.length >= 8 && 1197 1197 - String.fromCharCode( 1198 1198 - videoData[4], 1199 1199 - videoData[5], 1200 1200 - videoData[6], 1201 1201 - videoData[7], 1202 1202 - ) === "ftyp"; 1203 1203 - 1204 1204 - if (isInit) { 1205 1205 - debug("init segment received"); 1206 1206 - awaitingInit = false; 1207 1207 - pendingBoxes.length = 0; 1208 1208 - if (sourceBuffer && !sourceBuffer.updating) { 1209 1209 - try { 1210 1210 - sourceBuffer.abort(); 1211 1211 - if (sourceBuffer.buffered.length > 0) { 1212 1212 - const end = sourceBuffer.buffered.end( 1213 1213 - sourceBuffer.buffered.length - 1, 1214 1214 - ); 1215 1215 - sourceBuffer.remove(0, end); 1216 1216 - } 1217 1217 - } catch (_) {} 1218 1218 - } 1219 1219 - } else { 1220 1220 - pendingBoxes.push(videoData); 1221 1221 - off = buf.length; 1222 1222 - return; 1223 1223 - } 1224 1224 - } 1225 1225 - 1226 1226 - appendVideo(videoData); 1227 1227 - off = buf.length; 1228 1228 - } else if (trackId === 2) { 1193 1193 + if (trackId === 2) { 1229 1194 const payload = new TextDecoder().decode(buf.slice(o4)); 1230 1195 try { 1231 1196 const chat = JSON.parse(payload); ··· 1268 1233 1269 1234 connect(); 1270 1235 1236 1236 + // ── Video WebSocket (separate connection, no MoQ framing) ────────── 1237 1237 + let videoWs = null; 1238 1238 + let videoReconnectDelay = 1000; 1239 1239 + 1240 1240 + function connectVideoWs() { 1241 1241 + videoWs = new WebSocket( 1242 1242 + proto + "//" + location.host + "/ws/" + roomId + "/video", 1243 1243 + ); 1244 1244 + videoWs.binaryType = "arraybuffer"; 1245 1245 + 1246 1246 + videoWs.onopen = () => { 1247 1247 + debug("video WS connected"); 1248 1248 + videoReconnectDelay = 1000; 1249 1249 + awaitingInit = true; 1250 1250 + pendingBoxes.length = 0; 1251 1251 + seekInitDone = false; 1252 1252 + }; 1253 1253 + 1254 1254 + videoWs.onmessage = (event) => { 1255 1255 + const data = new Uint8Array(event.data); 1256 1256 + 1257 1257 + // Always detect init segments, even when not awaitingInit. 1258 1258 + // The video WS and control WS are independent connections with 1259 1259 + // no ordering guarantee — init can arrive before the state update. 1260 1260 + const isInit = 1261 1261 + data.length >= 8 && 1262 1262 + String.fromCharCode(data[4], data[5], data[6], data[7]) === "ftyp"; 1263 1263 + 1264 1264 + if (isInit) { 1265 1265 + debug("init segment received (video WS)"); 1266 1266 + awaitingInit = false; 1267 1267 + pendingBoxes.length = 0; 1268 1268 + if (sourceBuffer && !sourceBuffer.updating) { 1269 1269 + try { 1270 1270 + sourceBuffer.abort(); 1271 1271 + if (sourceBuffer.buffered.length > 0) { 1272 1272 + const end = sourceBuffer.buffered.end( 1273 1273 + sourceBuffer.buffered.length - 1, 1274 1274 + ); 1275 1275 + sourceBuffer.remove(0, end); 1276 1276 + } 1277 1277 + } catch (_) {} 1278 1278 + } 1279 1279 + appendVideo(data); 1280 1280 + return; 1281 1281 + } 1282 1282 + 1283 1283 + if (awaitingInit) { 1284 1284 + pendingBoxes.push(data); 1285 1285 + return; 1286 1286 + } 1287 1287 + 1288 1288 + appendVideo(data); 1289 1289 + }; 1290 1290 + 1291 1291 + videoWs.onclose = () => { 1292 1292 + setTimeout(() => { 1293 1293 + videoReconnectDelay = Math.min(videoReconnectDelay * 2, 30000); 1294 1294 + connectVideoWs(); 1295 1295 + }, videoReconnectDelay); 1296 1296 + }; 1297 1297 + } 1298 1298 + 1299 1299 + // Open video connection after a short delay to let the control WS 1300 1300 + // establish the room subscription first. 1301 1301 + setTimeout(connectVideoWs, 100); 1302 1302 + 1271 1303 function copyRoomLink() { 1272 1304 const url = window.location.href; 1273 1305 navigator.clipboard ··· 1355 1387 pill.textContent = c === 1 ? "1 viewer" : c + " viewers"; 1356 1388 } 1357 1389 1358 1358 - // Clear old interval when track changes. 1359 1359 - if (elapsedInterval) { 1360 1360 - clearInterval(elapsedInterval); 1361 1361 - elapsedInterval = null; 1362 1362 - } 1363 1363 - 1364 1390 if (state.current_track) { 1365 1391 const trackId = state.current_track.id; 1392 1392 + const trackChanged = currentTrackId !== trackId; 1366 1393 1367 1367 - // Detect track change — set awaitingInit so the next video object 1368 1368 - // is treated as the new init segment. 1369 1369 - if (currentTrackId !== null && currentTrackId !== trackId) { 1394 1394 + // Detect track change — reset video state. 1395 1395 + if (currentTrackId !== null && trackChanged) { 1370 1396 debug("track change", currentTrackId, "->", trackId); 1371 1397 awaitingInit = true; 1372 1398 pendingBoxes.length = 0; 1373 1399 seekInitDone = false; 1400 1400 + if (elapsedInterval) { 1401 1401 + clearInterval(elapsedInterval); 1402 1402 + elapsedInterval = null; 1403 1403 + } 1374 1404 } 1375 1405 currentTrackId = trackId; 1406 1406 + 1407 1407 + // Skip overlay DOM work when the track hasn't changed. 1408 1408 + // Queue display is still updated below. 1409 1409 + if (!trackChanged) { 1410 1410 + updateQueueDisplay(state); 1411 1411 + return; 1412 1412 + } 1376 1413 1377 1414 const startedAt = state.current_track.started_at; 1378 1415 const sourceUrl = state.current_track.url || ""; ··· 1455 1492 } 1456 1493 } 1457 1494 } else { 1495 1495 + if (elapsedInterval) { 1496 1496 + clearInterval(elapsedInterval); 1497 1497 + elapsedInterval = null; 1498 1498 + } 1458 1499 currentTrackId = null; 1459 1500 container.classList.add("idle"); 1460 1501 } 1461 1502 1462 1462 - // Update queue section from state. 1503 1503 + // Update queue section from state — inline, not deferred. 1463 1504 updateQueueDisplay(state); 1464 1505 } 1465 1506 ··· 1476 1517 esc(item.thumbnail) + 1477 1518 '" class="thumb-img" alt="">' 1478 1519 : '<div class="thumb"></div>'; 1520 1520 + const pendingClass = item.pending ? ' class="pending"' : ''; 1479 1521 html += 1480 1480 - "<li>" + 1522 1522 + '<li' + pendingClass + '>' + 1481 1523 thumb + 1482 1524 '<div class="title">' + 1483 1525 esc(item.title) +