adapter-local-files: media provider, index, transcoder

crates/adapters/local-files/src/config.rs

@@ -0,0 +1,13 @@
+use std::path::PathBuf;
+
+/// Configuration for the local files media provider.
+pub struct LocalFilesConfig {
+    /// Root directory containing video files. All files are served relative to this.
+    pub root_dir: PathBuf,
+    /// Public base URL of this API server, used to build stream URLs.
+    pub base_url: String,
+    /// Directory for FFmpeg HLS transcode cache. `None` disables transcoding.
+    pub transcode_dir: Option<PathBuf>,
+    /// How long (hours) to keep transcode cache entries. Passed to TranscodeManager.
+    pub cleanup_ttl_hours: u32,
+}

crates/adapters/local-files/src/index.rs

@@ -0,0 +1,203 @@
+use std::collections::HashMap;
+use std::path::PathBuf;
+use std::sync::Arc;
+
+use chrono::Utc;
+use tokio::sync::RwLock;
+use tracing::{error, info};
+
+use domain::MediaItemId;
+
+use crate::config::LocalFilesConfig;
+use crate::scanner::{scan_dir, LocalFileItem};
+
+/// Encode a rel-path string into a URL-safe, padding-free base64 MediaItemId.
+pub fn encode_id(rel_path: &str) -> MediaItemId {
+    use base64::Engine as _;
+    MediaItemId::new(
+        base64::engine::general_purpose::URL_SAFE_NO_PAD.encode(rel_path.as_bytes()),
+    )
+}
+
+/// Decode a MediaItemId back to a relative path string.
+pub fn decode_id(id: &MediaItemId) -> Option<String> {
+    use base64::Engine as _;
+    let bytes = base64::engine::general_purpose::URL_SAFE_NO_PAD
+        .decode(id.as_ref())
+        .ok()?;
+    String::from_utf8(bytes).ok()
+}
+
+/// In-memory (+ SQLite-backed) index of local video files.
+///
+/// On startup the index is populated from the SQLite cache so the provider can
+/// serve requests immediately. A background task calls `rescan()` to pick up
+/// any changes on disk and write them back to the cache.
+pub struct LocalIndex {
+    items: Arc<RwLock<HashMap<MediaItemId, LocalFileItem>>>,
+    pub root_dir: PathBuf,
+    provider_id: String,
+    pool: sqlx::SqlitePool,
+}
+
+impl LocalIndex {
+    /// Create the index, immediately loading persisted entries from SQLite.
+    pub async fn new(
+        config: &LocalFilesConfig,
+        pool: sqlx::SqlitePool,
+        provider_id: String,
+    ) -> Self {
+        let idx = Self {
+            items: Arc::new(RwLock::new(HashMap::new())),
+            root_dir: config.root_dir.clone(),
+            provider_id,
+            pool,
+        };
+        idx.load_from_db().await;
+        idx
+    }
+
+    /// Load previously scanned items from SQLite (instant on startup).
+    async fn load_from_db(&self) {
+        #[derive(sqlx::FromRow)]
+        struct Row {
+            id: String,
+            rel_path: String,
+            title: String,
+            duration_secs: i64,
+            year: Option<i64>,
+            tags: String,
+            top_dir: String,
+        }
+
+        let rows = sqlx::query_as::<_, Row>(
+            "SELECT id, rel_path, title, duration_secs, year, tags, top_dir \
+             FROM local_files_index WHERE provider_id = ?",
+        )
+        .bind(&self.provider_id)
+        .fetch_all(&self.pool)
+        .await;
+
+        match rows {
+            Ok(rows) => {
+                let mut map = self.items.write().await;
+                for row in rows {
+                    let tags: Vec<String> =
+                        serde_json::from_str(&row.tags).unwrap_or_default();
+                    let item = LocalFileItem {
+                        rel_path: row.rel_path,
+                        title: row.title,
+                        duration_secs: row.duration_secs as u32,
+                        year: row.year.map(|y| y as u16),
+                        tags,
+                        top_dir: row.top_dir,
+                    };
+                    map.insert(MediaItemId::new(row.id), item);
+                }
+                info!(
+                    "Local files index [{}]: loaded {} items from DB",
+                    self.provider_id,
+                    map.len()
+                );
+            }
+            Err(e) => {
+                // Table might not exist yet on first run -- that's fine.
+                tracing::debug!("Could not load local files index from DB: {}", e);
+            }
+        }
+    }
+
+    /// Scan the filesystem for video files and rebuild the index.
+    ///
+    /// Returns the number of items found. Called on startup (background task)
+    /// and via `POST /files/rescan`.
+    pub async fn rescan(&self) -> u32 {
+        info!(
+            "Local files [{}]: scanning {:?}",
+            self.provider_id, self.root_dir
+        );
+        let new_items = scan_dir(&self.root_dir).await;
+        let count = new_items.len() as u32;
+
+        // Swap in-memory map.
+        {
+            let mut map = self.items.write().await;
+            map.clear();
+            for item in &new_items {
+                let id = encode_id(&item.rel_path);
+                map.insert(id, item.clone());
+            }
+        }
+
+        // Persist to SQLite.
+        if let Err(e) = self.save_to_db(&new_items).await {
+            error!("Failed to persist local files index: {}", e);
+        }
+
+        info!(
+            "Local files [{}]: indexed {} items",
+            self.provider_id, count
+        );
+        count
+    }
+
+    async fn save_to_db(&self, items: &[LocalFileItem]) -> Result<(), sqlx::Error> {
+        // Rebuild the table in one transaction, scoped to this provider.
+        let mut tx = self.pool.begin().await?;
+
+        sqlx::query("DELETE FROM local_files_index WHERE provider_id = ?")
+            .bind(&self.provider_id)
+            .execute(&mut *tx)
+            .await?;
+
+        let now = Utc::now().to_rfc3339();
+        for item in items {
+            let id = encode_id(&item.rel_path).into_inner();
+            let tags_json =
+                serde_json::to_string(&item.tags).unwrap_or_else(|_| "[]".into());
+            sqlx::query(
+                "INSERT INTO local_files_index \
+                 (id, rel_path, title, duration_secs, year, tags, top_dir, scanned_at, provider_id) \
+                 VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)",
+            )
+            .bind(&id)
+            .bind(&item.rel_path)
+            .bind(&item.title)
+            .bind(item.duration_secs as i64)
+            .bind(item.year.map(|y| y as i64))
+            .bind(&tags_json)
+            .bind(&item.top_dir)
+            .bind(&now)
+            .bind(&self.provider_id)
+            .execute(&mut *tx)
+            .await?;
+        }
+
+        tx.commit().await
+    }
+
+    pub async fn get(&self, id: &MediaItemId) -> Option<LocalFileItem> {
+        self.items.read().await.get(id).cloned()
+    }
+
+    pub async fn get_all(&self) -> Vec<(MediaItemId, LocalFileItem)> {
+        self.items
+            .read()
+            .await
+            .iter()
+            .map(|(k, v)| (k.clone(), v.clone()))
+            .collect()
+    }
+
+    /// Return unique top-level directories as collection names.
+    pub async fn collections(&self) -> Vec<String> {
+        let map = self.items.read().await;
+        let mut seen = std::collections::HashSet::new();
+        for item in map.values() {
+            seen.insert(item.top_dir.clone());
+        }
+        let mut dirs: Vec<String> = seen.into_iter().collect();
+        dirs.sort();
+        dirs
+    }
+}

