//! Constrained inspection and archiving of exact, public YouTube videos.
//!
//! VideoForge delegates extraction to a trusted external `yt-dlp` executable.
//! It does not expose arbitrary arguments, authentication, or access-control
//! workarounds. Callers must explicitly confirm authorization before archiving.
use std::collections::{BTreeSet, VecDeque};
use std::ffi::OsStr;
use std::fs::{self, File, OpenOptions};
use std::io::{self, Read, Seek, Write};
use std::path::{Component, Path, PathBuf};
use std::process::{Command, ExitStatus, Stdio};
use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::{Arc, mpsc};
use std::thread;
use std::time::{Duration, Instant};
use fs2::FileExt;
#[cfg(windows)]
use process_wrap::std::JobObject;
#[cfg(unix)]
use process_wrap::std::ProcessGroup;
use process_wrap::std::{StdChildWrapper, StdCommandWrap};
use serde::{Deserialize, Deserializer, Serialize, de};
use tempfile::{Builder as TempBuilder, TempDir};
use thiserror::Error;
use url::Url;
/// Operational guidance for automated agents integrating VideoForge.
pub const AGENT_GUIDE: &str = include_str!("../AGENT_GUIDE.md");
const DEFAULT_MAX_BYTES: u64 = 2 * 1024 * 1024 * 1024;
const DEFAULT_TIMEOUT: Duration = Duration::from_secs(30 * 60);
const METADATA_LIMIT: usize = 8 * 1024 * 1024;
const MANIFEST_LIMIT: u64 = 2 * 1024 * 1024;
const DIAGNOSTIC_LIMIT: usize = 64 * 1024;
const LINE_LIMIT: usize = 256 * 1024;
const MANIFEST_NAME: &str = "videoforge-video.json";
const INSTALL_MARKER_NAME: &str = ".videoforge-installing";
const MANIFEST_SCHEMA: u32 = 1;
const POLL_INTERVAL: Duration = Duration::from_millis(25);
/// A video service supported by VideoForge.
#[derive(Clone, Copy, Debug, Eq, Hash, PartialEq, Serialize, Deserialize)]
#[serde(rename_all = "lowercase")]
pub enum VideoProvider {
/// YouTube public videos.
Youtube,
}
impl VideoProvider {
/// Human-readable provider name.
#[must_use]
pub const fn label(self) -> &'static str {
match self {
Self::Youtube => "YouTube",
}
}
/// Stable lowercase provider identifier.
#[must_use]
pub const fn slug(self) -> &'static str {
match self {
Self::Youtube => "youtube",
}
}
}
/// A validated canonical video identity.
#[derive(Clone, Debug, Eq, PartialEq, Hash, Serialize)]
pub struct VideoTarget {
provider: VideoProvider,
id: String,
canonical_url: String,
}
impl<'de> Deserialize<'de> for VideoTarget {
fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
where
D: Deserializer<'de>,
{
#[derive(Deserialize)]
struct Fields {
provider: VideoProvider,
id: String,
canonical_url: String,
}
let fields = Fields::deserialize(deserializer)?;
let target = parse_video_target(&fields.canonical_url).map_err(de::Error::custom)?;
if fields.provider != target.provider
|| fields.id != target.id
|| fields.canonical_url != target.canonical_url
{
return Err(de::Error::custom(
"video target fields are not one canonical identity",
));
}
Ok(target)
}
}
impl VideoTarget {
/// Provider for this target.
#[must_use]
pub const fn provider(&self) -> VideoProvider {
self.provider
}
/// Exact provider video identifier.
#[must_use]
pub fn id(&self) -> &str {
&self.id
}
/// Canonical HTTPS URL used for every extractor invocation.
#[must_use]
pub fn canonical_url(&self) -> &str {
&self.canonical_url
}
}
/// Parse and canonicalize one exact public YouTube video URL.
pub fn parse_video_target(input: &str) -> Result<VideoTarget, VideoForgeError> {
if input.trim() != input || input.chars().any(char::is_control) {
return Err(VideoForgeError::InvalidTarget(
"URL must not contain surrounding whitespace or control characters".into(),
));
}
let url = Url::parse(input)
.map_err(|error| VideoForgeError::InvalidTarget(format!("invalid URL: {error}")))?;
if url.scheme() != "https" {
return Err(VideoForgeError::InvalidTarget(
"only HTTPS URLs are accepted".into(),
));
}
if !url.username().is_empty() || url.password().is_some() {
return Err(VideoForgeError::InvalidTarget(
"URL credentials are not accepted".into(),
));
}
let authority = input
.split_once("://")
.and_then(|(_, rest)| rest.split(['/', '?', '#']).next())
.unwrap_or_default();
if authority.contains('@') {
return Err(VideoForgeError::InvalidTarget(
"URL credentials are not accepted".into(),
));
}
if url.port().is_some() || authority.contains(':') {
return Err(VideoForgeError::InvalidTarget(
"explicit URL ports are not accepted".into(),
));
}
if url.fragment().is_some() {
return Err(VideoForgeError::InvalidTarget(
"URL fragments are not accepted".into(),
));
}
let host = url
.host_str()
.ok_or_else(|| VideoForgeError::InvalidTarget("URL has no host".into()))?;
let is_short = host == "youtu.be";
if !is_short && host != "youtube.com" && host != "www.youtube.com" && host != "m.youtube.com" {
return Err(VideoForgeError::InvalidTarget(
"unsupported YouTube host".into(),
));
}
let path: Vec<_> = url
.path_segments()
.ok_or_else(|| VideoForgeError::InvalidTarget("invalid URL path".into()))?
.collect();
let id = if is_short {
if path.len() != 1 || path[0].is_empty() || url.query().is_some() {
return Err(VideoForgeError::InvalidTarget(
"youtu.be URLs must have exactly one ID path component".into(),
));
}
path[0].to_owned()
} else if path == ["watch"] {
let Some(id) = url.query().and_then(|query| query.strip_prefix("v=")) else {
return Err(VideoForgeError::InvalidTarget(
"watch URLs must have exactly one v parameter and no playlist".into(),
));
};
if id.contains(['&', '=', '%']) {
return Err(VideoForgeError::InvalidTarget(
"watch URLs must contain one literal video ID".into(),
));
}
id.to_owned()
} else if path.len() == 2 && matches!(path[0], "shorts" | "live" | "embed") {
if url.query().is_some() {
return Err(VideoForgeError::InvalidTarget(
"extra query parameters and playlists are not accepted".into(),
));
}
path[1].to_owned()
} else {
return Err(VideoForgeError::InvalidTarget(
"unsupported YouTube video URL shape".into(),
));
};
validate_video_id(&id)?;
Ok(VideoTarget {
provider: VideoProvider::Youtube,
canonical_url: format!("https://www.youtube.com/watch?v={id}"),
id,
})
}
fn validate_video_id(id: &str) -> Result<(), VideoForgeError> {
if id.len() != 11
|| !id
.bytes()
.all(|byte| byte.is_ascii_alphanumeric() || byte == b'_' || byte == b'-')
{
return Err(VideoForgeError::InvalidTarget(
"YouTube IDs must be exactly 11 ASCII letters, digits, underscores, or hyphens".into(),
));
}
Ok(())
}
/// A constrained quality policy with a fixed internal `yt-dlp` selector.
#[derive(Clone, Copy, Debug, Default, Eq, PartialEq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum VideoQuality {
/// Best available video and audio.
#[default]
Best,
/// Best available result no taller than 1080p.
Max1080p,
/// Best available result no taller than 720p.
Max720p,
}
impl VideoQuality {
fn selector(self) -> &'static str {
match self {
Self::Best => "bestvideo*+bestaudio/best",
Self::Max1080p => "bestvideo*[height<=1080]+bestaudio/best[height<=1080]",
Self::Max720p => "bestvideo*[height<=720]+bestaudio/best[height<=720]",
}
}
}
/// Rights information reported by the source, plus conservative usage notices.
#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
pub struct VideoRights {
/// Canonical source URL.
pub source: String,
/// Reported uploader name.
pub uploader: Option<String>,
/// Reported uploader identifier.
pub uploader_id: Option<String>,
/// Reported channel name.
pub channel: Option<String>,
/// Reported channel identifier.
pub channel_id: Option<String>,
/// Video license label reported by `yt-dlp`, if any.
pub reported_license: Option<String>,
/// Suggested source attribution, not a determination of sufficiency.
pub attribution: String,
/// Conservative warning about video content rights.
pub redistribution_warning: String,
/// Reminder that `yt-dlp` software licensing is unrelated to content rights.
pub yt_dlp_license_notice: String,
}
/// Stable metadata retained by VideoForge.
#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
pub struct VideoSummary {
/// Canonical video identity.
pub target: VideoTarget,
/// Video title.
pub title: String,
/// Video description, when reported.
pub description: Option<String>,
/// Upload date in `YYYYMMDD` form, when reported.
pub upload_date: Option<String>,
/// Duration in seconds, when reported.
pub duration_seconds: Option<f64>,
/// Best available size estimate for the selected formats.
pub estimated_bytes: Option<u64>,
/// Source and rights context.
pub rights: VideoRights,
}
/// Versions and locations of external executables.
#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)]
pub struct DependencyInfo {
/// Configured `yt-dlp` executable.
pub yt_dlp_executable: PathBuf,
/// Reported `yt-dlp` version.
pub yt_dlp_version: String,
/// Reported first line of `ffmpeg -version`.
pub ffmpeg_version: String,
/// Reported first line of `ffprobe -version`.
pub ffprobe_version: String,
}
/// Limits and policy for an archive operation.
#[derive(Clone, Debug)]
pub struct ArchiveOptions {
/// Parent directory containing one directory per video ID.
pub output: PathBuf,
/// Verify and skip an existing valid VideoForge archive.
pub skip_existing: bool,
/// Hard cap for recursive staging bytes and final media bytes.
pub max_download_bytes: u64,
/// Per-process timeout used for inspection and downloading.
pub timeout: Duration,
/// Fixed quality policy.
pub quality: VideoQuality,
/// Explicit caller confirmation that archiving is authorized.
pub authorized: bool,
}
impl Default for ArchiveOptions {
fn default() -> Self {
Self {
output: PathBuf::from("videoforge-archive"),
skip_existing: false,
max_download_bytes: DEFAULT_MAX_BYTES,
timeout: DEFAULT_TIMEOUT,
quality: VideoQuality::Best,
authorized: false,
}
}
}
/// Integrity information for the one media file in an archive.
#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)]
pub struct VideoIntegrity {
/// Safe direct-child media filename.
pub media_file: String,
/// Exact media file length.
pub bytes: u64,
/// Lowercase BLAKE3 digest.
pub blake3: String,
/// Validated container extension (`mp4`, `webm`, or `mkv`).
pub container: String,
}
/// Persistent archive manifest.
#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
pub struct VideoManifest {
/// Manifest schema version.
pub schema_version: u32,
/// Video provider.
pub provider: VideoProvider,
/// Provider video ID.
pub video_id: String,
/// Canonical source URL.
pub canonical_url: String,
/// Stable, filtered metadata without signed URLs or request headers.
pub summary: VideoSummary,
/// `yt-dlp` version used for this archive.
pub yt_dlp_version: String,
/// Quality policy used to select the archived media.
pub quality: VideoQuality,
/// Verified media integrity.
pub integrity: VideoIntegrity,
/// Written only after all archive fields are complete.
pub complete: bool,
}
/// A fully verified archived video.
#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
pub struct ArchivedVideo {
/// Canonical target.
pub target: VideoTarget,
/// Archive directory.
pub directory: PathBuf,
/// Parsed and verified manifest.
pub manifest: VideoManifest,
/// Non-fatal cleanup or durability warnings after a verified install.
pub warnings: Vec<String>,
}
/// One selected video that could not be archived.
#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)]
pub struct ArchiveFailure {
/// Canonical target.
pub target: VideoTarget,
/// Bounded error with URL-like and sensitive diagnostic tokens redacted.
pub error: String,
}
/// Complete result for a multi-video archive operation.
#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
pub struct ArchiveSummary {
/// IDs in the same order as the selected targets.
pub selected_ids: Vec<String>,
/// Number newly archived or repaired.
pub archived: usize,
/// Number verified and skipped.
pub skipped: usize,
/// Number that failed independently.
pub failed: usize,
/// Full records for newly archived or repaired videos.
pub archived_videos: Vec<ArchivedVideo>,
/// Full verified records for skipped videos.
pub skipped_videos: Vec<ArchivedVideo>,
/// Per-video failures.
pub failures: Vec<ArchiveFailure>,
}
/// Current phase of a synchronous archive operation.
#[derive(Clone, Copy, Debug, Eq, PartialEq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum ProgressPhase {
/// Metadata is being inspected.
Inspecting,
/// Media is being downloaded.
Downloading,
/// Staged or existing media is being verified.
Verifying,
/// A staged archive is being installed transactionally.
Installing,
/// A valid archive was skipped.
Skipped,
/// One target failed.
Failed,
/// One target completed.
Complete,
}
/// Bounded progress state for the active target and aggregate operation.
#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
pub struct ProgressSnapshot {
/// Active target, if any.
pub target: Option<VideoTarget>,
/// Current phase.
pub phase: ProgressPhase,
/// Bytes reported or observed for the active download.
pub active_bytes: u64,
/// Reported total bytes, when available.
pub active_total_bytes: Option<u64>,
/// Reported bytes per second, when available.
pub speed_bytes_per_second: Option<f64>,
/// Reported remaining duration, when available.
pub eta: Option<Duration>,
/// Completed archive count.
pub archived: usize,
/// Completed verified skip count.
pub skipped: usize,
/// Completed failure count.
pub failed: usize,
}
/// Errors from target validation, subprocess policy, or archive integrity.
#[derive(Debug, Error)]
#[non_exhaustive]
pub enum VideoForgeError {
/// A URL or video ID is outside the supported target grammar.
#[error("invalid video target: {0}")]
InvalidTarget(String),
/// Archive options are unsafe or nonsensical.
#[error("invalid archive options: {0}")]
InvalidOptions(String),
/// Archiving was not explicitly authorized.
#[error("archiving requires ArchiveOptions::authorized = true")]
Unauthorized,
/// An external dependency was missing or unusable.
#[error("dependency error: {0}")]
Dependency(String),
/// A subprocess failed or emitted invalid output.
#[error("subprocess error: {0}")]
Process(String),
/// A bounded operation exceeded its deadline.
#[error("operation timed out after {0:?}")]
Timeout(Duration),
/// Metadata output exceeded the retained JSON limit.
#[error("yt-dlp metadata exceeded the {METADATA_LIMIT}-byte limit")]
MetadataTooLarge,
/// Extractor metadata was malformed.
#[error("invalid yt-dlp metadata: {0}")]
Metadata(String),
/// Extractor metadata did not satisfy the public-video policy.
#[error("video policy rejected metadata: {0}")]
Policy(String),
/// Staging bytes or final media exceeded the configured cap.
#[error("download exceeded the configured {limit}-byte cap (observed {observed})")]
ByteLimit { limit: u64, observed: u64 },
/// A filesystem path, symlink, or manifest path was unsafe.
#[error("unsafe archive path: {0}")]
UnsafePath(String),
/// An existing destination was not a recognized VideoForge archive.
#[error("archive destination conflict: {0}")]
ArchiveConflict(String),
/// Archive bytes or structure failed verification.
#[error("archive integrity error: {0}")]
Integrity(String),
/// Transactional replacement or cleanup failed.
#[error("archive transaction error: {0}")]
Transaction(String),
/// Filesystem I/O failed.
#[error("I/O error: {0}")]
Io(#[from] io::Error),
/// Manifest JSON encoding or decoding failed.
#[error("JSON error: {0}")]
Json(#[from] serde_json::Error),
}
/// Cloneable handle for invoking one trusted `yt-dlp` executable.
#[derive(Clone, Debug)]
pub struct VideoForge {
executable: PathBuf,
}
impl Default for VideoForge {
fn default() -> Self {
Self::new()
}
}
impl VideoForge {
/// Use `yt-dlp` from `PATH`.
#[must_use]
pub fn new() -> Self {
Self {
executable: PathBuf::from("yt-dlp"),
}
}
/// Use a specific trusted local `yt-dlp` executable.
#[must_use]
pub fn with_executable(executable: impl Into<PathBuf>) -> Self {
Self {
executable: executable.into(),
}
}
/// Probe `yt-dlp`, `ffmpeg`, and `ffprobe` with bounded subprocesses.
pub fn dependency_info(&self) -> Result<DependencyInfo, VideoForgeError> {
let probe_timeout = Duration::from_secs(15);
let yt_dlp_version = self.yt_dlp_version(probe_timeout)?;
let ffmpeg_version = probe_program("ffmpeg", probe_timeout)?;
let ffprobe_version = probe_program("ffprobe", probe_timeout)?;
Ok(DependencyInfo {
yt_dlp_executable: self.executable.clone(),
yt_dlp_version,
ffmpeg_version,
ffprobe_version,
})
}
/// Inspect filtered metadata for a validated target.
pub fn inspect(
&self,
target: &VideoTarget,
quality: VideoQuality,
timeout: Duration,
) -> Result<VideoSummary, VideoForgeError> {
if timeout.is_zero() {
return Err(VideoForgeError::InvalidOptions(
"inspection timeout must be nonzero".into(),
));
}
self.inspect_internal(target, quality, timeout)
}
/// Archive all targets, accumulating independent per-video failures.
pub fn archive(
&self,
targets: &[VideoTarget],
options: &ArchiveOptions,
) -> Result<ArchiveSummary, VideoForgeError> {
self.archive_with_progress(targets, options, |_| {})
}
/// Archive all targets and synchronously report bounded progress snapshots.
pub fn archive_with_progress<F>(
&self,
targets: &[VideoTarget],
options: &ArchiveOptions,
mut progress: F,
) -> Result<ArchiveSummary, VideoForgeError>
where
F: FnMut(&ProgressSnapshot),
{
validate_archive_options(options)?;
if !options.authorized {
return Err(VideoForgeError::Unauthorized);
}
let unique_ids = targets.iter().map(VideoTarget::id).collect::<BTreeSet<_>>();
if unique_ids.len() != targets.len() {
return Err(VideoForgeError::InvalidOptions(
"selected video targets must have unique IDs".into(),
));
}
ensure_safe_directory(&options.output)?;
let output = fs::canonicalize(&options.output)?;
let mut summary = ArchiveSummary {
selected_ids: targets.iter().map(|target| target.id.clone()).collect(),
archived: 0,
skipped: 0,
failed: 0,
archived_videos: Vec::new(),
skipped_videos: Vec::new(),
failures: Vec::new(),
};
let mut dependencies: Option<DependencyInfo> = None;
for target in targets {
match self.archive_one(
target,
options,
&output,
&mut dependencies,
&summary,
&mut progress,
) {
Ok(ArchiveOutcome::Archived(video)) => {
summary.archived += 1;
summary.archived_videos.push(video);
emit_progress(
&mut progress,
Some(target),
ProgressPhase::Complete,
&summary,
DownloadProgress::default(),
);
}
Ok(ArchiveOutcome::Skipped(video)) => {
summary.skipped += 1;
summary.skipped_videos.push(video);
emit_progress(
&mut progress,
Some(target),
ProgressPhase::Skipped,
&summary,
DownloadProgress::default(),
);
}
Err(error) => {
summary.failed += 1;
summary.failures.push(ArchiveFailure {
target: target.clone(),
error: error.to_string(),
});
emit_progress(
&mut progress,
Some(target),
ProgressPhase::Failed,
&summary,
DownloadProgress::default(),
);
}
}
}
debug_assert_eq!(
summary.selected_ids.len(),
summary.archived + summary.skipped + summary.failed
);
Ok(summary)
}
fn archive_one<F>(
&self,
target: &VideoTarget,
options: &ArchiveOptions,
output: &Path,
dependencies: &mut Option<DependencyInfo>,
aggregate: &ArchiveSummary,
progress: &mut F,
) -> Result<ArchiveOutcome, VideoForgeError>
where
F: FnMut(&ProgressSnapshot),
{
let destination = output.join(target.id());
let mut transaction = DestinationTransaction::acquire(output, target, options.timeout)?;
transaction.recover(output, &destination, target)?;
let existing = recognized_manifest(&destination, target)?;
emit_progress(
progress,
Some(target),
ProgressPhase::Inspecting,
aggregate,
DownloadProgress::default(),
);
// Archive and skip decisions always use fresh provider metadata.
let video_summary = self.inspect_internal(target, options.quality, options.timeout)?;
if options.skip_existing && existing.is_some() {
emit_progress(
progress,
Some(target),
ProgressPhase::Verifying,
aggregate,
DownloadProgress::default(),
);
if let Ok(video) = verify_archive(&destination) {
if video.manifest.summary == video_summary
&& video.manifest.quality == options.quality
{
return Ok(ArchiveOutcome::Skipped(video));
}
}
}
if let Some(bytes) = video_summary.estimated_bytes {
if bytes > options.max_download_bytes {
return Err(VideoForgeError::ByteLimit {
limit: options.max_download_bytes,
observed: bytes,
});
}
}
let dependency = match dependencies {
Some(info) => info.clone(),
None => {
let info = self.dependency_info()?;
*dependencies = Some(info.clone());
info
}
};
let stage = TempBuilder::new()
.prefix(&format!(".videoforge-{}-stage-", target.id()))
.tempdir_in(output)?;
let stage_path = fs::canonicalize(stage.path())?;
let final_event =
self.download_to_stage(target, options, &stage_path, aggregate, progress)?;
emit_progress(
progress,
Some(target),
ProgressPhase::Verifying,
aggregate,
DownloadProgress::default(),
);
let integrity = validate_staged_media(&stage_path, target, &final_event, options)?;
let manifest = VideoManifest {
schema_version: MANIFEST_SCHEMA,
provider: target.provider(),
video_id: target.id().to_owned(),
canonical_url: target.canonical_url().to_owned(),
summary: video_summary,
yt_dlp_version: dependency.yt_dlp_version,
quality: options.quality,
integrity,
complete: true,
};
write_manifest(&stage_path, &manifest)?;
sync_directory(&stage_path)?;
emit_progress(
progress,
Some(target),
ProgressPhase::Installing,
aggregate,
DownloadProgress::default(),
);
transaction.mark_preparing()?;
let mut warnings = install_stage(
stage,
&destination,
output,
target,
existing.as_ref(),
&mut transaction,
)?;
let mut archived = verify_archive(&destination)?;
if let Err(error) = transaction.mark_idle() {
warnings.push(format!(
"archive is verified, but transaction state cleanup failed: {error}"
));
}
archived.warnings = warnings;
Ok(ArchiveOutcome::Archived(archived))
}
fn inspect_internal(
&self,
target: &VideoTarget,
quality: VideoQuality,
timeout: Duration,
) -> Result<VideoSummary, VideoForgeError> {
let mut command = Command::new(&self.executable);
configure_yt_dlp(&mut command);
command.args([
"--simulate",
"--dump-single-json",
"--format",
quality.selector(),
"--",
target.canonical_url(),
]);
let capture = run_captured(command, timeout, METADATA_LIMIT, true)?;
if capture.stdout_overflow {
return Err(VideoForgeError::MetadataTooLarge);
}
if !capture.status.success() {
return Err(VideoForgeError::Process(format!(
"yt-dlp metadata inspection failed with {}: {}",
capture.status,
diagnostic(&capture.stderr)
)));
}
let raw: RawMetadata = serde_json::from_slice(&capture.stdout)
.map_err(|error| VideoForgeError::Metadata(error.to_string()))?;
raw.into_summary(target)
}
fn yt_dlp_version(&self, timeout: Duration) -> Result<String, VideoForgeError> {
let mut command = Command::new(&self.executable);
configure_yt_dlp(&mut command);
command.arg("--version");
let capture = run_captured(command, timeout, 16 * 1024, true)?;
if capture.stdout_overflow || !capture.status.success() {
return Err(VideoForgeError::Dependency(format!(
"yt-dlp version probe failed with {}: {}",
capture.status,
diagnostic(&capture.stderr)
)));
}
first_version_line(&capture.stdout, "yt-dlp")
}
fn download_to_stage<F>(
&self,
target: &VideoTarget,
options: &ArchiveOptions,
stage: &Path,
aggregate: &ArchiveSummary,
progress_callback: &mut F,
) -> Result<FinalFileEvent, VideoForgeError>
where
F: FnMut(&ProgressSnapshot),
{
let mut command = Command::new(&self.executable);
configure_yt_dlp(&mut command);
command
.arg("--no-simulate")
.arg("--progress")
.arg("--newline")
.arg("--progress-delta")
.arg("0.25")
.arg("--format")
.arg(options.quality.selector())
.arg("--paths")
.arg(stage)
.arg("--output")
.arg("%(id)s.%(ext)s")
.arg("--max-filesize")
.arg(options.max_download_bytes.to_string())
.arg("--no-overwrites")
.arg("--continue")
.arg("--abort-on-unavailable-fragments")
.arg("--retries")
.arg("3")
.arg("--fragment-retries")
.arg("3")
.arg("--file-access-retries")
.arg("3")
.arg("--extractor-retries")
.arg("2")
.arg("--retry-sleep")
.arg("http:linear=1:3:1")
.arg("--retry-sleep")
.arg("fragment:linear=1:3:1")
.arg("--retry-sleep")
.arg("file_access:linear=1:3:1")
.arg("--retry-sleep")
.arg("extractor:linear=1:3:1")
.arg("--socket-timeout")
.arg("20")
.arg("--no-mtime")
.arg("--progress-template")
.arg(
"download:videoforge-progress:{\"downloaded_bytes\":%(progress.downloaded_bytes)j,\"total_bytes\":%(progress.total_bytes,progress.total_bytes_estimate)j,\"speed\":%(progress.speed)j,\"eta\":%(progress.eta)j}",
)
.arg("--print")
.arg("after_move:videoforge-file:{\"filepath\":%(filepath)j,\"id\":%(id)j}")
.arg("--")
.arg(target.canonical_url());
prepare_command(&mut command);
let mut child = spawn_process(command, "yt-dlp download")?;
let stdout = child
.stdout()
.take()
.ok_or_else(|| VideoForgeError::Process("yt-dlp stdout was not piped".into()))?;
let stderr = child
.stderr()
.take()
.ok_or_else(|| VideoForgeError::Process("yt-dlp stderr was not piped".into()))?;
let (line_tx, line_rx) = mpsc::sync_channel(1024);
let event_overflow = Arc::new(AtomicBool::new(false));
let event_overflow_reader = Arc::clone(&event_overflow);
let stdout_thread =
thread::spawn(move || drain_lines(stdout, line_tx, event_overflow_reader));
let stderr_thread = thread::spawn(move || drain_suffix(stderr, DIAGNOSTIC_LIMIT));
let started = Instant::now();
let mut final_events = Vec::new();
let mut current = DownloadProgress::default();
let mut monitor_error = None;
let status = loop {
let mut progress_changed = false;
while let Ok(line) = line_rx.try_recv() {
if let Some(parsed) = parse_progress_line(&line) {
current = parsed;
progress_changed = true;
} else {
match parse_file_line(&line) {
Ok(Some(event)) => final_events.push(event),
Ok(None) => {}
Err(error) => monitor_error = Some(error),
}
}
}
match recursive_staging_bytes(stage) {
Ok(bytes) => {
current.downloaded_bytes = current.downloaded_bytes.max(bytes);
if bytes > options.max_download_bytes {
monitor_error = Some(VideoForgeError::ByteLimit {
limit: options.max_download_bytes,
observed: bytes,
});
}
}
Err(error) => monitor_error = Some(error),
}
if event_overflow.load(Ordering::Relaxed) {
monitor_error = Some(VideoForgeError::Process(
"yt-dlp emitted excessive namespaced output".into(),
));
}
if monitor_error.is_some() || started.elapsed() >= options.timeout {
kill_process_tree(&mut child);
let _ = child.wait();
break None;
}
if progress_changed {
emit_progress(
progress_callback,
Some(target),
ProgressPhase::Downloading,
aggregate,
current,
);
}
if let Some(status) = child.try_wait()? {
break Some(status);
}
thread::sleep(POLL_INTERVAL);
};
// A successfully exited leader can still leave ffmpeg or another
// descendant holding inherited pipes. Terminate any remaining group
// members before bounded reader joins.
if !stdout_thread.is_finished() || !stderr_thread.is_finished() {
kill_process_tree(&mut child);
}
let stdout_result = join_reader(stdout_thread, "stdout")?;
let stderr = join_reader(stderr_thread, "stderr")?;
for line in line_rx.try_iter() {
if let Some(event) = parse_file_line(&line)? {
final_events.push(event);
}
}
if stdout_result.line_overflow || event_overflow.load(Ordering::Relaxed) {
return Err(VideoForgeError::Process(
"yt-dlp emitted excessive or overlong namespaced output".into(),
));
}
if let Some(error) = monitor_error {
return Err(error);
}
let Some(status) = status else {
return Err(VideoForgeError::Timeout(options.timeout));
};
if !status.success() {
let stdout_diagnostic = diagnostic(&stdout_result.diagnostics);
let detail = if stderr.is_empty() {
stdout_diagnostic
} else {
diagnostic(&stderr)
};
return Err(VideoForgeError::Process(format!(
"yt-dlp download failed with {status}: {detail}"
)));
}
if final_events.len() != 1 {
return Err(VideoForgeError::Integrity(format!(
"expected exactly one final filepath event, received {}",
final_events.len()
)));
}
Ok(final_events.remove(0))
}
}
/// Verify an archive directory, including identity, exact files, media magic,
/// size, and BLAKE3 digest.
pub fn verify_archive(path: impl AsRef<Path>) -> Result<ArchivedVideo, VideoForgeError> {
let path = path.as_ref();
validate_existing_path_components(path)?;
let canonical_directory = fs::canonicalize(path)?;
let manifest_path = canonical_directory.join(MANIFEST_NAME);
let manifest = read_manifest(&manifest_path)?;
if manifest.schema_version != MANIFEST_SCHEMA || !manifest.complete {
return Err(VideoForgeError::Integrity(
"manifest schema is unsupported or incomplete".into(),
));
}
validate_video_id(&manifest.video_id).map_err(|error| {
VideoForgeError::Integrity(format!("manifest has invalid video ID: {error}"))
})?;
if manifest.provider != VideoProvider::Youtube {
return Err(VideoForgeError::Integrity(
"manifest provider is unsupported".into(),
));
}
let target = parse_video_target(&manifest.canonical_url).map_err(|error| {
VideoForgeError::Integrity(format!("manifest canonical URL is invalid: {error}"))
})?;
if target.id() != manifest.video_id
|| manifest.summary.target != target
|| canonical_directory.file_name() != Some(OsStr::new(target.id()))
{
return Err(VideoForgeError::Integrity(
"manifest identity does not match its URL, summary, and directory".into(),
));
}
validate_relative_media_name(&manifest.integrity.media_file)?;
let media_path = canonical_directory.join(&manifest.integrity.media_file);
let media_metadata = fs::symlink_metadata(&media_path)?;
if media_metadata.file_type().is_symlink() || !media_metadata.is_file() {
return Err(VideoForgeError::UnsafePath(
"manifest media path is not a regular non-symlink file".into(),
));
}
if media_metadata.len() == 0 || media_metadata.len() != manifest.integrity.bytes {
return Err(VideoForgeError::Integrity(
"manifest media size does not match the file".into(),
));
}
let container = media_container(&media_path)?;
if container != manifest.integrity.container {
return Err(VideoForgeError::Integrity(
"manifest container does not match media extension and magic".into(),
));
}
let digest = hash_file(&media_path)?;
if digest != manifest.integrity.blake3 {
return Err(VideoForgeError::Integrity(
"media BLAKE3 digest does not match the manifest".into(),
));
}
let mut names = Vec::new();
for entry in fs::read_dir(&canonical_directory)? {
let entry = entry?;
let metadata = fs::symlink_metadata(entry.path())?;
if metadata.file_type().is_symlink() || !metadata.is_file() {
return Err(VideoForgeError::UnsafePath(
"archive contains a non-file or symlink entry".into(),
));
}
names.push(entry.file_name());
}
names.sort();
let mut expected = vec![
OsStr::new(MANIFEST_NAME).to_os_string(),
OsStr::new(&manifest.integrity.media_file).to_os_string(),
];
expected.sort();
if names != expected {
return Err(VideoForgeError::Integrity(
"archive must contain exactly one media file and its manifest".into(),
));
}
Ok(ArchivedVideo {
target,
directory: canonical_directory,
manifest,
warnings: Vec::new(),
})
}
#[derive(Debug, Deserialize)]
struct RawMetadata {
id: Option<String>,
title: Option<String>,
description: Option<String>,
uploader: Option<String>,
uploader_id: Option<String>,
channel: Option<String>,
channel_id: Option<String>,
upload_date: Option<String>,
duration: Option<f64>,
webpage_url: Option<String>,
extractor: Option<String>,
extractor_key: Option<String>,
availability: Option<String>,
age_limit: Option<u8>,
live_status: Option<String>,
license: Option<String>,
filesize: Option<u64>,
filesize_approx: Option<u64>,
requested_formats: Option<Vec<RawFormat>>,
#[serde(rename = "_has_drm", alias = "has_drm")]
has_drm: Option<bool>,
entries: Option<Vec<serde::de::IgnoredAny>>,
}
#[derive(Debug, Deserialize)]
struct RawFormat {
filesize: Option<u64>,
filesize_approx: Option<u64>,
has_drm: Option<bool>,
}
impl RawMetadata {
fn into_summary(self, target: &VideoTarget) -> Result<VideoSummary, VideoForgeError> {
if self.id.as_deref() != Some(target.id()) {
return Err(VideoForgeError::Policy(
"extractor ID does not match the selected target".into(),
));
}
if self.extractor.as_deref() != Some("youtube")
|| self.extractor_key.as_deref() != Some("Youtube")
{
return Err(VideoForgeError::Policy(
"metadata was not produced by the exact YouTube extractor".into(),
));
}
if self.entries.is_some() {
return Err(VideoForgeError::Policy(
"playlist or multi-entry metadata is not accepted".into(),
));
}
if self.availability.as_deref() != Some("public") {
return Err(VideoForgeError::Policy(
"video availability is not exactly public".into(),
));
}
if self.age_limit != Some(0) {
return Err(VideoForgeError::Policy(
"video age_limit is not exactly zero".into(),
));
}
if self.live_status.as_deref() != Some("not_live") {
return Err(VideoForgeError::Policy(
"live, upcoming, or post-live videos are not accepted".into(),
));
}
let selected_formats_are_clear = self
.requested_formats
.as_deref()
.filter(|formats| !formats.is_empty())
.is_some_and(|formats| formats.iter().all(|format| format.has_drm == Some(false)));
let single_format_is_clear =
self.requested_formats.is_none() && self.has_drm == Some(false);
if self.has_drm == Some(true) || !(selected_formats_are_clear || single_format_is_clear) {
return Err(VideoForgeError::Policy(
"DRM must not be reported for the video and must be explicitly false for every selected format"
.into(),
));
}
let webpage_url = self
.webpage_url
.as_deref()
.ok_or_else(|| VideoForgeError::Policy("reported webpage URL is missing".into()))?;
let reported = parse_video_target(webpage_url).map_err(|_| {
VideoForgeError::Policy("reported webpage URL is not an exact video URL".into())
})?;
if reported != *target {
return Err(VideoForgeError::Policy(
"reported webpage URL does not match the target".into(),
));
}
let title = required_text(self.title, "title", 4 * 1024)?;
let description = optional_text(self.description, "description", 512 * 1024)?;
let uploader = optional_text(self.uploader, "uploader", 4 * 1024)?;
let uploader_id = optional_text(self.uploader_id, "uploader_id", 4 * 1024)?;
let channel = optional_text(self.channel, "channel", 4 * 1024)?;
let channel_id = optional_text(self.channel_id, "channel_id", 4 * 1024)?;
let reported_license = optional_text(self.license, "license", 8 * 1024)?;
let upload_date = match self.upload_date {
Some(value) if value.len() == 8 && value.bytes().all(|byte| byte.is_ascii_digit()) => {
Some(value)
}
Some(_) => {
return Err(VideoForgeError::Metadata(
"upload_date must use YYYYMMDD digits".into(),
));
}
None => None,
};
if self
.duration
.is_some_and(|duration| !duration.is_finite() || duration < 0.0)
{
return Err(VideoForgeError::Metadata(
"duration must be finite and nonnegative".into(),
));
}
let format_bytes = self.requested_formats.as_ref().and_then(|formats| {
if formats.is_empty() {
return None;
}
formats.iter().try_fold(0_u64, |total, format| {
total.checked_add(format.filesize.or(format.filesize_approx)?)
})
});
let estimated_bytes = format_bytes.or(self.filesize).or(self.filesize_approx);
let creator = channel
.as_deref()
.or(uploader.as_deref())
.unwrap_or("the reported creator");
let attribution = format!("\"{title}\" by {creator} ({})", target.canonical_url());
let redistribution_warning = if reported_license
.as_deref()
.is_some_and(|license| license.to_ascii_lowercase().contains("creative commons"))
{
"A reported Creative Commons label still requires review of its exact terms, attribution requirements, and the uploader's authority to license the content."
} else {
"A Standard YouTube License or an unknown/missing license does not grant permission to redistribute this video."
};
Ok(VideoSummary {
target: target.clone(),
title,
description,
upload_date,
duration_seconds: self.duration,
estimated_bytes,
rights: VideoRights {
source: target.canonical_url().to_owned(),
uploader,
uploader_id,
channel,
channel_id,
reported_license,
attribution,
redistribution_warning: redistribution_warning.into(),
yt_dlp_license_notice: "The yt-dlp software license applies only to yt-dlp and grants no rights in video content.".into(),
},
})
}
}
fn required_text(
value: Option<String>,
field: &str,
max: usize,
) -> Result<String, VideoForgeError> {
let value = value
.filter(|value| !value.is_empty())
.ok_or_else(|| VideoForgeError::Metadata(format!("{field} is missing or empty")))?;
if value.len() > max {
return Err(VideoForgeError::Metadata(format!(
"{field} exceeds its {max}-byte limit"
)));
}
Ok(value)
}
fn optional_text(
value: Option<String>,
field: &str,
max: usize,
) -> Result<Option<String>, VideoForgeError> {
match value {
Some(value) if value.len() > max => Err(VideoForgeError::Metadata(format!(
"{field} exceeds its {max}-byte limit"
))),
Some(value) if value.is_empty() => Ok(None),
value => Ok(value),
}
}
fn validate_archive_options(options: &ArchiveOptions) -> Result<(), VideoForgeError> {
if options.max_download_bytes == 0 {
return Err(VideoForgeError::InvalidOptions(
"max_download_bytes must be nonzero".into(),
));
}
if options.timeout.is_zero() {
return Err(VideoForgeError::InvalidOptions(
"timeout must be nonzero".into(),
));
}
if options.output.as_os_str().is_empty() {
return Err(VideoForgeError::InvalidOptions(
"output path must not be empty".into(),
));
}
Ok(())
}
fn configure_yt_dlp(command: &mut Command) {
command.args([
"--ignore-config",
"--no-plugin-dirs",
"--no-remote-components",
"--use-extractors",
"youtube",
"--no-playlist",
"--abort-on-error",
"--no-color",
"--proxy",
"",
"--no-geo-bypass",
]);
}
fn prepare_command(command: &mut Command) {
command
.env("YTDLP_NO_PLUGINS", "1")
.stdin(Stdio::null())
.stdout(Stdio::piped())
.stderr(Stdio::piped());
}
struct CapturedProcess {
status: ExitStatus,
stdout: Vec<u8>,
stderr: Vec<u8>,
stdout_overflow: bool,
}
fn run_captured(
mut command: Command,
timeout: Duration,
stdout_limit: usize,
prefix_stdout: bool,
) -> Result<CapturedProcess, VideoForgeError> {
prepare_command(&mut command);
let mut child = spawn_process(command, "external process")?;
let stdout = child
.stdout()
.take()
.ok_or_else(|| VideoForgeError::Process("process stdout was not piped".into()))?;
let stderr = child
.stderr()
.take()
.ok_or_else(|| VideoForgeError::Process("process stderr was not piped".into()))?;
let overflow = Arc::new(AtomicBool::new(false));
let overflow_reader = Arc::clone(&overflow);
let stdout_thread = thread::spawn(move || {
if prefix_stdout {
drain_prefix(stdout, stdout_limit, overflow_reader)
} else {
drain_suffix(stdout, stdout_limit)
}
});
let stderr_thread = thread::spawn(move || drain_suffix(stderr, DIAGNOSTIC_LIMIT));
let started = Instant::now();
let status = loop {
if started.elapsed() >= timeout || overflow.load(Ordering::Relaxed) {
kill_process_tree(&mut child);
let _ = child.wait();
break None;
}
if let Some(status) = child.try_wait()? {
break Some(status);
}
thread::sleep(POLL_INTERVAL);
};
if !stdout_thread.is_finished() || !stderr_thread.is_finished() {
kill_process_tree(&mut child);
}
let stdout = join_reader(stdout_thread, "stdout")?;
let stderr = join_reader(stderr_thread, "stderr")?;
let stdout_overflow = overflow.load(Ordering::Relaxed);
match status {
Some(status) => Ok(CapturedProcess {
status,
stdout,
stderr,
stdout_overflow,
}),
None if stdout_overflow => Ok(CapturedProcess {
status: synthetic_failure_status()?,
stdout,
stderr,
stdout_overflow: true,
}),
None => Err(VideoForgeError::Timeout(timeout)),
}
}
type ManagedChild = Box<dyn StdChildWrapper>;
fn spawn_process(command: Command, description: &str) -> Result<ManagedChild, VideoForgeError> {
let mut wrapped = StdCommandWrap::from(command);
#[cfg(unix)]
wrapped.wrap(ProcessGroup::leader());
#[cfg(windows)]
wrapped.wrap(JobObject);
wrapped.spawn().map_err(|error| {
VideoForgeError::Dependency(format!("could not start {description}: {error}"))
})
}
fn join_reader<T>(
handle: thread::JoinHandle<io::Result<T>>,
stream: &str,
) -> Result<T, VideoForgeError> {
let deadline = Instant::now() + Duration::from_secs(5);
while !handle.is_finished() {
if Instant::now() >= deadline {
return Err(VideoForgeError::Process(format!(
"{stream} reader did not close after process termination"
)));
}
thread::sleep(POLL_INTERVAL);
}
handle
.join()
.map_err(|_| VideoForgeError::Process(format!("{stream} reader thread panicked")))?
.map_err(VideoForgeError::Io)
}
fn drain_prefix<R: Read>(
mut reader: R,
limit: usize,
overflow: Arc<AtomicBool>,
) -> io::Result<Vec<u8>> {
let mut retained = Vec::with_capacity(limit.min(64 * 1024));
let mut buffer = [0_u8; 16 * 1024];
loop {
let count = reader.read(&mut buffer)?;
if count == 0 {
break;
}
let remaining = limit.saturating_sub(retained.len());
retained.extend_from_slice(&buffer[..count.min(remaining)]);
if count > remaining {
overflow.store(true, Ordering::Relaxed);
}
}
Ok(retained)
}
fn drain_suffix<R: Read>(mut reader: R, limit: usize) -> io::Result<Vec<u8>> {
let mut retained = VecDeque::with_capacity(limit.min(64 * 1024));
let mut buffer = [0_u8; 16 * 1024];
loop {
let count = reader.read(&mut buffer)?;
if count == 0 {
break;
}
for byte in &buffer[..count] {
if retained.len() == limit {
retained.pop_front();
}
retained.push_back(*byte);
}
}
Ok(retained.into_iter().collect())
}
struct DrainedLines {
diagnostics: Vec<u8>,
line_overflow: bool,
}
fn drain_lines<R: Read>(
mut reader: R,
sender: mpsc::SyncSender<String>,
event_overflow: Arc<AtomicBool>,
) -> io::Result<DrainedLines> {
let mut diagnostics = VecDeque::with_capacity(DIAGNOSTIC_LIMIT);
let mut current = Vec::new();
let mut buffer = [0_u8; 8 * 1024];
let mut overflow = false;
let mut saw_overflow = false;
loop {
let count = reader.read(&mut buffer)?;
if count == 0 {
break;
}
for byte in &buffer[..count] {
if diagnostics.len() == DIAGNOSTIC_LIMIT {
diagnostics.pop_front();
}
diagnostics.push_back(*byte);
if *byte == b'\n' {
if !overflow {
let line = String::from_utf8_lossy(¤t)
.trim_end_matches('\r')
.to_owned();
if (line.starts_with("videoforge-progress:")
|| line.starts_with("videoforge-file:"))
&& sender.try_send(line).is_err()
{
event_overflow.store(true, Ordering::Relaxed);
}
}
saw_overflow |= overflow;
current.clear();
overflow = false;
} else if current.len() < LINE_LIMIT {
current.push(*byte);
} else {
overflow = true;
}
}
}
saw_overflow |= overflow;
if !current.is_empty() && !overflow {
let line = String::from_utf8_lossy(¤t).to_string();
if (line.starts_with("videoforge-progress:") || line.starts_with("videoforge-file:"))
&& sender.try_send(line).is_err()
{
event_overflow.store(true, Ordering::Relaxed);
}
}
Ok(DrainedLines {
diagnostics: diagnostics.into_iter().collect(),
line_overflow: saw_overflow,
})
}
fn kill_process_tree(child: &mut ManagedChild) {
let _ = child.kill();
}
#[cfg(unix)]
fn synthetic_failure_status() -> io::Result<ExitStatus> {
use std::os::unix::process::ExitStatusExt;
Ok(ExitStatus::from_raw(1 << 8))
}
#[cfg(windows)]
fn synthetic_failure_status() -> io::Result<ExitStatus> {
use std::os::windows::process::ExitStatusExt;
Ok(ExitStatus::from_raw(1))
}
#[cfg(not(any(unix, windows)))]
fn synthetic_failure_status() -> io::Result<ExitStatus> {
Err(io::Error::other("unsupported platform exit status"))
}
fn probe_program(program: &str, timeout: Duration) -> Result<String, VideoForgeError> {
let mut command = Command::new(program);
command.arg("-version");
let capture = run_captured(command, timeout, 64 * 1024, true)?;
if capture.stdout_overflow || !capture.status.success() {
return Err(VideoForgeError::Dependency(format!(
"{program} version probe failed with {}: {}",
capture.status,
diagnostic(&capture.stderr)
)));
}
first_version_line(&capture.stdout, program)
}
fn first_version_line(bytes: &[u8], program: &str) -> Result<String, VideoForgeError> {
let output = std::str::from_utf8(bytes)
.map_err(|_| VideoForgeError::Dependency(format!("{program} version was not UTF-8")))?;
let line = output
.lines()
.next()
.map(str::trim)
.filter(|line| !line.is_empty())
.ok_or_else(|| VideoForgeError::Dependency(format!("{program} reported no version")))?;
if line.len() > 4096 || line.chars().any(char::is_control) {
return Err(VideoForgeError::Dependency(format!(
"{program} reported an invalid version line"
)));
}
Ok(line.to_owned())
}
fn diagnostic(bytes: &[u8]) -> String {
let value = String::from_utf8_lossy(bytes);
let mut sanitized = Vec::new();
let mut redact_following = 0usize;
for token in value.split_whitespace() {
let lowercase = token.to_ascii_lowercase();
let contains_url = lowercase.contains("http://") || lowercase.contains("https://");
let sensitive = lowercase.contains("authorization")
|| lowercase.contains("cookie")
|| lowercase.contains("signature")
|| lowercase.contains("sig=")
|| lowercase.contains("token=")
|| lowercase.contains("headers=");
if redact_following > 0 || sensitive {
sanitized.push("[redacted-sensitive-value]");
redact_following = redact_following.saturating_sub(1);
if lowercase.contains("authorization") {
redact_following = redact_following.max(2);
}
} else if contains_url {
sanitized.push("[redacted-url]");
} else {
sanitized.push(token);
}
}
let sanitized = sanitized.join(" ");
let trimmed = sanitized.trim();
if trimmed.is_empty() {
"no diagnostic output".into()
} else {
trimmed.to_owned()
}
}
#[derive(Clone, Copy, Debug, Default)]
struct DownloadProgress {
downloaded_bytes: u64,
total_bytes: Option<u64>,
speed: Option<f64>,
eta_seconds: Option<u64>,
}
#[derive(Deserialize)]
struct RawProgress {
downloaded_bytes: Option<serde_json::Value>,
total_bytes: Option<serde_json::Value>,
speed: Option<serde_json::Value>,
eta: Option<serde_json::Value>,
}
fn parse_progress_line(line: &str) -> Option<DownloadProgress> {
let payload = line.strip_prefix("videoforge-progress:")?;
let raw: RawProgress = serde_json::from_str(payload).ok()?;
Some(DownloadProgress {
downloaded_bytes: json_u64(raw.downloaded_bytes.as_ref()).unwrap_or(0),
total_bytes: json_u64(raw.total_bytes.as_ref()),
speed: json_f64(raw.speed.as_ref()),
eta_seconds: json_u64(raw.eta.as_ref()),
})
}
fn json_u64(value: Option<&serde_json::Value>) -> Option<u64> {
match value? {
serde_json::Value::Number(number) => number.as_u64().or_else(|| {
number
.as_f64()
.filter(|value| value.is_finite() && *value >= 0.0)
.map(|value| value as u64)
}),
serde_json::Value::String(value) => value.parse().ok(),
_ => None,
}
}
fn json_f64(value: Option<&serde_json::Value>) -> Option<f64> {
let parsed = match value? {
serde_json::Value::Number(number) => number.as_f64(),
serde_json::Value::String(value) => value.parse().ok(),
_ => None,
}?;
(parsed.is_finite() && parsed >= 0.0).then_some(parsed)
}
#[derive(Clone, Debug, Deserialize)]
struct FinalFileEvent {
filepath: String,
id: String,
}
fn parse_file_line(line: &str) -> Result<Option<FinalFileEvent>, VideoForgeError> {
let Some(payload) = line.strip_prefix("videoforge-file:") else {
return Ok(None);
};
let event = serde_json::from_str(payload).map_err(|error| {
VideoForgeError::Process(format!("invalid namespaced filepath event: {error}"))
})?;
Ok(Some(event))
}
fn emit_progress<F>(
callback: &mut F,
target: Option<&VideoTarget>,
phase: ProgressPhase,
aggregate: &ArchiveSummary,
active: DownloadProgress,
) where
F: FnMut(&ProgressSnapshot),
{
callback(&ProgressSnapshot {
target: target.cloned(),
phase,
active_bytes: active.downloaded_bytes,
active_total_bytes: active.total_bytes,
speed_bytes_per_second: active.speed,
eta: active.eta_seconds.map(Duration::from_secs),
archived: aggregate.archived,
skipped: aggregate.skipped,
failed: aggregate.failed,
});
}
fn recursive_staging_bytes(root: &Path) -> Result<u64, VideoForgeError> {
let mut total = 0_u64;
let mut pending = vec![root.to_path_buf()];
while let Some(directory) = pending.pop() {
for entry in fs::read_dir(directory)? {
let entry = entry?;
let metadata = fs::symlink_metadata(entry.path())?;
if metadata.file_type().is_symlink() {
return Err(VideoForgeError::UnsafePath(
"a symlink appeared in the staging directory".into(),
));
}
if metadata.is_dir() {
pending.push(entry.path());
} else if metadata.is_file() {
total = total.checked_add(metadata.len()).ok_or_else(|| {
VideoForgeError::Integrity("staging byte count overflowed".into())
})?;
} else {
return Err(VideoForgeError::UnsafePath(
"a special file appeared in the staging directory".into(),
));
}
}
}
Ok(total)
}
fn validate_staged_media(
stage: &Path,
target: &VideoTarget,
event: &FinalFileEvent,
options: &ArchiveOptions,
) -> Result<VideoIntegrity, VideoForgeError> {
if event.id != target.id() {
return Err(VideoForgeError::Integrity(
"final filepath event ID does not match the target".into(),
));
}
let event_path = Path::new(&event.filepath);
if !event_path.is_absolute() || event_path.parent() != Some(stage) {
return Err(VideoForgeError::UnsafePath(
"final filepath is not an absolute direct child of staging".into(),
));
}
let file_name = event_path
.file_name()
.and_then(OsStr::to_str)
.ok_or_else(|| VideoForgeError::UnsafePath("media filename is not valid UTF-8".into()))?;
validate_relative_media_name(file_name)?;
let metadata = fs::symlink_metadata(event_path)?;
if metadata.file_type().is_symlink() || !metadata.is_file() {
return Err(VideoForgeError::UnsafePath(
"final media is not a regular non-symlink file".into(),
));
}
if metadata.len() == 0 {
return Err(VideoForgeError::Integrity("final media is empty".into()));
}
if metadata.len() > options.max_download_bytes {
return Err(VideoForgeError::ByteLimit {
limit: options.max_download_bytes,
observed: metadata.len(),
});
}
let entries: Vec<_> = fs::read_dir(stage)?.collect::<Result<_, _>>()?;
if entries.len() != 1 || entries[0].path() != event_path {
return Err(VideoForgeError::Integrity(
"staging must contain exactly the final media file before manifest creation".into(),
));
}
let container = media_container(event_path)?;
let blake3 = hash_file(event_path)?;
File::open(event_path)?.sync_all()?;
Ok(VideoIntegrity {
media_file: file_name.to_owned(),
bytes: metadata.len(),
blake3,
container,
})
}
fn validate_relative_media_name(name: &str) -> Result<(), VideoForgeError> {
let path = Path::new(name);
let mut components = path.components();
if name.is_empty()
|| name == MANIFEST_NAME
|| !matches!(components.next(), Some(Component::Normal(_)))
|| components.next().is_some()
{
return Err(VideoForgeError::UnsafePath(
"manifest media path must be one safe relative component".into(),
));
}
Ok(())
}
fn media_container(path: &Path) -> Result<String, VideoForgeError> {
let extension = path
.extension()
.and_then(OsStr::to_str)
.map(str::to_ascii_lowercase)
.ok_or_else(|| VideoForgeError::Integrity("media has no supported extension".into()))?;
let mut file = File::open(path)?;
let mut magic = [0_u8; 4096];
let count = file.read(&mut magic)?;
let valid = match extension.as_str() {
"mp4" => count >= 12 && &magic[4..8] == b"ftyp",
"webm" => {
count >= 4
&& magic[..4] == [0x1a, 0x45, 0xdf, 0xa3]
&& magic[..count].windows(4).any(|window| window == b"webm")
}
"mkv" => {
count >= 4
&& magic[..4] == [0x1a, 0x45, 0xdf, 0xa3]
&& magic[..count]
.windows(8)
.any(|window| window == b"matroska")
}
_ => false,
};
if !valid {
return Err(VideoForgeError::Integrity(format!(
"media extension or container magic is unsupported: {extension}"
)));
}
Ok(extension)
}
fn hash_file(path: &Path) -> Result<String, VideoForgeError> {
let mut file = File::open(path)?;
let mut hasher = blake3::Hasher::new();
let mut buffer = [0_u8; 128 * 1024];
loop {
let count = file.read(&mut buffer)?;
if count == 0 {
break;
}
hasher.update(&buffer[..count]);
}
Ok(hasher.finalize().to_hex().to_string())
}
fn write_manifest(directory: &Path, manifest: &VideoManifest) -> Result<(), VideoForgeError> {
let encoded = serde_json::to_vec_pretty(manifest)?;
if encoded.len() as u64 > MANIFEST_LIMIT {
return Err(VideoForgeError::Metadata(
"filtered manifest exceeds the 2 MiB limit".into(),
));
}
let path = directory.join(MANIFEST_NAME);
let mut file = OpenOptions::new().write(true).create_new(true).open(path)?;
file.write_all(&encoded)?;
file.write_all(b"\n")?;
file.sync_all()?;
Ok(())
}
fn read_manifest(path: &Path) -> Result<VideoManifest, VideoForgeError> {
let metadata = fs::symlink_metadata(path)?;
if metadata.file_type().is_symlink() || !metadata.is_file() {
return Err(VideoForgeError::UnsafePath(
"manifest is not a regular non-symlink file".into(),
));
}
if metadata.len() > MANIFEST_LIMIT {
return Err(VideoForgeError::Integrity(
"manifest exceeds the 2 MiB limit".into(),
));
}
let mut bytes = Vec::with_capacity(metadata.len() as usize);
File::open(path)?
.take(MANIFEST_LIMIT + 1)
.read_to_end(&mut bytes)?;
if bytes.len() as u64 > MANIFEST_LIMIT {
return Err(VideoForgeError::Integrity(
"manifest grew beyond the 2 MiB limit".into(),
));
}
Ok(serde_json::from_slice(&bytes)?)
}
fn recognized_manifest(
destination: &Path,
target: &VideoTarget,
) -> Result<Option<VideoManifest>, VideoForgeError> {
match fs::symlink_metadata(destination) {
Ok(metadata) => {
if metadata.file_type().is_symlink() || !metadata.is_dir() {
return Err(VideoForgeError::ArchiveConflict(format!(
"{} is not a regular archive directory",
destination.display()
)));
}
}
Err(error) if error.kind() == io::ErrorKind::NotFound => return Ok(None),
Err(error) => return Err(error.into()),
}
let manifest_path = destination.join(MANIFEST_NAME);
let manifest = read_manifest(&manifest_path).map_err(|error| {
VideoForgeError::ArchiveConflict(format!(
"{} has no recognized manifest: {error}",
destination.display()
))
})?;
if manifest.schema_version != MANIFEST_SCHEMA
|| manifest.provider != target.provider()
|| manifest.video_id != target.id()
|| manifest.canonical_url != target.canonical_url()
|| manifest.summary.target != *target
{
return Err(VideoForgeError::ArchiveConflict(format!(
"{} manifest belongs to a different archive identity",
destination.display()
)));
}
Ok(Some(manifest))
}
fn ensure_safe_directory(path: &Path) -> Result<(), VideoForgeError> {
let absolute = if path.is_absolute() {
path.to_path_buf()
} else {
std::env::current_dir()?.join(path)
};
let mut current = PathBuf::new();
for component in absolute.components() {
match component {
Component::Prefix(prefix) => current.push(prefix.as_os_str()),
Component::RootDir => current.push(component.as_os_str()),
Component::CurDir => {}
Component::ParentDir => {
return Err(VideoForgeError::UnsafePath(
"output path must not contain parent traversal".into(),
));
}
Component::Normal(part) => {
current.push(part);
match fs::symlink_metadata(¤t) {
Ok(metadata) => {
if metadata.file_type().is_symlink() || !metadata.is_dir() {
return Err(VideoForgeError::UnsafePath(format!(
"output component {} is not a regular directory",
current.display()
)));
}
}
Err(error) if error.kind() == io::ErrorKind::NotFound => {
fs::create_dir(¤t)?;
let metadata = fs::symlink_metadata(¤t)?;
if metadata.file_type().is_symlink() || !metadata.is_dir() {
return Err(VideoForgeError::UnsafePath(format!(
"new output component {} is unsafe",
current.display()
)));
}
}
Err(error) => return Err(error.into()),
}
}
}
}
Ok(())
}
fn validate_existing_path_components(path: &Path) -> Result<(), VideoForgeError> {
let absolute = if path.is_absolute() {
path.to_path_buf()
} else {
std::env::current_dir()?.join(path)
};
let mut current = PathBuf::new();
for component in absolute.components() {
match component {
Component::Prefix(prefix) => current.push(prefix.as_os_str()),
Component::RootDir => current.push(component.as_os_str()),
Component::CurDir => {}
Component::ParentDir => {
return Err(VideoForgeError::UnsafePath(
"archive path must not contain parent traversal".into(),
));
}
Component::Normal(part) => {
current.push(part);
let metadata = fs::symlink_metadata(¤t)?;
if metadata.file_type().is_symlink() || !metadata.is_dir() {
return Err(VideoForgeError::UnsafePath(format!(
"archive path component {} is not a regular directory",
current.display()
)));
}
}
}
}
Ok(())
}
struct DestinationTransaction {
file: File,
directory: PathBuf,
interrupted: bool,
}
impl DestinationTransaction {
fn acquire(
output: &Path,
target: &VideoTarget,
timeout: Duration,
) -> Result<Self, VideoForgeError> {
let path = output.join(format!(".videoforge-{}.lock", target.id()));
if let Ok(metadata) = fs::symlink_metadata(&path) {
if metadata.file_type().is_symlink() || !metadata.is_file() {
return Err(VideoForgeError::UnsafePath(format!(
"transaction lock {} is not a regular file",
path.display()
)));
}
}
let mut file = OpenOptions::new()
.read(true)
.write(true)
.create(true)
.truncate(false)
.open(&path)?;
let started = Instant::now();
loop {
match file.try_lock_exclusive() {
Ok(()) => break,
Err(error) if error.kind() == io::ErrorKind::WouldBlock => {
if started.elapsed() >= timeout {
return Err(VideoForgeError::Timeout(timeout));
}
thread::sleep(POLL_INTERVAL);
}
Err(error) => return Err(error.into()),
}
}
let mut state_log = String::new();
file.rewind()?;
Read::by_ref(&mut file)
.take(64 * 1024 + 1)
.read_to_string(&mut state_log)?;
if state_log.len() > 64 * 1024 {
return Err(VideoForgeError::Transaction(format!(
"transaction lock {} exceeds its state limit",
path.display()
)));
}
let complete_log = if state_log.ends_with('\n') {
state_log.as_str()
} else {
state_log
.rsplit_once('\n')
.map_or("", |(complete, _)| complete)
};
let state = complete_log.lines().next_back().unwrap_or("").trim();
if !matches!(state, "" | "idle" | "preparing" | "installing") {
return Err(VideoForgeError::Transaction(format!(
"transaction lock {} contains an unknown state",
path.display()
)));
}
Ok(Self {
file,
directory: output.to_path_buf(),
interrupted: matches!(state, "preparing" | "installing"),
})
}
fn recover(
&mut self,
output: &Path,
destination: &Path,
target: &VideoTarget,
) -> Result<(), VideoForgeError> {
if self.interrupted {
recover_interrupted_install(output, destination, target)?;
self.mark_idle()?;
}
Ok(())
}
fn mark_installing(&mut self) -> Result<(), VideoForgeError> {
self.write_state("installing")?;
self.interrupted = true;
Ok(())
}
fn mark_preparing(&mut self) -> Result<(), VideoForgeError> {
self.write_state("preparing")?;
self.interrupted = true;
Ok(())
}
fn mark_idle(&mut self) -> Result<(), VideoForgeError> {
self.write_state("idle")?;
if self.file.metadata()?.len() > 60 * 1024 {
self.file.set_len(0)?;
self.file.rewind()?;
self.file.write_all(b"idle\n")?;
self.file.sync_all()?;
sync_directory(&self.directory)?;
}
self.interrupted = false;
Ok(())
}
fn write_state(&mut self, state: &str) -> Result<(), VideoForgeError> {
self.file.seek(io::SeekFrom::End(0))?;
self.file.write_all(state.as_bytes())?;
self.file.write_all(b"\n")?;
self.file.sync_all()?;
sync_directory(&self.directory)?;
Ok(())
}
}
impl Drop for DestinationTransaction {
fn drop(&mut self) {
let _ = FileExt::unlock(&self.file);
}
}
fn recover_interrupted_install(
output: &Path,
destination: &Path,
target: &VideoTarget,
) -> Result<(), VideoForgeError> {
let backup_prefix = format!(".videoforge-{}-backup-", target.id());
let backups = transaction_directories(output, &backup_prefix)?;
if backups.len() > 1 {
return Err(VideoForgeError::Transaction(format!(
"multiple interrupted backups exist for {}; operator review is required",
target.id()
)));
}
if verify_archive(destination).is_ok() {
if let Some(backup) = backups.first() {
fs::remove_dir_all(backup)?;
sync_directory(output)?;
}
return Ok(());
}
if destination.exists() {
if !looks_like_interrupted_destination(destination, target)? {
return Err(VideoForgeError::ArchiveConflict(format!(
"{} is not recognizable as interrupted VideoForge state",
destination.display()
)));
}
fs::remove_dir_all(destination)?;
}
if let Some(backup) = backups.first() {
let old_archive = backup.join("archive");
recognized_manifest(&old_archive, target)?.ok_or_else(|| {
VideoForgeError::Transaction(format!(
"interrupted backup {} is not a recognized archive",
backup.display()
))
})?;
fs::rename(&old_archive, destination)?;
fs::remove_dir(backup)?;
}
sync_directory(output)?;
Ok(())
}
fn transaction_directories(output: &Path, prefix: &str) -> Result<Vec<PathBuf>, VideoForgeError> {
let mut directories = Vec::new();
for entry in fs::read_dir(output)? {
let entry = entry?;
let name = entry.file_name();
if name.to_str().is_some_and(|name| name.starts_with(prefix)) {
let metadata = fs::symlink_metadata(entry.path())?;
if metadata.file_type().is_symlink() || !metadata.is_dir() {
return Err(VideoForgeError::UnsafePath(format!(
"transaction path {} is not a regular directory",
entry.path().display()
)));
}
directories.push(entry.path());
}
}
directories.sort();
Ok(directories)
}
fn looks_like_interrupted_destination(
destination: &Path,
target: &VideoTarget,
) -> Result<bool, VideoForgeError> {
let metadata = fs::symlink_metadata(destination)?;
if metadata.file_type().is_symlink() || !metadata.is_dir() {
return Ok(false);
}
let marker = destination.join(INSTALL_MARKER_NAME);
let marker_metadata = match fs::symlink_metadata(&marker) {
Ok(metadata) if metadata.is_file() && !metadata.file_type().is_symlink() => metadata,
_ => return Ok(false),
};
if marker_metadata.len() > 64 {
return Ok(false);
}
let marker_id = fs::read_to_string(&marker)?;
if marker_id.trim() != target.id() {
return Ok(false);
}
for entry in fs::read_dir(destination)? {
let entry = entry?;
let metadata = fs::symlink_metadata(entry.path())?;
if metadata.file_type().is_symlink() || !metadata.is_file() {
return Ok(false);
}
let name = entry.file_name();
let name = name.to_str().unwrap_or_default();
let valid_media = ["mp4", "webm", "mkv"]
.iter()
.any(|extension| name == format!("{}.{extension}", target.id()));
if name != INSTALL_MARKER_NAME && name != MANIFEST_NAME && !valid_media {
return Ok(false);
}
}
Ok(true)
}
fn install_stage(
stage: TempDir,
destination: &Path,
output: &Path,
target: &VideoTarget,
expected_existing: Option<&VideoManifest>,
transaction: &mut DestinationTransaction,
) -> Result<Vec<String>, VideoForgeError> {
revalidate_destination(destination, target, expected_existing)?;
let stage_path = stage.keep();
if expected_existing.is_none() {
if let Err(error) =
install_staged_files(&stage_path, destination, || transaction.mark_installing())
{
let cleanup = fs::remove_dir_all(&stage_path);
return Err(VideoForgeError::Transaction(format!(
"install failed: {error}; staging cleanup: {}",
result_label(&cleanup)
)));
}
if let Err(sync_error) = sync_directory(output) {
let cleanup = fs::remove_dir_all(destination);
let cleanup_sync = sync_directory(output);
return Err(VideoForgeError::Transaction(format!(
"install directory sync failed: {sync_error}; new archive cleanup: {}; cleanup sync: {}",
result_label(&cleanup),
result_label(&cleanup_sync)
)));
}
return Ok(Vec::new());
}
let backup = TempBuilder::new()
.prefix(&format!(".videoforge-{}-backup-", target.id()))
.tempdir_in(output)?;
let backup_old = backup.path().join("archive");
if let Err(error) = fs::rename(destination, &backup_old) {
let cleanup = fs::remove_dir_all(&stage_path);
return Err(VideoForgeError::Transaction(format!(
"could not move old archive to backup: {error}; staging cleanup: {}",
result_label(&cleanup)
)));
}
if let Err(error) = sync_directory(backup.path()).and_then(|()| sync_directory(output)) {
let rollback = fs::rename(&backup_old, destination);
let cleanup = fs::remove_dir_all(&stage_path);
let backup_note = if rollback.is_err() {
let preserved = backup.keep();
format!("; old archive preserved at {}", preserved.display())
} else {
match backup.close() {
Ok(()) => String::new(),
Err(error) => format!("; backup cleanup failed: {error}"),
}
};
return Err(VideoForgeError::Transaction(format!(
"could not durably place old archive in backup: {error}; rollback: {}; staging cleanup: {}{backup_note}",
result_label(&rollback),
result_label(&cleanup)
)));
}
let moved_manifest = recognized_manifest(&backup_old, target);
if moved_manifest.as_ref().ok().and_then(Option::as_ref) != expected_existing {
let rollback = fs::rename(&backup_old, destination);
let cleanup = fs::remove_dir_all(&stage_path);
let backup_note = if rollback.is_err() {
let preserved = backup.keep();
format!("; old archive preserved at {}", preserved.display())
} else {
match backup.close() {
Ok(()) => String::new(),
Err(error) => format!("; backup cleanup failed: {error}"),
}
};
return Err(VideoForgeError::Transaction(format!(
"archive changed during replacement; rollback: {}; staging cleanup: {}{backup_note}",
result_label(&rollback),
result_label(&cleanup),
)));
}
if let Err(install_error) =
install_staged_files(&stage_path, destination, || transaction.mark_installing())
{
let rollback = fs::rename(&backup_old, destination);
let cleanup = fs::remove_dir_all(&stage_path);
let backup_note = if rollback.is_err() {
let preserved = backup.keep();
format!("; old archive preserved at {}", preserved.display())
} else {
match backup.close() {
Ok(()) => String::new(),
Err(error) => format!("; backup cleanup failed: {error}"),
}
};
return Err(VideoForgeError::Transaction(format!(
"replacement install failed: {install_error}; rollback: {}; staging cleanup: {}{backup_note}",
result_label(&rollback),
result_label(&cleanup),
)));
}
if let Err(error) = sync_directory(output) {
let failed_new = backup.path().join("failed-new-archive");
let move_new = fs::rename(destination, &failed_new);
let restore_old = if move_new.is_ok() {
fs::rename(&backup_old, destination)
} else {
Err(io::Error::other(
"new archive could not be moved aside for rollback",
))
};
let rollback_sync = if restore_old.is_ok() {
sync_directory(output)
} else {
Err(VideoForgeError::Transaction(
"old archive was not restored".into(),
))
};
let backup_note = if restore_old.is_ok() && rollback_sync.is_ok() {
match backup.close() {
Ok(()) => String::new(),
Err(error) => format!("; rollback cleanup failed: {error}"),
}
} else {
let preserved = backup.keep();
format!("; transaction state preserved at {}", preserved.display())
};
return Err(VideoForgeError::Transaction(format!(
"replacement directory sync failed: {error}; move new: {}; restore old: {}; rollback sync: {}{backup_note}",
result_label(&move_new),
result_label(&restore_old),
result_label(&rollback_sync)
)));
}
let mut warnings = Vec::new();
if let Err(error) = backup.close() {
warnings.push(format!(
"replacement is verified, but old archive backup cleanup failed: {error}"
));
}
if let Err(error) = sync_directory(output) {
warnings.push(format!(
"replacement is verified, but final backup-removal directory sync failed: {error}"
));
}
Ok(warnings)
}
fn install_staged_files<F>(
stage: &Path,
destination: &Path,
mark_installing: F,
) -> Result<(), VideoForgeError>
where
F: FnOnce() -> Result<(), VideoForgeError>,
{
fs::create_dir(destination).map_err(|error| {
VideoForgeError::ArchiveConflict(format!(
"could not reserve new destination {} without replacement: {error}",
destination.display()
))
})?;
let result = (|| {
let marker_path = destination.join(INSTALL_MARKER_NAME);
let mut marker = OpenOptions::new()
.write(true)
.create_new(true)
.open(&marker_path)?;
marker.write_all(target_id_from_stage(stage)?.as_bytes())?;
marker.write_all(b"\n")?;
marker.sync_all()?;
sync_directory(destination)?;
mark_installing()?;
let mut entries = fs::read_dir(stage)?.collect::<Result<Vec<_>, _>>()?;
entries.sort_by_key(|entry| entry.file_name() == OsStr::new(MANIFEST_NAME));
if entries.len() != 2
|| !entries
.iter()
.any(|entry| entry.file_name() == OsStr::new(MANIFEST_NAME))
{
return Err(VideoForgeError::Integrity(
"staging must contain one media file and one manifest".into(),
));
}
for entry in entries {
let metadata = fs::symlink_metadata(entry.path())?;
if metadata.file_type().is_symlink() || !metadata.is_file() {
return Err(VideoForgeError::UnsafePath(
"staging entry changed before installation".into(),
));
}
fs::rename(entry.path(), destination.join(entry.file_name()))?;
}
fs::remove_file(marker_path)?;
fs::remove_dir(stage)?;
sync_directory(destination)?;
Ok(())
})();
if let Err(error) = result {
let cleanup = fs::remove_dir_all(destination);
return Err(VideoForgeError::Transaction(format!(
"staged file installation failed: {error}; destination cleanup: {}",
result_label(&cleanup)
)));
}
Ok(())
}
fn target_id_from_stage(stage: &Path) -> Result<String, VideoForgeError> {
let manifest = read_manifest(&stage.join(MANIFEST_NAME))?;
validate_video_id(&manifest.video_id)?;
Ok(manifest.video_id)
}
fn revalidate_destination(
destination: &Path,
target: &VideoTarget,
expected: Option<&VideoManifest>,
) -> Result<(), VideoForgeError> {
match expected {
None => match fs::symlink_metadata(destination) {
Err(error) if error.kind() == io::ErrorKind::NotFound => Ok(()),
Ok(_) => Err(VideoForgeError::ArchiveConflict(format!(
"{} appeared while the video was being staged",
destination.display()
))),
Err(error) => Err(error.into()),
},
Some(expected) => {
let current = recognized_manifest(destination, target)?.ok_or_else(|| {
VideoForgeError::ArchiveConflict(format!(
"{} disappeared while the video was being staged",
destination.display()
))
})?;
if ¤t != expected {
return Err(VideoForgeError::ArchiveConflict(format!(
"{} changed while the video was being staged",
destination.display()
)));
}
Ok(())
}
}
}
fn result_label<E: std::fmt::Display>(result: &std::result::Result<(), E>) -> String {
match result {
Ok(()) => "ok".into(),
Err(error) => format!("failed ({error})"),
}
}
#[cfg(unix)]
fn sync_directory(path: &Path) -> Result<(), VideoForgeError> {
File::open(path)?.sync_all()?;
Ok(())
}
#[cfg(windows)]
fn sync_directory(path: &Path) -> Result<(), VideoForgeError> {
use std::os::windows::fs::OpenOptionsExt;
const FILE_FLAG_BACKUP_SEMANTICS: u32 = 0x0200_0000;
OpenOptions::new()
.read(true)
.custom_flags(FILE_FLAG_BACKUP_SEMANTICS)
.open(path)?
.sync_all()?;
Ok(())
}
#[cfg(not(any(unix, windows)))]
fn sync_directory(_path: &Path) -> Result<(), VideoForgeError> {
Ok(())
}
enum ArchiveOutcome {
Archived(ArchivedVideo),
Skipped(ArchivedVideo),
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn canonicalizes_all_supported_url_shapes() {
for input in [
"https://youtube.com/watch?v=BaW_jenozKc",
"https://www.youtube.com/watch?v=BaW_jenozKc",
"https://m.youtube.com/watch?v=BaW_jenozKc",
"https://youtu.be/BaW_jenozKc",
"https://www.youtube.com/shorts/BaW_jenozKc",
"https://www.youtube.com/live/BaW_jenozKc",
"https://www.youtube.com/embed/BaW_jenozKc",
] {
let target = parse_video_target(input).unwrap();
assert_eq!(target.id(), "BaW_jenozKc");
assert_eq!(
target.canonical_url(),
"https://www.youtube.com/watch?v=BaW_jenozKc"
);
assert_eq!(target.provider().label(), "YouTube");
assert_eq!(target.provider().slug(), "youtube");
}
}
#[test]
fn rejects_urls_outside_exact_grammar() {
for input in [
"http://www.youtube.com/watch?v=BaW_jenozKc",
"https://user@www.youtube.com/watch?v=BaW_jenozKc",
"https://@www.youtube.com/watch?v=BaW_jenozKc",
"https://www.youtube.com:443/watch?v=BaW_jenozKc",
"https://www.youtube.com/watch?v=BaW_jenozKc#x",
"https://www.youtube.com/watch?v=BaW_jenozKc&list=PL123",
"https://www.youtube.com/playlist?list=BaW_jenozKc",
"https://evil.example/watch?v=BaW_jenozKc",
"https://youtube.com.evil.example/watch?v=BaW_jenozKc",
"https://www.youtube.com/watch/BaW_jenozKc",
"https://youtu.be/BaW_jenozKc?t=1",
"https://www.youtube.com/shorts/BaW_jenozKc?feature=share",
"https://www.youtube.com/watch?v=short",
"https://www.youtube.com/watch?v=BaW%5FjenozKc",
" https://youtu.be/BaW_jenozKc",
] {
assert!(parse_video_target(input).is_err(), "accepted {input}");
}
}
#[test]
fn deserialization_cannot_bypass_target_validation() {
let forged = serde_json::json!({
"provider": "youtube",
"id": "BaW_jenozKc",
"canonical_url": "https://evil.example/watch?v=BaW_jenozKc"
});
assert!(serde_json::from_value::<VideoTarget>(forged).is_err());
let target = parse_video_target("https://youtu.be/BaW_jenozKc").unwrap();
let round_trip: VideoTarget =
serde_json::from_value(serde_json::to_value(&target).unwrap()).unwrap();
assert_eq!(round_trip, target);
}
#[test]
fn parses_progress_only_from_namespaced_json() {
let progress = parse_progress_line(
"videoforge-progress:{\"downloaded_bytes\":12,\"total_bytes\":\"20\",\"speed\":4.5,\"eta\":2}",
)
.unwrap();
assert_eq!(progress.downloaded_bytes, 12);
assert_eq!(progress.total_bytes, Some(20));
assert_eq!(progress.speed, Some(4.5));
assert_eq!(progress.eta_seconds, Some(2));
assert!(parse_progress_line("[download] 50%").is_none());
assert!(parse_progress_line("videoforge-progress:not-json").is_none());
}
#[test]
fn diagnostics_redact_urls_and_sensitive_tokens() {
let output = diagnostic(
b"ERROR from 'https://media.example/videoplayback?x=1&sig=secret': Authorization: Bearer secret cookie=value token=secret",
);
assert!(!output.contains("media.example"));
assert!(!output.contains("secret"));
assert!(!output.contains("cookie=value"));
assert!(output.contains("[redacted-sensitive-value]"));
}
#[test]
fn metadata_policy_rejects_nonpublic_and_drm() {
let target = parse_video_target("https://youtu.be/BaW_jenozKc").unwrap();
let mut raw = valid_raw_metadata();
raw.availability = Some("unlisted".into());
assert!(matches!(
raw.into_summary(&target),
Err(VideoForgeError::Policy(_))
));
let raw: RawMetadata = serde_json::from_value(serde_json::json!({
"id": "BaW_jenozKc",
"title": "Test video",
"webpage_url": "https://www.youtube.com/watch?v=BaW_jenozKc",
"extractor": "youtube",
"extractor_key": "Youtube",
"availability": "public",
"age_limit": 0,
"live_status": "not_live",
"_has_drm": true
}))
.unwrap();
assert!(matches!(
raw.into_summary(&target),
Err(VideoForgeError::Policy(_))
));
let mut raw = valid_raw_metadata();
raw.webpage_url = None;
assert!(matches!(
raw.into_summary(&target),
Err(VideoForgeError::Policy(_))
));
let mut raw = valid_raw_metadata();
raw.has_drm = None;
assert!(raw.into_summary(&target).is_ok());
let mut raw = valid_raw_metadata();
raw.requested_formats = None;
raw.has_drm = Some(false);
assert!(raw.into_summary(&target).is_ok());
let mut raw = valid_raw_metadata();
raw.requested_formats.as_mut().unwrap()[0].has_drm = Some(true);
assert!(matches!(
raw.into_summary(&target),
Err(VideoForgeError::Policy(_))
));
}
#[test]
fn metadata_estimates_selected_format_bytes_and_preserves_rights() {
let target = parse_video_target("https://youtu.be/BaW_jenozKc").unwrap();
let summary = valid_raw_metadata().into_summary(&target).unwrap();
assert_eq!(summary.estimated_bytes, Some(30));
assert_eq!(summary.rights.channel.as_deref(), Some("Channel"));
assert!(summary.rights.attribution.contains(target.canonical_url()));
assert!(
summary
.rights
.redistribution_warning
.contains("does not grant")
);
assert!(summary.rights.yt_dlp_license_notice.contains("no rights"));
}
#[test]
fn embeds_the_packaged_agent_guide() {
assert_eq!(AGENT_GUIDE, include_str!("../AGENT_GUIDE.md"));
}
fn valid_raw_metadata() -> RawMetadata {
RawMetadata {
id: Some("BaW_jenozKc".into()),
title: Some("Test video".into()),
description: Some("Description".into()),
uploader: Some("Uploader".into()),
uploader_id: Some("uploader-id".into()),
channel: Some("Channel".into()),
channel_id: Some("channel-id".into()),
upload_date: Some("20121002".into()),
duration: Some(10.0),
webpage_url: Some("https://www.youtube.com/watch?v=BaW_jenozKc".into()),
extractor: Some("youtube".into()),
extractor_key: Some("Youtube".into()),
availability: Some("public".into()),
age_limit: Some(0),
live_status: Some("not_live".into()),
license: None,
filesize: None,
filesize_approx: None,
requested_formats: Some(vec![
RawFormat {
filesize: Some(10),
filesize_approx: None,
has_drm: Some(false),
},
RawFormat {
filesize: Some(20),
filesize_approx: None,
has_drm: Some(false),
},
]),
has_drm: Some(false),
entries: None,
}
}
#[cfg(unix)]
mod unix {
use super::*;
use std::os::unix::fs::{PermissionsExt, symlink};
use std::sync::atomic::AtomicUsize;
use std::sync::{Mutex, MutexGuard};
static PATH_LOCK: Mutex<()> = Mutex::new(());
static SCRIPT_COUNTER: AtomicUsize = AtomicUsize::new(0);
fn write_script(directory: &Path, body: &str) -> PathBuf {
let script_id = SCRIPT_COUNTER.fetch_add(1, Ordering::Relaxed);
let path = directory.join(format!("fake-yt-dlp-{script_id}"));
let mut file = File::create(&path).unwrap();
file.write_all(format!("#!/bin/sh\nset -eu\n{body}\n").as_bytes())
.unwrap();
file.sync_all().unwrap();
drop(file);
let mut permissions = fs::metadata(&path).unwrap().permissions();
permissions.set_mode(0o700);
fs::set_permissions(&path, permissions).unwrap();
path
}
fn valid_json() -> String {
serde_json::json!({
"id": "BaW_jenozKc",
"title": "Test video",
"description": "Description",
"uploader": "Uploader",
"uploader_id": "uploader-id",
"channel": "Channel",
"channel_id": "channel-id",
"upload_date": "20121002",
"duration": 10.0,
"webpage_url": "https://www.youtube.com/watch?v=BaW_jenozKc",
"extractor": "youtube",
"extractor_key": "Youtube",
"availability": "public",
"age_limit": 0,
"live_status": "not_live",
"license": null,
"requested_formats": [{"filesize": 12, "has_drm": false}],
"_has_drm": false
})
.to_string()
}
fn fake_tools(directory: &Path) -> PathGuard {
let bin = directory.join("bin");
fs::create_dir(&bin).unwrap();
for name in ["ffmpeg", "ffprobe"] {
let path = bin.join(name);
fs::write(&path, format!("#!/bin/sh\necho '{name} version fake'\n")).unwrap();
let mut permissions = fs::metadata(&path).unwrap().permissions();
permissions.set_mode(0o700);
fs::set_permissions(path, permissions).unwrap();
}
PathGuard::prepend(&bin)
}
struct PathGuard {
_lock: MutexGuard<'static, ()>,
old: Option<std::ffi::OsString>,
}
impl PathGuard {
fn prepend(path: &Path) -> Self {
let lock = PATH_LOCK.lock().unwrap();
let old = std::env::var_os("PATH");
let mut paths = vec![path.to_path_buf()];
paths.extend(std::env::split_paths(old.as_deref().unwrap_or_default()));
unsafe {
std::env::set_var("PATH", std::env::join_paths(paths).unwrap());
}
Self { _lock: lock, old }
}
}
impl Drop for PathGuard {
fn drop(&mut self) {
unsafe {
match self.old.take() {
Some(value) => std::env::set_var("PATH", value),
None => std::env::remove_var("PATH"),
}
}
}
}
fn archive_options(output: PathBuf) -> ArchiveOptions {
ArchiveOptions {
output,
authorized: true,
max_download_bytes: 1024 * 1024,
timeout: Duration::from_secs(5),
..ArchiveOptions::default()
}
}
#[test]
fn executable_version_and_hardening_arguments() {
let temp = tempfile::tempdir().unwrap();
let arguments = temp.path().join("args");
let script = write_script(
temp.path(),
&format!(
"printf '%s\\n' \"$@\" > '{}'\nprintf '%s\\n' \"$YTDLP_NO_PLUGINS\" >> '{}'\necho 2026.07.01",
arguments.display(),
arguments.display()
),
);
let version = VideoForge::with_executable(script)
.yt_dlp_version(Duration::from_secs(2))
.unwrap();
assert_eq!(version, "2026.07.01");
let args = fs::read_to_string(arguments).unwrap();
for required in [
"--ignore-config",
"--no-plugin-dirs",
"--no-remote-components",
"--use-extractors",
"youtube",
"--no-playlist",
"--abort-on-error",
"--no-color",
"--proxy",
"--no-geo-bypass",
"--version",
"1",
] {
assert!(
args.lines().any(|line| line == required),
"missing {required}"
);
}
assert!(args.contains("--proxy\n\n--no-geo-bypass"));
}
#[test]
fn inspect_has_bounded_json_and_no_argument_injection() {
let temp = tempfile::tempdir().unwrap();
let arguments = temp.path().join("args");
let script = write_script(
temp.path(),
&format!(
"printf '%s\\n' \"$@\" > '{}'\nprintf '%s' '{}'",
arguments.display(),
valid_json()
),
);
let target = parse_video_target("https://youtu.be/BaW_jenozKc").unwrap();
let summary = VideoForge::with_executable(script)
.inspect(&target, VideoQuality::Max720p, Duration::from_secs(2))
.unwrap();
assert_eq!(summary.title, "Test video");
let args: Vec<_> = fs::read_to_string(arguments)
.unwrap()
.lines()
.map(str::to_owned)
.collect();
assert_eq!(args[args.len() - 2], "--");
assert_eq!(args.last().unwrap(), target.canonical_url());
assert!(args.contains(&VideoQuality::Max720p.selector().to_owned()));
let huge = write_script(
temp.path(),
"dd if=/dev/zero bs=1048576 count=9 2>/dev/null",
);
let error = VideoForge::with_executable(huge)
.inspect(&target, VideoQuality::Best, Duration::from_secs(2))
.unwrap_err();
assert!(matches!(error, VideoForgeError::MetadataTooLarge));
}
#[test]
fn inspect_terminates_descendants_that_hold_output_pipes() {
let temp = tempfile::tempdir().unwrap();
let script = write_script(
temp.path(),
&format!("(sleep 30) &\nprintf '%s' '{}'", valid_json()),
);
let target = parse_video_target("https://youtu.be/BaW_jenozKc").unwrap();
let started = Instant::now();
let summary = VideoForge::with_executable(script)
.inspect(&target, VideoQuality::Best, Duration::from_secs(5))
.unwrap();
assert_eq!(summary.title, "Test video");
assert!(started.elapsed() < Duration::from_secs(5));
}
#[test]
fn authorization_rejects_before_output_creation() {
let temp = tempfile::tempdir().unwrap();
let output = temp.path().join("must-not-exist");
let target = parse_video_target("https://youtu.be/BaW_jenozKc").unwrap();
let options = ArchiveOptions {
output: output.clone(),
..ArchiveOptions::default()
};
let error = VideoForge::new().archive(&[target], &options).unwrap_err();
assert!(matches!(error, VideoForgeError::Unauthorized));
assert!(!output.exists());
}
#[test]
fn duplicate_targets_are_rejected_before_output_creation() {
let temp = tempfile::tempdir().unwrap();
let output = temp.path().join("must-not-exist");
let target = parse_video_target("https://youtu.be/BaW_jenozKc").unwrap();
let error = VideoForge::new()
.archive(&[target.clone(), target], &archive_options(output.clone()))
.unwrap_err();
assert!(matches!(error, VideoForgeError::InvalidOptions(_)));
assert!(!output.exists());
}
#[test]
fn valid_archive_verify_skip_and_summary_invariants() {
let temp = tempfile::tempdir().unwrap();
let _path = fake_tools(temp.path());
let script = archive_script(temp.path(), false);
let target = parse_video_target("https://youtu.be/BaW_jenozKc").unwrap();
let options = archive_options(temp.path().join("archives"));
let forge = VideoForge::with_executable(script);
let first = forge
.archive(std::slice::from_ref(&target), &options)
.unwrap();
assert_eq!((first.archived, first.skipped, first.failed), (1, 0, 0));
let download_args = fs::read_to_string(temp.path().join("download-args")).unwrap();
for required in [
"--no-simulate",
"--progress",
"--no-mtime",
"--no-plugin-dirs",
"--no-remote-components",
"--no-geo-bypass",
"--abort-on-unavailable-fragments",
"--extractor-retries",
] {
assert!(
download_args.lines().any(|line| line == required),
"missing download argument {required}"
);
}
assert!(download_args.contains("--proxy\n\n--no-geo-bypass"));
assert!(!download_args.contains("--cookies"));
let verified = verify_archive(options.output.join(target.id())).unwrap();
assert_eq!(verified.target, target);
let skipped = forge
.archive(
std::slice::from_ref(&target),
&ArchiveOptions {
skip_existing: true,
..options.clone()
},
)
.unwrap();
assert_eq!(
(skipped.archived, skipped.skipped, skipped.failed),
(0, 1, 0)
);
assert_eq!(skipped.skipped_videos.len(), 1);
assert_eq!(
skipped.selected_ids.len(),
skipped.archived + skipped.skipped + skipped.failed
);
}
#[test]
fn corrupt_recognized_archive_is_repaired() {
let temp = tempfile::tempdir().unwrap();
let _path = fake_tools(temp.path());
let script = archive_script(temp.path(), false);
let target = parse_video_target("https://youtu.be/BaW_jenozKc").unwrap();
let options = archive_options(temp.path().join("archives"));
let forge = VideoForge::with_executable(script);
forge
.archive(std::slice::from_ref(&target), &options)
.unwrap();
let archive = options.output.join(target.id());
let verified = verify_archive(&archive).unwrap();
fs::write(
archive.join(&verified.manifest.integrity.media_file),
b"corrupt",
)
.unwrap();
let result = forge
.archive(
std::slice::from_ref(&target),
&ArchiveOptions {
skip_existing: true,
..options
},
)
.unwrap();
assert_eq!((result.archived, result.skipped, result.failed), (1, 0, 0));
verify_archive(archive).unwrap();
}
#[test]
fn a_different_quality_policy_prevents_skip() {
let temp = tempfile::tempdir().unwrap();
let _path = fake_tools(temp.path());
let script = archive_script(temp.path(), false);
let target = parse_video_target("https://youtu.be/BaW_jenozKc").unwrap();
let options = archive_options(temp.path().join("archives"));
let forge = VideoForge::with_executable(script);
forge
.archive(std::slice::from_ref(&target), &options)
.unwrap();
let result = forge
.archive(
std::slice::from_ref(&target),
&ArchiveOptions {
quality: VideoQuality::Max720p,
skip_existing: true,
..options
},
)
.unwrap();
assert_eq!((result.archived, result.skipped, result.failed), (1, 0, 0));
assert_eq!(
result.archived_videos[0].manifest.quality,
VideoQuality::Max720p
);
}
#[test]
fn stale_metadata_prevents_verified_skip_and_repairs_archive() {
let temp = tempfile::tempdir().unwrap();
let _path = fake_tools(temp.path());
let script = archive_script(temp.path(), false);
let target = parse_video_target("https://youtu.be/BaW_jenozKc").unwrap();
let options = archive_options(temp.path().join("archives"));
let forge = VideoForge::with_executable(script);
forge
.archive(std::slice::from_ref(&target), &options)
.unwrap();
let archive = options.output.join(target.id());
let manifest_path = archive.join(MANIFEST_NAME);
let mut manifest = read_manifest(&manifest_path).unwrap();
manifest.summary.title = "stale title".into();
fs::write(&manifest_path, serde_json::to_vec(&manifest).unwrap()).unwrap();
assert_eq!(
verify_archive(&archive).unwrap().manifest.summary.title,
"stale title"
);
let result = forge
.archive(
std::slice::from_ref(&target),
&ArchiveOptions {
skip_existing: true,
..options
},
)
.unwrap();
assert_eq!((result.archived, result.skipped, result.failed), (1, 0, 0));
assert_eq!(
verify_archive(archive).unwrap().manifest.summary.title,
"Test video"
);
}
#[test]
fn advertised_size_over_cap_fails_before_download() {
let temp = tempfile::tempdir().unwrap();
let script = archive_script(temp.path(), false);
let target = parse_video_target("https://youtu.be/BaW_jenozKc").unwrap();
let mut options = archive_options(temp.path().join("archives"));
options.max_download_bytes = 8;
let summary = VideoForge::with_executable(script)
.archive(&[target], &options)
.unwrap();
assert_eq!(
(summary.archived, summary.skipped, summary.failed),
(0, 0, 1)
);
assert!(summary.failures[0].error.contains("byte cap"));
}
#[test]
fn foreign_destination_is_refused_per_video() {
let temp = tempfile::tempdir().unwrap();
let output = temp.path().join("archives");
let destination = output.join("BaW_jenozKc");
fs::create_dir_all(&destination).unwrap();
fs::write(destination.join("foreign.txt"), b"keep").unwrap();
let target = parse_video_target("https://youtu.be/BaW_jenozKc").unwrap();
let summary = VideoForge::new()
.archive(&[target], &archive_options(output))
.unwrap();
assert_eq!(
(summary.archived, summary.skipped, summary.failed),
(0, 0, 1)
);
assert!(summary.failures[0].error.contains("destination conflict"));
assert_eq!(fs::read(destination.join("foreign.txt")).unwrap(), b"keep");
}
#[test]
fn byte_cap_kills_download_and_accumulates_failure() {
let temp = tempfile::tempdir().unwrap();
let _path = fake_tools(temp.path());
let script = archive_script(temp.path(), true);
let target = parse_video_target("https://youtu.be/BaW_jenozKc").unwrap();
let mut options = archive_options(temp.path().join("archives"));
options.max_download_bytes = 32;
let summary = VideoForge::with_executable(script)
.archive(&[target], &options)
.unwrap();
assert_eq!(summary.failed, 1);
assert!(summary.failures[0].error.contains("byte cap"));
}
#[test]
fn manifest_traversal_and_symlink_are_rejected() {
let temp = tempfile::tempdir().unwrap();
let archive = create_manual_archive(temp.path());
let manifest_path = archive.join(MANIFEST_NAME);
let mut manifest = read_manifest(&manifest_path).unwrap();
manifest.integrity.media_file = "../outside.mp4".into();
fs::write(&manifest_path, serde_json::to_vec(&manifest).unwrap()).unwrap();
assert!(matches!(
verify_archive(&archive),
Err(VideoForgeError::UnsafePath(_))
));
let archive = create_manual_archive(temp.path());
let manifest = read_manifest(archive.join(MANIFEST_NAME).as_path()).unwrap();
let media = archive.join(&manifest.integrity.media_file);
fs::remove_file(&media).unwrap();
let outside = temp.path().join("outside.mp4");
fs::write(&outside, mp4_bytes()).unwrap();
symlink(outside, media).unwrap();
assert!(matches!(
verify_archive(archive),
Err(VideoForgeError::UnsafePath(_))
));
}
#[test]
fn verification_rejects_extra_files_bad_magic_and_oversized_manifests() {
let temp = tempfile::tempdir().unwrap();
let archive = create_manual_archive(temp.path());
fs::write(archive.join("extra.txt"), b"unexpected").unwrap();
assert!(matches!(
verify_archive(&archive),
Err(VideoForgeError::Integrity(_))
));
let archive = create_manual_archive(temp.path());
let manifest = read_manifest(&archive.join(MANIFEST_NAME)).unwrap();
fs::write(archive.join(manifest.integrity.media_file), b"not an mp4").unwrap();
assert!(matches!(
verify_archive(&archive),
Err(VideoForgeError::Integrity(_))
));
let archive = create_manual_archive(temp.path());
let manifest_path = archive.join(MANIFEST_NAME);
let oversized = File::create(&manifest_path).unwrap();
oversized.set_len(MANIFEST_LIMIT + 1).unwrap();
assert!(matches!(
verify_archive(archive),
Err(VideoForgeError::Integrity(_))
));
}
#[test]
fn destination_changes_are_detected_before_install() {
let temp = tempfile::tempdir().unwrap();
let target = parse_video_target("https://youtu.be/BaW_jenozKc").unwrap();
let archive = create_manual_archive(temp.path());
let expected = read_manifest(&archive.join(MANIFEST_NAME)).unwrap();
let mut changed = expected.clone();
changed.summary.title = "changed concurrently".into();
fs::write(
archive.join(MANIFEST_NAME),
serde_json::to_vec(&changed).unwrap(),
)
.unwrap();
assert!(matches!(
revalidate_destination(&archive, &target, Some(&expected)),
Err(VideoForgeError::ArchiveConflict(_))
));
}
#[test]
fn interrupted_install_state_removes_partial_destination() {
let temp = tempfile::tempdir().unwrap();
let output = temp.path().join("archives");
fs::create_dir(&output).unwrap();
let target = parse_video_target("https://youtu.be/BaW_jenozKc").unwrap();
let destination = output.join(target.id());
{
let mut transaction =
DestinationTransaction::acquire(&output, &target, Duration::from_secs(1))
.unwrap();
transaction.mark_preparing().unwrap();
}
fs::create_dir(&destination).unwrap();
fs::write(destination.join(INSTALL_MARKER_NAME), target.id()).unwrap();
fs::write(destination.join("BaW_jenozKc.mp4"), mp4_bytes()).unwrap();
let mut recovered =
DestinationTransaction::acquire(&output, &target, Duration::from_secs(1)).unwrap();
recovered.recover(&output, &destination, &target).unwrap();
assert!(!destination.exists());
}
#[test]
fn interrupted_state_never_claims_an_unmarked_foreign_destination() {
let temp = tempfile::tempdir().unwrap();
let output = temp.path().join("archives");
fs::create_dir(&output).unwrap();
let target = parse_video_target("https://youtu.be/BaW_jenozKc").unwrap();
let destination = output.join(target.id());
{
let mut transaction =
DestinationTransaction::acquire(&output, &target, Duration::from_secs(1))
.unwrap();
transaction.mark_preparing().unwrap();
}
fs::create_dir(&destination).unwrap();
let mut recovered =
DestinationTransaction::acquire(&output, &target, Duration::from_secs(1)).unwrap();
assert!(matches!(
recovered.recover(&output, &destination, &target),
Err(VideoForgeError::ArchiveConflict(_))
));
assert!(destination.is_dir());
}
#[test]
fn interrupted_replacement_restores_recognized_backup() {
let temp = tempfile::tempdir().unwrap();
let output = temp.path().join("archives");
fs::create_dir(&output).unwrap();
let target = parse_video_target("https://youtu.be/BaW_jenozKc").unwrap();
let destination = create_manual_archive(&output);
let backup = output.join(format!(".videoforge-{}-backup-interrupted", target.id()));
{
let mut transaction =
DestinationTransaction::acquire(&output, &target, Duration::from_secs(1))
.unwrap();
transaction.mark_preparing().unwrap();
}
fs::create_dir(&backup).unwrap();
fs::rename(&destination, backup.join("archive")).unwrap();
fs::create_dir(&destination).unwrap();
fs::write(destination.join(INSTALL_MARKER_NAME), target.id()).unwrap();
fs::write(destination.join("BaW_jenozKc.mp4"), mp4_bytes()).unwrap();
let mut recovered =
DestinationTransaction::acquire(&output, &target, Duration::from_secs(1)).unwrap();
recovered.recover(&output, &destination, &target).unwrap();
assert!(!backup.exists());
verify_archive(destination).unwrap();
}
fn archive_script(directory: &Path, oversized: bool) -> PathBuf {
let json = valid_json().replace('"', "\\\"");
let arguments = directory.join("download-args");
let media_command = if oversized {
"dd if=/dev/zero of=\"$stage/BaW_jenozKc.mp4\" bs=1024 count=2 2>/dev/null; sleep 1"
} else {
"printf '\\000\\000\\000\\030ftypisomvideoforge' > \"$stage/BaW_jenozKc.mp4\""
};
write_script(
directory,
&format!(
"case \" $* \" in\n *' --version '*) echo 2026.07.01; exit 0;;\n *' --dump-single-json '*) printf '%s' \"{json}\"; exit 0;;\nesac\nprintf '%s\\n' \"$@\" > '{}'\nstage=''\nprevious=''\nfor argument in \"$@\"; do\n if [ \"$previous\" = '--paths' ]; then stage=$argument; fi\n previous=$argument\ndone\n{media_command}\nprintf '%s\\n' 'videoforge-progress:{{\"downloaded_bytes\":20,\"total_bytes\":20,\"speed\":10,\"eta\":0}}'\nprintf 'videoforge-file:{{\"filepath\":\"%s/BaW_jenozKc.mp4\",\"id\":\"BaW_jenozKc\"}}\\n' \"$stage\"",
arguments.display()
),
)
}
fn mp4_bytes() -> &'static [u8] {
b"\0\0\0\x18ftypisomvideoforge"
}
fn create_manual_archive(parent: &Path) -> PathBuf {
let target = parse_video_target("https://youtu.be/BaW_jenozKc").unwrap();
let directory = parent.join(target.id());
if directory.exists() {
fs::remove_dir_all(&directory).unwrap();
}
fs::create_dir(&directory).unwrap();
let media_file = "BaW_jenozKc.mp4";
fs::write(directory.join(media_file), mp4_bytes()).unwrap();
let summary = valid_raw_metadata().into_summary(&target).unwrap();
let manifest = VideoManifest {
schema_version: MANIFEST_SCHEMA,
provider: VideoProvider::Youtube,
video_id: target.id().into(),
canonical_url: target.canonical_url().into(),
summary,
yt_dlp_version: "test".into(),
quality: VideoQuality::Best,
integrity: VideoIntegrity {
media_file: media_file.into(),
bytes: mp4_bytes().len() as u64,
blake3: blake3::hash(mp4_bytes()).to_hex().to_string(),
container: "mp4".into(),
},
complete: true,
};
fs::write(
directory.join(MANIFEST_NAME),
serde_json::to_vec(&manifest).unwrap(),
)
.unwrap();
directory
}
}
#[test]
#[ignore = "requires network access and an installed current yt-dlp"]
fn live_inspect_public_test_video() {
let target = parse_video_target("https://www.youtube.com/watch?v=jNQXAC9IVRw").unwrap();
let summary = VideoForge::new()
.inspect(&target, VideoQuality::Best, Duration::from_secs(120))
.unwrap();
assert_eq!(summary.target, target);
}
}