diff --git a/Cargo.lock b/Cargo.lock index 034166a..fd8f574 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1767,6 +1767,29 @@ version = "0.2.3" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "b4596b6d070b27117e987119b4dac604f3c58cfb0b191112e24771b2faeac1a6" +[[package]] +name = "playout" +version = "0.1.0" +dependencies = [ + "anyhow", + "async-trait", + "axum", + "bytes", + "chrono", + "domain", + "dotenvy", + "futures", + "serde", + "serde_json", + "thiserror", + "tokio", + "tower", + "tower-http", + "tracing", + "tracing-subscriber", + "uuid", +] + [[package]] name = "potential_utf" version = "0.1.5" diff --git a/Cargo.toml b/Cargo.toml index 40654ab..5ed400a 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -1,5 +1,5 @@ [workspace] -members = ["crates/domain", "crates/application", "crates/api-types", "crates/infra-wiring", "crates/adapters/adapter-common", "crates/adapters/sqlite", "crates/adapters/auth", "crates/adapters/jellyfin", "crates/adapters/local-files", "crates/adapters/event-publisher", "crates/presentation", "crates/worker", "crates/mcp"] +members = ["crates/domain", "crates/application", "crates/api-types", "crates/infra-wiring", "crates/adapters/adapter-common", "crates/adapters/sqlite", "crates/adapters/auth", "crates/adapters/jellyfin", "crates/adapters/local-files", "crates/adapters/event-publisher", "crates/presentation", "crates/worker", "crates/mcp", "crates/playout"] exclude = ["k-tv-backend", "k-tv-frontend"] resolver = "2" diff --git a/crates/playout/Cargo.toml b/crates/playout/Cargo.toml new file mode 100644 index 0000000..b9ca7c2 --- /dev/null +++ b/crates/playout/Cargo.toml @@ -0,0 +1,33 @@ +[package] +name = "playout" +version = "0.1.0" +edition = "2024" + +[[bin]] +name = "k-tv-playout" +path = "src/main.rs" + +[dependencies] +domain = { workspace = true } + +axum = { workspace = true } +tower = { workspace = true } +tower-http = { workspace = true } +tokio = { workspace = true } +tracing = { workspace = true } +tracing-subscriber = { workspace = true } + +async-trait = { workspace = true } +chrono = { workspace = true } +serde = { workspace = true } +serde_json = { workspace = true } +uuid = { workspace = true } +thiserror = { workspace = true } + +anyhow = "1" +dotenvy = "0.15" +bytes = "1" +futures = "0.3" + +[dev-dependencies] +tokio = { workspace = true, features = ["test-util"] } diff --git a/crates/playout/src/config.rs b/crates/playout/src/config.rs new file mode 100644 index 0000000..cfb8f50 --- /dev/null +++ b/crates/playout/src/config.rs @@ -0,0 +1,34 @@ +use std::path::PathBuf; + +#[derive(Debug, Clone)] +pub struct PlayoutConfig { + pub listen_addr: String, + pub segment_duration_secs: u32, + pub window_size: usize, + pub storage_path: PathBuf, + pub tick_interval_ms: u64, +} + +impl PlayoutConfig { + pub fn from_env() -> Self { + Self { + listen_addr: std::env::var("PLAYOUT_LISTEN_ADDR") + .unwrap_or_else(|_| "0.0.0.0:9090".into()), + segment_duration_secs: std::env::var("PLAYOUT_SEGMENT_DURATION") + .ok() + .and_then(|v| v.parse().ok()) + .unwrap_or(6), + window_size: std::env::var("PLAYOUT_WINDOW_SIZE") + .ok() + .and_then(|v| v.parse().ok()) + .unwrap_or(10), + storage_path: std::env::var("PLAYOUT_STORAGE_PATH") + .map(PathBuf::from) + .unwrap_or_else(|_| PathBuf::from("/tmp/k-tv-playout")), + tick_interval_ms: std::env::var("PLAYOUT_TICK_INTERVAL_MS") + .ok() + .and_then(|v| v.parse().ok()) + .unwrap_or(1000), + } + } +} diff --git a/crates/playout/src/engine.rs b/crates/playout/src/engine.rs new file mode 100644 index 0000000..1f59971 --- /dev/null +++ b/crates/playout/src/engine.rs @@ -0,0 +1,263 @@ +use std::collections::HashMap; +use std::path::PathBuf; +use std::sync::Arc; + +use chrono::Utc; +use domain::models::{CurrentBroadcast, GeneratedSchedule}; +use domain::ports::ScheduleQuery; +use domain::value_objects::{ChannelId, SourceUri}; +use domain::{ScheduleEngineService, SlotId}; +use serde::Serialize; +use tokio::sync::RwLock; +use tracing::{error, info, warn}; + +use crate::config::PlayoutConfig; +use crate::ffmpeg::{FfmpegConfig, FfmpegHandle}; +use crate::segment_store::SegmentStore; + +pub trait SourceUriResolver: Send + Sync { + fn resolve(&self, provider_id: &str, external_id: &str) -> Option; +} + +struct ChannelState { + ffmpeg: FfmpegHandle, + current_slot_id: SlotId, + schedule_id: domain::value_objects::ScheduleId, +} + +#[derive(Debug, Clone, Serialize)] +pub struct ChannelStatus { + pub channel_id: String, + pub current_slot_id: String, + pub schedule_id: String, + pub ffmpeg_running: bool, +} + +pub struct PlayoutEngine { + config: PlayoutConfig, + schedule_query: Arc, + source_resolver: Arc, + store: Arc, + channels: Arc>>, +} + +impl PlayoutEngine { + pub fn new( + config: PlayoutConfig, + schedule_query: Arc, + source_resolver: Arc, + store: Arc, + ) -> Self { + Self { + config, + schedule_query, + source_resolver, + store, + channels: Arc::new(RwLock::new(HashMap::new())), + } + } + + pub async fn add_channel(&self, channel_id: ChannelId) -> anyhow::Result<()> { + { + let channels = self.channels.read().await; + if channels.contains_key(&channel_id) { + return Err(anyhow::anyhow!("channel {channel_id} already active")); + } + } + + let now = Utc::now(); + let schedule = self + .schedule_query + .find_active(channel_id, now) + .await + .map_err(|e| anyhow::anyhow!("{e}"))? + .ok_or_else(|| anyhow::anyhow!("no active schedule for {channel_id}"))?; + + let broadcast = ScheduleEngineService::get_current_broadcast(&schedule, now) + .ok_or_else(|| anyhow::anyhow!("no current slot for {channel_id}"))?; + + let handle = self.start_slot(channel_id, &broadcast, &schedule)?; + + let state = ChannelState { + ffmpeg: handle, + current_slot_id: broadcast.slot().id(), + schedule_id: schedule.id(), + }; + + self.channels.write().await.insert(channel_id, state); + info!(%channel_id, "channel added"); + Ok(()) + } + + pub async fn remove_channel(&self, channel_id: ChannelId) { + if let Some(mut state) = self.channels.write().await.remove(&channel_id) { + state.ffmpeg.stop().await; + info!(%channel_id, "channel removed"); + } + } + + pub async fn active_channels(&self) -> Vec { + self.channels.read().await.keys().copied().collect() + } + + pub async fn channel_statuses(&self) -> Vec { + let mut channels = self.channels.write().await; + channels + .iter_mut() + .map(|(id, state)| ChannelStatus { + channel_id: id.to_string(), + current_slot_id: state.current_slot_id.to_string(), + schedule_id: state.schedule_id.to_string(), + ffmpeg_running: state.ffmpeg.is_running(), + }) + .collect() + } + + pub async fn tick(&self) { + let now = Utc::now(); + let channel_ids: Vec = self.channels.read().await.keys().copied().collect(); + + let futures: Vec<_> = channel_ids + .into_iter() + .map(|id| self.tick_channel(id, now)) + .collect(); + + let results = futures::future::join_all(futures).await; + for result in results { + if let Err(e) = result { + warn!(%e, "channel tick failed"); + } + } + } + + async fn tick_channel( + &self, + channel_id: ChannelId, + now: chrono::DateTime, + ) -> anyhow::Result<()> { + let schedule = match self.schedule_query.find_active(channel_id, now).await { + Ok(Some(s)) => s, + Ok(None) => { + warn!(%channel_id, "no active schedule, removing"); + self.remove_channel(channel_id).await; + return Ok(()); + } + Err(e) => return Err(anyhow::anyhow!("{e}")), + }; + + let broadcast = match ScheduleEngineService::get_current_broadcast(&schedule, now) { + Some(b) => b, + None => return Ok(()), + }; + + let current_slot_id = broadcast.slot().id(); + + let needs_transition = { + let channels = self.channels.read().await; + channels + .get(&channel_id) + .map(|s| s.current_slot_id != current_slot_id || s.schedule_id != schedule.id()) + .unwrap_or(false) + }; + + let ffmpeg_dead = { + let mut channels = self.channels.write().await; + channels + .get_mut(&channel_id) + .map(|s| !s.ffmpeg.is_running()) + .unwrap_or(false) + }; + + if needs_transition || ffmpeg_dead { + if needs_transition { + info!(%channel_id, %current_slot_id, "slot transition"); + } else { + warn!(%channel_id, "ffmpeg process died, restarting"); + } + + let mut channels = self.channels.write().await; + if let Some(state) = channels.get_mut(&channel_id) { + state.ffmpeg.stop().await; + state.ffmpeg = self.start_slot(channel_id, &broadcast, &schedule)?; + state.current_slot_id = current_slot_id; + state.schedule_id = schedule.id(); + } + } + + self.cleanup_old_segments(channel_id).await; + + Ok(()) + } + + fn start_slot( + &self, + channel_id: ChannelId, + broadcast: &CurrentBroadcast, + _schedule: &GeneratedSchedule, + ) -> anyhow::Result { + let slot = broadcast.slot(); + let item = slot.item(); + + let source_uri = self + .source_resolver + .resolve(item.provider_id(), item.external_id()) + .ok_or_else(|| { + anyhow::anyhow!( + "no source uri for {}::{}", + item.provider_id(), + item.external_id() + ) + })?; + + let uri_str = match &source_uri { + SourceUri::NetworkUrl { url } => url.clone(), + SourceUri::FilePath { path } => path.clone(), + }; + + let output_dir = self.output_dir(channel_id); + + let config = FfmpegConfig { + source_uri: uri_str, + start_offset_secs: broadcast.offset_secs(), + segment_duration_secs: self.config.segment_duration_secs, + output_dir, + segment_prefix: "seg".into(), + }; + + let handle = FfmpegHandle::spawn(config, self.store.clone(), channel_id.to_string()); + + Ok(handle) + } + + fn output_dir(&self, channel_id: ChannelId) -> PathBuf { + self.config.storage_path.join(channel_id.to_string()) + } + + async fn cleanup_old_segments(&self, channel_id: ChannelId) { + let channel_str = channel_id.to_string(); + let segments = match self.store.list_segments(&channel_str).await { + Ok(s) => s, + Err(_) => return, + }; + + if segments.len() <= self.config.window_size { + return; + } + + let to_delete = segments.len() - self.config.window_size; + for name in segments.iter().take(to_delete) { + if let Err(e) = self.store.delete_segment(&channel_str, name).await { + error!(%e, segment = %name, "failed to delete old segment"); + } + } + } + + pub async fn shutdown(&self) { + let mut channels = self.channels.write().await; + for (id, state) in channels.iter_mut() { + info!(%id, "stopping channel"); + state.ffmpeg.stop().await; + } + channels.clear(); + } +} diff --git a/crates/playout/src/ffmpeg.rs b/crates/playout/src/ffmpeg.rs new file mode 100644 index 0000000..fa70c18 --- /dev/null +++ b/crates/playout/src/ffmpeg.rs @@ -0,0 +1,161 @@ +use std::path::PathBuf; +use std::sync::Arc; + +use bytes::Bytes; +use tokio::process::{Child, Command}; +use tokio::sync::Notify; +use tracing::{error, info, warn}; + +use crate::segment_store::SegmentStore; + +pub struct FfmpegConfig { + pub source_uri: String, + pub start_offset_secs: u32, + pub segment_duration_secs: u32, + pub output_dir: PathBuf, + pub segment_prefix: String, +} + +pub struct FfmpegHandle { + child: Option, + stop: Arc, +} + +impl FfmpegHandle { + pub fn spawn( + config: FfmpegConfig, + store: Arc, + channel_id: String, + ) -> Self { + let stop = Arc::new(Notify::new()); + let stop_clone = stop.clone(); + + let child = match Self::start_ffmpeg(&config) { + Ok(child) => { + let output_dir = config.output_dir.clone(); + tokio::spawn(Self::ingest_loop( + store, + channel_id, + output_dir, + stop_clone, + )); + Some(child) + } + Err(e) => { + error!(%e, "failed to start ffmpeg"); + None + } + }; + + Self { + child, + stop, + } + } + + fn start_ffmpeg(config: &FfmpegConfig) -> std::io::Result { + std::fs::create_dir_all(&config.output_dir)?; + + let mut cmd = Command::new("ffmpeg"); + cmd.args(["-re"]) + .args(["-ss", &config.start_offset_secs.to_string()]) + .args(["-i", &config.source_uri]) + .args(["-map", "0:v?"]) + .args(["-map", "0:a?"]) + .args(["-map", "0:s?"]) + .args(["-c:v", "copy"]) + .args(["-c:a", "aac"]) + .args(["-c:s", "webvtt"]) + .args(["-f", "hls"]) + .args([ + "-hls_time", + &config.segment_duration_secs.to_string(), + ]) + .args(["-hls_list_size", "0"]) + .args(["-hls_flags", "independent_segments"]) + .args([ + "-hls_segment_filename", + &config + .output_dir + .join(format!("{}%05d.ts", config.segment_prefix)) + .to_string_lossy(), + ]) + .arg( + config + .output_dir + .join("live.m3u8") + .to_string_lossy() + .to_string(), + ) + .stdin(std::process::Stdio::null()) + .stdout(std::process::Stdio::null()) + .stderr(std::process::Stdio::piped()) + .kill_on_drop(true); + + info!(source = %config.source_uri, "starting ffmpeg"); + cmd.spawn() + } + + async fn ingest_loop( + store: Arc, + channel_id: String, + output_dir: PathBuf, + stop: Arc, + ) { + let mut interval = tokio::time::interval(tokio::time::Duration::from_millis(500)); + let mut known_segments: std::collections::HashSet = + std::collections::HashSet::new(); + + loop { + tokio::select! { + _ = stop.notified() => break, + _ = interval.tick() => {} + } + + let entries = match tokio::fs::read_dir(&output_dir).await { + Ok(e) => e, + Err(_) => continue, + }; + + let mut entries = entries; + while let Ok(Some(entry)) = entries.next_entry().await { + let name = entry.file_name().to_string_lossy().to_string(); + if !name.ends_with(".ts") || known_segments.contains(&name) { + continue; + } + + match tokio::fs::read(entry.path()).await { + Ok(data) => { + if let Err(e) = store + .write_segment(&channel_id, &name, Bytes::from(data)) + .await + { + warn!(%e, segment = %name, "failed to write segment to store"); + } else { + known_segments.insert(name); + } + } + Err(e) => { + warn!(%e, segment = %name, "failed to read segment file"); + } + } + } + } + } + + pub async fn stop(&mut self) { + self.stop.notify_one(); + if let Some(ref mut child) = self.child { + let _ = child.kill().await; + let _ = child.wait().await; + } + self.child = None; + } + + pub fn is_running(&mut self) -> bool { + match &mut self.child { + Some(child) => child.try_wait().ok().flatten().is_none(), + None => false, + } + } +} diff --git a/crates/playout/src/http.rs b/crates/playout/src/http.rs new file mode 100644 index 0000000..91f9301 --- /dev/null +++ b/crates/playout/src/http.rs @@ -0,0 +1,120 @@ +use std::sync::Arc; + +use axum::extract::{Path, State}; +use axum::http::{HeaderValue, StatusCode, header}; +use axum::response::{IntoResponse, Response}; +use axum::routing::{delete, get, post}; +use axum::Router; +use domain::value_objects::ChannelId; + +use crate::config::PlayoutConfig; +use crate::engine::PlayoutEngine; +use crate::playlist::generate_m3u8; +use crate::segment_store::SegmentStore; + +#[derive(Clone)] +pub struct AppState { + pub engine: Arc, + pub store: Arc, + pub config: PlayoutConfig, +} + +pub fn router(state: AppState) -> Router { + Router::new() + .route( + "/playout/{channel_id}/playlist.m3u8", + get(get_playlist), + ) + .route("/playout/{channel_id}/{segment}", get(get_segment)) + .route("/playout/channels", get(list_channels)) + .route("/playout/channels/status", get(channel_statuses)) + .route("/playout/channels/{channel_id}", post(add_channel)) + .route( + "/playout/channels/{channel_id}", + delete(remove_channel), + ) + .with_state(state) +} + +async fn get_playlist( + State(state): State, + Path(channel_id): Path, +) -> Response { + let segments = match state.store.list_segments(&channel_id).await { + Ok(s) => s, + Err(_) => return StatusCode::NOT_FOUND.into_response(), + }; + + let body = generate_m3u8( + &channel_id, + &segments, + state.config.segment_duration_secs, + state.config.window_size, + ); + + let mut response = body.into_response(); + response.headers_mut().insert( + header::CONTENT_TYPE, + HeaderValue::from_static("application/vnd.apple.mpegurl"), + ); + response +} + +async fn get_segment( + State(state): State, + Path((channel_id, segment)): Path<(String, String)>, +) -> Response { + match state.store.read_segment(&channel_id, &segment).await { + Ok(data) => { + let mut response = data.into_response(); + response.headers_mut().insert( + header::CONTENT_TYPE, + HeaderValue::from_static("video/mp2t"), + ); + response + } + Err(_) => StatusCode::NOT_FOUND.into_response(), + } +} + +async fn list_channels(State(state): State) -> Response { + let channels = state.engine.active_channels().await; + let ids: Vec = channels.into_iter().map(|c| c.to_string()).collect(); + axum::Json(ids).into_response() +} + +async fn channel_statuses(State(state): State) -> Response { + let statuses = state.engine.channel_statuses().await; + axum::Json(statuses).into_response() +} + +async fn add_channel( + State(state): State, + Path(channel_id): Path, +) -> Response { + let id: ChannelId = match channel_id.parse() { + Ok(id) => id, + Err(_) => return StatusCode::BAD_REQUEST.into_response(), + }; + + match state.engine.add_channel(id).await { + Ok(()) => StatusCode::CREATED.into_response(), + Err(e) => { + tracing::error!(%e, "add channel failed"); + (StatusCode::INTERNAL_SERVER_ERROR, e.to_string()).into_response() + } + } +} + +async fn remove_channel( + State(state): State, + Path(channel_id): Path, +) -> Response { + let id: ChannelId = match channel_id.parse() { + Ok(id) => id, + Err(_) => return StatusCode::BAD_REQUEST.into_response(), + }; + + state.engine.remove_channel(id).await; + StatusCode::NO_CONTENT.into_response() +} diff --git a/crates/playout/src/lib.rs b/crates/playout/src/lib.rs new file mode 100644 index 0000000..189d1af --- /dev/null +++ b/crates/playout/src/lib.rs @@ -0,0 +1,6 @@ +pub mod config; +pub mod engine; +pub mod ffmpeg; +pub mod http; +pub mod playlist; +pub mod segment_store; diff --git a/crates/playout/src/main.rs b/crates/playout/src/main.rs new file mode 100644 index 0000000..a331630 --- /dev/null +++ b/crates/playout/src/main.rs @@ -0,0 +1,132 @@ +use std::sync::Arc; + +use tokio::net::TcpListener; +use tracing::info; + +use playout::config::PlayoutConfig; +use playout::engine::{PlayoutEngine, SourceUriResolver}; +use playout::http::{self, AppState}; +use playout::segment_store::filesystem::FilesystemSegmentStore; + +use domain::value_objects::SourceUri; + +struct StubResolver; + +impl SourceUriResolver for StubResolver { + fn resolve(&self, _provider_id: &str, _external_id: &str) -> Option { + None + } +} + +#[tokio::main] +async fn main() -> anyhow::Result<()> { + dotenvy::dotenv().ok(); + + tracing_subscriber::fmt() + .with_env_filter( + tracing_subscriber::EnvFilter::try_from_default_env() + .unwrap_or_else(|_| "info".into()), + ) + .init(); + + let config = PlayoutConfig::from_env(); + info!(addr = %config.listen_addr, "starting k-tv-playout"); + + let store = Arc::new(FilesystemSegmentStore::new(config.storage_path.clone())); + let resolver: Arc = Arc::new(StubResolver); + + // TODO: wire real ScheduleQuery from adapter-sqlite once DB URL is configured + let schedule_query: Arc = + Arc::new(NoopScheduleQuery); + + let engine = Arc::new(PlayoutEngine::new( + config.clone(), + schedule_query, + resolver, + store.clone(), + )); + + let tick_engine = engine.clone(); + let tick_interval = config.tick_interval_ms; + tokio::spawn(async move { + let mut interval = + tokio::time::interval(tokio::time::Duration::from_millis(tick_interval)); + loop { + interval.tick().await; + tick_engine.tick().await; + } + }); + + let state = AppState { + engine: engine.clone(), + store, + config: config.clone(), + }; + + let app = http::router(state); + let listener = TcpListener::bind(&config.listen_addr).await?; + info!("listening on {}", config.listen_addr); + + axum::serve(listener, app) + .with_graceful_shutdown(shutdown_signal(engine)) + .await?; + + Ok(()) +} + +async fn shutdown_signal(engine: Arc) { + tokio::signal::ctrl_c() + .await + .expect("failed to install ctrl+c handler"); + info!("shutting down"); + engine.shutdown().await; +} + +struct NoopScheduleQuery; + +#[async_trait::async_trait] +impl domain::ports::ScheduleQuery for NoopScheduleQuery { + async fn find_active( + &self, + _channel_id: domain::value_objects::ChannelId, + _at: chrono::DateTime, + ) -> domain::DomainResult> { + Ok(None) + } + + async fn find_latest( + &self, + _channel_id: domain::value_objects::ChannelId, + ) -> domain::DomainResult> { + Ok(None) + } + + async fn find_playback_history( + &self, + _channel_id: domain::value_objects::ChannelId, + ) -> domain::DomainResult> { + Ok(vec![]) + } + + async fn find_last_slot_per_block( + &self, + _channel_id: domain::value_objects::ChannelId, + ) -> domain::DomainResult> { + Ok(std::collections::HashMap::new()) + } + + async fn list_schedule_history( + &self, + _channel_id: domain::value_objects::ChannelId, + ) -> domain::DomainResult> { + Ok(vec![]) + } + + async fn get_schedule_by_id( + &self, + _channel_id: domain::value_objects::ChannelId, + _schedule_id: domain::value_objects::ScheduleId, + ) -> domain::DomainResult> { + Ok(None) + } +} diff --git a/crates/playout/src/playlist.rs b/crates/playout/src/playlist.rs new file mode 100644 index 0000000..f7deffb --- /dev/null +++ b/crates/playout/src/playlist.rs @@ -0,0 +1,67 @@ +pub fn generate_m3u8( + channel_id: &str, + segments: &[String], + segment_duration_secs: u32, + window_size: usize, +) -> String { + let visible: Vec<&String> = if segments.len() > window_size { + segments[segments.len() - window_size..].iter().collect() + } else { + segments.iter().collect() + }; + + let target_duration = segment_duration_secs; + let media_sequence = if segments.len() > window_size { + segments.len() - window_size + } else { + 0 + }; + + let mut out = String::new(); + out.push_str("#EXTM3U\n"); + out.push_str("#EXT-X-VERSION:3\n"); + out.push_str(&format!("#EXT-X-TARGETDURATION:{target_duration}\n")); + out.push_str(&format!("#EXT-X-MEDIA-SEQUENCE:{media_sequence}\n")); + + for seg in &visible { + out.push_str(&format!("#EXTINF:{target_duration},\n")); + out.push_str(&format!("/playout/{channel_id}/{seg}\n")); + } + + out +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn generates_valid_m3u8() { + let segments: Vec = (0..5).map(|i| format!("seg{i:05}.ts")).collect(); + let playlist = generate_m3u8("ch1", &segments, 6, 10); + + assert!(playlist.starts_with("#EXTM3U\n")); + assert!(playlist.contains("#EXT-X-TARGETDURATION:6")); + assert!(playlist.contains("#EXT-X-MEDIA-SEQUENCE:0")); + assert!(playlist.contains("/playout/ch1/seg00000.ts")); + assert!(playlist.contains("/playout/ch1/seg00004.ts")); + } + + #[test] + fn sliding_window_trims_old_segments() { + let segments: Vec = (0..15).map(|i| format!("seg{i:05}.ts")).collect(); + let playlist = generate_m3u8("ch1", &segments, 6, 5); + + assert!(playlist.contains("#EXT-X-MEDIA-SEQUENCE:10")); + assert!(!playlist.contains("seg00000.ts")); + assert!(playlist.contains("seg00010.ts")); + assert!(playlist.contains("seg00014.ts")); + } + + #[test] + fn empty_segments() { + let playlist = generate_m3u8("ch1", &[], 6, 10); + assert!(playlist.contains("#EXTM3U")); + assert!(playlist.contains("#EXT-X-MEDIA-SEQUENCE:0")); + } +} diff --git a/crates/playout/src/segment_store/filesystem.rs b/crates/playout/src/segment_store/filesystem.rs new file mode 100644 index 0000000..b0f5a2c --- /dev/null +++ b/crates/playout/src/segment_store/filesystem.rs @@ -0,0 +1,78 @@ +use std::path::PathBuf; + +use async_trait::async_trait; +use bytes::Bytes; +use tokio::fs; + +use super::{SegmentStore, SegmentStoreError, SegmentStoreResult}; + +pub struct FilesystemSegmentStore { + base_path: PathBuf, +} + +impl FilesystemSegmentStore { + pub fn new(base_path: PathBuf) -> Self { + Self { base_path } + } + + fn channel_dir(&self, channel_id: &str) -> PathBuf { + self.base_path.join(channel_id) + } + + fn segment_path(&self, channel_id: &str, name: &str) -> PathBuf { + self.channel_dir(channel_id).join(name) + } +} + +#[async_trait] +impl SegmentStore for FilesystemSegmentStore { + async fn write_segment( + &self, + channel_id: &str, + name: &str, + data: Bytes, + ) -> SegmentStoreResult<()> { + let dir = self.channel_dir(channel_id); + fs::create_dir_all(&dir).await?; + fs::write(self.segment_path(channel_id, name), &data).await?; + Ok(()) + } + + async fn read_segment(&self, channel_id: &str, name: &str) -> SegmentStoreResult { + let path = self.segment_path(channel_id, name); + match fs::read(&path).await { + Ok(data) => Ok(Bytes::from(data)), + Err(e) if e.kind() == std::io::ErrorKind::NotFound => { + Err(SegmentStoreError::NotFound(name.to_string())) + } + Err(e) => Err(e.into()), + } + } + + async fn list_segments(&self, channel_id: &str) -> SegmentStoreResult> { + let dir = self.channel_dir(channel_id); + if !dir.exists() { + return Ok(Vec::new()); + } + let mut entries = fs::read_dir(&dir).await?; + let mut names = Vec::new(); + while let Some(entry) = entries.next_entry().await? { + if let Some(name) = entry.file_name().to_str() + && name.ends_with(".ts") + { + names.push(name.to_string()); + } + } + names.sort(); + Ok(names) + } + + async fn delete_segment(&self, channel_id: &str, name: &str) -> SegmentStoreResult<()> { + let path = self.segment_path(channel_id, name); + match fs::remove_file(&path).await { + Ok(()) => Ok(()), + Err(e) if e.kind() == std::io::ErrorKind::NotFound => Ok(()), + Err(e) => Err(e.into()), + } + } +} diff --git a/crates/playout/src/segment_store/memory.rs b/crates/playout/src/segment_store/memory.rs new file mode 100644 index 0000000..1c50e41 --- /dev/null +++ b/crates/playout/src/segment_store/memory.rs @@ -0,0 +1,125 @@ +use std::collections::HashMap; +use std::sync::Arc; + +use async_trait::async_trait; +use bytes::Bytes; +use tokio::sync::RwLock; + +use super::{SegmentStore, SegmentStoreError, SegmentStoreResult}; + +type ChannelSegments = HashMap>; + +pub struct InMemorySegmentStore { + data: Arc>, +} + +impl InMemorySegmentStore { + pub fn new() -> Self { + Self { + data: Arc::new(RwLock::new(HashMap::new())), + } + } +} + +impl Default for InMemorySegmentStore { + fn default() -> Self { + Self::new() + } +} + +#[async_trait] +impl SegmentStore for InMemorySegmentStore { + async fn write_segment( + &self, + channel_id: &str, + name: &str, + data: Bytes, + ) -> SegmentStoreResult<()> { + let mut store = self.data.write().await; + store + .entry(channel_id.to_string()) + .or_default() + .insert(name.to_string(), data); + Ok(()) + } + + async fn read_segment(&self, channel_id: &str, name: &str) -> SegmentStoreResult { + let store = self.data.read().await; + store + .get(channel_id) + .and_then(|segs| segs.get(name)) + .cloned() + .ok_or_else(|| SegmentStoreError::NotFound(name.to_string())) + } + + async fn list_segments(&self, channel_id: &str) -> SegmentStoreResult> { + let store = self.data.read().await; + let mut names: Vec = store + .get(channel_id) + .map(|segs| segs.keys().filter(|k| k.ends_with(".ts")).cloned().collect()) + .unwrap_or_default(); + names.sort(); + Ok(names) + } + + async fn delete_segment(&self, channel_id: &str, name: &str) -> SegmentStoreResult<()> { + let mut store = self.data.write().await; + if let Some(segs) = store.get_mut(channel_id) { + segs.remove(name); + } + Ok(()) + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[tokio::test] + async fn write_read_delete_lifecycle() { + let store = InMemorySegmentStore::new(); + let channel = "test-chan"; + + store + .write_segment(channel, "seg0.ts", Bytes::from_static(b"aaa")) + .await + .unwrap(); + store + .write_segment(channel, "seg1.ts", Bytes::from_static(b"bbb")) + .await + .unwrap(); + + let data = store.read_segment(channel, "seg0.ts").await.unwrap(); + assert_eq!(data.as_ref(), b"aaa"); + + let list = store.list_segments(channel).await.unwrap(); + assert_eq!(list, vec!["seg0.ts", "seg1.ts"]); + + store.delete_segment(channel, "seg0.ts").await.unwrap(); + let list = store.list_segments(channel).await.unwrap(); + assert_eq!(list, vec!["seg1.ts"]); + + let err = store.read_segment(channel, "seg0.ts").await; + assert!(err.is_err()); + } + + #[tokio::test] + async fn read_nonexistent_returns_not_found() { + let store = InMemorySegmentStore::new(); + let err = store.read_segment("chan", "nope.ts").await.unwrap_err(); + assert!(matches!(err, SegmentStoreError::NotFound(_))); + } + + #[tokio::test] + async fn delete_nonexistent_is_ok() { + let store = InMemorySegmentStore::new(); + store.delete_segment("chan", "nope.ts").await.unwrap(); + } + + #[tokio::test] + async fn list_empty_channel() { + let store = InMemorySegmentStore::new(); + let list = store.list_segments("empty").await.unwrap(); + assert!(list.is_empty()); + } +} diff --git a/crates/playout/src/segment_store/mod.rs b/crates/playout/src/segment_store/mod.rs new file mode 100644 index 0000000..0a7b6e7 --- /dev/null +++ b/crates/playout/src/segment_store/mod.rs @@ -0,0 +1,28 @@ +pub mod filesystem; +pub mod memory; + +use async_trait::async_trait; +use bytes::Bytes; +use thiserror::Error; + +#[derive(Debug, Error)] +pub enum SegmentStoreError { + #[error("segment not found: {0}")] + NotFound(String), + #[error("io error: {0}")] + Io(#[from] std::io::Error), +} + +pub type SegmentStoreResult = Result; + +#[async_trait] +pub trait SegmentStore: Send + Sync { + async fn write_segment(&self, channel_id: &str, name: &str, data: Bytes) + -> SegmentStoreResult<()>; + + async fn read_segment(&self, channel_id: &str, name: &str) -> SegmentStoreResult; + + async fn list_segments(&self, channel_id: &str) -> SegmentStoreResult>; + + async fn delete_segment(&self, channel_id: &str, name: &str) -> SegmentStoreResult<()>; +}