Added Cloudflare R2 media storage support

src/bin/normalize_videos_to_r2.rs

@@ -5,6 +5,8 @@ mod models;
 #[path = "../uploads.rs"]
 mod uploads;
 
+use std::path::{Path, PathBuf};
+
 use cot::db::{Database, Model};
 use models::Media;
 
@@ -14,8 +16,147 @@ fn database_url() -> String {
         .unwrap_or_else(|_| "postgresql://postgres:postgres@localhost:5432/web_petting".to_string())
 }
 
-async fn local_file_exists(path: &std::path::Path) -> Result<bool, std::io::Error> {
-    tokio::fs::try_exists(path).await
+async fn local_file_exists(path: &Path) -> Result<bool, std::io::Error> {
+    match tokio::fs::metadata(path).await {
+        Ok(metadata) => Ok(metadata.is_file() && metadata.len() > 0),
+        Err(error) if error.kind() == std::io::ErrorKind::NotFound => Ok(false),
+        Err(error) => Err(error),
+    }
+}
+
+async fn valid_local_video(path: &Path, media_id: i64) -> Result<bool, std::io::Error> {
+    if !local_file_exists(path).await? {
+        return Ok(false);
+    }
+    match uploads::validate_video_file(path).await {
+        Ok(()) => Ok(true),
+        Err(error) => {
+            eprintln!(
+                "Ignoring incomplete local output for media {media_id}: {} ({error})",
+                path.display()
+            );
+            Ok(false)
+        }
+    }
+}
+
+async fn download_to_workspace(
+    r2: &uploads::Storage,
+    db_path: &str,
+    workspace: &Path,
+    filename: &str,
+) -> Result<PathBuf, Box<dyn std::error::Error>> {
+    let destination = workspace.join(filename);
+    r2.download_to_path(db_path, &destination).await?;
+    Ok(destination)
+}
+
+async fn prepare_missing_objects(
+    r2: &uploads::Storage,
+    media_id: i64,
+    original_db_path: &str,
+    target_db_path: &str,
+    target_thumbnail_db_path: &str,
+    target_in_r2: bool,
+    thumbnail_in_r2: bool,
+) -> Result<(), Box<dyn std::error::Error>> {
+    let workspace =
+        std::env::temp_dir().join(format!("web-petting-normalize-{}", uuid::Uuid::new_v4()));
+    tokio::fs::create_dir(&workspace).await?;
+
+    let result = async {
+        let original_path = uploads::resolve_db_path(original_db_path);
+        let local_target_path = uploads::resolve_db_path(target_db_path);
+        let generated_video_path = workspace.join("video.web.mp4");
+        let generated_thumbnail_path = workspace.join("video.web.thumb.jpg");
+        let mut prepared_video: Option<PathBuf> = None;
+        let mut prepared_thumbnail: Option<PathBuf> = None;
+
+        if !target_in_r2 {
+            if valid_local_video(&local_target_path, media_id).await? {
+                println!("Reusing valid compact video on PVC for media {media_id}");
+                prepared_video = Some(local_target_path.clone());
+            } else {
+                let source = if original_db_path != target_db_path
+                    && local_file_exists(&original_path).await?
+                {
+                    original_path.clone()
+                } else if original_db_path != target_db_path && r2.exists(original_db_path).await? {
+                    println!("Downloading source from R2 for media {media_id}");
+                    let extension = original_db_path.rsplit('.').next().unwrap_or("mov");
+                    download_to_workspace(
+                        r2,
+                        original_db_path,
+                        &workspace,
+                        &format!("source.{extension}"),
+                    )
+                    .await?
+                } else {
+                    return Err(format!(
+                        "source is absent from both PVC and R2: {}",
+                        original_path.display()
+                    )
+                    .into());
+                };
+
+                println!("Converting media {media_id}: {original_db_path}");
+                uploads::create_compact_video_files(
+                    &source,
+                    &generated_video_path,
+                    &generated_thumbnail_path,
+                )
+                .await?;
+                prepared_video = Some(generated_video_path.clone());
+                prepared_thumbnail = Some(generated_thumbnail_path.clone());
+            }
+
+            let video_path = prepared_video
+                .as_ref()
+                .ok_or("compact video was not prepared")?;
+            uploads::validate_video_file(video_path).await?;
+            r2.upload_local_copy(target_db_path, video_path).await?;
+            println!("Uploaded compact video: {target_db_path}");
+        }
+
+        if !thumbnail_in_r2 {
+            let video_for_preview = if let Some(video_path) = prepared_video.as_ref() {
+                video_path.clone()
+            } else if valid_local_video(&local_target_path, media_id).await? {
+                local_target_path
+            } else {
+                println!("Downloading compact video from R2 for media {media_id}");
+                let downloaded =
+                    download_to_workspace(r2, target_db_path, &workspace, "existing-video.web.mp4")
+                        .await?;
+                uploads::validate_video_file(&downloaded).await?;
+                downloaded
+            };
+
+            let thumbnail_path = if let Some(thumbnail_path) = prepared_thumbnail.as_ref() {
+                thumbnail_path.clone()
+            } else {
+                uploads::create_video_thumbnail_file(&video_for_preview, &generated_thumbnail_path)
+                    .await?;
+                generated_thumbnail_path
+            };
+            if !local_file_exists(&thumbnail_path).await? {
+                return Err(format!(
+                    "video preview was not created: {}",
+                    thumbnail_path.display()
+                )
+                .into());
+            }
+            r2.upload_local_copy(target_thumbnail_db_path, &thumbnail_path)
+                .await?;
+            println!("Uploaded video preview: {target_thumbnail_db_path}");
+        }
+
+        Ok(())
+    }
+    .await;
+
+    let _ = tokio::fs::remove_dir_all(&workspace).await;
+    result
 }
 
 async fn run() -> Result<(), Box<dyn std::error::Error>> {
@@ -35,95 +176,42 @@ async fn run() -> Result<(), Box<dyn std::error::Error>> {
         let original_db_path = media.file_path.clone();
         let target_db_path = uploads::normalized_video_db_path(&original_db_path);
         let target_thumbnail_db_path = uploads::thumbnail_db_path(&target_db_path);
-        let target_in_r2 = r2.exists(&target_db_path).await?;
-        let thumbnail_in_r2 = r2.exists(&target_thumbnail_db_path).await?;
+        let mut target_in_r2 = r2.exists(&target_db_path).await?;
+        let mut thumbnail_in_r2 = r2.exists(&target_thumbnail_db_path).await?;
 
         if !target_in_r2 || !thumbnail_in_r2 {
-            let source_path = uploads::resolve_db_path(&original_db_path);
-            let target_path = uploads::resolve_db_path(&target_db_path);
-            let target_thumbnail_path = uploads::resolve_db_path(&target_thumbnail_db_path);
-            let source_exists = local_file_exists(&source_path).await?;
-            let target_exists = local_file_exists(&target_path).await?;
-            let target_thumbnail_exists = local_file_exists(&target_thumbnail_path).await?;
-            let mut temporary_source_dir = None;
-
-            let conversion_source = if target_exists {
-                target_path.clone()
-            } else if source_exists {
-                source_path.clone()
-            } else if r2.exists(&original_db_path).await? {
-                let directory = std::env::temp_dir()
-                    .join(format!("web-petting-normalize-{}", uuid::Uuid::new_v4()));
-                tokio::fs::create_dir(&directory).await?;
-                let extension = original_db_path.rsplit('.').next().unwrap_or("mov");
-                let temporary_source = directory.join(format!("source.{extension}"));
-                println!("Downloading missing PVC source from R2 for media {media_id}");
-                if let Err(error) = r2
-                    .download_to_path(&original_db_path, &temporary_source)
-                    .await
-                {
-                    eprintln!("Could not download media {media_id} from R2: {error}");
-                    let _ = tokio::fs::remove_dir_all(&directory).await;
-                    failed += 1;
-                    continue;
-                }
-                temporary_source_dir = Some(directory);
-                temporary_source
-            } else {
-                eprintln!(
-                    "Missing source for media {media_id}: {}",
-                    source_path.display()
-                );
+            if let Err(error) = prepare_missing_objects(
+                &r2,
+                media_id,
+                &original_db_path,
+                &target_db_path,
+                &target_thumbnail_db_path,
+                target_in_r2,
+                thumbnail_in_r2,
+            )
+            .await
+            {
+                eprintln!("Could not prepare media {media_id}: {error}");
                 if media.status == "active" {
                     failed += 1;
                 } else {
                     missing_archived += 1;
                 }
                 continue;
-            };
+            }
+            target_in_r2 = r2.exists(&target_db_path).await?;
+            thumbnail_in_r2 = r2.exists(&target_thumbnail_db_path).await?;
+        }
 
-            if let Some(parent) = target_path.parent() {
-                tokio::fs::create_dir_all(parent).await?;
-            }
-            let preparation = if original_db_path == target_db_path || target_exists {
-                if target_thumbnail_exists {
-                    Ok(())
-                } else {
-                    uploads::create_video_thumbnail_file(&conversion_source, &target_thumbnail_path)
-                        .await
-                }
-            } else {
-                println!("Converting media {media_id}: {original_db_path}");
-                uploads::create_compact_video_files(
-                    &conversion_source,
-                    &target_path,
-                    &target_thumbnail_path,
-                )
-                .await
-            };
-            if let Some(directory) = temporary_source_dir {
-                let _ = tokio::fs::remove_dir_all(directory).await;
-            }
-            if let Err(error) = preparation {
-                eprintln!("Could not prepare media {media_id}: {error}");
-                failed += 1;
-                continue;
-            }
-
-            if !target_in_r2 {
-                r2.upload_local_copy(&target_db_path, &target_path).await?;
-                println!("Uploaded compact video: {target_db_path}");
-            }
-            if !thumbnail_in_r2 {
-                r2.upload_local_copy(&target_thumbnail_db_path, &target_thumbnail_path)
-                    .await?;
-                println!("Uploaded video preview: {target_thumbnail_db_path}");
-            }
+        if !target_in_r2 || !thumbnail_in_r2 {
+            eprintln!("Could not verify normalized R2 objects for media {media_id}");
+            failed += 1;
+            continue;
         }
 
         if original_db_path != target_db_path {
-            // New objects are safely present before the old R2 keys are removed.
-            // The source on the PVC is deliberately left untouched.
+            // Both replacements are verified before old R2 keys are removed.
+            // Original files on the PVC are deliberately left untouched.
             r2.remove(&original_db_path).await?;
             let original_thumbnail = uploads::thumbnail_db_path(&original_db_path);
             if original_thumbnail != target_thumbnail_db_path {

src/uploads.rs

@@ -370,9 +370,30 @@ impl Storage {
                 "R2 storage is not configured",
             ));
         };
-        let body = ByteStream::from_path(local_path)
-            .await
-            .map_err(|error| StorageError::new("failed to open local file for migration", error))?;
+        let metadata = tokio::fs::metadata(local_path).await.map_err(|error| {
+            StorageError::new(
+                &format!(
+                    "failed to inspect local migration file {}",
+                    local_path.display()
+                ),
+                error,
+            )
+        })?;
+        if !metadata.is_file() || metadata.len() == 0 {
+            return Err(StorageError::new(
+                "failed to open local file for migration",
+                format!("{} is not a non-empty file", local_path.display()),
+            ));
+        }
+        let body = ByteStream::from_path(local_path).await.map_err(|error| {
+            StorageError::new(
+                &format!(
+                    "failed to open local file for migration {}",
+                    local_path.display()
+                ),
+                error,
+            )
+        })?;
         storage.put_stream(db_path, body).await
     }
 }
@@ -673,6 +694,11 @@ async fn video_duration(source: &Path) -> StorageResult<f64> {
     Ok(duration)
 }
 
+#[allow(dead_code)]
+pub async fn validate_video_file(source: &Path) -> StorageResult<()> {
+    video_duration(source).await.map(|_| ())
+}
+
 async fn create_video_sprite(source: &Path, destination: &Path) -> StorageResult<()> {
     let duration = video_duration(source).await?;
     let frame_times = [0.08, 0.34, 0.60, 0.86].map(|position| duration * position);
@@ -710,8 +736,16 @@ pub async fn create_compact_video_files(
     video_destination: &Path,
     thumbnail_destination: &Path,
 ) -> StorageResult<()> {
-    transcode_video_for_browser(source, video_destination).await?;
-    create_video_sprite(video_destination, thumbnail_destination).await
+    let result = async {
+        transcode_video_for_browser(source, video_destination).await?;
+        create_video_sprite(video_destination, thumbnail_destination).await
+    }
+    .await;
+    if result.is_err() {
+        let _ = tokio::fs::remove_file(video_destination).await;
+        let _ = tokio::fs::remove_file(thumbnail_destination).await;
+    }
+    result
 }
 
 /// Convert an incoming upload and store only the compact browser MP4 and its