chore: send notification for file upload state (#6738)

* chore: send notification for file upload state

* chore: fix test

Author: appflowy

frontend/appflowy_flutter/lib/startup/tasks/file_storage_task.dart

@@ -37,7 +37,7 @@ class FileStorageService {
         final fileProgress = FileProgress.fromJsonString(event);
         if (fileProgress != null) {
           Log.debug(
-            "Upload progress: file: ${fileProgress.fileUrl} ${fileProgress.progress}",
+            "FileStorageService upload file: ${fileProgress.fileUrl} ${fileProgress.progress}",
           );
           final notifier = _notifierList[fileProgress.fileUrl];
           if (notifier != null) {

frontend/rust-lib/flowy-ai/src/local_ai/local_llm_resource.rs

@@ -411,7 +411,7 @@ impl LocalAIResourceController {
     Ok(())
   }
 
-  #[instrument(level = "info", skip_all, err)]
+  #[instrument(level = "info", skip_all)]
   pub fn get_chat_config(&self, rag_enabled: bool) -> FlowyResult<AIPluginConfig> {
     if !self.is_resource_ready() {
       return Err(FlowyError::local_ai().with_context("Local AI resources are not ready"));

frontend/rust-lib/flowy-storage-pub/src/storage.rs

@@ -7,7 +7,6 @@ use serde::Serialize;
 use std::fmt::Display;
 use std::ops::{Deref, DerefMut};
 use tokio::sync::broadcast;
-use tracing::warn;
 
 #[async_trait]
 pub trait StorageService: Send + Sync {
@@ -68,22 +67,25 @@ pub enum FileUploadState {
 #[derive(Clone, Debug, Serialize)]
 pub struct FileProgress {
   pub file_url: String,
+  pub file_id: String,
   pub progress: f64,
   pub error: Option<String>,
 }
 
 impl FileProgress {
-  pub fn new_progress(file_url: String, progress: f64) -> Self {
+  pub fn new_progress(file_url: String, file_id: String, progress: f64) -> Self {
     FileProgress {
       file_url,
-      progress,
+      file_id,
+      progress: (progress * 10.0).round() / 10.0,
       error: None,
     }
   }
 
-  pub fn new_error(file_url: String, error: String) -> Self {
+  pub fn new_error(file_url: String, file_id: String, error: String) -> Self {
     FileProgress {
       file_url,
+      file_id,
       progress: 0.0,
       error: Some(error),
     }
@@ -92,7 +94,7 @@ impl FileProgress {
 
 impl Display for FileProgress {
   fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
-    write!(f, "FileProgress: {} - {}", self.file_url, self.progress)
+    write!(f, "FileProgress: {} - {}", self.file_id, self.progress)
   }
 }
 
@@ -122,9 +124,7 @@ impl ProgressNotifier {
 
   pub async fn notify(&mut self, progress: FileUploadState) {
     self.current_value = Some(progress.clone());
-    if let Err(err) = self.tx.send(progress) {
-      warn!("Failed to send progress notification: {:?}", err);
-    }
+    let _ = self.tx.send(progress);
   }
 }
 

frontend/rust-lib/flowy-storage/src/manager.rs

@@ -101,6 +101,31 @@ impl StorageManager {
       }
     });
 
+    let mut rx = global_notifier.subscribe();
+    let weak_notifier = Arc::downgrade(&progress_notifiers);
+    tokio::spawn(async move {
+      while let Ok(progress) = rx.recv().await {
+        if let Some(notifiers) = weak_notifier.upgrade() {
+          if let Some(mut notifier) = notifiers.get_mut(&progress.file_id) {
+            if progress.progress >= 1.0 {
+              let finish = FileUploadState::Finished {
+                file_id: progress.file_id,
+              };
+              notifier.notify(finish).await;
+            } else {
+              let progress = FileUploadState::Uploading {
+                progress: progress.progress,
+              };
+              notifier.notify(progress).await;
+            }
+          }
+        } else {
+          info!("progress notifiers is dropped");
+          break;
+        }
+      }
+    });
+
     Self {
       storage_service,
       cloud_service,
@@ -139,6 +164,7 @@ impl StorageManager {
 
     if let Err(err) = self.global_notifier.send(FileProgress::new_progress(
       url.to_string(),
+      file_id.clone(),
       if is_finish { 1.0 } else { 0.0 },
     )) {
       error!("[File] send global notifier failed: {}", err);
@@ -194,7 +220,7 @@ async fn prepare_upload_task(
 ) -> FlowyResult<()> {
   let uid = user_service.user_id()?;
   let conn = user_service.sqlite_connection(uid)?;
-  let upload_files = batch_select_upload_file(conn, 100)?;
+  let upload_files = batch_select_upload_file(conn, 100, false)?;
   let tasks = upload_files
     .into_iter()
     .map(|upload_file| UploadTask::BackgroundTask {
@@ -317,7 +343,6 @@ impl StorageService for StorageServiceImpl {
 
     // 1. create a file record and chunk the file
     let (chunks, record) = create_upload_record(workspace_id, parent_dir, local_file_path).await?;
-
     // 2. save the record to sqlite
     let conn = self
       .user_service
@@ -379,7 +404,6 @@ impl StorageService for StorageServiceImpl {
       &self.temp_storage,
       chunks,
       file_record,
-      self.progress_notifiers.clone(),
       self.global_notifier.clone(),
     )
     .await?;
@@ -404,7 +428,6 @@ impl StorageService for StorageServiceImpl {
         &self.user_service,
         &self.temp_storage,
         upload_file,
-        self.progress_notifiers.clone(),
         self.global_notifier.clone(),
       )
       .await?;
@@ -460,6 +483,7 @@ async fn create_upload_record(
   let record = UploadFileTable {
     workspace_id,
     file_id,
+    // When the upload_id is empty string, we will create a new upload using [Self::start_upload] method
     upload_id: "".to_string(),
     parent_dir,
     local_file_path,
@@ -479,7 +503,6 @@ async fn start_upload(
   temp_storage: &Arc<FileTempStorage>,
   mut chunked_bytes: ChunkedBytes,
   upload_file: &UploadFileTable,
-  progress_notifiers: Arc<DashMap<String, ProgressNotifier>>,
   global_notifier: GlobalNotifier,
 ) -> FlowyResult<()> {
   // 4. gather existing completed parts
@@ -503,50 +526,47 @@ async fn start_upload(
   );
 
   let mut upload_file = upload_file.clone();
-  if upload_file.upload_id.is_empty() {
-    // 1. create upload
-    trace!(
-      "[File] create upload for workspace: {}, parent_dir: {}, file_id: {}",
-      upload_file.workspace_id,
-      upload_file.parent_dir,
-      upload_file.file_id
-    );
-
-    let create_upload_resp_result = cloud_service
-      .create_upload(
-        &upload_file.workspace_id,
-        &upload_file.parent_dir,
-        &upload_file.file_id,
-        &upload_file.content_type,
-      )
-      .await;
-    if let Err(err) = create_upload_resp_result.as_ref() {
-      if err.is_file_limit_exceeded() {
-        make_notification(StorageNotification::FileStorageLimitExceeded)
-          .payload(err.clone())
-          .send();
-      }
-    }
-    let create_upload_resp = create_upload_resp_result?;
+  // 1. create upload
+  trace!(
+    "[File] create upload for workspace: {}, parent_dir: {}, file_id: {}",
+    upload_file.workspace_id,
+    upload_file.parent_dir,
+    upload_file.file_id
+  );
 
-    // 2. update upload_id
-    let conn = user_service.sqlite_connection(user_service.user_id()?)?;
-    update_upload_file_upload_id(
-      conn,
+  let create_upload_resp_result = cloud_service
+    .create_upload(
       &upload_file.workspace_id,
       &upload_file.parent_dir,
       &upload_file.file_id,
-      &create_upload_resp.upload_id,
-    )?;
-
-    trace!(
-      "[File] {} update upload_id: {}",
-      upload_file.file_id,
-      create_upload_resp.upload_id
-    );
-    // temporary store the upload_id
-    upload_file.upload_id = create_upload_resp.upload_id;
+      &upload_file.content_type,
+    )
+    .await;
+  if let Err(err) = create_upload_resp_result.as_ref() {
+    if err.is_file_limit_exceeded() {
+      make_notification(StorageNotification::FileStorageLimitExceeded)
+        .payload(err.clone())
+        .send();
+    }
   }
+  let create_upload_resp = create_upload_resp_result?;
+
+  // 2. update upload_id
+  let conn = user_service.sqlite_connection(user_service.user_id()?)?;
+  update_upload_file_upload_id(
+    conn,
+    &upload_file.workspace_id,
+    &upload_file.parent_dir,
+    &upload_file.file_id,
+    &create_upload_resp.upload_id,
+  )?;
+
+  trace!(
+    "[File] {} update upload_id: {}",
+    upload_file.file_id,
+    create_upload_resp.upload_id
+  );
+  upload_file.upload_id = create_upload_resp.upload_id;
 
   // 3. start uploading parts
   info!(
@@ -586,22 +606,17 @@ async fn start_upload(
     .await
     {
       Ok(resp) => {
-        let progress = (part_number as f64 / total_parts as f64).clamp(0.0, 1.0);
-        trace!(
-          "[File] {} upload progress:{}, etag: {}",
-          upload_file.file_id,
-          progress,
-          resp.e_tag
-        );
-
-        if let Err(err) = global_notifier.send(FileProgress::new_progress(file_url, progress)) {
-          error!("[File] send global notifier failed: {}", err);
+        let mut progress_value = (part_number as f64 / total_parts as f64).clamp(0.0, 1.0);
+        // The 0.1 is reserved for the complete_upload progress
+        if progress_value >= 0.9 {
+          progress_value = 0.9;
         }
+        let progress =
+          FileProgress::new_progress(file_url, upload_file.file_id.clone(), progress_value);
+        trace!("[File] upload progress: {}", progress);
 
-        if let Some(mut notifier) = progress_notifiers.get_mut(&upload_file.file_id) {
-          notifier
-            .notify(FileUploadState::Uploading { progress })
-            .await;
+        if let Err(err) = global_notifier.send(progress) {
+          error!("[File] send global notifier failed: {}", err);
         }
 
         // gather completed part
@@ -617,7 +632,11 @@ async fn start_upload(
             .send();
         }
 
-        if let Err(err) = global_notifier.send(FileProgress::new_error(file_url, err.msg.clone())) {
+        if let Err(err) = global_notifier.send(FileProgress::new_error(
+          file_url,
+          upload_file.file_id.clone(),
+          err.msg.clone(),
+        )) {
           error!("[File] send global notifier failed: {}", err);
         }
         return Err(err);
@@ -632,7 +651,6 @@ async fn start_upload(
     temp_storage,
     &upload_file,
     completed_parts,
-    &progress_notifiers,
     &global_notifier,
   )
   .await;
@@ -655,7 +673,6 @@ async fn resume_upload(
   user_service: &Arc<dyn StorageUserService>,
   temp_storage: &Arc<FileTempStorage>,
   upload_file: UploadFileTable,
-  progress_notifiers: Arc<DashMap<String, ProgressNotifier>>,
   global_notifier: GlobalNotifier,
 ) -> FlowyResult<()> {
   trace!(
@@ -675,7 +692,6 @@ async fn resume_upload(
         temp_storage,
         chunked_bytes,
         &upload_file,
-        progress_notifiers,
         global_notifier,
       )
       .await?;
@@ -740,14 +756,8 @@ async fn complete_upload(
   temp_storage: &Arc<FileTempStorage>,
   upload_file: &UploadFileTable,
   parts: Vec<CompletedPartRequest>,
-  progress_notifiers: &Arc<DashMap<String, ProgressNotifier>>,
   global_notifier: &GlobalNotifier,
 ) -> Result<(), FlowyError> {
-  trace!(
-    "[File]: completing file upload: {}, num parts: {}",
-    upload_file.file_id,
-    parts.len()
-  );
   let file_url = cloud_service
     .get_object_url_v1(
       &upload_file.workspace_id,
@@ -756,6 +766,12 @@ async fn complete_upload(
     )
     .await?;
 
+  info!(
+    "[File]: completing file upload: {}, num parts: {}, url:{}",
+    upload_file.file_id,
+    parts.len(),
+    file_url
+  );
   match cloud_service
     .complete_upload(
       &upload_file.workspace_id,
@@ -768,15 +784,7 @@ async fn complete_upload(
   {
     Ok(_) => {
       info!("[File] completed upload file: {}", upload_file.file_id);
-      if let Some(mut notifier) = progress_notifiers.get_mut(&upload_file.file_id) {
-        notifier
-          .notify(FileUploadState::Finished {
-            file_id: upload_file.file_id.clone(),
-          })
-          .await;
-      }
-
-      let progress = FileProgress::new_progress(file_url, 1.0);
+      let progress = FileProgress::new_progress(file_url, upload_file.file_id.clone(), 1.0);
       info!(
         "[File]: notify upload progress:{}, {}",
         upload_file.file_id, progress
@@ -796,7 +804,10 @@ async fn complete_upload(
       }
     },
     Err(err) => {
-      let progress = FileProgress::new_error(file_url, err.msg.clone());
+      error!("[File] complete upload failed: {}", err);
+
+      let progress =
+        FileProgress::new_error(file_url, upload_file.file_id.clone(), err.msg.clone());
       if let Err(send_err) = global_notifier.send(progress) {
         error!("[File] send global notifier failed: {}", send_err);
       }

frontend/rust-lib/flowy-storage/src/sqlite_sql.rs

@@ -180,11 +180,14 @@ pub fn select_upload_parts(
 pub fn batch_select_upload_file(
   mut conn: DBConnection,
   limit: i32,
+  is_finish: bool,
 ) -> FlowyResult<Vec<UploadFileTable>> {
   let results = upload_file_table::dsl::upload_file_table
+    .filter(upload_file_table::is_finish.eq(is_finish))
     .order(upload_file_table::created_at.desc())
     .limit(limit.into())
-    .load::<UploadFileTable>(&mut conn)?;
+    .load::<UploadFileTable>(&mut *conn)?;
+
   Ok(results)
 }