crates/adapters/local-files/src/lib.rs

@@ -0,0 +1,52 @@
+//! Local-files media provider adapter.
+//!
+//! Implements [`domain::ports::IMediaProvider`] by scanning a local filesystem
+//! directory for video files. Optional FFmpeg HLS transcoding via
+//! [`TranscodeManager`].
+
+pub mod config;
+pub mod index;
+pub mod provider;
+pub mod scanner;
+pub mod transcoder;
+
+pub use config::LocalFilesConfig;
+pub use index::LocalIndex;
+pub use provider::{LocalFilesProvider, decode_stream_id};
+pub use transcoder::TranscodeManager;
+
+use std::sync::Arc;
+
+/// Bundle of all local-files components, constructed once at startup.
+pub struct LocalFilesBundle {
+    pub provider: LocalFilesProvider,
+    pub local_index: Arc<LocalIndex>,
+    pub transcode_manager: Option<Arc<TranscodeManager>>,
+}
+
+impl LocalFilesBundle {
+    /// Build the bundle from config and a SQLite pool.
+    ///
+    /// If `config.transcode_dir` is `Some`, a `TranscodeManager` is created
+    /// with its background cleanup task.
+    pub async fn build(
+        config: LocalFilesConfig,
+        pool: sqlx::SqlitePool,
+        provider_id: String,
+    ) -> Self {
+        let local_index = Arc::new(LocalIndex::new(&config, pool, provider_id).await);
+
+        let transcode_manager = config.transcode_dir.as_ref().map(|dir| {
+            TranscodeManager::new(dir.clone(), config.cleanup_ttl_hours)
+        });
+
+        let provider =
+            LocalFilesProvider::new(Arc::clone(&local_index), &config, transcode_manager.clone());
+
+        Self {
+            provider,
+            local_index,
+            transcode_manager,
+        }
+    }
+}

