Local upload tracking (#1092)

* cleanup `upload_video` arguments

* wip

* wip

* make `progress_upload` required for instant mode

* fix uploading of thumbnails

* absolute mess

* store recordings into project meta?

* cursed realtime upload progress

* break out API definitions + use tracing for logging in `upload.rs`

* working stream-based uploader

* bring back progress tracking to upload

* implement `from_file` abstraction

* abstract another endpoint into `api` module

* abstract more API + retry on S3 upload request

* finish overhauling upload code

* restructure project file again

* UI for uploading state + errors

* show recording and pending status

* merge in changes from #1077

* upload progress working in sync pogggg

* wip

* polish off todo's in `recordings.tsx`

* a bunch of random fixes

* fixes

* fix

* resumable system

* format

* feature flag the new uploader

* fixes to some remaining issues

* fix button visibility on `CapCard`

* wip

* Clippy improvements

* cleanup

* fix CI

* fix Typescript

* fixes

* make `CapCard` more conservative with upload status

* CodeRabbit fixes

* fix some `CapCard` states

* fix retry policy

* wip

* fixes

* emit first chunk multiple times

* potentially fix the missing video toolbar sometimes

* improve chunk emit logic

* fix retry policy for s3 upload

* stttttrrrriiiinnnnggg

Author: oscartbeaumont

Cargo.lock

@@ -104,6 +104,9 @@ wgpu.workspace = true
 bytemuck = "1.23.1"
 kameo = "0.17.2"
 tauri-plugin-sentry = "0.5.0"
+thiserror.workspace = true
+bytes = "1.10.1"
+async-stream = "0.3.6"
 
 [target.'cfg(target_os = "macos")'.dependencies]
 core-graphics = "0.24.0"

apps/desktop/src-tauri/Cargo.toml

@@ -0,0 +1,244 @@
+//! TODO: We should investigate generating this with OpenAPI.
+//! This will come part of the EffectTS rewrite work.
+
+use serde::{Deserialize, Serialize};
+use serde_json::json;
+use tauri::AppHandle;
+
+use crate::web_api::ManagerExt;
+
+pub async fn upload_multipart_initiate(app: &AppHandle, video_id: &str) -> Result<String, String> {
+    #[derive(Deserialize)]
+    #[serde(rename_all = "camelCase")]
+    pub struct Response {
+        upload_id: String,
+    }
+
+    let resp = app
+        .authed_api_request("/api/upload/multipart/initiate", |c, url| {
+            c.post(url)
+                .header("Content-Type", "application/json")
+                .json(&serde_json::json!({
+                    "videoId": video_id,
+                    "contentType": "video/mp4"
+                }))
+        })
+        .await
+        .map_err(|err| format!("api/upload_multipart_initiate/request: {err}"))?;
+
+    if !resp.status().is_success() {
+        let status = resp.status();
+        let error_body = resp
+            .text()
+            .await
+            .unwrap_or_else(|_| "<no response body>".to_string());
+        return Err(format!(
+            "api/upload_multipart_initiate/{status}: {error_body}"
+        ));
+    }
+
+    resp.json::<Response>()
+        .await
+        .map_err(|err| format!("api/upload_multipart_initiate/response: {err}"))
+        .map(|data| data.upload_id)
+}
+
+pub async fn upload_multipart_presign_part(
+    app: &AppHandle,
+    video_id: &str,
+    upload_id: &str,
+    part_number: u32,
+    md5_sum: &str,
+) -> Result<String, String> {
+    #[derive(Deserialize)]
+    #[serde(rename_all = "camelCase")]
+    pub struct Response {
+        presigned_url: String,
+    }
+
+    let resp = app
+        .authed_api_request("/api/upload/multipart/presign-part", |c, url| {
+            c.post(url)
+                .header("Content-Type", "application/json")
+                .json(&serde_json::json!({
+                    "videoId": video_id,
+                    "uploadId": upload_id,
+                    "partNumber": part_number,
+                    "md5Sum": md5_sum
+                }))
+        })
+        .await
+        .map_err(|err| format!("api/upload_multipart_presign_part/request: {err}"))?;
+
+    if !resp.status().is_success() {
+        let status = resp.status();
+        let error_body = resp
+            .text()
+            .await
+            .unwrap_or_else(|_| "<no response body>".to_string());
+        return Err(format!(
+            "api/upload_multipart_presign_part/{status}: {error_body}"
+        ));
+    }
+
+    resp.json::<Response>()
+        .await
+        .map_err(|err| format!("api/upload_multipart_presign_part/response: {err}"))
+        .map(|data| data.presigned_url)
+}
+
+#[derive(Serialize)]
+#[serde(rename_all = "camelCase")]
+pub struct UploadedPart {
+    pub part_number: u32,
+    pub etag: String,
+    pub size: usize,
+    #[serde(skip)]
+    pub total_size: u64,
+}
+
+#[derive(Serialize, Debug, Clone)]
+#[serde(rename_all = "camelCase")]
+pub struct S3VideoMeta {
+    #[serde(rename = "durationInSecs")]
+    pub duration_in_secs: f64,
+    pub width: u32,
+    pub height: u32,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub fps: Option<f32>,
+}
+
+pub async fn upload_multipart_complete(
+    app: &AppHandle,
+    video_id: &str,
+    upload_id: &str,
+    parts: &[UploadedPart],
+    meta: Option<S3VideoMeta>,
+) -> Result<Option<String>, String> {
+    #[derive(Serialize)]
+    #[serde(rename_all = "camelCase")]
+    pub struct MultipartCompleteRequest<'a> {
+        video_id: &'a str,
+        upload_id: &'a str,
+        parts: &'a [UploadedPart],
+        #[serde(flatten)]
+        meta: Option<S3VideoMeta>,
+    }
+
+    #[derive(Deserialize)]
+    pub struct Response {
+        location: Option<String>,
+    }
+
+    let resp = app
+        .authed_api_request("/api/upload/multipart/complete", |c, url| {
+            c.post(url)
+                .header("Content-Type", "application/json")
+                .json(&MultipartCompleteRequest {
+                    video_id,
+                    upload_id,
+                    parts,
+                    meta,
+                })
+        })
+        .await
+        .map_err(|err| format!("api/upload_multipart_complete/request: {err}"))?;
+
+    if !resp.status().is_success() {
+        let status = resp.status();
+        let error_body = resp
+            .text()
+            .await
+            .unwrap_or_else(|_| "<no response body>".to_string());
+        return Err(format!(
+            "api/upload_multipart_complete/{status}: {error_body}"
+        ));
+    }
+
+    resp.json::<Response>()
+        .await
+        .map_err(|err| format!("api/upload_multipart_complete/response: {err}"))
+        .map(|data| data.location)
+}
+
+#[derive(Serialize)]
+#[serde(rename_all = "lowercase")]
+pub enum PresignedS3PutRequestMethod {
+    #[allow(unused)]
+    Post,
+    Put,
+}
+
+#[derive(Serialize)]
+#[serde(rename_all = "camelCase")]
+pub struct PresignedS3PutRequest {
+    pub video_id: String,
+    pub subpath: String,
+    pub method: PresignedS3PutRequestMethod,
+    #[serde(flatten)]
+    pub meta: Option<S3VideoMeta>,
+}
+
+pub async fn upload_signed(app: &AppHandle, body: PresignedS3PutRequest) -> Result<String, String> {
+    #[derive(Deserialize)]
+    struct Data {
+        url: String,
+    }
+
+    #[derive(Deserialize)]
+    #[serde(rename_all = "camelCase")]
+    pub struct Response {
+        presigned_put_data: Data,
+    }
+
+    let resp = app
+        .authed_api_request("/api/upload/signed", |client, url| {
+            client.post(url).json(&body)
+        })
+        .await
+        .map_err(|err| format!("api/upload_signed/request: {err}"))?;
+
+    if !resp.status().is_success() {
+        let status = resp.status();
+        let error_body = resp
+            .text()
+            .await
+            .unwrap_or_else(|_| "<no response body>".to_string());
+        return Err(format!("api/upload_signed/{status}: {error_body}"));
+    }
+
+    resp.json::<Response>()
+        .await
+        .map_err(|err| format!("api/upload_signed/response: {err}"))
+        .map(|data| data.presigned_put_data.url)
+}
+
+pub async fn desktop_video_progress(
+    app: &AppHandle,
+    video_id: &str,
+    uploaded: u64,
+    total: u64,
+) -> Result<(), String> {
+    let resp = 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
+        .map_err(|err| format!("api/desktop_video_progress/request: {err}"))?;
+
+    if !resp.status().is_success() {
+        let status = resp.status();
+        let error_body = resp
+            .text()
+            .await
+            .unwrap_or_else(|_| "<no response body>".to_string());
+        return Err(format!("api/desktop_video_progress/{status}: {error_body}"));
+    }
+
+    Ok(())
+}

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

@@ -97,6 +97,8 @@ pub struct GeneralSettingsStore {
     pub enable_new_recording_flow: bool,
     #[serde(default)]
     pub post_deletion_behaviour: PostDeletionBehaviour,
+    #[serde(default = "default_enable_new_uploader", skip_serializing_if = "no")]
+    pub enable_new_uploader: bool,
 }
 
 fn default_enable_native_camera_preview() -> bool {
@@ -108,6 +110,10 @@ fn default_enable_new_recording_flow() -> bool {
     cfg!(debug_assertions)
 }
 
+fn default_enable_new_uploader() -> bool {
+    cfg!(debug_assertions)
+}
+
 fn no(_: &bool) -> bool {
     false
 }
@@ -155,6 +161,7 @@ impl Default for GeneralSettingsStore {
             auto_zoom_on_clicks: false,
             enable_new_recording_flow: default_enable_new_recording_flow(),
             post_deletion_behaviour: PostDeletionBehaviour::DoNothing,
+            enable_new_uploader: default_enable_new_uploader(),
         }
     }
 }