diff --git a/Cargo.toml b/Cargo.toml index 5cf04e9..04d5639 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -34,6 +34,9 @@ futures-util = "0.3" rustls = { version = "0.23", features = ["ring"] } rcgen = "0.13" +# ── Async ───────────────────────────────────────────────────────── +async-trait = "0.1" + # ── Identifiers ──────────────────────────────────────────────────── uuid = { version = "1", features = ["v4", "serde"] } diff --git a/src/db.rs b/src/db.rs index 15ea749..2964b67 100644 --- a/src/db.rs +++ b/src/db.rs @@ -47,7 +47,7 @@ pub(crate) async fn run_migrations(pool: &Pool) -> Result<(), sqlx::Erro title TEXT NOT NULL, duration TEXT NOT NULL DEFAULT '--:--', thumbnail TEXT, - extractor TEXT NOT NULL DEFAULT '', + source TEXT NOT NULL DEFAULT '', added_by TEXT, position INTEGER NOT NULL, added_at TEXT NOT NULL, @@ -86,6 +86,37 @@ pub(crate) async fn run_migrations(pool: &Pool) -> Result<(), sqlx::Erro .await?; } + // Migration: rename `extractor` column to `source` (schema cleanup). + let has_source: bool = sqlx::query_scalar( + "SELECT COUNT(*) FROM pragma_table_info('queue_items') WHERE name = 'source'", + ) + .fetch_one(pool) + .await + .map(|c: i64| c > 0) + .unwrap_or(false); + + if !has_source { + // For pre-refactor databases: rename extractor → source. + let has_extractor: bool = sqlx::query_scalar( + "SELECT COUNT(*) FROM pragma_table_info('queue_items') WHERE name = 'extractor'", + ) + .fetch_one(pool) + .await + .map(|c: i64| c > 0) + .unwrap_or(false); + + if has_extractor { + sqlx::query("ALTER TABLE queue_items RENAME COLUMN extractor TO source") + .execute(pool) + .await?; + } else { + // Safety net: neither column exists (shouldn't happen). + sqlx::query("ALTER TABLE queue_items ADD COLUMN source TEXT NOT NULL DEFAULT ''") + .execute(pool) + .await?; + } + } + Ok(()) } diff --git a/src/main.rs b/src/main.rs index b10f06e..7bae696 100644 --- a/src/main.rs +++ b/src/main.rs @@ -5,8 +5,10 @@ mod db; mod media; +mod playback; mod playlist; mod room; +mod source; mod transport; mod types; mod web; @@ -14,6 +16,7 @@ mod web; use clap::Parser; use std::net::SocketAddr; use std::path::PathBuf; +use std::sync::Arc; use std::time::Duration; use tracing_subscriber::EnvFilter; @@ -55,7 +58,13 @@ async fn main() -> anyhow::Result<()> { db::run_migrations(&pool).await?; tracing::info!("database ready at {}", args.db_path.display()); - let rooms = room::Registry::new(pool.clone(), args.cache_dir.clone()); + // ── Source registry ────────────────────────────────────────────── + let mut sources = source::SourceRegistry::new(); + sources.register(Box::new(source::ytdlp::YtdlpSource::new())); + sources.register(Box::new(source::direct::DirectSource::new())); + let sources = Arc::new(sources); + + let rooms = room::Registry::new(pool.clone(), args.cache_dir.clone(), sources.clone()); // Idle room sweeper: every 60s, remove rooms idle for 600s (10 min). let registry_clone = rooms.clone(); @@ -70,7 +79,7 @@ async fn main() -> anyhow::Result<()> { } }); - let app = web::router(rooms, pool, args.cache_dir); + let app = web::router(rooms, pool, args.cache_dir, sources); let listener = tokio::net::TcpListener::bind(args.http_addr).await?; tracing::info!("HTTP server listening on {}", args.http_addr); diff --git a/src/media.rs b/src/media.rs index 458d40f..aa54888 100644 --- a/src/media.rs +++ b/src/media.rs @@ -1,23 +1,24 @@ -//! Media extraction and transcoding pipeline. +//! FFmpeg transcoding to fragmented MP4. //! -//! Wraps `yt-dlp` for metadata extraction and stream URL resolution, -//! and `ffmpeg` for transcoding to raw fMP4 chunks on stdout. -//! No fMP4 parsing — just reads stdout chunks and publishes as MoQ objects. - -use std::path::Path; +//! Takes a playable stream URL, spawns `ffmpeg` to transcode to fMP4 at +//! real-time speed (`-re`), reads fMP4 boxes from stdout as moof+mdat pairs, +//! and publishes them over per-room broadcast channels. +//! +//! This module does **not** resolve URLs or extract metadata — that's the +//! responsibility of [`crate::source`]. -use bytes::Bytes; -use serde::Deserialize; use tokio::io::AsyncReadExt; use tokio::process::Command; -use crate::types::{TrackMeta, TrackPublishers}; +use crate::transport::TrackPublishers; -/// Errors that can occur during media pipeline operations. +// --------------------------------------------------------------------------- +// Error +// --------------------------------------------------------------------------- + +/// Errors that can occur during FFmpeg transcoding. #[derive(Debug, thiserror::Error)] -pub(crate) enum MediaError { - #[error("extraction failed: {0}")] - Extraction(String), +pub(crate) enum TranscodeError { #[error("transcoding failed: {0}")] Transcode(String), #[error("io error: {0}")] @@ -26,74 +27,41 @@ pub(crate) enum MediaError { Aborted, } -/// Raw yt-dlp `--dump-json` output fields we care about. -#[derive(Debug, Deserialize)] -#[serde(rename_all = "snake_case")] -struct YtdlpMetadata { - title: String, - duration: Option, - thumbnail: Option, - webpage_url: String, - #[serde(default)] - extractor: String, -} +// --------------------------------------------------------------------------- +// Public API +// --------------------------------------------------------------------------- -/// Spawn FFmpeg pipeline and publish raw fMP4 chunks as MoQ objects. +/// Transcode a playable stream URL to fMP4, publishing raw boxes as MoQ objects. +/// +/// Spawns `ffmpeg` with the stream URL as input, transcodes to fragmented MP4 +/// on stdout, and publishes each media segment (moof+mdat pair) as a single +/// MoQ object. The init segment (ftyp+moov) is cached so late-joining clients +/// can initialise their MediaSource. /// /// Returns `Ok(())` on normal EOF, `Err` on failure. -pub(crate) async fn spawn_pipeline( - url: &str, - cache_dir: &Path, +/// +/// ## Abort behaviour +/// +/// When `abort` is signalled, the pipeline exits early with +/// [`TranscodeError::Aborted`] and the ffmpeg process is killed. +pub(crate) async fn transcode_to_fmp4( + stream_url: &str, publishers: TrackPublishers, mut abort: tokio::sync::watch::Receiver, group_id: u64, -) -> Result<(), MediaError> { - tracing::debug!(url, "spawning ffmpeg pipeline"); - - // 1. Get direct stream URL from yt-dlp (best single-file format). - // Using -f b ensures a combined audio+video URL; -g alone returns - // separate audio and video URLs which ffmpeg can't use with one -i. - let output = Command::new("yt-dlp") - .args([ - "-f", - "b", - "-g", - "--no-playlist", - "--no-warnings", - "--cache-dir", - ]) - .arg(cache_dir) - .arg(url) - .kill_on_drop(true) - .output() - .await - .map_err(MediaError::Io)?; - - if !output.status.success() { - let stderr = String::from_utf8_lossy(&output.stderr); - return Err(MediaError::Extraction(stderr.trim().to_string())); - } - - let stream_url = String::from_utf8(output.stdout) - .map_err(|_| MediaError::Extraction("invalid UTF-8 from yt-dlp".into()))? - .trim() - .to_string(); - - if stream_url.is_empty() { - return Err(MediaError::Extraction( - "yt-dlp returned empty stream URL".into(), - )); - } - // 2. Spawn FFmpeg transcoding to fMP4 on stdout. - // -re throttles the input to native frame rate so output - // stays in sync with wall-clock time. Without it FFmpeg - // transcodes faster than real-time, making late-join - // position calculations impossible. +) -> Result<(), TranscodeError> { + tracing::debug!(url = %stream_url, "spawning ffmpeg pipeline"); + + // Spawn FFmpeg transcoding to fMP4 on stdout. + // -re throttles the input to native frame rate so output + // stays in sync with wall-clock time. Without it FFmpeg + // transcodes faster than real-time, making late-join + // position calculations impossible. let mut ffmpeg = Command::new("ffmpeg") .args([ "-re", "-i", - &stream_url, + stream_url, "-c:v", "libx264", "-profile:v", @@ -122,14 +90,14 @@ pub(crate) async fn spawn_pipeline( .stderr(std::process::Stdio::piped()) .kill_on_drop(true) .spawn() - .map_err(|e| MediaError::Transcode(format!("failed to spawn ffmpeg: {e}")))?; + .map_err(|e| TranscodeError::Transcode(format!("failed to spawn ffmpeg: {e}")))?; tracing::debug!("ffmpeg pipeline started"); let mut stdout = ffmpeg .stdout .take() - .ok_or_else(|| MediaError::Transcode("no stdout from ffmpeg".into()))?; + .ok_or_else(|| TranscodeError::Transcode("no stdout from ffmpeg".into()))?; // Spawn background task to drain stderr (prevents pipe deadlock at 64KB). let stderr_drain = ffmpeg.stderr.take().map(|stderr| { @@ -141,18 +109,17 @@ pub(crate) async fn spawn_pipeline( }) }); - // 3. Read and publish each fMP4 box as a MoQ object. - // Box format: [4-byte big-endian size][4-byte type][payload] + // 1. Read and publish each fMP4 box as a MoQ object. // Publishing complete boxes ensures MSE can consume them directly. read_and_publish_boxes(&mut stdout, &publishers, &mut abort, group_id).await?; - // 4. Wait for FFmpeg to exit + // 2. Wait for FFmpeg to exit let status = ffmpeg .wait() .await - .map_err(|e| MediaError::Transcode(format!("ffmpeg wait failed: {e}")))?; + .map_err(|e| TranscodeError::Transcode(format!("ffmpeg wait failed: {e}")))?; - // 5. Collect stderr output (background task already completed since + // 3. Collect stderr output (background task already completed since // ffmpeg closed stderr on exit). let stderr_output = match stderr_drain { Some(handle) => handle.await.unwrap_or_default(), @@ -169,26 +136,26 @@ pub(crate) async fn spawn_pipeline( Ok(()) } +// --------------------------------------------------------------------------- +// fMP4 box parsing +// --------------------------------------------------------------------------- + /// Read fMP4 boxes from FFmpeg stdout, buffer moof+mdat pairs, and publish /// each complete segment as a single MoQ object. /// -/// Box format: [4-byte big-endian size][4-byte type][payload] -/// /// MSE requires moof and its following mdat to be appended as one buffer. /// This function buffers moof, waits for the next mdat, concatenates them, /// and publishes the pair as a single object. -/// -/// Returns Ok on EOF, Err on read failure. async fn read_and_publish_boxes( stdout: &mut (impl tokio::io::AsyncRead + Unpin), publishers: &TrackPublishers, abort: &mut tokio::sync::watch::Receiver, group_id: u64, -) -> Result<(), MediaError> { +) -> Result<(), TranscodeError> { // Check abort signal before starting (handles pre-signalled aborts). if *abort.borrow() { tracing::debug!("pipeline aborted before starting"); - return Err(MediaError::Aborted); + return Err(TranscodeError::Aborted); } let mut object_id: u64 = 0; @@ -199,11 +166,11 @@ async fn read_and_publish_boxes( // Read 8-byte box header let mut header = [0u8; 8]; tokio::select! { - _ = abort.changed() => return Err(MediaError::Aborted), + _ = abort.changed() => return Err(TranscodeError::Aborted), result = stdout.read_exact(&mut header) => { match result { Err(ref e) if e.kind() == std::io::ErrorKind::UnexpectedEof => return Ok(()), - Err(e) => return Err(MediaError::Io(e)), + Err(e) => return Err(TranscodeError::Io(e)), Ok(_) => {} } } @@ -214,20 +181,21 @@ async fn read_and_publish_boxes( match &box_type { b"ftyp" => { - // Buffer ftyp — wait for moov before publishing init segment. init_data = box_data.to_vec(); } b"moov" => { - // Concatenate with buffered ftyp and publish as one init segment. if !init_data.is_empty() { let mut init = std::mem::take(&mut init_data); init.extend_from_slice(&box_data); - publishers.cache_init_segment(Bytes::copy_from_slice(&init), group_id); - // Init is NOT published to broadcast — the forward task sends - // it explicitly from cache so late joiners can initialise. + let payload = bytes::Bytes::copy_from_slice(&init); + // Cache for late-joining clients (forward task sends from cache). + publishers.cache_init_segment(payload.clone(), group_id); + // Also publish to broadcast so existing clients on track change + // (e.g. after skip) receive the new init and can initialise. + publishers.publish_video(group_id, object_id, payload); object_id += 1; } else { - publishers.publish_video(group_id, object_id, Bytes::from(box_data)); + publishers.publish_video(group_id, object_id, bytes::Bytes::from(box_data)); object_id += 1; } } @@ -238,15 +206,15 @@ async fn read_and_publish_boxes( if let Some(moof) = pending_moof.take() { let mut combined = moof; combined.extend_from_slice(&box_data); - publishers.publish_video(group_id, object_id, Bytes::from(combined)); + publishers.publish_video(group_id, object_id, bytes::Bytes::from(combined)); object_id += 1; } else { - publishers.publish_video(group_id, object_id, Bytes::from(box_data)); + publishers.publish_video(group_id, object_id, bytes::Bytes::from(box_data)); object_id += 1; } } _ => { - publishers.publish_video(group_id, object_id, Bytes::from(box_data)); + publishers.publish_video(group_id, object_id, bytes::Bytes::from(box_data)); object_id += 1; } } @@ -254,13 +222,10 @@ async fn read_and_publish_boxes( } /// Read the complete bytes of an ISOBMFF box given its 8-byte header. -/// -/// Returns the full box bytes (header + payload, including extended-size -/// field if applicable), suitable for MSE SourceBuffer.appendBuffer(). async fn read_box_payload( reader: &mut R, header: &[u8; 8], -) -> Result, MediaError> { +) -> Result, TranscodeError> { let size = u32::from_be_bytes([header[0], header[1], header[2], header[3]]); if size == 0 { @@ -277,7 +242,7 @@ async fn read_box_payload( let mut ext = [0u8; 8]; reader.read_exact(&mut ext).await?; let total = u64::from_be_bytes(ext) as usize; - let payload_len = total - 16; // header(8) + ext(8) + let payload_len = total - 16; let mut payload = vec![0u8; payload_len]; reader.read_exact(&mut payload).await?; let mut full = Vec::with_capacity(total); @@ -296,54 +261,12 @@ async fn read_box_payload( Ok(full) } -/// Run `yt-dlp --dump-json ` and parse the result. -/// -/// Returns a [`TrackMeta`] suitable for the room queue. -/// -/// # Errors -/// -/// Returns an error if yt-dlp is not installed, the URL is not -/// extractable, or the process times out. -pub(crate) async fn extract(url: &str, cache_dir: &Path) -> anyhow::Result { - if !url.starts_with("http://") && !url.starts_with("https://") { - anyhow::bail!("URL must start with http:// or https://"); - } +// --------------------------------------------------------------------------- +// Formatting +// --------------------------------------------------------------------------- - let output = Command::new("yt-dlp") - .args([ - "--dump-json", - "--no-playlist", - "--no-warnings", - "--cache-dir", - ]) - .arg(cache_dir) - .arg(url) - .kill_on_drop(true) - .output() - .await?; - - if !output.status.success() { - let stderr = String::from_utf8_lossy(&output.stderr); - anyhow::bail!("yt-dlp failed: {}", stderr.trim()); - } - - let raw: YtdlpMetadata = serde_json::from_slice(&output.stdout)?; - - let duration = raw - .duration - .map(format_duration) - .unwrap_or_else(|| "--:--".into()); - - Ok(TrackMeta { - title: raw.title, - duration, - thumbnail: raw.thumbnail, - url: raw.webpage_url, - extractor: raw.extractor, - }) -} - -fn format_duration(total_secs: f64) -> String { +/// Format a duration in seconds to a human-readable `M:SS` or `H:MM:SS` string. +pub(crate) fn format_duration(total_secs: f64) -> String { let total = total_secs as u64; let hours = total / 3600; let minutes = (total % 3600) / 60; @@ -356,10 +279,13 @@ fn format_duration(total_secs: f64) -> String { } } +// --------------------------------------------------------------------------- +// Tests +// --------------------------------------------------------------------------- + #[cfg(test)] mod tests { use super::*; - use std::path::Path; #[test] fn test_format_duration() { @@ -368,26 +294,17 @@ mod tests { assert_eq!(format_duration(3661.0), "1:01:01"); } - #[test] - fn test_extract_rejects_bad_urls() { - let rt = tokio::runtime::Runtime::new().unwrap(); - let result = rt.block_on(extract("javascript:alert(1)", Path::new("/tmp"))); - assert!(result.is_err()); - assert!(result.unwrap_err().to_string().contains("http")); + fn make_box(box_type: &[u8; 4], payload: &[u8]) -> Vec { + let size = 8 + payload.len() as u32; + let mut buf = Vec::with_capacity(size as usize); + buf.extend_from_slice(&size.to_be_bytes()); + buf.extend_from_slice(box_type); + buf.extend_from_slice(payload); + buf } #[tokio::test] async fn test_moof_mdat_combining() { - fn make_box(box_type: &[u8; 4], payload: &[u8]) -> Vec { - let size = 8 + payload.len() as u32; - let mut buf = Vec::with_capacity(size as usize); - buf.extend_from_slice(&size.to_be_bytes()); - buf.extend_from_slice(box_type); - buf.extend_from_slice(payload); - buf - } - - // Build a minimal fMP4 sequence: ftyp, moov, moof, mdat. let ftyp = make_box(b"ftyp", b"iso5"); let moov = make_box(b"moov", b"moovdata"); let moof = make_box(b"moof", b"moofdata"); @@ -408,8 +325,7 @@ mod tests { .await .expect("read_and_publish_boxes should succeed on valid fMP4"); - // Init segment is NOT published to broadcast — only cached. - // Verify the cache contains ftyp+moov. + // Init is cached (for late joiners) AND published to broadcast (for track changes). let cached = publishers.get_init_segment(); assert!(cached.is_some(), "init cache should contain ftyp+moov"); let (cached_group, cached_data) = cached.unwrap(); @@ -418,8 +334,14 @@ mod tests { expected_init.extend_from_slice(&moov); assert_eq!(cached_data.to_vec(), expected_init); - // Object 1: moof+mdat combined into one media segment - // (object_id=0 was skipped — it's reserved for the forward task's cached init) + // Object 0: init (ftyp+moov) published to broadcast. + let init_obj = rx.recv().await.expect("should receive init via broadcast"); + assert_eq!(init_obj.object_id, 0); + let mut expected_init_broadcast = ftyp.clone(); + expected_init_broadcast.extend_from_slice(&moov); + assert_eq!(init_obj.payload.to_vec(), expected_init_broadcast); + + // Object 1: moof+mdat combined. let obj1 = rx.recv().await.expect("should receive combined moof+mdat"); assert_eq!(obj1.object_id, 1); let mut expected_media = moof.clone(); @@ -427,20 +349,8 @@ mod tests { assert_eq!(obj1.payload.to_vec(), expected_media); } - /// Verify init segment (ftyp+moov) is cached but NOT published to broadcast. - /// Only the moof+mdat media segment should appear on the broadcast channel, - /// with object_id=1 (object_id=0 is reserved for the forward task's cached init). #[tokio::test] - async fn test_init_not_published_to_broadcast() { - fn make_box(box_type: &[u8; 4], payload: &[u8]) -> Vec { - let size = 8 + payload.len() as u32; - let mut buf = Vec::with_capacity(size as usize); - buf.extend_from_slice(&size.to_be_bytes()); - buf.extend_from_slice(box_type); - buf.extend_from_slice(payload); - buf - } - + async fn test_init_broadcast_and_cache() { let ftyp = make_box(b"ftyp", b"iso5"); let moov = make_box(b"moov", b"moovdata"); let moof = make_box(b"moof", b"moofdata"); @@ -461,45 +371,28 @@ mod tests { .await .expect("read_and_publish_boxes should succeed"); - // Verify init is cached but NOT broadcast. + // Init is both cached and broadcast. let cached = publishers.get_init_segment(); assert!(cached.is_some(), "init cache should contain ftyp+moov"); - let (cached_group, cached_data) = cached.unwrap(); - assert_eq!(cached_group, 0); + + // Object 0: init via broadcast. + let init = rx.recv().await.expect("should receive init via broadcast"); + assert_eq!(init.object_id, 0); + assert_eq!(init.group_id, 0); let mut expected_init = ftyp.clone(); expected_init.extend_from_slice(&moov); - assert_eq!(cached_data.to_vec(), expected_init); + assert_eq!(init.payload.to_vec(), expected_init); - // Only one broadcast message: moof+mdat at object_id=1. + // Object 1: moof+mdat media segment. let media = rx.recv().await.expect("should receive media segment"); - assert_eq!( - media.object_id, 1, - "media segment should have object_id=1 (0 reserved for cached init)" - ); - assert_eq!(media.group_id, 0); + assert_eq!(media.object_id, 1); let mut expected_media = moof.clone(); expected_media.extend_from_slice(&mdat); assert_eq!(media.payload.to_vec(), expected_media); - - // Confirm no second broadcast message. - match tokio::time::timeout(std::time::Duration::from_millis(100), rx.recv()).await { - Err(tokio::time::error::Elapsed { .. }) => {} // expected: no more messages - Ok(_) => panic!("should not receive a second broadcast message"), - } } - /// Verify get_init_segment returns the correct group_id. #[tokio::test] async fn test_init_cache_has_group_id() { - fn make_box(box_type: &[u8; 4], payload: &[u8]) -> Vec { - let size = 8 + payload.len() as u32; - let mut buf = Vec::with_capacity(size as usize); - buf.extend_from_slice(&size.to_be_bytes()); - buf.extend_from_slice(box_type); - buf.extend_from_slice(payload); - buf - } - let ftyp = make_box(b"ftyp", b"iso5"); let moov = make_box(b"moov", b"moovdata"); diff --git a/src/playback.rs b/src/playback.rs new file mode 100644 index 0000000..de9199b --- /dev/null +++ b/src/playback.rs @@ -0,0 +1,299 @@ +//! Playback lifecycle management. +//! +//! The [`Player`] struct owns the lifecycle of the currently-playing track: +//! starting a new track (resolving its stream URL and transcoding to fMP4), +//! aborting playback, and detecting when a track has ended. +//! +//! This separates playback concerns from room orchestration — the room actor +//! tells the player *what* to play, not *how* to play it. + +use std::path::PathBuf; +use std::sync::Arc; +use std::time::Duration; + +use tokio::sync::{mpsc, watch}; + +use crate::media; +use crate::playlist::QueueItem; +use crate::source::SourceRegistry; +use crate::transport::TrackPublishers; +use crate::types::{ActiveTrackInfo, RoomCommand}; + +/// Manages the lifecycle of the currently-playing track. +/// +/// # Lifecycle +/// +/// 1. **Start**: [`Player::start`] resolves the URL via the source registry, +/// spawns ffmpeg transcoding, and stores an abort handle. +/// 2. **End**: The spawned task sends [`RoomCommand::TrackEnded`] when ffmpeg +/// finishes or errors. The room actor calls [`Player::on_track_ended`] to +/// acknowledge. +/// 3. **Skip**: [`Player::abort`] signals the transcoding task to stop. +/// +/// The player is **not** responsible for queue management — that belongs to +/// the room actor. +pub(crate) struct Player { + source_registry: Arc, + cache_dir: PathBuf, + /// Channel to send [`RoomCommand::TrackEnded`] back to the room actor. + pipeline_tx: mpsc::Sender, + active_track: Option, + abort_tx: Option>, +} + +impl Player { + pub(crate) fn new( + source_registry: Arc, + cache_dir: PathBuf, + pipeline_tx: mpsc::Sender, + ) -> Self { + Self { + source_registry, + cache_dir, + pipeline_tx, + active_track: None, + abort_tx: None, + } + } + + /// Start playing a queue item. + /// + /// Resolves its URL to a playable stream via the source registry, spawns + /// ffmpeg transcoding to fMP4, and returns the active track info for state + /// publishing. + /// + /// If a track is already playing, it is **not** aborted — call [`abort`] + /// first. + /// + /// [`abort`]: Player::abort + pub(crate) fn start( + &mut self, + item: &QueueItem, + publishers: TrackPublishers, + ) -> ActiveTrackInfo { + let (abort_tx, abort_rx) = watch::channel(false); + let info = ActiveTrackInfo::from_item(item); + let track_id = item.id; + + let source_registry = self.source_registry.clone(); + let url = item.url.clone(); + let cache_dir = self.cache_dir.clone(); + let pipeline_tx = self.pipeline_tx.clone(); + + tokio::spawn(async move { + // 1. Resolve URL to a playable stream. + let stream_url = match source_registry.resolve(&url, &cache_dir).await { + Ok(url) => url, + Err(e) => { + tracing::error!(%url, error = %e, "player: failed to resolve stream"); + send_track_ended(&pipeline_tx, track_id).await; + return; + } + }; + + // 2. Transcode to fMP4 via ffmpeg. + let group_id = track_id as u64; + let result = media::transcode_to_fmp4(&stream_url, publishers, abort_rx, group_id) + .await; + + if let Err(e) = &result { + tracing::warn!(%url, error = %e, "player: transcoding finished with error"); + } + + // Small delay to give Skip time to clear state first. + tokio::time::sleep(Duration::from_millis(200)).await; + send_track_ended(&pipeline_tx, track_id).await; + }); + + self.active_track = Some(info.clone()); + self.abort_tx = Some(abort_tx); + + info + } + + /// Abort the currently playing track. + /// + /// Returns the DB id of the aborted track, or `None` if nothing was playing. + pub(crate) fn abort(&mut self) -> Option { + let id = self.active_track.take().map(|t| t.id); + if let Some(abort) = self.abort_tx.take() { + let _ = abort.send(true); + } + id + } + + /// Acknowledge that a track has ended. + /// + /// Returns the track's DB id if `item_id` matches the currently active + /// track, or `None` if it doesn't match (e.g. it was already skipped). + pub(crate) fn on_track_ended(&mut self, item_id: i64) -> Option { + let is_current = self + .active_track + .as_ref() + .map(|t| t.id == item_id) + .unwrap_or(false); + if is_current { + self.active_track = None; + self.abort_tx = None; + Some(item_id) + } else { + None + } + } + + /// Is a track currently playing? + pub(crate) fn is_playing(&self) -> bool { + self.active_track.is_some() + } + + /// Get a snapshot of the currently playing track, if any. + pub(crate) fn active_track(&self) -> Option { + self.active_track.clone() + } +} + +/// Send [`RoomCommand::TrackEnded`] with best-effort logging on failure. +async fn send_track_ended(tx: &mpsc::Sender, item_id: i64) { + if let Err(e) = tx.send(RoomCommand::TrackEnded { item_id }).await { + tracing::warn!(item_id, error = %e, "player: failed to send TrackEnded"); + } +} + +// --------------------------------------------------------------------------- +// Tests +// --------------------------------------------------------------------------- + +#[cfg(test)] +mod tests { + use super::*; + use crate::db; + use crate::playlist::Playlist; + use crate::types::TrackMeta; + use sqlx::sqlite::SqlitePoolOptions; + use sqlx::Pool; + use std::path::Path; + use std::sync::Arc; + + /// A mock source that resolves all URLs to a fake stream. + struct MockPlayerSource; + + #[async_trait::async_trait] + impl crate::source::MediaSource for MockPlayerSource { + async fn extract( + &self, + url: &str, + _cache_dir: &Path, + ) -> Result { + Err(crate::source::SourceError::Unsupported(url.into())) + } + + async fn resolve( + &self, + url: &str, + _cache_dir: &Path, + ) -> Result { + // Fake resolve: return the URL with /stream appended. + Ok(format!("{url}/stream")) + } + } + + fn test_registry() -> Arc { + let mut reg = SourceRegistry::new(); + reg.register(Box::new(MockPlayerSource)); + Arc::new(reg) + } + + async fn test_pool() -> Pool { + let pool = SqlitePoolOptions::new() + .max_connections(1) + .connect(":memory:") + .await + .unwrap(); + db::run_migrations(&pool).await.unwrap(); + db::room_insert(&pool, "test-room", false) + .await + .unwrap(); + pool + } + + async fn make_queue_item(pool: &Pool) -> QueueItem { + let pl = Playlist::new(pool.clone(), "test-room"); + pl.push(&TrackMeta { + title: "Test".into(), + duration: "3:45".into(), + thumbnail: None, + url: "https://example.com/track".into(), + source: crate::types::SourceKind::Direct, + }) + .await + .unwrap() + } + + #[tokio::test] + async fn test_player_start_sets_active_track() { + let pool = test_pool().await; + let item = make_queue_item(&pool).await; + let (tx, _rx) = mpsc::channel(256); + + let mut player = Player::new(test_registry(), PathBuf::from("/tmp"), tx); + assert!(!player.is_playing()); + + let info = player.start(&item, TrackPublishers::new()); + assert!(player.is_playing()); + assert_eq!(info.title, "Test"); + assert_eq!(info.id, item.id); + } + + #[tokio::test] + async fn test_player_abort_clears_active_track() { + let pool = test_pool().await; + let item = make_queue_item(&pool).await; + let (tx, _rx) = mpsc::channel(256); + + let mut player = Player::new(test_registry(), PathBuf::from("/tmp"), tx); + player.start(&item, TrackPublishers::new()); + assert!(player.is_playing()); + + let aborted_id = player.abort(); + assert_eq!(aborted_id, Some(item.id)); + assert!(!player.is_playing()); + assert!(player.active_track().is_none()); + } + + #[tokio::test] + async fn test_player_abort_when_idle_returns_none() { + let (tx, _rx) = mpsc::channel(256); + let mut player = Player::new(test_registry(), PathBuf::from("/tmp"), tx); + assert!(player.abort().is_none()); + } + + #[tokio::test] + async fn test_player_on_track_ended_matches_current() { + let pool = test_pool().await; + let item = make_queue_item(&pool).await; + let (tx, _rx) = mpsc::channel(256); + + let mut player = Player::new(test_registry(), PathBuf::from("/tmp"), tx); + player.start(&item, TrackPublishers::new()); + + let result = player.on_track_ended(item.id); + assert_eq!(result, Some(item.id)); + assert!(!player.is_playing()); + } + + #[tokio::test] + async fn test_player_on_track_ended_ignores_old_id() { + let pool = test_pool().await; + let item = make_queue_item(&pool).await; + let (tx, _rx) = mpsc::channel(256); + + let mut player = Player::new(test_registry(), PathBuf::from("/tmp"), tx); + player.start(&item, TrackPublishers::new()); + assert!(player.is_playing()); + + // An old/stale item_id should be ignored. + let result = player.on_track_ended(99999); + assert!(result.is_none()); + assert!(player.is_playing()); + } +} diff --git a/src/playlist.rs b/src/playlist.rs index 442c18d..6827845 100644 --- a/src/playlist.rs +++ b/src/playlist.rs @@ -21,7 +21,7 @@ pub(crate) struct QueueItem { pub(crate) title: String, pub(crate) duration: String, pub(crate) thumbnail: Option, - pub(crate) extractor: String, + pub(crate) source: String, pub(crate) position: i32, pub(crate) played: bool, pub(crate) played_at: Option, @@ -51,7 +51,7 @@ impl Playlist { let position = count as i32; let result = sqlx::query( - "INSERT INTO queue_items (room_id, url, title, duration, thumbnail, extractor, added_by, position, added_at, played) + "INSERT INTO queue_items (room_id, url, title, duration, thumbnail, source, added_by, position, added_at, played) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, 0)", ) .bind(&self.room_id) @@ -59,7 +59,7 @@ impl Playlist { .bind(&meta.title) .bind(&meta.duration) .bind(&meta.thumbnail) - .bind(&meta.extractor) + .bind(meta.source.to_string()) .bind(None::) // added_by .bind(position) .bind(now_iso()) @@ -69,7 +69,7 @@ impl Playlist { let id = result.last_insert_rowid(); let item = sqlx::query_as::<_, QueueItem>( - "SELECT id, url, title, duration, thumbnail, extractor, position, played, played_at, started_playing_at + "SELECT id, url, title, duration, thumbnail, source, position, played, played_at, started_playing_at FROM queue_items WHERE id = ?", ) .bind(id) @@ -88,7 +88,7 @@ impl Playlist { let mut tx = self.pool.begin().await?; let item = sqlx::query_as::<_, QueueItem>( - "SELECT id, url, title, duration, thumbnail, extractor, position, played, played_at, started_playing_at + "SELECT id, url, title, duration, thumbnail, source, position, played, played_at, started_playing_at FROM queue_items WHERE room_id = ? AND played = 0 AND started_playing_at IS NULL ORDER BY position ASC LIMIT 1", ) @@ -119,7 +119,7 @@ impl Playlist { /// List queued items (not playing, not finished) ordered by position ASC. pub(crate) async fn upcoming(&self) -> Result, sqlx::Error> { let items = sqlx::query_as::<_, QueueItem>( - "SELECT id, url, title, duration, thumbnail, extractor, position, played, played_at, started_playing_at + "SELECT id, url, title, duration, thumbnail, source, position, played, played_at, started_playing_at FROM queue_items WHERE room_id = ? AND played = 0 AND started_playing_at IS NULL ORDER BY position ASC", ) @@ -132,7 +132,7 @@ impl Playlist { /// List played items ordered by played_at DESC (newest first), limited to `limit`. pub(crate) async fn history(&self, limit: i64) -> Result, sqlx::Error> { let items = sqlx::query_as::<_, QueueItem>( - "SELECT id, url, title, duration, thumbnail, extractor, position, played, played_at, started_playing_at + "SELECT id, url, title, duration, thumbnail, source, position, played, played_at, started_playing_at FROM queue_items WHERE room_id = ? AND played = 1 ORDER BY played_at DESC, id DESC LIMIT ?", ) .bind(&self.room_id) @@ -181,7 +181,7 @@ mod tests { duration: "3:45".into(), thumbnail: None, url: "https://example.com/track".into(), - extractor: "test".into(), + source: crate::types::SourceKind::Direct, } } diff --git a/src/room.rs b/src/room.rs index 7df5089..d1818c1 100644 --- a/src/room.rs +++ b/src/room.rs @@ -21,11 +21,13 @@ use tokio::sync::oneshot; use tokio::sync::{RwLock, mpsc, watch}; use crate::db; -use crate::media; -use crate::playlist::{Playlist, QueueItem}; +use crate::playback::Player; +use crate::playlist::Playlist; +use crate::source::SourceRegistry; +use crate::transport::TrackPublishers; use crate::types::{ ChatMessage, HistoryEntry, QueueSummary, RoomCommand, RoomHandle, RoomId, TrackMeta, - TrackPublishers, TrackState, + TrackState, }; /// Distributed unique ID generator (Discord-style snowflake). @@ -107,41 +109,15 @@ impl SnowflakeIdGen { } } -/// Ephemeral info about the active track, stored in-memory only. -struct ActiveTrackInfo { - id: i64, - title: String, - url: String, - duration: String, - /// Epoch ms when the track started — for client-side elapsed computation. - started_at_wall: i64, -} - -impl ActiveTrackInfo { - fn from_item(item: &QueueItem) -> Self { - Self { - id: item.id, - title: item.title.clone(), - url: item.url.clone(), - duration: item.duration.clone(), - started_at_wall: chrono::Utc::now().timestamp_millis(), - } - } -} - /// Per-room event-loop actor. Owns all mutable state. struct RoomActor { rx: mpsc::Receiver, - /// Clone of tx passed to pipeline tasks so they can send TrackEnded back. - pipeline_tx: mpsc::Sender, room_id: RoomId, publishers: TrackPublishers, pool: Pool, - cache_dir: PathBuf, // Mutable state (owned by this actor, no locks needed) - active_track: Option, - pipeline_abort: Option>, + player: Player, client_count: u64, next_user_id: u64, last_active_tx: watch::Sender>, @@ -153,6 +129,7 @@ pub(crate) struct Registry { pool: Pool, id_gen: Arc, cache_dir: PathBuf, + sources: Arc, inner: Arc>, } @@ -161,7 +138,11 @@ struct Inner { } impl Registry { - pub(crate) fn new(pool: Pool, cache_dir: PathBuf) -> Self { + pub(crate) fn new( + pool: Pool, + cache_dir: PathBuf, + sources: Arc, + ) -> Self { Self { pool, id_gen: Arc::new(SnowflakeIdGen::new(1)), @@ -169,6 +150,7 @@ impl Registry { rooms: HashMap::new(), })), cache_dir, + sources, } } @@ -182,15 +164,17 @@ impl Registry { let (tx, rx) = mpsc::channel(256); let (last_active_tx, last_active_rx) = watch::channel(now); + let player = Player::new( + self.sources.clone(), + self.cache_dir.clone(), + tx.clone(), + ); let actor = RoomActor { rx, - pipeline_tx: tx.clone(), room_id: id.clone(), publishers: publishers.clone(), pool: self.pool.clone(), - cache_dir: self.cache_dir.clone(), - active_track: None, - pipeline_abort: None, + player, client_count: 0, next_user_id: 1, last_active_tx, @@ -233,7 +217,13 @@ impl Registry { duration: i.duration, thumbnail: i.thumbnail, url: i.url, - extractor: i.extractor, + source: i + .source + .parse() + .unwrap_or_else(|s: String| { + tracing::warn!(source = %s, "unknown source kind in queue_items"); + crate::types::SourceKind::Direct + }), }) .collect() } @@ -416,9 +406,7 @@ impl RoomActor { } } // On shutdown, abort any active pipeline. - if let Some(abort) = self.pipeline_abort.take() { - let _ = abort.send(true); - } + self.player.abort(); tracing::info!(room = %self.room_id, "room actor shut down"); } @@ -430,17 +418,14 @@ impl RoomActor { if let Err(e) = playlist.push(&meta).await { tracing::warn!("failed to persist queue item: {e}"); } - if self.active_track.is_none() { + if !self.player.is_playing() { self.start_next().await; } self.publish_state_snapshot().await; } RoomCommand::Skip => { tracing::info!(room = %self.room_id, "skip requested"); - let finished_id = self.active_track.take().map(|t| t.id); - if let Some(abort) = self.pipeline_abort.take() { - let _ = abort.send(true); - } + let finished_id = self.player.abort(); if let Some(id) = finished_id { let playlist = Playlist::new(self.pool.clone(), &self.room_id.0); let _ = playlist.mark_finished(id).await; @@ -450,14 +435,7 @@ impl RoomActor { } RoomCommand::TrackEnded { item_id } => { tracing::debug!(room = %self.room_id, item_id, "track ended"); - let is_current = self - .active_track - .as_ref() - .map(|t| t.id == item_id) - .unwrap_or(false); - if is_current { - self.active_track = None; - self.pipeline_abort = None; + if self.player.on_track_ended(item_id).is_some() { let playlist = Playlist::new(self.pool.clone(), &self.room_id.0); let _ = playlist.mark_finished(item_id).await; self.start_next().await; @@ -505,9 +483,7 @@ impl RoomActor { } RoomCommand::Shutdown => { tracing::info!(room = %self.room_id, "actor shutdown requested"); - if let Some(abort) = self.pipeline_abort.take() { - let _ = abort.send(true); - } + self.player.abort(); return true; } } @@ -525,33 +501,11 @@ impl RoomActor { } }; tracing::info!(room = %self.room_id, title = %item.title, item_id = item.id, "starting track"); - let (abort_tx, abort_rx) = watch::channel(false); - let info = ActiveTrackInfo::from_item(&item); - self.active_track = Some(info); - self.pipeline_abort = Some(abort_tx); - - let url = item.url.clone(); - let cache = self.cache_dir.clone(); - let publishers = self.publishers.clone(); - let pipeline_tx = self.pipeline_tx.clone(); - let track_item_id = item.id; - let group_id = item.id as u64; - - tokio::spawn(async move { - let _result = media::spawn_pipeline(&url, &cache, publishers, abort_rx, group_id).await; - // Small delay before sending TrackEnded (avoids tight loop, - // gives Skip time to clear state first). - tokio::time::sleep(Duration::from_millis(200)).await; - let _ = pipeline_tx - .send(RoomCommand::TrackEnded { - item_id: track_item_id, - }) - .await; - }); + self.player.start(&item, self.publishers.clone()); } async fn publish_state_snapshot(&self) { - let current = self.active_track.as_ref().map(|t| TrackState { + let current = self.player.active_track().map(|t| TrackState { id: t.id, title: t.title.clone(), url: t.url.clone(), @@ -615,6 +569,40 @@ mod tests { pool } + /// A mock source that handles all URLs (for tests that need playback). + struct PanaceaSource; + + #[async_trait::async_trait] + impl crate::source::MediaSource for PanaceaSource { + async fn extract( + &self, + url: &str, + _cache_dir: &std::path::Path, + ) -> Result { + Ok(TrackMeta { + title: "test".into(), + duration: "3:45".into(), + thumbnail: None, + url: url.into(), + source: crate::types::SourceKind::Direct, + }) + } + + async fn resolve( + &self, + url: &str, + _cache_dir: &std::path::Path, + ) -> Result { + Ok(format!("{url}/stream")) + } + } + + fn test_registry() -> Arc { + let mut reg = SourceRegistry::new(); + reg.register(Box::new(PanaceaSource)); + Arc::new(reg) + } + #[tokio::test] async fn test_snowflake_uniqueness() { let gen_id = Arc::new(SnowflakeIdGen::new(1)); @@ -642,7 +630,7 @@ mod tests { #[tokio::test] async fn test_create_and_exists() { let pool = test_pool().await; - let reg = Registry::new(pool, PathBuf::from("/tmp/moqbox")); + let reg = Registry::new(pool, PathBuf::from("/tmp/moqbox"), test_registry()); let room = reg.create().await; assert!(reg.exists(&room.id).await); } @@ -650,7 +638,7 @@ mod tests { #[tokio::test] async fn test_push_and_queue() { let pool = test_pool().await; - let reg = Registry::new(pool, PathBuf::from("/tmp/moqbox")); + let reg = Registry::new(pool, PathBuf::from("/tmp/moqbox"), test_registry()); let room = reg.create().await; let meta_a = TrackMeta { @@ -658,14 +646,14 @@ mod tests { duration: "3:45".into(), thumbnail: None, url: "https://example.com/a".into(), - extractor: "test".into(), + source: crate::types::SourceKind::Direct, }; let meta_b = TrackMeta { title: "Track B".into(), duration: "4:20".into(), thumbnail: None, url: "https://example.com/b".into(), - extractor: "test".into(), + source: crate::types::SourceKind::Direct, }; // Push first track — it gets popped and starts playing immediately. @@ -684,7 +672,7 @@ mod tests { #[tokio::test] async fn test_remove() { let pool = test_pool().await; - let reg = Registry::new(pool, PathBuf::from("/tmp/moqbox")); + let reg = Registry::new(pool, PathBuf::from("/tmp/moqbox"), test_registry()); let room = reg.create().await; reg.remove(&room.id).await; @@ -694,7 +682,7 @@ mod tests { #[tokio::test] async fn test_send_chat() { let pool = test_pool().await; - let reg = Registry::new(pool, PathBuf::from("/tmp/moqbox")); + let reg = Registry::new(pool, PathBuf::from("/tmp/moqbox"), test_registry()); let room = reg.create().await; let msg = reg @@ -710,7 +698,7 @@ mod tests { #[tokio::test] async fn test_send_chat_persistence() { let pool = test_pool().await; - let reg = Registry::new(pool, PathBuf::from("/tmp/moqbox")); + let reg = Registry::new(pool, PathBuf::from("/tmp/moqbox"), test_registry()); let room = reg.create().await; let _ = reg @@ -737,7 +725,7 @@ mod tests { #[tokio::test] async fn test_send_chat_empty_content() { let pool = test_pool().await; - let reg = Registry::new(pool, PathBuf::from("/tmp/moqbox")); + let reg = Registry::new(pool, PathBuf::from("/tmp/moqbox"), test_registry()); let room = reg.create().await; let result = reg.send_chat(&room.id, "alice", "", "message").await; @@ -747,7 +735,7 @@ mod tests { #[tokio::test] async fn test_send_chat_long_content() { let pool = test_pool().await; - let reg = Registry::new(pool, PathBuf::from("/tmp/moqbox")); + let reg = Registry::new(pool, PathBuf::from("/tmp/moqbox"), test_registry()); let room = reg.create().await; let long = "a".repeat(2001); @@ -758,7 +746,7 @@ mod tests { #[tokio::test] async fn test_send_chat_long_username() { let pool = test_pool().await; - let reg = Registry::new(pool, PathBuf::from("/tmp/moqbox")); + let reg = Registry::new(pool, PathBuf::from("/tmp/moqbox"), test_registry()); let room = reg.create().await; let long = "a".repeat(33); @@ -769,7 +757,7 @@ mod tests { #[tokio::test] async fn test_register_client() { let pool = test_pool().await; - let reg = Registry::new(pool, PathBuf::from("/tmp/moqbox")); + let reg = Registry::new(pool, PathBuf::from("/tmp/moqbox"), test_registry()); let room = reg.create().await; let id1 = reg.register_client(&room.id).await; @@ -782,7 +770,7 @@ mod tests { #[tokio::test] async fn test_register_client_nonexistent_room() { let pool = test_pool().await; - let reg = Registry::new(pool, PathBuf::from("/tmp/moqbox")); + let reg = Registry::new(pool, PathBuf::from("/tmp/moqbox"), test_registry()); let fake_id = RoomId("nonexistent".into()); assert_eq!(reg.register_client(&fake_id).await, None); } @@ -790,7 +778,7 @@ mod tests { #[tokio::test] async fn test_unregister_client() { let pool = test_pool().await; - let reg = Registry::new(pool, PathBuf::from("/tmp/moqbox")); + let reg = Registry::new(pool, PathBuf::from("/tmp/moqbox"), test_registry()); let room = reg.create().await; // Register and unregister — subsequent register should still work. @@ -804,7 +792,7 @@ mod tests { #[tokio::test] async fn test_skip_when_idle_does_not_panic() { let pool = test_pool().await; - let reg = Registry::new(pool, PathBuf::from("/tmp/moqbox")); + let reg = Registry::new(pool, PathBuf::from("/tmp/moqbox"), test_registry()); let room = reg.create().await; // Skip on an empty, idle room should not panic or deadlock. reg.skip(&room.id).await; @@ -814,7 +802,7 @@ mod tests { #[tokio::test] async fn test_double_skip_does_not_deadlock() { let pool = test_pool().await; - let reg = Registry::new(pool, PathBuf::from("/tmp/moqbox")); + let reg = Registry::new(pool, PathBuf::from("/tmp/moqbox"), test_registry()); let room = reg.create().await; let meta = TrackMeta { @@ -822,7 +810,7 @@ mod tests { duration: "3:45".into(), thumbnail: None, url: "https://example.com/a".into(), - extractor: "test".into(), + source: crate::types::SourceKind::Direct, }; reg.push(&room.id, meta).await; tokio::time::sleep(Duration::from_millis(100)).await; @@ -840,7 +828,7 @@ mod tests { #[tokio::test] async fn test_send_chat_returns_id_zero() { let pool = test_pool().await; - let reg = Registry::new(pool, PathBuf::from("/tmp/moqbox")); + let reg = Registry::new(pool, PathBuf::from("/tmp/moqbox"), test_registry()); let room = reg.create().await; let msg = reg .send_chat(&room.id, "alice", "hello", "message") @@ -853,7 +841,7 @@ mod tests { #[tokio::test] async fn test_sweep_idle_removes_inactive_rooms() { let pool = test_pool().await; - let reg = Registry::new(pool, PathBuf::from("/tmp/moqbox")); + let reg = Registry::new(pool, PathBuf::from("/tmp/moqbox"), test_registry()); let room = reg.create().await; // Tiny sleep so the room has a non-zero idle duration. @@ -871,7 +859,7 @@ mod tests { #[tokio::test] async fn test_sweep_idle_preserves_active_rooms() { let pool = test_pool().await; - let reg = Registry::new(pool, PathBuf::from("/tmp/moqbox")); + let reg = Registry::new(pool, PathBuf::from("/tmp/moqbox"), test_registry()); let room = reg.create().await; // Register a client (triggers state snapshot, updates last_active). @@ -885,7 +873,7 @@ mod tests { #[tokio::test] async fn test_push_starts_playback_when_idle() { let pool = test_pool().await; - let reg = Registry::new(pool, PathBuf::from("/tmp/moqbox")); + let reg = Registry::new(pool, PathBuf::from("/tmp/moqbox"), test_registry()); let room = reg.create().await; let meta = TrackMeta { @@ -893,7 +881,7 @@ mod tests { duration: "3:45".into(), thumbnail: None, url: "https://example.com/t".into(), - extractor: "test".into(), + source: crate::types::SourceKind::Direct, }; reg.push(&room.id, meta).await; @@ -911,7 +899,7 @@ mod tests { #[tokio::test] async fn test_push_appends_when_already_playing() { let pool = test_pool().await; - let reg = Registry::new(pool, PathBuf::from("/tmp/moqbox")); + let reg = Registry::new(pool, PathBuf::from("/tmp/moqbox"), test_registry()); let room = reg.create().await; // First track starts playing immediately. @@ -922,7 +910,7 @@ mod tests { duration: "3:45".into(), thumbnail: None, url: "https://example.com/a".into(), - extractor: "test".into(), + source: crate::types::SourceKind::Direct, }, ) .await; @@ -936,7 +924,7 @@ mod tests { duration: "3:45".into(), thumbnail: None, url: "https://example.com/b".into(), - extractor: "test".into(), + source: crate::types::SourceKind::Direct, }, ) .await; @@ -958,7 +946,7 @@ mod tests { #[tokio::test] async fn test_skip_when_queue_empty_after_track_ends() { let pool = test_pool().await; - let reg = Registry::new(pool, PathBuf::from("/tmp/moqbox")); + let reg = Registry::new(pool, PathBuf::from("/tmp/moqbox"), test_registry()); let room = reg.create().await; // Push a track, let it start playing, then skip. @@ -969,7 +957,7 @@ mod tests { duration: "3:45".into(), thumbnail: None, url: "https://example.com/a".into(), - extractor: "test".into(), + source: crate::types::SourceKind::Direct, }, ) .await; @@ -989,7 +977,7 @@ mod tests { #[tokio::test] async fn test_double_skip_does_not_corrupt_state() { let pool = test_pool().await; - let reg = Registry::new(pool, PathBuf::from("/tmp/moqbox")); + let reg = Registry::new(pool, PathBuf::from("/tmp/moqbox"), test_registry()); let room = reg.create().await; // Push two tracks and play the first. @@ -1000,7 +988,7 @@ mod tests { duration: "3:45".into(), thumbnail: None, url: "https://example.com/a".into(), - extractor: "test".into(), + source: crate::types::SourceKind::Direct, }, ) .await; @@ -1014,7 +1002,7 @@ mod tests { duration: "3:45".into(), thumbnail: None, url: "https://example.com/b".into(), - extractor: "test".into(), + source: crate::types::SourceKind::Direct, }, ) .await; @@ -1040,7 +1028,7 @@ mod tests { #[tokio::test] async fn test_register_client_overflow() { let pool = test_pool().await; - let reg = Registry::new(pool, PathBuf::from("/tmp/moqbox")); + let reg = Registry::new(pool, PathBuf::from("/tmp/moqbox"), test_registry()); let room = reg.create().await; // Register many clients — overflow should not panic. @@ -1058,7 +1046,7 @@ mod tests { #[tokio::test] async fn test_remove_while_playing_does_not_panic() { let pool = test_pool().await; - let reg = Registry::new(pool, PathBuf::from("/tmp/moqbox")); + let reg = Registry::new(pool, PathBuf::from("/tmp/moqbox"), test_registry()); let room = reg.create().await; reg.push( @@ -1068,7 +1056,7 @@ mod tests { duration: "3:45".into(), thumbnail: None, url: "https://example.com/a".into(), - extractor: "test".into(), + source: crate::types::SourceKind::Direct, }, ) .await; @@ -1087,7 +1075,7 @@ mod tests { #[tokio::test] async fn test_sweep_idle_keeps_active_room() { let pool = test_pool().await; - let reg = Registry::new(pool, PathBuf::from("/tmp/moqbox")); + let reg = Registry::new(pool, PathBuf::from("/tmp/moqbox"), test_registry()); let room = reg.create().await; // Register a client (updates last_active via actor while we await). @@ -1105,7 +1093,7 @@ mod tests { #[tokio::test] async fn test_client_count_tracking() { let pool = test_pool().await; - let reg = Registry::new(pool, PathBuf::from("/tmp/moqbox")); + let reg = Registry::new(pool, PathBuf::from("/tmp/moqbox"), test_registry()); let room = reg.create().await; // Register 3 clients. @@ -1127,7 +1115,7 @@ mod tests { #[tokio::test] async fn test_multiple_rooms_dont_interfere() { let pool = test_pool().await; - let reg = Registry::new(pool, PathBuf::from("/tmp/moqbox")); + let reg = Registry::new(pool, PathBuf::from("/tmp/moqbox"), test_registry()); let room_a = reg.create().await; let room_b = reg.create().await; @@ -1142,7 +1130,7 @@ mod tests { duration: "1:00".into(), thumbnail: None, url: "https://example.com/a".into(), - extractor: "test".into(), + source: crate::types::SourceKind::Direct, }, ) .await; @@ -1153,7 +1141,7 @@ mod tests { duration: "2:00".into(), thumbnail: None, url: "https://example.com/b".into(), - extractor: "test".into(), + source: crate::types::SourceKind::Direct, }, ) .await; @@ -1175,7 +1163,7 @@ mod tests { #[tokio::test] async fn test_skip_on_empty_queue_no_crash() { let pool = test_pool().await; - let reg = Registry::new(pool, PathBuf::from("/tmp/moqbox")); + let reg = Registry::new(pool, PathBuf::from("/tmp/moqbox"), test_registry()); let room = reg.create().await; // Skip with no tracks ever queued. reg.skip(&room.id).await; @@ -1186,7 +1174,7 @@ mod tests { #[tokio::test] async fn test_push_after_skip_works() { let pool = test_pool().await; - let reg = Registry::new(pool, PathBuf::from("/tmp/moqbox")); + let reg = Registry::new(pool, PathBuf::from("/tmp/moqbox"), test_registry()); let room = reg.create().await; // Push and skip. @@ -1197,7 +1185,7 @@ mod tests { duration: "3:45".into(), thumbnail: None, url: "https://a".into(), - extractor: "test".into(), + source: crate::types::SourceKind::Direct, }, ) .await; @@ -1213,7 +1201,7 @@ mod tests { duration: "3:45".into(), thumbnail: None, url: "https://b".into(), - extractor: "test".into(), + source: crate::types::SourceKind::Direct, }, ) .await; @@ -1232,7 +1220,7 @@ mod tests { #[tokio::test] async fn test_client_count_does_not_underflow() { let pool = test_pool().await; - let reg = Registry::new(pool, PathBuf::from("/tmp/moqbox")); + let reg = Registry::new(pool, PathBuf::from("/tmp/moqbox"), test_registry()); let room = reg.create().await; // Unregister with no clients registered — should not underflow. @@ -1246,7 +1234,7 @@ mod tests { #[tokio::test] async fn test_multiple_skips_sequential() { let pool = test_pool().await; - let reg = Registry::new(pool, PathBuf::from("/tmp/moqbox")); + let reg = Registry::new(pool, PathBuf::from("/tmp/moqbox"), test_registry()); let room = reg.create().await; // Push 3 tracks, skip through all. @@ -1258,7 +1246,7 @@ mod tests { duration: "1:00".into(), thumbnail: None, url: "https://x".into(), - extractor: "test".into(), + source: crate::types::SourceKind::Direct, }, ) .await; diff --git a/src/source/direct.rs b/src/source/direct.rs new file mode 100644 index 0000000..f135b5e --- /dev/null +++ b/src/source/direct.rs @@ -0,0 +1,275 @@ +//! Direct URL media source implementation. +//! +//! Handles URLs that point directly to media files (`.mp4`, `.webm`, `.mp3`, +//! etc.) without going through yt-dlp. Metadata is extracted heuristically +//! from the URL path — no network requests are made during extraction. + +use std::path::Path; + +use crate::source::{MediaSource, SourceError}; +use crate::types::TrackMeta; + +/// Media file extensions that DirectSource can handle. +const MEDIA_EXTENSIONS: &[&str] = &[ + "mp4", "webm", "mkv", "avi", "mov", "m4v", + "mp3", "ogg", "flac", "wav", "m4a", "aac", "opus", "wma", +]; + +// --------------------------------------------------------------------------- +// Source +// --------------------------------------------------------------------------- + +/// A media source that handles direct media file URLs. +/// +/// Detects media files by URL path extension. If the URL has a recognised +/// media file extension (`mp4`, `webm`, `mp3`, etc.), it is treated as a +/// direct-play URL: +/// +/// - **extract** derives the title from the filename (strips extension, +/// replaces separators with spaces) +/// - **resolve** returns the URL unchanged +/// +/// This source is intended as a fallback after [`YtdlpSource`](super::ytdlp::YtdlpSource) +/// — it only matches URLs with media extensions whose filenames look like +/// media files. +pub(crate) struct DirectSource; + +impl DirectSource { + pub(crate) fn new() -> Self { + Self + } +} + +#[async_trait::async_trait] +impl MediaSource for DirectSource { + async fn extract(&self, url: &str, _cache_dir: &Path) -> Result { + let Some(ext) = extension(url) else { + return Err(SourceError::Unsupported(format!("no file extension in URL: {url}"))); + }; + + if !MEDIA_EXTENSIONS.contains(&ext.as_str()) { + return Err(SourceError::Unsupported(format!( + "unrecognised media extension '.{ext}' in URL", + ))); + } + + let title = extract_title(url, &ext); + + Ok(TrackMeta { + title, + duration: "--:--".into(), + thumbnail: None, + url: url.to_string(), + source: crate::types::SourceKind::Direct, + }) + } + + async fn resolve(&self, url: &str, _cache_dir: &Path) -> Result { + if !url.starts_with("http://") && !url.starts_with("https://") { + return Err(SourceError::Unsupported(format!("not an HTTP(S) URL: {url}"))); + } + + // Direct URLs are already playable as-is. + Ok(url.to_string()) + } +} + +// --------------------------------------------------------------------------- +// Helpers +// --------------------------------------------------------------------------- + +/// Extract the file extension (without dot) from a URL path, if any. +fn extension(url: &str) -> Option { + let path = url.split('?').next().unwrap_or(url); + let path = path.split('#').next().unwrap_or(path); + let dot = path.rfind('.')?; + + // Extension must be after the last path segment. + let after_dot = &path[dot + 1..]; + if after_dot.is_empty() || after_dot.contains('/') { + return None; + } + + Some(after_dot.to_ascii_lowercase()) +} + +/// Derive a human-readable title from a URL path. +/// +/// Strips the extension, takes the last path segment, and replaces common +/// URL separators (`-`, `_`, `+`, `.`) with spaces. +fn extract_title(url: &str, ext: &str) -> String { + let path = url.split('?').next().unwrap_or(url); + let path = path.split('#').next().unwrap_or(path); + + // Take the last non-empty path segment. + let filename = path + .rsplit('/') + .find(|s| !s.is_empty()) + .unwrap_or("untitled"); + + // Strip the extension from the filename. + let dot_len = ext.len() + 1; // +1 for the dot + let stem = if filename.len() >= dot_len && filename.ends_with(&format!(".{ext}")) { + &filename[..filename.len() - dot_len] + } else { + filename + }; + + // Replace URL separators with spaces, then collapse whitespace. + let separators = ['-', '_', '+', '.']; + let result: String = stem + .replace(&separators[..], " ") + .split_whitespace() + .collect::>() + .join(" "); + + if result.is_empty() { "Untitled".into() } else { result } +} + +// --------------------------------------------------------------------------- +// Tests +// --------------------------------------------------------------------------- + +#[cfg(test)] +mod tests { + use super::*; + + // -- Extension detection ------------------------------------------------- + + #[test] + fn test_extension_mp4() { + assert_eq!(extension("https://example.com/video.mp4"), Some("mp4".into())); + } + + #[test] + fn test_extension_with_query() { + assert_eq!( + extension("https://example.com/v.mp4?t=30"), + Some("mp4".into()) + ); + } + + #[test] + fn test_extension_no_ext() { + assert_eq!(extension("https://example.com/video"), None); + } + + #[test] + fn test_extension_empty_path() { + assert_eq!(extension("https://example.com/"), None); + } + + #[test] + fn test_extension_uppercase() { + assert_eq!(extension("https://example.com/video.MP4"), Some("mp4".into())); + } + + // -- Title extraction ---------------------------------------------------- + + #[test] + fn test_title_basic() { + assert_eq!(extract_title("https://example.com/my-video.mp4", "mp4"), "my video"); + } + + #[test] + fn test_title_with_underscores() { + assert_eq!( + extract_title("https://example.com/great_song.mp3", "mp3"), + "great song" + ); + } + + #[test] + fn test_title_with_dots() { + assert_eq!( + extract_title("https://cdn.example.com/file.name.with.dots.mp4", "mp4"), + "file name with dots" + ); + } + + #[test] + fn test_title_no_stem() { + assert_eq!(extract_title("https://example.com/.mp4", "mp4"), "Untitled"); + } + + #[test] + fn test_title_deep_path() { + assert_eq!( + extract_title("https://cdn.example.com/path/to/some/deep-file.mp4", "mp4"), + "deep file" + ); + } + + #[test] + fn test_title_no_extension_match() { + // If the filename doesn't actually end with the given extension + // (shouldn't happen in practice), use the whole filename. + assert_eq!( + extract_title("https://example.com/video.mkv", "mp4"), + "video mkv" + ); + } + + #[test] + fn test_title_multi_space_collapse() { + assert_eq!( + extract_title("https://example.com/a---b.mp4", "mp4"), + "a b" + ); + } + + // -- MediaSource implementation ------------------------------------------ + + #[tokio::test] + async fn test_direct_source_accepts_mp4() { + let src = DirectSource::new(); + let meta = src + .extract("https://example.com/video.mp4", Path::new("/tmp")) + .await + .expect("DirectSource should accept .mp4 URLs"); + assert_eq!(meta.title, "video"); + assert_eq!(meta.source, crate::types::SourceKind::Direct); + assert_eq!(meta.duration, "--:--"); + assert!(meta.thumbnail.is_none()); + } + + #[tokio::test] + async fn test_direct_source_rejects_no_extension() { + let src = DirectSource::new(); + let result = src + .extract("https://example.com/video", Path::new("/tmp")) + .await; + assert!(matches!(result, Err(SourceError::Unsupported(_)))); + } + + #[tokio::test] + async fn test_direct_source_rejects_unknown_extension() { + let src = DirectSource::new(); + let result = src + .extract("https://example.com/file.pdf", Path::new("/tmp")) + .await; + assert!(matches!(result, Err(SourceError::Unsupported(_)))); + } + + #[tokio::test] + async fn test_direct_source_resolve_returns_url() { + let src = DirectSource::new(); + let stream = src + .resolve("https://example.com/video.mp4", Path::new("/tmp")) + .await + .expect("DirectSource should resolve .mp4 URLs"); + assert_eq!(stream, "https://example.com/video.mp4"); + } + + #[tokio::test] + async fn test_direct_source_resolve_rejects_non_http() { + let src = DirectSource::new(); + let result = src + .resolve("ftp://example.com/video.mp4", Path::new("/tmp")) + .await; + assert!(matches!(result, Err(SourceError::Unsupported(_)))); + } + + // Integration test with yt-dlp is in src/tests/ — this file focuses on + // DirectSource unit tests without network dependencies. +} diff --git a/src/source/mod.rs b/src/source/mod.rs new file mode 100644 index 0000000..c1e3ced --- /dev/null +++ b/src/source/mod.rs @@ -0,0 +1,275 @@ +//! Pluggable media source resolution. +//! +//! Defines [`MediaSource`] — the seam between URL-based track selection and +//! playable stream production. Different implementations handle different URL +//! patterns (yt-dlp, direct media URLs, etc.). +//! +//! # Adding a new source +//! +//! 1. Implement [`MediaSource`] for your source type. +//! 2. Register it with [`SourceRegistry::register`]. +//! 3. Sources are tried in registration order — first match wins. + +pub(crate) mod direct; +pub(crate) mod ytdlp; + +use std::path::Path; + +use crate::types::TrackMeta; + +// --------------------------------------------------------------------------- +// Error +// --------------------------------------------------------------------------- + +/// Errors from media source resolution. +#[derive(Debug, thiserror::Error)] +pub(crate) enum SourceError { + /// The source cannot handle this URL (try the next source in the registry). + #[error("unsupported URL: {0}")] + Unsupported(String), + + /// The source recognised the URL but couldn't extract metadata. + #[error("extraction failed: {0}")] + Extraction(String), + + /// The source recognised the URL but couldn't resolve a playable stream. + #[error("resolution failed: {0}")] + Resolution(String), + + /// I/O error during source operations. + #[error("io error: {0}")] + Io(#[from] std::io::Error), +} + +// --------------------------------------------------------------------------- +// Trait +// --------------------------------------------------------------------------- + +/// A media source knows how to extract metadata from a URL and resolve it to a +/// playable stream URL. +#[async_trait::async_trait] +pub(crate) trait MediaSource: Send + Sync { + /// Extract metadata from `url`. + /// + /// Returns [`SourceError::Unsupported`] if this source cannot handle the + /// URL — the registry will try the next source. Any other error is + /// terminal for this URL. + async fn extract(&self, url: &str, cache_dir: &Path) -> Result; + + /// Resolve `url` to a playable stream URL. + /// + /// Returns [`SourceError::Unsupported`] if this source cannot handle the + /// URL — the registry will try the next source. + async fn resolve(&self, url: &str, cache_dir: &Path) -> Result; +} + +// --------------------------------------------------------------------------- +// Registry +// --------------------------------------------------------------------------- + +/// A registry of media sources tried in registration order. +/// +/// When a URL arrives, each source is asked via +/// [`MediaSource::extract`] / [`MediaSource::resolve`]. The first source +/// that does **not** return [`SourceError::Unsupported`] wins. +/// +/// # Example +/// +/// ```ignore +/// let mut registry = SourceRegistry::new(); +/// registry.register(YtdlpSource::new()); +/// registry.register(DirectSource::new()); +/// +/// let meta = registry.extract("https://youtu.be/...", &cache_dir).await?; +/// ``` +pub(crate) struct SourceRegistry { + sources: Vec>, +} + +impl Default for SourceRegistry { + fn default() -> Self { + Self::new() + } +} + +impl SourceRegistry { + pub(crate) fn new() -> Self { + Self { + sources: Vec::new(), + } + } + + /// Register a source. Sources are tried in registration order. + pub(crate) fn register(&mut self, source: Box) { + self.sources.push(source); + } + + /// Extract metadata from a URL by trying each registered source in order. + pub(crate) async fn extract( + &self, + url: &str, + cache_dir: &Path, + ) -> Result { + let mut unsupported_hint = String::new(); + for source in &self.sources { + match source.extract(url, cache_dir).await { + Ok(meta) => return Ok(meta), + Err(SourceError::Unsupported(hint)) => { + unsupported_hint = hint; + continue; + } + Err(e) => return Err(e), + } + } + Err(SourceError::Unsupported(format!( + "no registered source could handle URL: {unsupported_hint}" + ))) + } + + /// Resolve a URL to a playable stream URL by trying each registered source. + pub(crate) async fn resolve( + &self, + url: &str, + cache_dir: &Path, + ) -> Result { + let mut unsupported_hint = String::new(); + for source in &self.sources { + match source.resolve(url, cache_dir).await { + Ok(stream) => return Ok(stream), + Err(SourceError::Unsupported(hint)) => { + unsupported_hint = hint; + continue; + } + Err(e) => return Err(e), + } + } + Err(SourceError::Unsupported(format!( + "no registered source could resolve URL: {unsupported_hint}" + ))) + } +} + +// --------------------------------------------------------------------------- +// Tests +// --------------------------------------------------------------------------- + +#[cfg(test)] +mod tests { + use super::*; + + /// A mock source that handles URLs starting with "mock://". + struct MockSource; + + #[async_trait::async_trait] + impl MediaSource for MockSource { + async fn extract(&self, url: &str, _cache_dir: &Path) -> Result { + if url.starts_with("mock://") { + Ok(TrackMeta { + title: "Mock Track".into(), + duration: "3:00".into(), + thumbnail: None, + url: url.into(), + source: crate::types::SourceKind::Ytdlp, + }) + } else { + Err(SourceError::Unsupported(url.into())) + } + } + + async fn resolve(&self, url: &str, _cache_dir: &Path) -> Result { + if url.starts_with("mock://") { + Ok(format!("{url}/stream")) + } else { + Err(SourceError::Unsupported(url.into())) + } + } + } + + /// A mock source that only matches a specific domain. + struct SpecificSource(&'static str); + + #[async_trait::async_trait] + impl MediaSource for SpecificSource { + async fn extract(&self, url: &str, _cache_dir: &Path) -> Result { + if url.contains(self.0) { + Ok(TrackMeta { + title: format!("From {}", self.0), + duration: "4:20".into(), + thumbnail: None, + url: url.into(), + source: crate::types::SourceKind::Direct, + }) + } else { + Err(SourceError::Unsupported(url.into())) + } + } + + async fn resolve(&self, url: &str, _cache_dir: &Path) -> Result { + if url.contains(self.0) { + Ok(url.into()) + } else { + Err(SourceError::Unsupported(url.into())) + } + } + } + + #[tokio::test] + async fn test_registry_tries_sources_in_order() { + let mut reg = SourceRegistry::new(); + reg.register(Box::new(MockSource)); + + let meta = reg + .extract("mock://example.com/track", Path::new("/tmp")) + .await + .expect("mock source should handle mock://"); + assert_eq!(meta.title, "Mock Track"); + } + + #[tokio::test] + async fn test_registry_unsupported_url() { + let mut reg = SourceRegistry::new(); + reg.register(Box::new(MockSource)); + + let result = reg.extract("unknown://url", Path::new("/tmp")).await; + assert!(matches!(result, Err(SourceError::Unsupported(_)))); + } + + #[tokio::test] + async fn test_registry_resolve_mock_url() { + let mut reg = SourceRegistry::new(); + reg.register(Box::new(MockSource)); + + let stream = reg + .resolve("mock://example.com/track", Path::new("/tmp")) + .await + .expect("mock source should resolve mock://"); + assert_eq!(stream, "mock://example.com/track/stream"); + } + + #[tokio::test] + async fn test_registry_falls_through_sources() { + let mut reg = SourceRegistry::new(); + reg.register(Box::new(SpecificSource("alpha"))); + reg.register(Box::new(SpecificSource("beta"))); + + let meta = reg + .extract("https://beta.com/video", Path::new("/tmp")) + .await + .expect("second source should handle beta URLs"); + assert_eq!(meta.title, "From beta"); + + // First source should still work for its URLs. + let meta = reg + .extract("https://alpha.com/video", Path::new("/tmp")) + .await + .expect("first source should handle alpha URLs"); + assert_eq!(meta.title, "From alpha"); + } + + #[tokio::test] + async fn test_registry_empty() { + let reg = SourceRegistry::new(); + let result = reg.extract("anything", Path::new("/tmp")).await; + assert!(matches!(result, Err(SourceError::Unsupported(_)))); + } +} diff --git a/src/source/ytdlp.rs b/src/source/ytdlp.rs new file mode 100644 index 0000000..e4e2773 --- /dev/null +++ b/src/source/ytdlp.rs @@ -0,0 +1,199 @@ +//! yt-dlp based media source implementation. +//! +//! Handles URLs that yt-dlp recognises (YouTube, SoundCloud, Bandcamp, etc.): +//! - `extract` calls `yt-dlp --dump-json` for metadata +//! - `resolve` calls `yt-dlp -f b -g` for a direct stream URL + +use std::path::Path; + +use serde::Deserialize; +use tokio::process::Command; + +use crate::media; +use crate::source::{MediaSource, SourceError}; +use crate::types::TrackMeta; + +// --------------------------------------------------------------------------- +// yt-dlp JSON output fields +// --------------------------------------------------------------------------- + +#[derive(Debug, Deserialize)] +#[serde(rename_all = "snake_case")] +struct YtdlpMetadata { + title: String, + duration: Option, + thumbnail: Option, + webpage_url: String, +} + +// --------------------------------------------------------------------------- +// Source +// --------------------------------------------------------------------------- + +/// A media source backed by yt-dlp. +/// +/// Handles URLs from YouTube, SoundCloud, Bandcamp, and hundreds of other +/// sites supported by yt-dlp. Requires `yt-dlp` on `$PATH`. +pub(crate) struct YtdlpSource; + +impl YtdlpSource { + pub(crate) fn new() -> Self { + Self + } +} + +#[async_trait::async_trait] +impl MediaSource for YtdlpSource { + async fn extract(&self, url: &str, cache_dir: &Path) -> Result { + if !url.starts_with("http://") && !url.starts_with("https://") { + return Err(SourceError::Unsupported(format!("not an HTTP(S) URL: {url}"))); + } + + let output = Command::new("yt-dlp") + .args([ + "--dump-json", + "--no-playlist", + "--no-warnings", + "--cache-dir", + ]) + .arg(cache_dir) + .arg(url) + .kill_on_drop(true) + .output() + .await + .map_err(|e| { + if e.kind() == std::io::ErrorKind::NotFound { + SourceError::Unsupported("yt-dlp not found on PATH".into()) + } else { + SourceError::Io(e) + } + })?; + + if !output.status.success() { + let stderr = String::from_utf8_lossy(&output.stderr); + if stderr.contains("Unsupported URL") { + return Err(SourceError::Unsupported(stderr.trim().to_string())); + } + return Err(SourceError::Extraction(stderr.trim().to_string())); + } + + let raw: YtdlpMetadata = serde_json::from_slice(&output.stdout) + .map_err(|e| SourceError::Extraction(format!("invalid yt-dlp JSON: {e}")))?; + + let duration = raw + .duration + .map(media::format_duration) + .unwrap_or_else(|| "--:--".into()); + + Ok(TrackMeta { + title: raw.title, + duration, + thumbnail: raw.thumbnail, + url: raw.webpage_url, + source: crate::types::SourceKind::Ytdlp, + }) + } + + async fn resolve(&self, url: &str, cache_dir: &Path) -> Result { + if !url.starts_with("http://") && !url.starts_with("https://") { + return Err(SourceError::Unsupported(format!("not an HTTP(S) URL: {url}"))); + } + + let output = Command::new("yt-dlp") + .args([ + "-f", + "b", + "-g", + "--no-playlist", + "--no-warnings", + "--cache-dir", + ]) + .arg(cache_dir) + .arg(url) + .kill_on_drop(true) + .output() + .await + .map_err(|e| { + if e.kind() == std::io::ErrorKind::NotFound { + SourceError::Unsupported("yt-dlp not found on PATH".into()) + } else { + SourceError::Io(e) + } + })?; + + if !output.status.success() { + let stderr = String::from_utf8_lossy(&output.stderr); + if stderr.contains("Unsupported URL") { + return Err(SourceError::Unsupported(stderr.trim().to_string())); + } + return Err(SourceError::Resolution(stderr.trim().to_string())); + } + + let stream_url = String::from_utf8(output.stdout) + .map_err(|_| SourceError::Resolution("invalid UTF-8 from yt-dlp".into()))? + .trim() + .to_string(); + + if stream_url.is_empty() { + return Err(SourceError::Resolution( + "yt-dlp returned empty stream URL".into(), + )); + } + + Ok(stream_url) + } +} + +// --------------------------------------------------------------------------- +// Tests +// --------------------------------------------------------------------------- + +#[cfg(test)] +mod tests { + use super::*; + use std::path::Path; + + #[tokio::test] + async fn test_extract_rejects_non_http() { + let source = YtdlpSource::new(); + let result = source + .extract("javascript:alert(1)", Path::new("/tmp")) + .await; + assert!( + matches!(result, Err(SourceError::Unsupported(_))), + "non-HTTP URLs should be rejected as unsupported" + ); + } + + #[tokio::test] + async fn test_extract_rejects_empty_url() { + let source = YtdlpSource::new(); + let result = source.extract("", Path::new("/tmp")).await; + assert!( + matches!(result, Err(SourceError::Unsupported(_))), + "empty URLs should be rejected as unsupported" + ); + } + + #[tokio::test] + async fn test_resolve_rejects_non_http() { + let source = YtdlpSource::new(); + let result = source + .resolve("ftp://example.com/video.mp4", Path::new("/tmp")) + .await; + assert!( + matches!(result, Err(SourceError::Unsupported(_))), + "non-HTTP URLs should be rejected as unsupported" + ); + } + + #[tokio::test] + async fn test_resolve_rejects_empty_url() { + let source = YtdlpSource::new(); + let result = source.resolve("", Path::new("/tmp")).await; + assert!( + matches!(result, Err(SourceError::Unsupported(_))), + "empty URLs should be rejected as unsupported" + ); + } +} diff --git a/src/transport.rs b/src/transport.rs index 577d4c3..13706b0 100644 --- a/src/transport.rs +++ b/src/transport.rs @@ -3,14 +3,17 @@ //! Provides varint encoding/decoding, MoQ message types, track publishers //! with broadcast channels, and a WebSocket session handler. +use std::sync::atomic::{AtomicU64, Ordering}; +use std::sync::{Arc, Mutex}; +use std::time::Duration; + use axum::extract::ws::{CloseFrame, Message, WebSocket}; use bytes::Bytes; use futures_util::{SinkExt, StreamExt}; -use std::time::Duration; -use tokio::sync::{broadcast, mpsc}; +use tokio::sync::{broadcast, mpsc, watch}; use crate::room; -use crate::types::{RoomCommand, RoomId, TrackId, TrackPublishers}; +use crate::types::{MoqObject, RoomCommand, RoomId, StateValue, TrackId}; pub(crate) fn encode_varint(value: u64) -> Vec { if value < 64 { @@ -249,6 +252,101 @@ pub(crate) fn room_namespace(room_id: &str) -> String { format!("moqbox/room/{room_id}") } +// --------------------------------------------------------------------------- +// Track publishers +// --------------------------------------------------------------------------- + +/// Per-room MoQ track publishers, shared across all client sessions. +/// +/// Cloning is cheap (all inner channels are `Arc`-based). +#[derive(Clone)] +pub(crate) struct TrackPublishers { + pub(crate) audio: broadcast::Sender, + pub(crate) video: broadcast::Sender, + pub(crate) chat: broadcast::Sender, + pub(crate) state: watch::Sender, + /// Cached fMP4 init segment (ftyp+moov boxes) for late-joining clients. + video_init: Arc>>, // (group_id, payload) + video_init_watch: watch::Sender, + chat_seq: Arc, +} + +impl TrackPublishers { + pub(crate) fn new() -> Self { + let (audio, _) = broadcast::channel(256); + let (video, _) = broadcast::channel(256); + let (chat, _) = broadcast::channel(256); + let (state, _) = watch::channel(StateValue { + payload: Bytes::new(), + seq: 0, + }); + let (video_init_watch, _) = watch::channel(false); + TrackPublishers { + audio, + video, + chat, + state, + video_init: Arc::new(Mutex::new(None)), + video_init_watch, + chat_seq: Arc::new(AtomicU64::new(0)), + } + } + + pub(crate) fn publish_video(&self, group_id: u64, object_id: u64, payload: Bytes) { + let _ = self.video.send(MoqObject { + track_id: TrackId::Video, + group_id, + object_id, + payload, + }); + } + + /// Publish a chat object with an auto-incrementing object_id. + pub(crate) fn publish_chat(&self, group_id: u64, payload: Bytes) { + let object_id = self.chat_seq.fetch_add(1, Ordering::Relaxed); + let _ = self.chat.send(MoqObject { + track_id: TrackId::Chat, + group_id, + object_id, + payload, + }); + } + + /// Publish a state update (latest-only, atomically increments seq). + pub(crate) fn publish_state(&self, payload: Bytes) { + let prev = self.state.borrow().clone(); + let new = StateValue { + payload, + seq: prev.seq + 1, + }; + let _ = self.state.send(new); + } + + /// Cache the fMP4 init segment (ftyp+moov boxes) so late-joining clients + /// can initialise their MediaSource before receiving live fragments. + pub(crate) fn cache_init_segment(&self, payload: Bytes, group_id: u64) { + tracing::debug!(group_id, size = payload.len(), "caching init segment"); + if let Ok(mut guard) = self.video_init.lock() { + *guard = Some((group_id, payload.clone())); + } + self.video_init_watch.send_replace(true); + } + + pub(crate) fn get_init_segment(&self) -> Option<(u64, Bytes)> { + let result = self.video_init.lock().ok().and_then(|g| g.clone()); + if let Some((group_id, ref data)) = result { + tracing::debug!(group_id, size = data.len(), "retrieved cached init segment"); + } else { + tracing::debug!("no cached init segment found"); + } + result + } + + pub(crate) fn video_init_waiter(&self) -> watch::Receiver { + self.video_init_watch.subscribe() + } +} + /// Handle a WebSocket upgrade and run the MoQ session lifecycle. /// /// 1. Sends `ANNOUNCE` for the room namespace. @@ -867,4 +965,75 @@ mod tests { assert_eq!(obj1.payload, Bytes::from("fanout")); assert_eq!(obj2.payload, Bytes::from("fanout")); } + + // ── Init segment cache tests ─────────────────────────────────────────── + + #[test] + fn test_init_cache_and_retrieve() { + let publishers = TrackPublishers::new(); + + assert!( + publishers.get_init_segment().is_none(), + "init should be None before caching" + ); + + let payload = Bytes::from("ftyp-moov-data"); + publishers.cache_init_segment(payload.clone(), 42); + + let retrieved = publishers.get_init_segment(); + assert!(retrieved.is_some(), "init should be Some after caching"); + let (group_id, data) = retrieved.unwrap(); + assert_eq!(group_id, 42, "group_id should match what was cached"); + assert_eq!(data, payload, "payload should match what was cached"); + } + + #[tokio::test] + async fn test_init_watch_signals() { + let publishers = TrackPublishers::new(); + + let mut waiter = publishers.video_init_waiter(); + assert!(!*waiter.borrow(), "initial watch value should be false"); + + let pubs = publishers.clone(); + tokio::spawn(async move { + tokio::time::sleep(std::time::Duration::from_millis(50)).await; + pubs.cache_init_segment(Bytes::from("init-payload"), 1); + }); + + let _ = tokio::time::timeout(std::time::Duration::from_secs(2), waiter.changed()) + .await + .expect("watch should fire within timeout when init is cached"); + + assert!( + *waiter.borrow(), + "watch value should be true after cache_init_segment" + ); + + let retrieved = publishers.get_init_segment(); + assert!( + retrieved.is_some(), + "init should be retrievable after watch fires" + ); + let (group_id, data) = retrieved.unwrap(); + assert_eq!(group_id, 1, "group_id should match"); + assert_eq!(data, Bytes::from("init-payload"), "payload should match"); + } + + #[test] + fn test_cache_overwrite_on_new_group() { + let publishers = TrackPublishers::new(); + + publishers.cache_init_segment(Bytes::from("init-1"), 1); + publishers.cache_init_segment(Bytes::from("init-2"), 2); + + let retrieved = publishers.get_init_segment(); + assert!(retrieved.is_some(), "init should be cached"); + let (group_id, data) = retrieved.unwrap(); + assert_eq!(group_id, 2, "group_id should be from the latest cache call"); + assert_eq!( + data, + Bytes::from("init-2"), + "payload should be from the latest cache call" + ); + } } diff --git a/src/types.rs b/src/types.rs index 2b1edb0..ce01602 100644 --- a/src/types.rs +++ b/src/types.rs @@ -1,17 +1,13 @@ //! Shared types used across moqbox modules. -//! -//! Consolidates room state types, MoQ transport types, and track -//! publishing infrastructure into a single module to avoid circular -//! dependencies and keep the module graph acyclic. use std::fmt; -use std::sync::atomic::{AtomicU64, Ordering}; -use std::sync::{Arc, Mutex}; use bytes::Bytes; use chrono::{DateTime, Utc}; use serde::Serialize; -use tokio::sync::{broadcast, mpsc, oneshot, watch}; +use tokio::sync::{mpsc, oneshot, watch}; + +pub(crate) use crate::transport::TrackPublishers; /// Opaque room identifier, human-readable and URL-safe. #[derive(Debug, Clone, Hash, Eq, PartialEq, Serialize)] @@ -23,14 +19,46 @@ impl fmt::Display for RoomId { } } -/// Metadata for a single queued track, extracted by yt-dlp. +/// Identifies which media source resolved a track. +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize)] +#[serde(rename_all = "snake_case")] +pub(crate) enum SourceKind { + Ytdlp, + Direct, +} + +impl std::fmt::Display for SourceKind { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + match self { + SourceKind::Ytdlp => write!(f, "ytdlp"), + SourceKind::Direct => write!(f, "direct"), + } + } +} + +impl std::str::FromStr for SourceKind { + type Err = String; + fn from_str(s: &str) -> Result { + match s { + "ytdlp" => Ok(SourceKind::Ytdlp), + "direct" => Ok(SourceKind::Direct), + _ => Err(format!("unknown source kind: {s}")), + } + } +} + +/// Metadata for a single queued track. +/// +/// Source-agnostic — produced by any [`MediaSource`](crate::source::MediaSource) +/// implementation (yt-dlp, direct URL detection, etc.). #[derive(Debug, Clone, Serialize)] pub(crate) struct TrackMeta { pub title: String, pub duration: String, pub thumbnail: Option, pub url: String, - pub extractor: String, + /// How this track was resolved (yt-dlp, direct URL, etc.). + pub source: SourceKind, } /// Snapshot of the currently playing track for state publishing. @@ -110,97 +138,6 @@ pub(crate) struct StateValue { pub(crate) seq: u64, } -/// Per-room MoQ track publishers, shared across all client sessions. -/// -/// Cloning is cheap (all inner channels are `Arc`-based). -#[derive(Clone)] -pub(crate) struct TrackPublishers { - pub(crate) audio: broadcast::Sender, - pub(crate) video: broadcast::Sender, - pub(crate) chat: broadcast::Sender, - pub(crate) state: watch::Sender, - /// Cached fMP4 init segment (ftyp+moov boxes) for late-joining clients. - video_init: Arc>>, // (group_id, payload) - video_init_watch: watch::Sender, - chat_seq: Arc, -} - -impl TrackPublishers { - pub(crate) fn new() -> Self { - let (audio, _) = broadcast::channel(256); - let (video, _) = broadcast::channel(256); - let (chat, _) = broadcast::channel(256); - let (state, _) = watch::channel(StateValue { - payload: Bytes::new(), - seq: 0, - }); - let (video_init_watch, _) = watch::channel(false); - TrackPublishers { - audio, - video, - chat, - state, - video_init: Arc::new(Mutex::new(None)), - video_init_watch, - chat_seq: Arc::new(AtomicU64::new(0)), - } - } - - pub(crate) fn publish_video(&self, group_id: u64, object_id: u64, payload: Bytes) { - let _ = self.video.send(MoqObject { - track_id: TrackId::Video, - group_id, - object_id, - payload, - }); - } - - /// Publish a chat object with an auto-incrementing object_id. - pub(crate) fn publish_chat(&self, group_id: u64, payload: Bytes) { - let object_id = self.chat_seq.fetch_add(1, Ordering::Relaxed); - let _ = self.chat.send(MoqObject { - track_id: TrackId::Chat, - group_id, - object_id, - payload, - }); - } - - /// Publish a state update (latest-only, atomically increments seq). - pub(crate) fn publish_state(&self, payload: Bytes) { - let prev = self.state.borrow().clone(); - let new = StateValue { - payload, - seq: prev.seq + 1, - }; - let _ = self.state.send(new); - } - - /// Cache the fMP4 init segment (ftyp+moov boxes) so late-joining clients - /// can initialise their MediaSource before receiving live fragments. - pub(crate) fn cache_init_segment(&self, payload: Bytes, group_id: u64) { - tracing::debug!(group_id, size = payload.len(), "caching init segment"); - if let Ok(mut guard) = self.video_init.lock() { - *guard = Some((group_id, payload.clone())); - } - self.video_init_watch.send_replace(true); - } - - pub(crate) fn get_init_segment(&self) -> Option<(u64, Bytes)> { - let result = self.video_init.lock().ok().and_then(|g| g.clone()); - if let Some((group_id, ref data)) = result { - tracing::debug!(group_id, size = data.len(), "retrieved cached init segment"); - } else { - tracing::debug!("no cached init segment found"); - } - result - } - - pub(crate) fn video_init_waiter(&self) -> watch::Receiver { - self.video_init_watch.subscribe() - } -} - /// All room operations — processed sequentially by the per-room actor. pub(crate) enum RoomCommand { QueueTrack(TrackMeta), @@ -222,6 +159,29 @@ pub(crate) enum RoomCommand { Shutdown, } +/// Ephemeral info about the active track, stored in-memory only. +#[derive(Debug, Clone)] +pub(crate) struct ActiveTrackInfo { + pub(crate) id: i64, + pub(crate) title: String, + pub(crate) url: String, + pub(crate) duration: String, + /// Epoch ms when the track started — for client-side elapsed computation. + pub(crate) started_at_wall: i64, +} + +impl ActiveTrackInfo { + pub(crate) fn from_item(item: &crate::playlist::QueueItem) -> Self { + Self { + id: item.id, + title: item.title.clone(), + url: item.url.clone(), + duration: item.duration.clone(), + started_at_wall: chrono::Utc::now().timestamp_millis(), + } + } +} + /// Public handle to a room — allows sending commands and reading publishers. #[derive(Clone)] pub(crate) struct RoomHandle { @@ -233,7 +193,6 @@ pub(crate) struct RoomHandle { #[cfg(test)] mod tests { use super::*; - use std::time::Duration; #[test] fn test_room_id_display() { @@ -250,86 +209,4 @@ mod tests { assert_eq!(TrackId::from_name("invalid"), None); assert_eq!(TrackId::from_name(""), None); } - - /// Test 1: Verify cache_init_segment stores and get_init_segment retrieves, - /// including correct group_id matching. - #[test] - fn test_init_cache_and_retrieve() { - let publishers = TrackPublishers::new(); - - // Should be None initially - assert!( - publishers.get_init_segment().is_none(), - "init should be None before caching" - ); - - // Cache with group_id=42 - let payload = Bytes::from("ftyp-moov-data"); - publishers.cache_init_segment(payload.clone(), 42); - - // Retrieve and verify - let retrieved = publishers.get_init_segment(); - assert!(retrieved.is_some(), "init should be Some after caching"); - let (group_id, data) = retrieved.unwrap(); - assert_eq!(group_id, 42, "group_id should match what was cached"); - assert_eq!(data, payload, "payload should match what was cached"); - } - - /// Test 2: Verify that the video_init_watch fires when cache_init_segment is called, - /// so a task waiting on it can proceed. - #[tokio::test] - async fn test_init_watch_signals() { - let publishers = TrackPublishers::new(); - - // Get a waiter before the init is cached - let mut waiter = publishers.video_init_waiter(); - assert!(!*waiter.borrow(), "initial watch value should be false"); - - // Spawn a task that caches init after a short delay - let pubs = publishers.clone(); - tokio::spawn(async move { - tokio::time::sleep(Duration::from_millis(50)).await; - pubs.cache_init_segment(Bytes::from("init-payload"), 1); - }); - - // Wait for the watch to fire (with timeout) - let _ = tokio::time::timeout(Duration::from_secs(2), waiter.changed()) - .await - .expect("watch should fire within timeout when init is cached"); - - // After the watch fires, the value should be true - assert!( - *waiter.borrow(), - "watch value should be true after cache_init_segment" - ); - - // And the init should be retrievable - let retrieved = publishers.get_init_segment(); - assert!( - retrieved.is_some(), - "init should be retrievable after watch fires" - ); - let (group_id, data) = retrieved.unwrap(); - assert_eq!(group_id, 1, "group_id should match"); - assert_eq!(data, Bytes::from("init-payload"), "payload should match"); - } - - /// Verify caching a new init segment for a new group overwrites the old one. - #[test] - fn test_cache_overwrite_on_new_group() { - let publishers = TrackPublishers::new(); - - publishers.cache_init_segment(Bytes::from("init-1"), 1); - publishers.cache_init_segment(Bytes::from("init-2"), 2); - - let retrieved = publishers.get_init_segment(); - assert!(retrieved.is_some(), "init should be cached"); - let (group_id, data) = retrieved.unwrap(); - assert_eq!(group_id, 2, "group_id should be from the latest cache call"); - assert_eq!( - data, - Bytes::from("init-2"), - "payload should be from the latest cache call" - ); - } } diff --git a/src/web.rs b/src/web.rs index b9ad477..f9ff39f 100644 --- a/src/web.rs +++ b/src/web.rs @@ -16,8 +16,8 @@ use axum::{ use sqlx::{Pool, Sqlite}; use tokio::sync::Mutex; -use crate::media; use crate::room; +use crate::source::SourceRegistry; use crate::types::{RoomId, TrackMeta}; /// Per-IP rate limiter for POST /api/ingest (max 1 req/5s per IP). @@ -58,10 +58,16 @@ pub(crate) struct AppState { pub rooms: room::Registry, pub pool: Pool, pub cache_dir: PathBuf, + pub sources: Arc, pub rate_limiter: RateLimiter, } -pub(crate) fn router(rooms: room::Registry, pool: Pool, cache_dir: PathBuf) -> Router { +pub(crate) fn router( + rooms: room::Registry, + pool: Pool, + cache_dir: PathBuf, + sources: Arc, +) -> Router { let rate_limiter = RateLimiter::new(); // Periodic cleanup for rate limiter entries (every 30s). @@ -78,6 +84,7 @@ pub(crate) fn router(rooms: room::Registry, pool: Pool, cache_dir: PathB rooms, pool, cache_dir, + sources, rate_limiter, }; @@ -141,7 +148,13 @@ async fn room_view( duration: i.duration, thumbnail: i.thumbnail, url: i.url, - extractor: i.extractor, + source: i + .source + .parse() + .unwrap_or_else(|s: String| { + tracing::warn!(source = %s, "unknown source kind in queue_items"); + crate::types::SourceKind::Direct + }), }) .collect::>() }; @@ -196,16 +209,13 @@ async fn ingest_url( } // Rate limiting — max 1 request per 5 seconds per IP. - match state.rate_limiter.check(addr).await { - Err(retry_after) => { - let mut res = err_response(StatusCode::TOO_MANY_REQUESTS, "rate limited"); - res.headers_mut().insert( - axum::http::header::RETRY_AFTER, - retry_after.as_secs().to_string().parse().unwrap(), - ); - return Err(res); - } - Ok(()) => {} + if let Err(retry_after) = state.rate_limiter.check(addr).await { + let mut res = err_response(StatusCode::TOO_MANY_REQUESTS, "rate limited"); + res.headers_mut().insert( + axum::http::header::RETRY_AFTER, + retry_after.as_secs().to_string().parse().expect("valid retry-after seconds"), + ); + return Err(res); } let room_id = RoomId(req.room_id); @@ -215,8 +225,10 @@ async fn ingest_url( return Err(err_response(StatusCode::NOT_FOUND, "room not found")); } - // Run yt-dlp extraction. - let meta = media::extract(&req.url, &state.cache_dir) + // Extract metadata via the source registry. + let meta = state + .sources + .extract(&req.url, &state.cache_dir) .await .map_err(|e| err_response(StatusCode::BAD_REQUEST, &e.to_string()))?; diff --git a/templates/room.html b/templates/room.html index a6ab924..6d162ae 100644 --- a/templates/room.html +++ b/templates/room.html @@ -193,6 +193,12 @@ button:active { transform: scale(0.96); } + button:disabled { + opacity: 0.5; + cursor: not-allowed; + transform: none; + filter: none; + } button.btn-primary { background: var(--green); color: var(--crust); @@ -274,16 +280,38 @@ color: var(--blue); text-decoration: none; font-size: 0.8rem; - padding: 2px 8px; + padding: 6px 10px; border-radius: var(--radius-sm); background: color-mix(in srgb, var(--blue) 15%, transparent); transition: background 150ms ease-out; flex-shrink: 0; + min-height: 40px; + display: inline-flex; + align-items: center; } .source-link:hover { background: color-mix(in srgb, var(--blue) 25%, transparent); } + .history-item { + opacity: 0.7; + } + + /* Connection status */ + .conn-status { + font-size: 0.75rem; + color: var(--overlay0); + margin-bottom: 0.5rem; + text-align: right; + transition: color 300ms ease-out; + } + .conn-status.connected { + color: var(--green); + } + .conn-status.disconnected { + color: var(--red); + } + /* History header */ .history-header { margin-top: 1rem; @@ -295,11 +323,42 @@ } /* Empty state */ - .empty { - color: var(--overlay0); + .empty-state { + display: flex; + flex-direction: column; + align-items: center; + justify-content: center; + gap: 0.75rem; + padding: 3rem 1.5rem; text-align: center; - padding: 3rem 1rem; - font-size: 0.9375rem; + color: var(--overlay1); + font-size: 0.875rem; + line-height: 1.5; + background: var(--surface0); + border-radius: var(--radius-lg); + border: 1px dashed var(--surface2); + margin-bottom: 0.5rem; + } + .empty-state .icon { + font-size: 1.75rem; + line-height: 1; + opacity: 0.5; + } + .empty-state .label { + color: var(--subtext0); + font-weight: 500; + } + .empty-state .hint { + color: var(--overlay0); + font-size: 0.8125rem; + max-width: 24em; + } + + /* Now-playing empty */ + #now-playing.empty { + color: var(--overlay0); + font-size: 0.875rem; + font-style: italic; } /* Chat */ @@ -477,8 +536,10 @@
{% if queue_empty %} -
- Queue is empty. Paste a link above to get started. +
+
🎵
+
Queue is empty
+
Paste a link from YouTube, SoundCloud, or Bandcamp above to get started.
{% else %}
    @@ -509,7 +570,7 @@
    Played
      {% for item in history %} -
    • +
    • {% if let Some(thumb) = item.thumbnail %} {% else %} @@ -532,6 +593,7 @@