crates/adapters/local-files/src/provider.rs

@@ -0,0 +1,200 @@
+use std::sync::Arc;
+
+use async_trait::async_trait;
+use domain::ports::{
+    Collection, IMediaProvider, ProviderCapabilities, StreamQuality, StreamingProtocol,
+};
+use domain::{ContentType, DomainError, DomainResult, MediaFilter, MediaItem, MediaItemId};
+
+use crate::config::LocalFilesConfig;
+use crate::index::{decode_id, LocalIndex};
+use crate::scanner::LocalFileItem;
+use crate::transcoder::TranscodeManager;
+
+pub struct LocalFilesProvider {
+    pub index: Arc<LocalIndex>,
+    base_url: String,
+    transcode_manager: Option<Arc<TranscodeManager>>,
+}
+
+const SHORT_DURATION_SECS: u32 = 1200; // 20 minutes
+
+impl LocalFilesProvider {
+    pub fn new(
+        index: Arc<LocalIndex>,
+        config: &LocalFilesConfig,
+        transcode_manager: Option<Arc<TranscodeManager>>,
+    ) -> Self {
+        Self {
+            index,
+            base_url: config.base_url.trim_end_matches('/').to_string(),
+            transcode_manager,
+        }
+    }
+}
+
+fn to_media_item(id: MediaItemId, item: &LocalFileItem) -> MediaItem {
+    let content_type = if item.duration_secs < SHORT_DURATION_SECS {
+        ContentType::Short
+    } else {
+        ContentType::Movie
+    };
+    MediaItem::from_persistence(
+        id,
+        item.title.clone(),
+        content_type,
+        item.duration_secs,
+        None,          // description
+        vec![],        // genres
+        item.year,
+        item.tags.clone(),
+        None,          // series_name
+        None,          // season_number
+        None,          // episode_number
+        None,          // thumbnail_url
+        None,          // collection_id
+    )
+}
+
+#[async_trait]
+impl IMediaProvider for LocalFilesProvider {
+    fn capabilities(&self) -> ProviderCapabilities {
+        ProviderCapabilities {
+            collections: true,
+            series: false,
+            genres: false,
+            tags: true,
+            decade: true,
+            search: true,
+            streaming_protocol: if self.transcode_manager.is_some() {
+                StreamingProtocol::Hls
+            } else {
+                StreamingProtocol::DirectFile
+            },
+            rescan: true,
+            transcode: self.transcode_manager.is_some(),
+        }
+    }
+
+    async fn fetch_items(&self, filter: &MediaFilter) -> DomainResult<Vec<MediaItem>> {
+        let all = self.index.get_all().await;
+
+        let results = all
+            .into_iter()
+            .filter_map(|(id, item)| {
+                // content_type: derive heuristically, then filter
+                let content_type = if item.duration_secs < SHORT_DURATION_SECS {
+                    ContentType::Short
+                } else {
+                    ContentType::Movie
+                };
+                if let Some(ref ct) = filter.content_type {
+                    if &content_type != ct {
+                        return None;
+                    }
+                }
+
+                // collections: match against top_dir
+                if !filter.collections.is_empty()
+                    && !filter.collections.contains(&item.top_dir)
+                {
+                    return None;
+                }
+
+                // tags: OR -- item must have at least one matching tag
+                if !filter.tags.is_empty() {
+                    let has = filter
+                        .tags
+                        .iter()
+                        .any(|tag| item.tags.iter().any(|t| t.eq_ignore_ascii_case(tag)));
+                    if !has {
+                        return None;
+                    }
+                }
+
+                // decade: year in [decade, decade+9]
+                if let Some(decade) = filter.decade {
+                    match item.year {
+                        Some(y) if y >= decade && y <= decade + 9 => {}
+                        _ => return None,
+                    }
+                }
+
+                // duration bounds
+                if let Some(min) = filter.min_duration_secs {
+                    if item.duration_secs < min {
+                        return None;
+                    }
+                }
+                if let Some(max) = filter.max_duration_secs {
+                    if item.duration_secs > max {
+                        return None;
+                    }
+                }
+
+                // search_term: case-insensitive substring in title
+                if let Some(ref q) = filter.search_term {
+                    if !item.title.to_lowercase().contains(&q.to_lowercase()) {
+                        return None;
+                    }
+                }
+
+                Some(to_media_item(id, &item))
+            })
+            .collect();
+
+        Ok(results)
+    }
+
+    async fn fetch_by_id(&self, item_id: &MediaItemId) -> DomainResult<Option<MediaItem>> {
+        Ok(self
+            .index
+            .get(item_id)
+            .await
+            .map(|item| to_media_item(item_id.clone(), &item)))
+    }
+
+    async fn get_stream_url(
+        &self,
+        item_id: &MediaItemId,
+        quality: &StreamQuality,
+    ) -> DomainResult<String> {
+        match quality {
+            StreamQuality::Transcode(_) if self.transcode_manager.is_some() => {
+                let tm = self.transcode_manager.as_ref().unwrap();
+                let rel = decode_id(item_id).ok_or_else(|| {
+                    DomainError::InfrastructureError("invalid item id encoding".into())
+                })?;
+                let src = self.index.root_dir.join(&rel);
+                tm.ensure_transcoded(item_id.as_ref(), &src).await?;
+                Ok(format!(
+                    "{}/api/v1/files/transcode/{}/playlist.m3u8",
+                    self.base_url,
+                    item_id.as_ref()
+                ))
+            }
+            _ => Ok(format!(
+                "{}/api/v1/files/stream/{}",
+                self.base_url,
+                item_id.as_ref()
+            )),
+        }
+    }
+
+    async fn list_collections(&self) -> DomainResult<Vec<Collection>> {
+        let dirs = self.index.collections().await;
+        Ok(dirs
+            .into_iter()
+            .map(|d| Collection {
+                id: d.clone(),
+                name: d,
+                collection_type: None,
+            })
+            .collect())
+    }
+}
+
+/// Decode an encoded ID from a URL path segment to its relative path string.
+pub fn decode_stream_id(encoded: &str) -> Option<String> {
+    decode_id(&MediaItemId::new(encoded))
+}

