use crate::content_hash; use crate::database::PreviewDao; use crate::libraries::Library; use crate::otel::global_tracer; use crate::video::ffmpeg::{generate_preview_clip, get_duration_seconds_blocking}; use crate::video::hls_paths; use actix::prelude::*; use log::{debug, error, info, warn}; use opentelemetry::KeyValue; use opentelemetry::trace::{Span, Status, Tracer}; use std::io::Result; use std::path::{Path, PathBuf}; use std::process::{Command, Stdio}; use std::sync::{Arc, Mutex}; use tokio::sync::Semaphore; // ffmpeg -i test.mp4 -c:v h264 -flags +cgop -g 30 -hls_time 3 out.m3u8 // ffmpeg -i "filename.mp4" -preset veryfast -c:v libx264 -f hls -hls_list_size 100 -hls_time 2 -crf 24 -vf scale=1080:-2,setsar=1:1 attempt/vid_out.m3u8 pub struct StreamActor; impl Actor for StreamActor { type Context = Context; } /// A video paired with its content hash, ready to be queued for HLS /// playlist generation. Hash is required because all output paths are /// keyed on it; callers that lack a hash (rows mid-backfill) must skip /// the video rather than fabricate one. #[derive(Debug, Clone)] pub struct VideoToQueue { pub video_path: PathBuf, pub content_hash: String, } pub fn generate_video_thumbnail(path: &Path, destination: &Path) -> std::io::Result<()> { // Probe duration up front and seek to ~50% — gives a more // representative frame than a fixed offset (skipping title cards on // long videos, landing inside the clip on 1–2s Snapchat MP4s) and // sidesteps the seek-past-EOF class of bug entirely. When duration // probing fails (LRV files, fragmented MP4s, ffprobe missing) fall // back to the first frame: ugly but reliable. // // -vf scale + -c:v mjpeg mirrors `generate_image_thumbnail_ffmpeg`. The // filter chain matters as much as the scale does: without it, ffmpeg // hands the decoded frame straight to the mjpeg encoder, which rejects // any non-yuvj420p source ("Non full-range YUV is non-standard"). The // filter chain lets ffmpeg auto-insert the pix_fmt converter the // encoder needs, which is how the image-thumbnail path already handles // the same class of source. let seek = get_duration_seconds_blocking(path).map(|d| format!("{:.3}", d / 2.0)); let mut cmd = Command::new("ffmpeg"); cmd.arg("-y"); if let Some(s) = &seek { cmd.arg("-ss").arg(s); } let output = cmd .arg("-i") .arg(path) .arg("-vframes") .arg("1") .arg("-vf") .arg("scale=200:-1") .arg("-f") .arg("image2") .arg("-c:v") .arg("mjpeg") .arg(destination) .output()?; if !output.status.success() { return Err(std::io::Error::other(format!( "ffmpeg failed ({}): {}", output.status, String::from_utf8_lossy(&output.stderr).trim() ))); } // ffmpeg can exit 0 without writing a frame for malformed files where // the probe duration lies. Confirm a non-empty file actually landed — // returning Err makes the caller write the `.unsupported` sentinel so // we stop re-detecting on every scan. let wrote = std::fs::metadata(destination) .map(|m| m.len() > 0) .unwrap_or(false); if !wrote { return Err(std::io::Error::other( "ffmpeg exited successfully but produced no thumbnail output", )); } Ok(()) } /// Use ffmpeg to extract a 200px-wide thumbnail from formats the `image` crate /// can't decode (RAW: NEF/ARW, HEIC/HEIF). Writes JPEG bytes to `destination` /// regardless of its extension. pub fn generate_image_thumbnail_ffmpeg(path: &Path, destination: &Path) -> std::io::Result<()> { let output = Command::new("ffmpeg") .arg("-y") .arg("-i") .arg(path) .arg("-vframes") .arg("1") .arg("-vf") .arg("scale=200:-1") .arg("-f") .arg("image2") .arg("-c:v") .arg("mjpeg") .arg(destination) .output()?; if !output.status.success() { return Err(std::io::Error::other(format!( "ffmpeg failed ({}): {}", output.status, String::from_utf8_lossy(&output.stderr).trim() ))); } Ok(()) } /// Video stream metadata needed to pick HLS encode settings. Populated by /// a single ffprobe call to avoid spawning multiple subprocesses per video. #[derive(Debug, Default)] struct VideoStreamMeta { is_h264: bool, /// Rotation in degrees (0/90/180/270). Checks both the legacy `rotate` /// stream tag and the modern display-matrix side data. rotation: i32, } /// Probe video stream metadata in one ffprobe call. Returns default (codec /// unknown, rotation 0) on any failure — callers fall back to transcoding. async fn probe_video_stream_meta(video_path: &str) -> VideoStreamMeta { let output = tokio::process::Command::new("ffprobe") .arg("-v") .arg("error") .arg("-select_streams") .arg("v:0") .arg("-print_format") .arg("json") .arg("-show_entries") .arg("stream=codec_name:stream_tags=rotate:side_data_list") .arg(video_path) .output() .await; let Ok(output) = output else { warn!("Failed to run ffprobe for {}", video_path); return VideoStreamMeta::default(); }; if !output.status.success() { warn!( "ffprobe failed for {}: {}", video_path, String::from_utf8_lossy(&output.stderr).trim() ); return VideoStreamMeta::default(); } let Ok(json) = serde_json::from_slice::(&output.stdout) else { warn!("ffprobe returned non-JSON for {}", video_path); return VideoStreamMeta::default(); }; let stream = &json["streams"][0]; let is_h264 = stream .get("codec_name") .and_then(|v| v.as_str()) .map(|s| s == "h264") .unwrap_or(false); // Prefer legacy `tags.rotate` (older containers); fall back to the // display-matrix side data (iPhone and other modern recorders). let rotation = stream .get("tags") .and_then(|t| t.get("rotate")) .and_then(|r| r.as_str()) .and_then(|s| s.parse::().ok()) .filter(|r| *r != 0) .or_else(|| { stream .get("side_data_list") .and_then(|l| l.as_array()) .and_then(|arr| { arr.iter() .find_map(|sd| sd.get("rotation").and_then(|r| r.as_f64())) }) .map(|f| f.abs() as i32) .filter(|r| *r != 0) }) .unwrap_or(0); debug!( "Probed {}: codec_h264={}, rotation={}°", video_path, is_h264, rotation ); VideoStreamMeta { is_h264, rotation } } /// Probe the max keyframe interval (GOP) in the first ~30s of a video. /// Returns `None` on probe failure or if we couldn't see at least two keyframes. /// /// Used to decide between stream-copy and transcode: HLS needs segments to /// start on keyframes, so if the source GOP exceeds `hls_time`, copying /// produces oversized/glitchy segments and we need to re-encode. async fn get_max_gop_seconds(video_path: &str) -> Option { let output = tokio::process::Command::new("ffprobe") .arg("-v") .arg("error") .arg("-select_streams") .arg("v:0") .arg("-skip_frame") .arg("nokey") .arg("-show_entries") .arg("frame=pts_time") .arg("-of") .arg("csv=p=0") .arg("-read_intervals") .arg("%+30") .arg(video_path) .output() .await .ok()?; if !output.status.success() { warn!( "ffprobe GOP check failed for {}: {}", video_path, String::from_utf8_lossy(&output.stderr).trim() ); return None; } let times: Vec = String::from_utf8_lossy(&output.stdout) .lines() .filter_map(|l| l.trim().parse::().ok()) .collect(); if times.len() < 2 { return None; } let max_gop = times .windows(2) .map(|w| w[1] - w[0]) .fold(0.0_f64, f64::max); debug!( "Max GOP in first {} keyframes of {}: {:.2}s", times.len(), video_path, max_gop ); Some(max_gop) } pub struct VideoPlaylistManager { video_dir: PathBuf, playlist_generator: Addr, } impl VideoPlaylistManager { pub fn new>( video_dir: P, playlist_generator: Addr, ) -> Self { Self { video_dir: video_dir.into(), playlist_generator, } } } impl Actor for VideoPlaylistManager { type Context = Context; } impl Handler for VideoPlaylistManager { type Result = (); fn handle(&mut self, msg: QueueVideosMessage, _ctx: &mut Self::Context) -> Self::Result { if msg.videos.is_empty() { return; } let video_dir = self.video_dir.clone(); let playlist_generator = self.playlist_generator.clone(); let mut queued = 0usize; let mut already_present = 0usize; for VideoToQueue { video_path, content_hash, } in msg.videos { let playlist = hls_paths::playlist_for_hash(&video_dir, &content_hash); let sentinel = hls_paths::sentinel_for_hash(&video_dir, &content_hash); if playlist.exists() || sentinel.exists() { already_present += 1; continue; } debug!( "Queueing playlist generation for {} (hash={})", video_path.display(), short_hash(&content_hash) ); playlist_generator.do_send(GeneratePlaylistMessage { video_path, content_hash, }); queued += 1; } info!( "Queue tick: {} queued, {} skipped (playlist or sentinel already on disk)", queued, already_present ); } } #[derive(Message)] #[rtype(result = "()")] pub struct QueueVideosMessage { pub videos: Vec, } #[derive(Message)] #[rtype(result = "Result<()>")] pub struct GeneratePlaylistMessage { pub video_path: PathBuf, pub content_hash: String, } pub struct PlaylistGenerator { semaphore: Arc, video_dir: PathBuf, } impl PlaylistGenerator { pub(crate) fn new>(video_dir: P) -> Self { // Concurrency is tunable via HLS_CONCURRENCY so operators can dial // it to their hardware: 1 on weak Synology boxes to avoid thermal // throttling, higher on desktops with spare cores. let concurrency = std::env::var("HLS_CONCURRENCY") .ok() .and_then(|v| v.parse::().ok()) .filter(|&n| n > 0) .unwrap_or(2); info!("PlaylistGenerator: concurrency={}", concurrency); PlaylistGenerator { semaphore: Arc::new(Semaphore::new(concurrency)), video_dir: video_dir.into(), } } } impl Actor for PlaylistGenerator { type Context = Context; } impl Handler for PlaylistGenerator { type Result = ResponseFuture>; fn handle(&mut self, msg: GeneratePlaylistMessage, _ctx: &mut Self::Context) -> Self::Result { let video_file = msg.video_path.to_str().unwrap().to_owned(); let content_hash_str = msg.content_hash.clone(); let semaphore = self.semaphore.clone(); let video_dir = self.video_dir.clone(); let hash_dir = content_hash::hls_dir(&video_dir, &content_hash_str); let playlist_path = hls_paths::playlist_for_hash(&video_dir, &content_hash_str); let sentinel_path = hls_paths::sentinel_for_hash(&video_dir, &content_hash_str); let segment_template = hls_paths::segment_template_for_hash(&video_dir, &content_hash_str); let playlist_file = playlist_path.to_string_lossy().to_string(); let segment_pattern = segment_template.to_string_lossy().to_string(); let tracer = global_tracer(); let mut span = tracer .span_builder("playlistgenerator.generate_playlist") .with_attributes(vec![ KeyValue::new("video_file", video_file.clone()), KeyValue::new("content_hash", content_hash_str.clone()), KeyValue::new("playlist_file", playlist_file.clone()), ]) .start(&tracer); Box::pin(async move { let wait_start = std::time::Instant::now(); let permit = semaphore .acquire_owned() .await .expect("Unable to acquire semaphore"); debug!( "Waited for {:?} before starting ffmpeg", wait_start.elapsed() ); span.add_event( "Waited for FFMPEG semaphore", vec![KeyValue::new( "wait_time", wait_start.elapsed().as_secs_f64(), )], ); if playlist_path.exists() { debug!("Playlist already exists: {}", playlist_file); span.set_status(Status::error(format!( "Playlist already exists: {}", playlist_file ))); return Err(std::io::Error::from(std::io::ErrorKind::AlreadyExists)); } // Ensure the shard + hash directory exist. Idempotent — the // dir may already be present from a prior attempt that wrote // a sentinel before being cleared for retry. if let Err(e) = tokio::fs::create_dir_all(&hash_dir).await { error!( "Failed to create HLS hash dir {}: {}", hash_dir.display(), e ); span.set_status(Status::error(format!("mkdir failed: {}", e))); return Err(e); } // One ffprobe call for codec + rotation metadata. let stream_meta = probe_video_stream_meta(&video_file).await; let is_h264 = stream_meta.is_h264; let rotation = stream_meta.rotation; let has_rotation = rotation != 0; // Stream-copy is only safe when the source GOP fits inside a // single HLS segment. Otherwise ffmpeg has to extend segments // past hls_time to land on a keyframe, producing uneven // segments and seeking glitches. const HLS_SEGMENT_SECONDS: f64 = 3.0; let gop_ok = if is_h264 && !has_rotation { match get_max_gop_seconds(&video_file).await { Some(g) if g > HLS_SEGMENT_SECONDS => { info!( "Video {} has long GOP ({:.1}s > {}s), transcoding for segment alignment", video_file, g, HLS_SEGMENT_SECONDS ); false } Some(_) => true, None => { // Probe failed — be conservative and transcode rather // than risk broken segments from a mystery source. debug!( "GOP probe failed for {}, transcoding to be safe", video_file ); false } } } else { false }; let use_copy = is_h264 && !has_rotation && gop_ok; if has_rotation { info!( "Video {} has rotation metadata ({}°), transcoding to apply rotation", video_file, rotation ); span.add_event( "Transcoding due to rotation", vec![KeyValue::new("rotation_degrees", rotation as i64)], ); } else if use_copy { info!("Video {} is already h264, using stream copy", video_file); span.add_event("Using stream copy (h264 detected)", vec![]); } else if is_h264 { info!( "Video {} is h264 but needs transcoding for GOP alignment", video_file ); span.add_event("Transcoding for GOP alignment", vec![]); } else { info!("Video {} needs transcoding to h264", video_file); span.add_event("Transcoding to h264", vec![]); } // Encode to a .tmp playlist alongside the final inside the // hash dir, so a concurrent scan never sees a half-written // .m3u8 as "done". Segments use the hash-keyed template; // ffmpeg writes them next to the playlist (relative refs). let playlist_tmp = format!("{}.tmp", playlist_file); let mut cmd = tokio::process::Command::new("ffmpeg"); cmd.arg("-y").arg("-i").arg(&video_file); if use_copy { cmd.arg("-c:v").arg("copy"); cmd.arg("-c:a").arg("aac"); } else { let nvenc = crate::video::ffmpeg::is_nvenc_available().await; if nvenc { // NVENC: no CRF, use VBR + target CQ. p1 = fastest // preset — prioritizes encoder throughput over bitrate // efficiency. CQ 23 roughly matches libx264 crf 21 // visually; NVENC has slightly lower compression // efficiency per quality. cmd.arg("-c:v").arg("h264_nvenc"); cmd.arg("-preset").arg("p1"); cmd.arg("-rc").arg("vbr"); cmd.arg("-cq").arg("23"); cmd.arg("-pix_fmt").arg("yuv420p"); } else { cmd.arg("-c:v").arg("h264"); cmd.arg("-crf").arg("21"); cmd.arg("-preset").arg("veryfast"); } cmd.arg("-vf").arg("scale='min(1080,iw)':-2,setsar=1:1"); cmd.arg("-c:a").arg("aac"); // Force an IDR frame every hls_time seconds so each HLS // segment starts on a keyframe — accurate seeking without // players having to decode from a prior segment. cmd.arg("-force_key_frames").arg("expr:gte(t,n_forced*3)"); } // -f hls is required because the playlist is written to a .tmp // path during encoding — ffmpeg normally infers the muxer from // the output extension and doesn't recognize ".m3u8.tmp". cmd.arg("-f").arg("hls"); cmd.arg("-hls_time").arg("3"); cmd.arg("-hls_list_size").arg("0"); cmd.arg("-hls_playlist_type").arg("vod"); // independent_segments advertises that each segment can be // decoded without reference to any other — the matching guarantee // for the forced keyframes above. cmd.arg("-hls_flags").arg("independent_segments"); cmd.arg("-hls_segment_filename").arg(&segment_pattern); cmd.arg(&playlist_tmp); cmd.stdout(Stdio::null()); cmd.stderr(Stdio::piped()); cmd.kill_on_drop(true); // Spawn + wait under a timeout so a hung ffmpeg (corrupt source, // NFS stall, etc.) doesn't permanently hold a semaphore slot. // Default is generous — a long 4K transcode on CPU can take hours. let timeout_secs = std::env::var("HLS_TIMEOUT_SECONDS") .ok() .and_then(|v| v.parse::().ok()) .unwrap_or(7200); let ffmpeg_result = match cmd.spawn() { Ok(child) => { match tokio::time::timeout( std::time::Duration::from_secs(timeout_secs), child.wait_with_output(), ) .await { Ok(res) => res .inspect_err(|e| { error!("Failed to wait on ffmpeg child process: {}", e) }) .map_err(|e| std::io::Error::other(e.to_string())), Err(_) => Err(std::io::Error::other(format!( "ffmpeg exceeded {}s timeout", timeout_secs ))), } } Err(e) => { error!("Failed to spawn ffmpeg: {}", e); Err(std::io::Error::other(e.to_string())) } }; drop(permit); let success = matches!(&ffmpeg_result, Ok(out) if out.status.success()); if success { if let Err(e) = tokio::fs::rename(&playlist_tmp, &playlist_path).await { error!( "ffmpeg succeeded but rename {} -> {} failed: {}", playlist_tmp, playlist_file, e ); cleanup_partial_hls(&hash_dir).await; span.set_status(Status::error(format!("rename failed: {}", e))); return Err(e); } debug!("Playlist complete: {}", playlist_file); span.set_status(Status::Ok); Ok(()) } else { let detail = match &ffmpeg_result { Ok(out) => format!( "exit {}: {}", out.status, String::from_utf8_lossy(&out.stderr).trim() ), Err(e) => format!("ffmpeg failed: {}", e), }; error!("ffmpeg failed for {}: {}", video_file, detail); cleanup_partial_hls(&hash_dir).await; if let Err(se) = tokio::fs::write(&sentinel_path, b"").await { warn!( "Failed to write playlist sentinel {}: {}", sentinel_path.display(), se ); } else { info!( "Wrote playlist sentinel {} so future scans skip {}", sentinel_path.display(), video_file ); } span.set_status(Status::error(detail.clone())); Err(std::io::Error::other(detail)) } }) } } /// Delete the partial playlist (.tmp) and any segment files left behind by /// a failed ffmpeg run. Wipes every non-sentinel file in the hash dir; /// retains the sentinel if one has already been written by an earlier /// caller in the same path (today there is none, but kept defensively so /// the function is safe to call after sentinel write too). async fn cleanup_partial_hls(hash_dir: &Path) { let Ok(mut entries) = tokio::fs::read_dir(hash_dir).await else { return; }; while let Ok(Some(entry)) = entries.next_entry().await { let path = entry.path(); let is_sentinel = path .file_name() .and_then(|n| n.to_str()) .map(|n| n == hls_paths::UNSUPPORTED_SENTINEL_FILENAME) .unwrap_or(false); if is_sentinel { continue; } if let Err(e) = tokio::fs::remove_file(&path).await { warn!( "Failed to remove partial HLS file {}: {}", path.display(), e ); } } } /// First 16 chars of a content hash for log lines. Short enough to keep /// log volume sane, long enough that distinct hashes don't collide in /// practice. fn short_hash(hash: &str) -> &str { let end = hash .char_indices() .nth(16) .map(|(i, _)| i) .unwrap_or(hash.len()); &hash[..end] } #[derive(Message)] #[rtype(result = "()")] pub struct GeneratePreviewClipMessage { pub video_path: String, } pub struct PreviewClipGenerator { semaphore: Arc, preview_clips_dir: String, libraries: Vec, preview_dao: Arc>>, } impl PreviewClipGenerator { pub fn new( preview_clips_dir: String, libraries: Vec, preview_dao: Arc>>, ) -> Self { PreviewClipGenerator { semaphore: Arc::new(Semaphore::new(2)), preview_clips_dir, libraries, preview_dao, } } /// Strip whichever library root actually contains `video_path`. /// Falls back to the first library if none match, so we never /// accidentally emit the absolute input path as the output path /// (which ffmpeg rejects as "cannot edit existing files in place"). fn relativize(&self, video_path: &str) -> String { for lib in &self.libraries { if let Some(stripped) = video_path.strip_prefix(&lib.root_path) { return stripped.trim_start_matches(['/', '\\']).replace('\\', "/"); } } video_path .trim_start_matches(['/', '\\']) .replace('\\', "/") } } impl Actor for PreviewClipGenerator { type Context = Context; } impl Handler for PreviewClipGenerator { type Result = ResponseFuture<()>; fn handle( &mut self, msg: GeneratePreviewClipMessage, _ctx: &mut Self::Context, ) -> Self::Result { let semaphore = self.semaphore.clone(); let preview_clips_dir = self.preview_clips_dir.clone(); let preview_dao = self.preview_dao.clone(); let video_path = msg.video_path; // Resolve against whichever library actually owns this video. let relative_path = self.relativize(&video_path); Box::pin(async move { let permit = semaphore .acquire_owned() .await .expect("Unable to acquire preview semaphore"); // Update status to processing { let otel_ctx = opentelemetry::Context::current(); let mut dao = preview_dao.lock().expect("Unable to lock PreviewDao"); let _ = dao.update_status(&otel_ctx, &relative_path, "processing", None, None, None); } // Compute output path: join preview_clips_dir with relative path, change ext to .mp4 let output_path = PathBuf::from(&preview_clips_dir) .join(&relative_path) .with_extension("mp4"); let output_str = output_path.to_string_lossy().to_string(); let video_path_owned = video_path.clone(); let relative_path_owned = relative_path.clone(); tokio::spawn(async move { match generate_preview_clip(&video_path_owned, &output_str).await { Ok((duration, size)) => { info!( "Preview clip complete for '{}' ({:.1}s, {} bytes)", relative_path_owned, duration, size ); let otel_ctx = opentelemetry::Context::current(); let mut dao = preview_dao.lock().expect("Unable to lock PreviewDao"); let _ = dao.update_status( &otel_ctx, &relative_path_owned, "complete", Some(duration as f32), Some(size as i32), None, ); } Err(e) => { error!( "Failed to generate preview clip for '{}': {}", relative_path_owned, e ); let otel_ctx = opentelemetry::Context::current(); let mut dao = preview_dao.lock().expect("Unable to lock PreviewDao"); let _ = dao.update_status( &otel_ctx, &relative_path_owned, "failed", None, None, Some(&e.to_string()), ); } } drop(permit); }); }) } }