Upload Progress in UI

apps/desktop/src-tauri/src/lib.rs

@@ -71,7 +71,7 @@ use std::{
     str::FromStr,
     sync::Arc,
 };
-use tauri::{AppHandle, Manager, State, Window, WindowEvent};
+use tauri::{AppHandle, Manager, State, Window, WindowEvent, ipc::Channel};
 use tauri_plugin_deep_link::DeepLinkExt;
 use tauri_plugin_dialog::DialogExt;
 use tauri_plugin_global_shortcut::GlobalShortcutExt;
@@ -1034,7 +1034,7 @@ async fn list_audio_devices() -> Result<Vec<String>, ()> {
     Ok(MicrophoneFeed::list().keys().cloned().collect())
 }
 
-#[derive(Serialize, Type, tauri_specta::Event, Debug, Clone)]
+#[derive(Serialize, Type, Debug, Clone)]
 pub struct UploadProgress {
     progress: f64,
 }
@@ -1053,6 +1053,7 @@ async fn upload_exported_video(
     app: AppHandle,
     path: PathBuf,
     mode: UploadMode,
+    channel: Channel<UploadProgress>,
 ) -> Result<UploadResult, String> {
     let Ok(Some(auth)) = AuthStore::get(&app) else {
         AuthStore::set(&app, None).map_err(|e| e.to_string())?;
@@ -1074,7 +1075,7 @@ async fn upload_exported_video(
         return Ok(UploadResult::UpgradeRequired);
     }
 
-    UploadProgress { progress: 0.0 }.emit(&app).ok();
+    channel.send(UploadProgress { progress: 0.0 }).ok();
 
     let s3_config = async {
         let video_id = match mode {
@@ -1113,11 +1114,12 @@ async fn upload_exported_video(
         Some(s3_config),
         Some(meta.project_path.join("screenshots/display.jpg")),
         Some(metadata),
+        Some(channel.clone()),
     )
     .await
     {
         Ok(uploaded_video) => {
-            UploadProgress { progress: 1.0 }.emit(&app).ok();
+            channel.send(UploadProgress { progress: 1.0 }).ok();
 
             meta.sharing = Some(SharingMeta {
                 link: uploaded_video.link.clone(),
@@ -1938,7 +1940,6 @@ pub async fn run(recording_logging_handle: LoggingHandle) {
             NewNotification,
             AuthenticationInvalid,
             audio_meter::AudioInputLevelChange,
-            UploadProgress,
             captions::DownloadProgress,
             recording::RecordingEvent,
             RecordingDeleted,
@@ -2310,13 +2311,12 @@ pub async fn run(recording_logging_handle: LoggingHandle) {
                         }
                     }
 
-                    if *focused {
-                        if let Ok(window_id) = window_id
-                            && window_id.activates_dock()
-                        {
-                            app.set_activation_policy(tauri::ActivationPolicy::Regular)
-                                .ok();
-                        }
+                    if *focused
+                        && let Ok(window_id) = window_id
+                        && window_id.activates_dock()
+                    {
+                        app.set_activation_policy(tauri::ActivationPolicy::Regular)
+                            .ok();
                     }
                 }
                 WindowEvent::DragDrop(tauri::DragDropEvent::Drop { paths, .. }) => {

apps/desktop/src-tauri/src/recording.rs

@@ -19,7 +19,7 @@ use std::{path::PathBuf, str::FromStr, sync::Arc, time::Duration};
 use tauri::{AppHandle, Manager};
 use tauri_plugin_dialog::{DialogExt, MessageDialogBuilder};
 use tauri_specta::Event;
-use tracing::{error, info};
+use tracing::{debug, error, info};
 
 use crate::{
     App, CurrentRecordingChanged, MutableState, NewStudioRecordingAdded, RecordingState,
@@ -355,7 +355,7 @@ pub async fn start_recording(
             )
         });
 
-    println!("spawning actor");
+    debug!("spawning start_recording actor");
 
     // done in spawn to catch panics just in case
     let spawn_actor_res = async {
@@ -812,6 +812,7 @@ async fn handle_recording_finish(
                                 Some(video_upload_info.config.clone()),
                                 Some(display_screenshot.clone()),
                                 meta,
+                                None,
                             )
                             .await
                             {

apps/desktop/src-tauri/src/upload.rs

@@ -11,22 +11,29 @@ use image::codecs::jpeg::JpegEncoder;
 use reqwest::StatusCode;
 use reqwest::header::CONTENT_LENGTH;
 use serde::{Deserialize, Serialize};
+use serde_json::json;
 use specta::Type;
-use std::path::PathBuf;
-use std::time::Duration;
-use tauri::AppHandle;
+use std::{
+    path::PathBuf,
+    time::{Duration, Instant},
+};
+use tauri::{AppHandle, ipc::Channel};
 use tauri_plugin_clipboard_manager::ClipboardExt;
-use tauri_specta::Event;
 use tokio::io::{AsyncReadExt, AsyncSeekExt};
-use tokio::task;
+use tokio::task::{self, JoinHandle};
 use tokio::time::sleep;
-use tracing::{error, info, warn};
+use tracing::{debug, error, info, trace, warn};
 
 #[derive(Deserialize, Serialize, Clone, Type, Debug)]
 pub struct S3UploadMeta {
     id: String,
 }
 
+#[derive(Deserialize, Clone, Debug)]
+pub struct CreateErrorResponse {
+    error: String,
+}
+
 // fn deserialize_empty_object_as_string<'de, D>(deserializer: D) -> Result<String, D::Error>
 // where
 //     D: Deserializer<'de>,
@@ -105,13 +112,107 @@ pub struct UploadedImage {
 //     pub config: S3UploadMeta,
 // }
 
+pub struct UploadProgressUpdater {
+    video_state: Option<VideoProgressState>,
+    app: AppHandle,
+    video_id: String,
+}
+
+struct VideoProgressState {
+    uploaded: u64,
+    total: u64,
+    pending_task: Option<JoinHandle<()>>,
+    last_update_time: Instant,
+}
+
+impl UploadProgressUpdater {
+    pub fn new(app: AppHandle, video_id: String) -> Self {
+        Self {
+            video_state: None,
+            app,
+            video_id,
+        }
+    }
+
+    pub fn update(&mut self, uploaded: u64, total: u64) {
+        let should_send_immediately = {
+            let state = self.video_state.get_or_insert_with(|| VideoProgressState {
+                uploaded,
+                total,
+                pending_task: None,
+                last_update_time: Instant::now(),
+            });
+
+            // Cancel any pending task
+            if let Some(handle) = state.pending_task.take() {
+                handle.abort();
+            }
+
+            state.uploaded = uploaded;
+            state.total = total;
+            state.last_update_time = Instant::now();
+
+            // Send immediately if upload is complete
+            uploaded >= total
+        };
+
+        let app = self.app.clone();
+        if should_send_immediately {
+            tokio::spawn({
+                let video_id = self.video_id.clone();
+                async move {
+                    Self::send_api_update(&app, video_id, uploaded, total).await;
+                }
+            });
+
+            // Clear state since upload is complete
+            self.video_state = None;
+        } else {
+            // Schedule delayed update
+            let handle = {
+                let video_id = self.video_id.clone();
+                tokio::spawn(async move {
+                    tokio::time::sleep(Duration::from_secs(2)).await;
+                    Self::send_api_update(&app, video_id, uploaded, total).await;
+                })
+            };
+
+            if let Some(state) = &mut self.video_state {
+                state.pending_task = Some(handle);
+            }
+        }
+    }
+
+    async fn send_api_update(app: &AppHandle, video_id: String, uploaded: u64, total: u64) {
+        let response = app
+            .authed_api_request("/api/desktop/video/progress", |client, url| {
+                client.post(url).json(&json!({
+                    "videoId": video_id,
+                    "uploaded": uploaded,
+                    "total": total,
+                    "updatedAt": chrono::Utc::now().to_rfc3339()
+                }))
+            })
+            .await;
+
+        match response {
+            Ok(resp) if resp.status().is_success() => {
+                trace!("Progress update sent successfully");
+            }
+            Ok(resp) => error!("Failed to send progress update: {}", resp.status()),
+            Err(err) => error!("Failed to send progress update: {err}"),
+        }
+    }
+}
+
 pub async fn upload_video(
     app: &AppHandle,
     video_id: String,
     file_path: PathBuf,
     existing_config: Option<S3UploadMeta>,
     screenshot_path: Option<PathBuf>,
     meta: Option<S3VideoMeta>,
+    channel: Option<Channel<UploadProgress>>,
 ) -> Result<UploadedVideo, String> {
     println!("Uploading video {video_id}...");
 
@@ -145,20 +246,24 @@ pub async fn upload_video(
 
     let reader_stream = tokio_util::io::ReaderStream::new(file);
 
-    let mut bytes_uploaded = 0;
-    let progress_stream = reader_stream.inspect({
-        let app = app.clone();
-        move |chunk| {
-            if bytes_uploaded > 0 {
-                let _ = UploadProgress {
-                    progress: bytes_uploaded as f64 / total_size as f64,
-                }
-                .emit(&app);
-            }
+    let mut bytes_uploaded = 0u64;
+    let mut progress = UploadProgressUpdater::new(app.clone(), video_id);
 
-            if let Ok(chunk) = chunk {
-                bytes_uploaded += chunk.len();
+    let progress_stream = reader_stream.inspect(move |chunk| {
+        if let Ok(chunk) = chunk {
+            bytes_uploaded += chunk.len() as u64;
+        }
+
+        if bytes_uploaded > 0 {
+            if let Some(channel) = &channel {
+                channel
+                    .send(UploadProgress {
+                        progress: bytes_uploaded as f64 / total_size as f64,
+                    })
+                    .ok();
             }
+
+            progress.update(bytes_uploaded, total_size);
         }
     });
 
@@ -316,6 +421,21 @@ pub async fn create_or_get_video(
         return Err("Failed to authenticate request; please log in again".into());
     }
 
+    if response.status() != StatusCode::OK {
+        if let Ok(error) = response.json::<CreateErrorResponse>().await {
+            if error.error == "upgrade_required" {
+                return Err(
+                    "You must upgrade to Cap Pro to upload recordings over 5 minutes in length"
+                        .into(),
+                );
+            }
+
+            return Err(format!("server error: {}", error.error));
+        }
+
+        return Err("Unknown error uploading video".into());
+    }
+
     let response_text = response
         .text()
         .await
@@ -544,13 +664,12 @@ impl InstantMultipartUpload {
         let mut uploaded_parts = Vec::new();
         let mut part_number = 1;
         let mut last_uploaded_position: u64 = 0;
-
-        println!("Starting multipart upload for {video_id}...");
+        let mut progress = UploadProgressUpdater::new(app.clone(), pre_created_video.id.clone());
 
         // --------------------------------------------
         // initiate the multipart upload
         // --------------------------------------------
-        println!("Initiating multipart upload for {video_id}...");
+        debug!("Initiating multipart upload for {video_id}...");
         let initiate_response = match app
             .authed_api_request("/api/upload/multipart/initiate", |c, url| {
                 c.post(url)
@@ -654,6 +773,7 @@ impl InstantMultipartUpload {
                     &mut part_number,
                     &mut last_uploaded_position,
                     new_data_size.min(CHUNK_SIZE),
+                    &mut progress,
                 )
                 .await
                 {
@@ -680,6 +800,7 @@ impl InstantMultipartUpload {
                         &mut 1,
                         &mut 0,
                         uploaded_parts[0].size as u64,
+                        &mut progress,
                     )
                     .await
                     .map_err(|err| format!("Failed to re-upload first chunk: {err}"))?;
@@ -726,6 +847,7 @@ impl InstantMultipartUpload {
         part_number: &mut i32,
         last_uploaded_position: &mut u64,
         chunk_size: u64,
+        progress: &mut UploadProgressUpdater,
     ) -> Result<UploadedPart, String> {
         let file_size = match tokio::fs::metadata(file_path).await {
             Ok(metadata) => metadata.len(),
@@ -838,6 +960,8 @@ impl InstantMultipartUpload {
             }
         };
 
+        progress.update(expected_pos, file_size);
+
         if !presign_response.status().is_success() {
             let status = presign_response.status();
             let error_body = presign_response