fix: end of file when unzip file

Cargo.toml

@@ -293,13 +293,13 @@ debug = true
 [patch.crates-io]
 # It's diffcult to resovle different version with the same crate used in AppFlowy Frontend and the Client-API crate.
 # So using patch to workaround this issue.
-collab = { git = "https://github.com/AppFlowy-IO/AppFlowy-Collab", rev = "907ca13edd948effde6a4cbd0f5c25e16519b3cb" }
-collab-entity = { git = "https://github.com/AppFlowy-IO/AppFlowy-Collab", rev = "907ca13edd948effde6a4cbd0f5c25e16519b3cb" }
-collab-folder = { git = "https://github.com/AppFlowy-IO/AppFlowy-Collab", rev = "907ca13edd948effde6a4cbd0f5c25e16519b3cb" }
-collab-document = { git = "https://github.com/AppFlowy-IO/AppFlowy-Collab", rev = "907ca13edd948effde6a4cbd0f5c25e16519b3cb" }
-collab-user = { git = "https://github.com/AppFlowy-IO/AppFlowy-Collab", rev = "907ca13edd948effde6a4cbd0f5c25e16519b3cb" }
-collab-database = { git = "https://github.com/AppFlowy-IO/AppFlowy-Collab", rev = "907ca13edd948effde6a4cbd0f5c25e16519b3cb" }
-collab-importer = { git = "https://github.com/AppFlowy-IO/AppFlowy-Collab", rev = "907ca13edd948effde6a4cbd0f5c25e16519b3cb" }
+collab = { git = "https://github.com/AppFlowy-IO/AppFlowy-Collab", rev = "2934851d337c7d322c5dd7a944c05cb7ffb930d4" }
+collab-entity = { git = "https://github.com/AppFlowy-IO/AppFlowy-Collab", rev = "2934851d337c7d322c5dd7a944c05cb7ffb930d4" }
+collab-folder = { git = "https://github.com/AppFlowy-IO/AppFlowy-Collab", rev = "2934851d337c7d322c5dd7a944c05cb7ffb930d4" }
+collab-document = { git = "https://github.com/AppFlowy-IO/AppFlowy-Collab", rev = "2934851d337c7d322c5dd7a944c05cb7ffb930d4" }
+collab-user = { git = "https://github.com/AppFlowy-IO/AppFlowy-Collab", rev = "2934851d337c7d322c5dd7a944c05cb7ffb930d4" }
+collab-database = { git = "https://github.com/AppFlowy-IO/AppFlowy-Collab", rev = "2934851d337c7d322c5dd7a944c05cb7ffb930d4" }
+collab-importer = { git = "https://github.com/AppFlowy-IO/AppFlowy-Collab", rev = "2934851d337c7d322c5dd7a944c05cb7ffb930d4" }
 
 [features]
 history = []

libs/workspace-template/src/database/database_collab.rs

@@ -12,8 +12,8 @@ pub async fn create_database_collab(
     collab_service,
     notifier: Default::default(),
   };
-  Database::create_with_view(params, context)
-    .await?
+  let database = Database::create_with_view(params, context).await?;
+  database
     .encode_database_collabs()
     .await
     .map_err(|e| anyhow::anyhow!("Failed to encode database collabs: {:?}", e))

libs/workspace-template/src/document/getting_started.rs

@@ -1,10 +1,9 @@
 use std::collections::HashMap;
-use std::sync::Arc;
 
 use anyhow::Error;
 use async_trait::async_trait;
 use collab::core::origin::CollabOrigin;
-use collab::lock::RwLock;
+
 use collab::preclude::Collab;
 use collab_database::database::{timestamp, DatabaseData};
 use collab_database::entity::CreateDatabaseParams;