crates/adapters/local-files/src/scanner.rs

@@ -0,0 +1,177 @@
+use std::path::Path;
+use tokio::process::Command;
+
+const VIDEO_EXTENSIONS: &[&str] = &["mp4", "mkv", "avi", "mov", "webm", "m4v"];
+
+/// In-memory representation of a scanned local video file.
+#[derive(Debug, Clone)]
+pub struct LocalFileItem {
+    /// Relative path from root, with forward slashes (used as the stable ID source).
+    pub rel_path: String,
+    pub title: String,
+    pub duration_secs: u32,
+    pub year: Option<u16>,
+    /// Ancestor directory names between root and file (excluding root itself).
+    pub tags: Vec<String>,
+    /// First path component under root (used as collection id/name).
+    pub top_dir: String,
+}
+
+/// Walk `root` and return all recognised video files with metadata.
+///
+/// ffprobe is called for each file to determine duration. Files that cannot be
+/// probed are included with `duration_secs = 0` so they still appear in the index.
+pub async fn scan_dir(root: &Path) -> Vec<LocalFileItem> {
+    let mut items = Vec::new();
+
+    let walker = walkdir::WalkDir::new(root).follow_links(true);
+    for entry in walker.into_iter().filter_map(|e| e.ok()) {
+        if !entry.file_type().is_file() {
+            continue;
+        }
+        let path = entry.path();
+        let ext = path
+            .extension()
+            .and_then(|e| e.to_str())
+            .map(|e| e.to_lowercase());
+        let ext = match ext {
+            Some(ref e) if VIDEO_EXTENSIONS.contains(&e.as_str()) => e.clone(),
+            _ => continue,
+        };
+        let _ = ext; // extension validated, not needed further
+
+        let rel = match path.strip_prefix(root) {
+            Ok(r) => r,
+            Err(_) => continue,
+        };
+
+        // Normalise to forward-slash string for cross-platform stability.
+        let rel_path: String = rel
+            .components()
+            .map(|c| c.as_os_str().to_string_lossy().into_owned())
+            .collect::<Vec<_>>()
+            .join("/");
+
+        // Top-level directory under root.
+        let top_dir = rel
+            .components()
+            .next()
+            .filter(|_| rel.components().count() > 1) // skip if file is at root level
+            .map(|c| c.as_os_str().to_string_lossy().into_owned())
+            .unwrap_or_else(|| "__root__".to_string());
+
+        // Title: stem with separator chars replaced by spaces.
+        let stem = path
+            .file_stem()
+            .and_then(|s| s.to_str())
+            .unwrap_or("")
+            .to_string();
+        let title = stem.replace(['_', '-', '.'], " ");
+        let title = title.trim().to_string();
+
+        // Year: first 4-digit number starting with 19xx or 20xx in filename or parent dirs.
+        let search_str = format!(
+            "{} {}",
+            stem,
+            rel.parent()
+                .and_then(|p| p.to_str())
+                .unwrap_or("")
+        );
+        let year = extract_year(&search_str);
+
+        // Tags: ancestor directory components between root and the file.
+        let tags: Vec<String> = rel
+            .parent()
+            .into_iter()
+            .flat_map(|p| p.components())
+            .map(|c| c.as_os_str().to_string_lossy().into_owned())
+            .filter(|s| !s.is_empty())
+            .collect();
+
+        let duration_secs = get_duration(path).await.unwrap_or(0);
+
+        items.push(LocalFileItem {
+            rel_path,
+            title,
+            duration_secs,
+            year,
+            tags,
+            top_dir,
+        });
+    }
+
+    items
+}
+
+/// Extract the first plausible 4-digit year (1900-2099) from `s`.
+fn extract_year(s: &str) -> Option<u16> {
+    let chars: Vec<char> = s.chars().collect();
+    let n = chars.len();
+    if n < 4 {
+        return None;
+    }
+    for i in 0..=(n - 4) {
+        // All four chars must be ASCII digits.
+        if !chars[i..i + 4].iter().all(|c| c.is_ascii_digit()) {
+            continue;
+        }
+        // Parse and range-check.
+        let s4: String = chars[i..i + 4].iter().collect();
+        let num: u16 = s4.parse().ok()?;
+        if !(1900..=2099).contains(&num) {
+            continue;
+        }
+        // Word-boundary: char before and after must not be digits.
+        let before_ok = i == 0 || !chars[i - 1].is_ascii_digit();
+        let after_ok = i + 4 >= n || !chars[i + 4].is_ascii_digit();
+        if before_ok && after_ok {
+            return Some(num);
+        }
+    }
+    None
+}
+
+/// Run ffprobe to get the duration of `path` in whole seconds.
+async fn get_duration(path: &Path) -> Option<u32> {
+    #[derive(serde::Deserialize)]
+    struct Fmt {
+        duration: Option<String>,
+    }
+    #[derive(serde::Deserialize)]
+    struct Out {
+        format: Fmt,
+    }
+
+    let output = Command::new("ffprobe")
+        .args([
+            "-v",
+            "quiet",
+            "-print_format",
+            "json",
+            "-show_format",
+            path.to_str()?,
+        ])
+        .output()
+        .await
+        .ok()?;
+
+    let parsed: Out = serde_json::from_slice(&output.stdout).ok()?;
+    let dur: f64 = parsed.format.duration?.parse().ok()?;
+    Some(dur as u32)
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn extract_year_basic() {
+        assert_eq!(extract_year("Movie 2024 HD"), Some(2024));
+        assert_eq!(extract_year("1999_classic"), Some(1999));
+        assert_eq!(extract_year("no year here"), None);
+        assert_eq!(extract_year("12345"), None); // 5-digit number
+        assert_eq!(extract_year("2100"), None); // out of range
+        assert_eq!(extract_year("1900"), Some(1900));
+        assert_eq!(extract_year("2099"), Some(2099));
+    }
+}

