add playout service: HLS streaming, FFmpeg, SegmentStore port (#9)

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),
+        }
+    }
+}

crates/playout/src/engine.rs

@@ -0,0 +1,219 @@
+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 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<SourceUri>;
+}
+
+struct ChannelState {
+    ffmpeg: FfmpegHandle,
+    current_slot_id: SlotId,
+    schedule_id: domain::value_objects::ScheduleId,
+}
+
+pub struct PlayoutEngine {
+    config: PlayoutConfig,
+    schedule_query: Arc<dyn ScheduleQuery>,
+    source_resolver: Arc<dyn SourceUriResolver>,
+    store: Arc<dyn SegmentStore>,
+    channels: Arc<RwLock<HashMap<ChannelId, ChannelState>>>,
+}
+
+impl PlayoutEngine {
+    pub fn new(
+        config: PlayoutConfig,
+        schedule_query: Arc<dyn ScheduleQuery>,
+        source_resolver: Arc<dyn SourceUriResolver>,
+        store: Arc<dyn SegmentStore>,
+    ) -> 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 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<ChannelId> {
+        self.channels.read().await.keys().copied().collect()
+    }
+
+    pub async fn tick(&self) {
+        let now = Utc::now();
+        let channel_ids: Vec<ChannelId> = self.channels.read().await.keys().copied().collect();
+
+        for channel_id in channel_ids {
+            if let Err(e) = self.tick_channel(channel_id, now).await {
+                warn!(%channel_id, %e, "tick failed");
+            }
+        }
+    }
+
+    async fn tick_channel(
+        &self,
+        channel_id: ChannelId,
+        now: chrono::DateTime<Utc>,
+    ) -> 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)
+        };
+
+        if needs_transition {
+            info!(%channel_id, %current_slot_id, "slot transition");
+            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<FfmpegHandle> {
+        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();
+    }
+}

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<Child>,
+    stop: Arc<Notify>,
+}
+
+impl FfmpegHandle {
+    pub fn spawn(
+        config: FfmpegConfig,
+        store: Arc<dyn SegmentStore>,
+        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<Child> {
+        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<dyn SegmentStore>,
+        channel_id: String,
+        output_dir: PathBuf,
+        stop: Arc<Notify>,
+    ) {
+        let mut interval = tokio::time::interval(tokio::time::Duration::from_millis(500));
+        let mut known_segments: std::collections::HashSet<String> =
+            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,
+        }
+    }
+}

crates/playout/src/http.rs

@@ -0,0 +1,117 @@
+use std::sync::Arc;
+
+use axum::extract::{Path, State};
+use axum::http::{HeaderValue, StatusCode, header};
+use axum::response::{IntoResponse, Response};
+use axum::routing::{get, post, delete};
+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<PlayoutEngine>,
+    pub store: Arc<dyn SegmentStore>,
+    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/{channel_id}", post(add_channel))
+        .route(
+            "/playout/channels/{channel_id}",
+            delete(remove_channel),
+        )
+        .with_state(state)
+}
+
+async fn get_playlist(
+    State(state): State<AppState>,
+    Path(channel_id): Path<String>,
+) -> 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<AppState>,
+    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<AppState>) -> Response {
+    let channels = state.engine.active_channels().await;
+    let ids: Vec<String> = channels.into_iter().map(|c| c.to_string()).collect();
+    axum::Json(ids).into_response()
+}
+
+async fn add_channel(
+    State(state): State<AppState>,
+    Path(channel_id): Path<String>,
+) -> 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<AppState>,
+    Path(channel_id): Path<String>,
+) -> 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()
+}

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;

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<SourceUri> {
+        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<dyn SourceUriResolver> = Arc::new(StubResolver);
+
+    // TODO: wire real ScheduleQuery from adapter-sqlite once DB URL is configured
+    let schedule_query: Arc<dyn domain::ports::ScheduleQuery> =
+        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<PlayoutEngine>) {
+    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<chrono::Utc>,
+    ) -> domain::DomainResult<Option<domain::GeneratedSchedule>> {
+        Ok(None)
+    }
+
+    async fn find_latest(
+        &self,
+        _channel_id: domain::value_objects::ChannelId,
+    ) -> domain::DomainResult<Option<domain::GeneratedSchedule>> {
+        Ok(None)
+    }
+
+    async fn find_playback_history(
+        &self,
+        _channel_id: domain::value_objects::ChannelId,
+    ) -> domain::DomainResult<Vec<domain::PlaybackRecord>> {
+        Ok(vec![])
+    }
+
+    async fn find_last_slot_per_block(
+        &self,
+        _channel_id: domain::value_objects::ChannelId,
+    ) -> domain::DomainResult<std::collections::HashMap<domain::BlockId, domain::MediaItemId>> {
+        Ok(std::collections::HashMap::new())
+    }
+
+    async fn list_schedule_history(
+        &self,
+        _channel_id: domain::value_objects::ChannelId,
+    ) -> domain::DomainResult<Vec<domain::GeneratedSchedule>> {
+        Ok(vec![])
+    }
+
+    async fn get_schedule_by_id(
+        &self,
+        _channel_id: domain::value_objects::ChannelId,
+        _schedule_id: domain::value_objects::ScheduleId,
+    ) -> domain::DomainResult<Option<domain::GeneratedSchedule>> {
+        Ok(None)
+    }
+}

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<String> = (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<String> = (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"));
+    }
+}

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<Bytes> {
+        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<Vec<String>> {
+        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()),
+        }
+    }
+}

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<String, HashMap<String, Bytes>>;
+
+pub struct InMemorySegmentStore {
+    data: Arc<RwLock<ChannelSegments>>,
+}
+
+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<Bytes> {
+        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<Vec<String>> {
+        let store = self.data.read().await;
+        let mut names: Vec<String> = 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());
+    }
+}

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<T> = Result<T, SegmentStoreError>;
+
+#[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<Bytes>;
+
+    async fn list_segments(&self, channel_id: &str) -> SegmentStoreResult<Vec<String>>;
+
+    async fn delete_segment(&self, channel_id: &str, name: &str) -> SegmentStoreResult<()>;
+}