@@ -66,12 +65,7 @@ impl GettingStartedTemplate {
     create_database_params: CreateDatabaseParams,
   ) -> anyhow::Result<Vec<TemplateData>> {
     let database_id = create_database_params.database_id.clone();
-    let encoded_database = tokio::task::spawn_blocking({
-      let create_database_params = create_database_params.clone();
-      move || create_database_collab(create_database_params)
-    })
-    .await?
-    .await?;
+    let encoded_database = create_database_collab(create_database_params).await?;
 
     let encoded_database_collab = encoded_database
       .encoded_database_collab
@@ -236,7 +230,7 @@ impl WorkspaceTemplate for GettingStartedTemplate {
   async fn create_workspace_view(
     &self,
     _uid: i64,
-    workspace_view_builder: Arc<RwLock<WorkspaceViewBuilder>>,
+    workspace_view_builder: &mut WorkspaceViewBuilder,
   ) -> anyhow::Result<Vec<TemplateData>> {
     let general_view_uuid = gen_view_id().to_string();
     let shared_view_uuid = gen_view_id().to_string();
@@ -263,12 +257,10 @@ impl WorkspaceTemplate for GettingStartedTemplate {
       )
       .await?;
 
-    let mut builder = workspace_view_builder.write().await;
-
     // Create general space with 2 built-in views: Getting started, To-dos
     //    The Getting started view is a document view, and the To-dos view is a board view
     //    The Getting started view contains 2 sub views: Desktop guide, Mobile guide
-    builder
+    workspace_view_builder
       .with_view_builder(|view_builder| async {
         let created_at = timestamp();
         let mut view_builder = view_builder
@@ -305,7 +297,7 @@ impl WorkspaceTemplate for GettingStartedTemplate {
       .await;
 
     // Create shared space without any built-in views
-    builder
+    workspace_view_builder
       .with_view_builder(|view_builder| async {
         let created_at = timestamp();
         let view_builder = view_builder
@@ -371,12 +363,11 @@ impl WorkspaceTemplate for DocumentTemplate {
   async fn create_workspace_view(
     &self,
     _uid: i64,
-    workspace_view_builder: Arc<RwLock<WorkspaceViewBuilder>>,
+    workspace_view_builder: &mut WorkspaceViewBuilder,
   ) -> anyhow::Result<Vec<TemplateData>> {
     let view_id = gen_view_id().to_string();
 
-    let mut builder = workspace_view_builder.write().await;
-    builder
+    workspace_view_builder
       .with_view_builder(|view_builder| async {
         view_builder
           .with_name("Getting started")

libs/workspace-template/src/lib.rs

@@ -5,7 +5,7 @@ pub use anyhow::Result;
 use async_trait::async_trait;
 use collab::core::origin::CollabOrigin;
 use collab::entity::EncodedCollab;
-use collab::lock::RwLock;
+
 use collab::preclude::Collab;
 use collab_entity::CollabType;
 use collab_folder::{
@@ -30,7 +30,7 @@ pub trait WorkspaceTemplate {
   async fn create_workspace_view(
     &self,
     uid: i64,
-    workspace_view_builder: Arc<RwLock<WorkspaceViewBuilder>>,
+    workspace_view_builder: &mut WorkspaceViewBuilder,
   ) -> Result<Vec<TemplateData>>;
 }
 
@@ -91,22 +91,18 @@ impl WorkspaceTemplateBuilder {
   }
 
   pub async fn build(&self) -> Result<Vec<TemplateData>> {
-    let workspace_view_builder = Arc::new(RwLock::from(WorkspaceViewBuilder::new(
-      self.workspace_id.clone(),
-      self.uid,
-    )));
-
+    let mut workspace_view_builder = WorkspaceViewBuilder::new(self.workspace_id.clone(), self.uid);
     let mut templates: Vec<TemplateData> = vec![];
     for handler in self.handlers.values() {
       if let Ok(template) = handler
-        .create_workspace_view(self.uid, workspace_view_builder.clone())
+        .create_workspace_view(self.uid, &mut workspace_view_builder)
         .await
       {
         templates.extend(template);
       }
     }
 
-    let views = workspace_view_builder.write().await.build();
+    let views = workspace_view_builder.build();
     // Safe to unwrap because we have at least one view.
     let first_view = views.first().unwrap().parent_view.clone();
     let first_level_views = views

libs/workspace-template/src/tests/getting_started_tests.rs

@@ -1,7 +1,5 @@
 use std::collections::HashMap;
-use std::sync::Arc;
 
-use collab::lock::RwLock;
 use collab::preclude::uuid_v4;
 use collab_database::database::DatabaseData;
 use collab_database::entity::CreateDatabaseParams;
@@ -95,18 +93,17 @@ mod tests {
   #[tokio::test]
   async fn create_workspace_view_with_getting_started_template_test() {
     let template = GettingStartedTemplate;
-    let workspace_view_builder = Arc::new(RwLock::new(WorkspaceViewBuilder::new(generate_id(), 1)));
+    let mut workspace_view_builder = WorkspaceViewBuilder::new(generate_id(), 1);
 
     let result = template
-      .create_workspace_view(1, workspace_view_builder.clone())
+      .create_workspace_view(1, &mut workspace_view_builder)
       .await
       .unwrap();
 
     // 2 spaces + 3 documents + 1 database + 5 database rows
     assert_eq!(result.len(), 11);
 
-    let mut builder = workspace_view_builder.write().await;
-    let views = builder.build();
+    let views = workspace_view_builder.build();
 
     // check the number of spaces
     assert_eq!(views.len(), 2);

services/appflowy-worker/Cargo.toml

@@ -46,3 +46,4 @@ mime_guess = "2.0"
 bytes.workspace = true
 uuid.workspace = true
 
+

services/appflowy-worker/src/application.rs

@@ -97,7 +97,7 @@ pub async fn create_app(listener: TcpListener, config: Config) -> Result<(), Err
     Arc::new(state.s3_client.clone()),
     Arc::new(email_notifier),
     "import_task_stream",
-    30,
+    10,
   ));
 
   let app = Router::new().with_state(state);

services/appflowy-worker/src/import_worker/unzip.rs

@@ -1,55 +1,86 @@
-use anyhow::Result;
+use anyhow::{Context, Result};
 use async_zip::base::read::stream::{Ready, ZipFileReader};
 use async_zip::{StringEncoding, ZipString};
 use futures::io::{AsyncBufRead, AsyncReadExt};
 use std::ffi::OsString;
 use std::os::unix::ffi::OsStringExt;
 use std::path::PathBuf;
-use tokio::fs::{self, File};
+use tokio::fs;
+use tokio::fs::File;
 use tokio::io::AsyncWriteExt;
 
+use tracing::error;
+
 pub struct UnzipFile {
   pub file_name: String,
   pub unzip_dir_path: PathBuf,
 }
 
 pub async fn unzip_async<R: AsyncBufRead + Unpin>(
   mut zip_reader: ZipFileReader<Ready<R>>,
-  out: PathBuf,
+  out_dir: PathBuf,
 ) -> Result<UnzipFile, anyhow::Error> {
   let mut real_file_name = None;
   while let Some(mut next_reader) = zip_reader.next_with_entry().await? {
     let entry_reader = next_reader.reader_mut();
-    let filename = get_filename(entry_reader.entry().filename())?;
+    let filename = get_filename(entry_reader.entry().filename())
+      .with_context(|| "Failed to extract filename from entry".to_string())?;
 
+    // Save the root folder name if we haven't done so yet
     if real_file_name.is_none() && filename.ends_with('/') {
       real_file_name = Some(filename.split('/').next().unwrap_or(&filename).to_string());
     }
 
-    let output_path = out.join(&filename);
+    let output_path = out_dir.join(&filename);
     if filename.ends_with('/') {
-      fs::create_dir_all(&output_path).await?;
+      fs::create_dir_all(&output_path)
+        .await
+        .with_context(|| format!("Failed to create directory: {}", output_path.display()))?;
     } else {
+      // Ensure parent directories exist
       if let Some(parent) = output_path.parent() {
         if !parent.exists() {
-          fs::create_dir_all(parent).await?;
+          fs::create_dir_all(parent)
+            .await
+            .with_context(|| format!("Failed to create parent directory: {}", parent.display()))?;
         }
       }
 
-      let mut outfile = File::create(&output_path).await?;
+      // Write file contents
+      let mut outfile = File::create(&output_path)
+        .await
+        .with_context(|| format!("Failed to create file: {}", output_path.display()))?;
       let mut buffer = vec![];
-      entry_reader.read_to_end(&mut buffer).await?;
-      outfile.write_all(&buffer).await?;
+      match entry_reader.read_to_end(&mut buffer).await {
+        Ok(_) => {
+          outfile
+            .write_all(&buffer)
+            .await
+            .with_context(|| format!("Failed to write data to file: {}", output_path.display()))?;
+        },
+        Err(err) => {
+          error!(
+            "Failed to read entry: {:?}. Error: {:?}",
+            entry_reader.entry(),
+            err,
+          );
+          return Err(anyhow::anyhow!(
+            "Unexpected EOF while reading: {}",
+            filename
+          ));
+        },
+      }
     }
 
+    // Move to the next file in the zip
     zip_reader = next_reader.done().await?;
   }
 
   match real_file_name {
-    None => Err(anyhow::anyhow!("No files found in zip archive")),
+    None => Err(anyhow::anyhow!("No files found in the zip archive")),
     Some(file_name) => Ok(UnzipFile {
       file_name: file_name.clone(),
-      unzip_dir_path: out.join(file_name),
+      unzip_dir_path: out_dir.join(file_name),
     }),
   }
 }