crates/adapters/local-files/src/transcoder.rs

@@ -0,0 +1,263 @@
+//! FFmpeg HLS transcoder for local video files.
+//!
+//! `TranscodeManager` orchestrates on-demand transcoding: the first request for
+//! an item spawns an ffmpeg process and returns once the initial HLS playlist
+//! appears. Concurrent requests for the same item subscribe to a watch channel
+//! and wait without spawning duplicate processes. Transcoded segments are cached
+//! in `transcode_dir/{item_id}/` and cleaned up by a background task.
+
+use std::collections::HashMap;
+use std::path::{Path, PathBuf};
+use std::sync::{
+    Arc,
+    atomic::{AtomicU32, Ordering},
+};
+use std::time::{Duration, Instant};
+
+use tokio::sync::{Mutex, watch};
+use tracing::{error, info, warn};
+
+use domain::{DomainError, DomainResult};
+
+// ============================================================================
+// Types
+// ============================================================================
+
+#[derive(Clone, Debug)]
+pub enum TranscodeStatus {
+    Ready,
+    Failed(String),
+}
+
+// ============================================================================
+// Manager
+// ============================================================================
+
+pub struct TranscodeManager {
+    pub transcode_dir: PathBuf,
+    cleanup_ttl_hours: Arc<AtomicU32>,
+    active: Arc<Mutex<HashMap<String, watch::Sender<Option<TranscodeStatus>>>>>,
+}
+
+impl TranscodeManager {
+    pub fn new(transcode_dir: PathBuf, cleanup_ttl_hours: u32) -> Arc<Self> {
+        let mgr = Arc::new(Self {
+            transcode_dir,
+            cleanup_ttl_hours: Arc::new(AtomicU32::new(cleanup_ttl_hours)),
+            active: Arc::new(Mutex::new(HashMap::new())),
+        });
+        // Background cleanup task -- uses Weak to avoid keeping manager alive.
+        let weak = Arc::downgrade(&mgr);
+        tokio::spawn(async move {
+            let mut interval = tokio::time::interval(Duration::from_secs(3600));
+            loop {
+                interval.tick().await;
+                match weak.upgrade() {
+                    Some(m) => m.run_cleanup().await,
+                    None => break,
+                }
+            }
+        });
+        mgr
+    }
+
+    /// Update the cleanup TTL (also persisted to DB by the route handler).
+    pub fn set_cleanup_ttl(&self, hours: u32) {
+        self.cleanup_ttl_hours.store(hours, Ordering::Relaxed);
+    }
+
+    pub fn get_cleanup_ttl(&self) -> u32 {
+        self.cleanup_ttl_hours.load(Ordering::Relaxed)
+    }
+
+    /// Ensure `item_id` has been transcoded to HLS. Blocks until the initial
+    /// playlist appears or an error occurs. Concurrent callers share the result.
+    pub async fn ensure_transcoded(&self, item_id: &str, src_path: &Path) -> DomainResult<()> {
+        let out_dir = self.transcode_dir.join(item_id);
+        let playlist = out_dir.join("playlist.m3u8");
+
+        if playlist.exists() {
+            return Ok(());
+        }
+
+        let mut rx = {
+            let mut map = self.active.lock().await;
+            if let Some(tx) = map.get(item_id) {
+                tx.subscribe()
+            } else {
+                let (tx, rx) = watch::channel::<Option<TranscodeStatus>>(None);
+                map.insert(item_id.to_string(), tx.clone());
+
+                let item_id_owned = item_id.to_string();
+                let src_owned = src_path.to_path_buf();
+                let out_dir_owned = out_dir.clone();
+                let playlist_owned = playlist.clone();
+                let active_ref = Arc::clone(&self.active);
+
+                tokio::spawn(async move {
+                    let _ = tokio::fs::create_dir_all(&out_dir_owned).await;
+                    let status =
+                        do_transcode(&src_owned, &out_dir_owned, &playlist_owned).await;
+                    if matches!(status, TranscodeStatus::Ready) {
+                        info!("transcode ready: {}", item_id_owned);
+                    } else if let TranscodeStatus::Failed(ref e) = status {
+                        error!("transcode failed for {}: {}", item_id_owned, e);
+                    }
+                    let _ = tx.send(Some(status));
+                    active_ref.lock().await.remove(&item_id_owned);
+                });
+
+                rx
+            }
+        };
+
+        // Wait for Ready or Failed.
+        loop {
+            rx.changed().await.map_err(|_| {
+                DomainError::InfrastructureError(
+                    "transcode task dropped unexpectedly".into(),
+                )
+            })?;
+            if let Some(status) = &*rx.borrow() {
+                return match status {
+                    TranscodeStatus::Ready => Ok(()),
+                    TranscodeStatus::Failed(e) => Err(DomainError::InfrastructureError(
+                        format!("transcode failed: {}", e),
+                    )),
+                };
+            }
+        }
+    }
+
+    /// Remove all cached transcode directories.
+    pub async fn clear_cache(&self) -> std::io::Result<()> {
+        if self.transcode_dir.exists() {
+            tokio::fs::remove_dir_all(&self.transcode_dir).await?;
+        }
+        tokio::fs::create_dir_all(&self.transcode_dir).await
+    }
+
+    /// Return `(total_bytes, item_count)` for the cache directory.
+    pub async fn cache_stats(&self) -> (u64, usize) {
+        let mut total_bytes = 0u64;
+        let mut item_count = 0usize;
+        let Ok(mut entries) = tokio::fs::read_dir(&self.transcode_dir).await else {
+            return (0, 0);
+        };
+        while let Ok(Some(entry)) = entries.next_entry().await {
+            if !entry.path().is_dir() {
+                continue;
+            }
+            item_count += 1;
+            if let Ok(mut sub) = tokio::fs::read_dir(entry.path()).await {
+                while let Ok(Some(f)) = sub.next_entry().await {
+                    if let Ok(meta) = f.metadata().await {
+                        total_bytes += meta.len();
+                    }
+                }
+            }
+        }
+        (total_bytes, item_count)
+    }
+
+    async fn run_cleanup(&self) {
+        let ttl_hours = self.cleanup_ttl_hours.load(Ordering::Relaxed) as u64;
+        let ttl = Duration::from_secs(ttl_hours * 3600);
+        let now = std::time::SystemTime::now();
+
+        let Ok(mut entries) = tokio::fs::read_dir(&self.transcode_dir).await else {
+            return;
+        };
+        while let Ok(Some(entry)) = entries.next_entry().await {
+            let path = entry.path();
+            if !path.is_dir() {
+                continue;
+            }
+            let playlist = path.join("playlist.m3u8");
+            if let Ok(meta) = tokio::fs::metadata(&playlist).await {
+                if let Ok(modified) = meta.modified() {
+                    if let Ok(age) = now.duration_since(modified) {
+                        if age > ttl {
+                            warn!("cleanup: removing stale transcode {:?}", path);
+                            let _ = tokio::fs::remove_dir_all(&path).await;
+                        }
+                    }
+                }
+            }
+        }
+    }
+}
+
+// ============================================================================
+// FFmpeg helper
+// ============================================================================
+
+async fn do_transcode(src: &Path, out_dir: &Path, playlist: &Path) -> TranscodeStatus {
+    let segment_pattern = out_dir.join("seg%05d.ts");
+
+    let mut child = match tokio::process::Command::new("ffmpeg")
+        .args([
+            "-i",
+            src.to_str().unwrap_or(""),
+            "-c:v",
+            "libx264",
+            "-preset",
+            "fast",
+            "-crf",
+            "23",
+            "-c:a",
+            "aac",
+            "-b:a",
+            "128k",
+            "-hls_time",
+            "6",
+            "-hls_list_size",
+            "0",
+            "-hls_flags",
+            "independent_segments",
+            "-hls_segment_filename",
+            segment_pattern.to_str().unwrap_or(""),
+            playlist.to_str().unwrap_or(""),
+        ])
+        .stdout(std::process::Stdio::null())
+        .stderr(std::process::Stdio::null())
+        .spawn()
+    {
+        Ok(c) => c,
+        Err(e) => return TranscodeStatus::Failed(format!("ffmpeg spawn error: {}", e)),
+    };
+
+    // Poll for playlist.m3u8 -- it appears after the first segment is written,
+    // allowing the client to start playback before transcoding is complete.
+    let start = Instant::now();
+    let timeout = Duration::from_secs(60);
+    loop {
+        if playlist.exists() {
+            return TranscodeStatus::Ready;
+        }
+        if start.elapsed() > timeout {
+            let _ = child.kill().await;
+            return TranscodeStatus::Failed(
+                "timeout waiting for transcode to start".into(),
+            );
+        }
+        match child.try_wait() {
+            Ok(Some(status)) => {
+                return if playlist.exists() {
+                    TranscodeStatus::Ready
+                } else if status.success() {
+                    TranscodeStatus::Failed(
+                        "ffmpeg exited but produced no playlist".into(),
+                    )
+                } else {
+                    TranscodeStatus::Failed(
+                        "ffmpeg exited with non-zero status".into(),
+                    )
+                };
+            }
+            Err(e) => return TranscodeStatus::Failed(e.to_string()),
+            Ok(None) => {}
+        }
+        tokio::time::sleep(Duration::from_millis(100)).await;
+    }
+}