+
disconnected
{ - console.log( - "MediaSource sourceopen fired, readyState:", - mediaSource.readyState, - ); + debug("MediaSource sourceopen"); try { sourceBuffer = mediaSource.addSourceBuffer( 'video/mp4; codecs="avc1.4d0028,mp4a.40.2"', ); sourceBuffer.mode = "segments"; - console.log( - "SourceBuffer created, mode:", - sourceBuffer.mode, - "updating:", - sourceBuffer.updating, - ); sourceBuffer.addEventListener("updateend", () => { - console.log( - "SourceBuffer updateend, pendingBoxes.length:", - pendingBoxes.length, - ); if (pendingBoxes.length > 0) { const next = pendingBoxes.shift(); try { @@ -681,31 +728,15 @@ console.warn("SourceBuffer append failed:", e); } } - // Late-join seek: after first media segment is appended, - // seek to the start of buffered data to get out of the - // t=0 stall. The periodic sync in updateNowPlaying handles - // drift correction to the live position. if (!seekInitDone) { const v = document.getElementById("video-player"); if (v && v.buffered.length > 0) { seekInitDone = true; v.currentTime = v.buffered.start(0); - console.log( - "Initial seek to:", - v.currentTime, - "buffered:", - v.buffered.start(0), - "-", - v.buffered.end(0), - ); } } }); - // Flush any boxes that arrived before sourceopen. - console.log( - "sourceopen flush: pendingBoxes.length before flush:", - pendingBoxes.length, - ); + // Flush boxes that arrived before sourceopen. while (pendingBoxes.length > 0) { const box = pendingBoxes.shift(); if (!sourceBuffer.updating) { @@ -729,21 +760,8 @@ function appendVideo(data) { if (sourceBuffer && !sourceBuffer.updating) { - console.log( - "appendVideo: appending to sourceBuffer, size:", - data.length, - ); sourceBuffer.appendBuffer(data); } else { - // Buffer if sourceBuffer is still null or busy updating. - console.log( - "appendVideo: buffering to pendingBoxes, size:", - data.length, - "sourceBuffer:", - !!sourceBuffer, - "updating:", - sourceBuffer ? sourceBuffer.updating : "N/A", - ); pendingBoxes.push(data); } } @@ -775,6 +793,13 @@ ); } + function updateConnStatus(connected) { + const el = document.getElementById("conn-status"); + if (!el) return; + el.className = "conn-status " + (connected ? "connected" : "disconnected"); + el.textContent = connected ? "connected" : "disconnected"; + } + function connect() { ws = new WebSocket( proto + @@ -788,11 +813,9 @@ ws.binaryType = "arraybuffer"; ws.onopen = () => { - console.log( - "WS onopen: resetting awaitingInit=true, clearing pendingBoxes", - ); + debug("WS connected"); + updateConnStatus(true); reconnectDelay = 1000; - // Full reset on reconnect — ensures init is re-sent and accepted. awaitingInit = true; pendingBoxes.length = 0; currentTrackId = null; @@ -811,14 +834,6 @@ const [objectId, o4] = decVarint(buf, o3); if (trackId === 1) { const videoData = new Uint8Array(buf.slice(o4)); - console.log( - "Video obj received, size:", - videoData.length, - "awaitingInit:", - awaitingInit, - "pendingBoxes:", - pendingBoxes.length, - ); if (awaitingInit) { const isInit = @@ -830,22 +845,8 @@ videoData[7], ) === "ftyp"; - console.log( - "awaitingInit check: isInit:", - isInit, - "first 4 bytes:", - String.fromCharCode( - videoData[4], - videoData[5], - videoData[6], - videoData[7], - ), - ); - if (isInit) { - console.log( - "INIT RECEIVED — setting awaitingInit=false, clearing pendingBoxes", - ); + debug("init segment received"); awaitingInit = false; pendingBoxes.length = 0; if (sourceBuffer && !sourceBuffer.updating) { @@ -860,11 +861,6 @@ } catch (_) {} } } else { - // Buffer until init arrives. - console.log( - "Buffering non-init data, pendingBoxes now:", - pendingBoxes.length + 1, - ); pendingBoxes.push(videoData); off = buf.length; return; @@ -891,12 +887,10 @@ off = buf.length; } } else if (msgType === 0x02) { - // SUBSCRIBE_OK — just advance past it. const [ns, o2] = decString(buf, o1); const [tr, o3] = decString(buf, o2); off = o3; } else if (msgType === 0x06) { - // SUBSCRIBE_RST — advance past it. const [ns, o2] = decString(buf, o1); const [tr, o3] = decString(buf, o2); const [reason, o4] = decString(buf, o3); @@ -908,7 +902,7 @@ }; ws.onclose = () => { - console.log("WS closed, reconnecting in " + reconnectDelay + "ms"); + updateConnStatus(false); setTimeout(() => { reconnectDelay = Math.min(reconnectDelay * 2, 30000); connect(); @@ -972,25 +966,11 @@ // Detect track change — set awaitingInit so the next video object // is treated as the new init segment. if (currentTrackId !== null && currentTrackId !== trackId) { - console.log( - "Track change detected:", - currentTrackId, - "->", - trackId, - "setting awaitingInit=true", - ); + debug("track change", currentTrackId, "->", trackId); awaitingInit = true; pendingBoxes.length = 0; seekInitDone = false; } - console.log( - "updateNowPlaying: trackId=", - trackId, - "currentTrackId=", - currentTrackId, - "awaitingInit=", - awaitingInit, - ); currentTrackId = trackId; const startedAt = state.current_track.started_at; @@ -1038,7 +1018,7 @@ } else { currentTrackId = null; el.innerHTML = - 'Waiting for tracks...'; + '
◔
Waiting for tracks
Paste a link above or wait for the next track to start.
'; } // Update queue section from state. @@ -1076,10 +1056,10 @@ qSection.innerHTML = html; } else if (state.current_track) { qSection.innerHTML = - '
No upcoming tracks. Add another link to the queue.
'; + '
▶
No upcoming tracks
Add another link to keep the queue going.
'; } else { qSection.innerHTML = - '
Queue is empty. Paste a link above to get started.
'; + '
🎵
Queue is empty
Paste a link from YouTube, SoundCloud, or Bandcamp above to get started.
'; } if (hSection && state.history && state.history.length > 0) { @@ -1115,7 +1095,7 @@ function formatTime(secs) { const m = Math.floor(secs / 60); const s = Math.floor(secs % 60); - return m + ":" + (s < 10 ? "0" : "") + s; + return m + ":" + String(s).padStart(2, "0"); } function esc(s) { @@ -1158,17 +1138,25 @@ // ── Ingest helper ─────────────────────────────────────────────── async function ingest() { const input = document.getElementById("url-input"); - const resp = await fetch("/api/ingest", { - method: "POST", - headers: { "Content-Type": "application/json" }, - body: JSON.stringify({ url: input.value, room_id: roomId }), - }); - if (resp.ok) { - input.value = ""; - // Queue will update via state track broadcast — no reload needed. - } else { - const err = await resp.json(); - alert(err.error || "ingest failed"); + const btn = document.querySelector(".input-row .btn-primary"); + if (!input.value.trim()) return; + btn.disabled = true; + btn.textContent = "Adding…"; + try { + const resp = await fetch("/api/ingest", { + method: "POST", + headers: { "Content-Type": "application/json" }, + body: JSON.stringify({ url: input.value, room_id: roomId }), + }); + if (resp.ok) { + input.value = ""; + } else { + const err = await resp.json(); + alert(err.error || "ingest failed"); + } + } finally { + btn.disabled = false; + btn.textContent = "Add"; } }