Add api -> video processor job request after video is uploaded

.gitignore

@@ -2,3 +2,5 @@
 uploads/
 node_modules/
 .DS_Store
+.env*
+!.env.example

api/Cargo.toml

@@ -16,3 +16,6 @@ tower-http = { version = "0.6", features = ["fs", "trace"] }
 tracing = "0.1"
 tracing-subscriber = { version = "0.3", features = ["env-filter"] }
 db = { path = "../db" }
+grpc = { path = "../grpc" }
+tonic = "0.12"
+prost = "0.13"

api/src/main.rs

@@ -24,6 +24,10 @@ pub struct AppState {
     config: Config,
 }
 
+pub enum AppError {
+    QueueConnectionError,
+}
+
 #[tokio::main]
 async fn main() {
     // Start the tracer

api/src/upload/mod.rs

@@ -2,11 +2,14 @@ use std::sync::Arc;
 
 use axum::extract::Multipart;
 use axum::extract::State;
+use grpc::video_processing::raw_video_processor_client::RawVideoProcessorClient;
+use grpc::video_processing::ProcessRawVideoRequest;
 use sqlx::types::chrono;
 use sqlx::Pool;
 use sqlx::Postgres;
 use tokio::fs::File;
 use tokio::io::AsyncWriteExt;
+use tonic::Request;
 
 use crate::templates::UploadTemplate;
 use crate::AppState;
@@ -31,10 +34,17 @@ pub async fn handle_upload_mp4(
     }
 
     if file_written {
+        // First, save the raw video
         let video = save_raw_upload(&state.db, file_id, file_path)
             .await
             .map_err(|err| return UploadTemplate::new().upload_error(&err))
-            .unwrap();
+            .expect("Could not save raw video");
+        // Second, send a job to the video processing queue to process this video
+        send_video_processing_request(video.id.clone(), video.raw_file_path)
+            .await
+            .expect("Could send processing request");
+        // Lastly, render the upload complete template
+        // TODO: Add processing status into template
         UploadTemplate::new().upload_id(&video.id)
     } else {
         UploadTemplate::new().upload_error("Some error uploading file")
@@ -82,3 +92,23 @@ async fn save_raw_upload(
 
     Ok(video)
 }
+
+/// Sends a job to the video processor service to queue a video processing job
+async fn send_video_processing_request(video_id: String, file_path: String) -> Result<(), String> {
+    let queue_url = std::env::var("QUEUE_URL").expect("QUEUE_URL must be set");
+    let mut client = RawVideoProcessorClient::connect(queue_url)
+        .await
+        .map_err(|e| format!("Could not connect to RawVideoProcessorClient {}", e))?;
+
+    let request = Request::new(ProcessRawVideoRequest {
+        id: video_id,
+        path: file_path,
+    });
+    let response = client
+        .process_raw_video(request)
+        .await
+        .map_err(|e| format!("Could not send video processing request {}", e));
+
+    tracing::debug!("Job processing response: {:?}", response);
+    Ok(())
+}