diff --git a/Cargo.lock b/Cargo.lock index 0310f6e..b99e5e6 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -43,6 +43,24 @@ dependencies = [ "tracing", ] +[[package]] +name = "adapter-local-files" +version = "0.1.0" +dependencies = [ + "async-trait", + "base64 0.22.1", + "chrono", + "domain", + "infra-wiring", + "serde", + "serde_json", + "sqlx", + "tokio", + "tracing", + "uuid", + "walkdir", +] + [[package]] name = "adapter-postgres" version = "0.1.0" @@ -2102,6 +2120,15 @@ version = "1.0.23" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "9774ba4a74de5f7b1c1451ed6cd5285a32eddb5cccb8cc655a4e50009e06477f" +[[package]] +name = "same-file" +version = "1.0.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "93fc1dc3aaa9bfed95e02e6eadabb4baf7e3078b0bd1b4d7b6b0b68378900502" +dependencies = [ + "winapi-util", +] + [[package]] name = "schannel" version = "0.1.29" @@ -3057,6 +3084,16 @@ version = "0.9.5" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "0b928f33d975fc6ad9f86c8f283853ad26bdd5b10b7f1542aa2fa15e2289105a" +[[package]] +name = "walkdir" +version = "2.5.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "29790946404f91d9c5d06f9874efddea1dc06c5efe94541a7d6863108e3a5e4b" +dependencies = [ + "same-file", + "winapi-util", +] + [[package]] name = "want" version = "0.3.1" @@ -3172,6 +3209,15 @@ dependencies = [ "wasite", ] +[[package]] +name = "winapi-util" +version = "0.1.11" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c2a7b1c03c876122aa43f3020e6c3c3ee5c05081c9a00739faf7503aeba10d22" +dependencies = [ + "windows-sys 0.61.2", +] + [[package]] name = "windows-core" version = "0.62.2" diff --git a/Cargo.toml b/Cargo.toml index 91eb2d2..0c3ebb0 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/postgres", "crates/adapters/auth", "crates/adapters/jellyfin"] +members = ["crates/domain", "crates/application", "crates/api-types", "crates/infra-wiring", "crates/adapters/adapter-common", "crates/adapters/sqlite", "crates/adapters/postgres", "crates/adapters/auth", "crates/adapters/jellyfin", "crates/adapters/local-files"] exclude = ["k-tv-backend", "k-tv-frontend"] resolver = "2" diff --git a/crates/adapters/local-files/Cargo.toml b/crates/adapters/local-files/Cargo.toml new file mode 100644 index 0000000..9e3850a --- /dev/null +++ b/crates/adapters/local-files/Cargo.toml @@ -0,0 +1,18 @@ +[package] +name = "adapter-local-files" +version = "0.1.0" +edition = "2024" + +[dependencies] +domain = { workspace = true } +infra-wiring = { workspace = true, features = ["sqlite"] } +async-trait = { workspace = true } +sqlx = { workspace = true, features = ["sqlite"] } +tokio = { workspace = true } +tracing = { workspace = true } +serde = { workspace = true } +serde_json = { workspace = true } +chrono = { workspace = true } +uuid = { workspace = true } +walkdir = "2" +base64 = "0.22" diff --git a/crates/adapters/local-files/src/config.rs b/crates/adapters/local-files/src/config.rs new file mode 100644 index 0000000..f34e746 --- /dev/null +++ b/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, + /// How long (hours) to keep transcode cache entries. Passed to TranscodeManager. + pub cleanup_ttl_hours: u32, +} diff --git a/crates/adapters/local-files/src/index.rs b/crates/adapters/local-files/src/index.rs new file mode 100644 index 0000000..0b3616e --- /dev/null +++ b/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 { + 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>>, + 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, + 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 = + 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 { + 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 { + 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 = seen.into_iter().collect(); + dirs.sort(); + dirs + } +} diff --git a/crates/adapters/local-files/src/lib.rs b/crates/adapters/local-files/src/lib.rs new file mode 100644 index 0000000..bf9d426 --- /dev/null +++ b/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, + pub transcode_manager: Option>, +} + +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, + } + } +} diff --git a/crates/adapters/local-files/src/provider.rs b/crates/adapters/local-files/src/provider.rs new file mode 100644 index 0000000..1734b56 --- /dev/null +++ b/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, + base_url: String, + transcode_manager: Option>, +} + +const SHORT_DURATION_SECS: u32 = 1200; // 20 minutes + +impl LocalFilesProvider { + pub fn new( + index: Arc, + config: &LocalFilesConfig, + transcode_manager: Option>, + ) -> 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> { + 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> { + 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 { + 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> { + 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 { + decode_id(&MediaItemId::new(encoded)) +} diff --git a/crates/adapters/local-files/src/scanner.rs b/crates/adapters/local-files/src/scanner.rs new file mode 100644 index 0000000..b8fb73f --- /dev/null +++ b/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, + /// Ancestor directory names between root and file (excluding root itself). + pub tags: Vec, + /// 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 { + 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::>() + .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 = 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 { + let chars: Vec = 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 { + #[derive(serde::Deserialize)] + struct Fmt { + duration: Option, + } + #[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)); + } +} diff --git a/crates/adapters/local-files/src/transcoder.rs b/crates/adapters/local-files/src/transcoder.rs new file mode 100644 index 0000000..ca2bf57 --- /dev/null +++ b/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, + active: Arc>>>>, +} + +impl TranscodeManager { + pub fn new(transcode_dir: PathBuf, cleanup_ttl_hours: u32) -> Arc { + 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::>(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; + } +}