chk
/
AppFlowy-Cloud
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

fix: Support notion database row document (#929) 

* chore: fix import notion database
This commit is contained in:
Nathan.fooo 2024-10-24 19:04:06 +08:00 committed by GitHub
parent 3247c8c9c6
commit e738508d79
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
4 changed files with 47 additions and 31 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

15
Cargo.lock generated
View File

@ -2239,7 +2239,7 @@ dependencies = [
[[package]] [[package]]
name = "collab" name = "collab"
version = "0.2.0" version = "0.2.0"
source = "git+https://github.com/AppFlowy-IO/AppFlowy-Collab?rev=e77fd84e0e32b4dc1dcfa271547517e5b0d8e987#e77fd84e0e32b4dc1dcfa271547517e5b0d8e987" source = "git+https://github.com/AppFlowy-IO/AppFlowy-Collab?rev=3493d8ca30850d1ee84f5ba1f6d7b673614b172e#3493d8ca30850d1ee84f5ba1f6d7b673614b172e"
dependencies = [ dependencies = [
"anyhow", "anyhow",
"arc-swap", "arc-swap",
@ -2264,7 +2264,7 @@ dependencies = [
[[package]] [[package]]
name = "collab-database" name = "collab-database"
version = "0.2.0" version = "0.2.0"
source = "git+https://github.com/AppFlowy-IO/AppFlowy-Collab?rev=e77fd84e0e32b4dc1dcfa271547517e5b0d8e987#e77fd84e0e32b4dc1dcfa271547517e5b0d8e987" source = "git+https://github.com/AppFlowy-IO/AppFlowy-Collab?rev=3493d8ca30850d1ee84f5ba1f6d7b673614b172e#3493d8ca30850d1ee84f5ba1f6d7b673614b172e"
dependencies = [ dependencies = [
"anyhow", "anyhow",
"async-trait", "async-trait",
@ -2303,7 +2303,7 @@ dependencies = [
[[package]] [[package]]
name = "collab-document" name = "collab-document"
version = "0.2.0" version = "0.2.0"
source = "git+https://github.com/AppFlowy-IO/AppFlowy-Collab?rev=e77fd84e0e32b4dc1dcfa271547517e5b0d8e987#e77fd84e0e32b4dc1dcfa271547517e5b0d8e987" source = "git+https://github.com/AppFlowy-IO/AppFlowy-Collab?rev=3493d8ca30850d1ee84f5ba1f6d7b673614b172e#3493d8ca30850d1ee84f5ba1f6d7b673614b172e"
dependencies = [ dependencies = [
"anyhow", "anyhow",
"arc-swap", "arc-swap",
@ -2324,7 +2324,7 @@ dependencies = [
[[package]] [[package]]
name = "collab-entity" name = "collab-entity"
version = "0.2.0" version = "0.2.0"
source = "git+https://github.com/AppFlowy-IO/AppFlowy-Collab?rev=e77fd84e0e32b4dc1dcfa271547517e5b0d8e987#e77fd84e0e32b4dc1dcfa271547517e5b0d8e987" source = "git+https://github.com/AppFlowy-IO/AppFlowy-Collab?rev=3493d8ca30850d1ee84f5ba1f6d7b673614b172e#3493d8ca30850d1ee84f5ba1f6d7b673614b172e"
dependencies = [ dependencies = [
"anyhow", "anyhow",
"bytes", "bytes",
@ -2344,7 +2344,7 @@ dependencies = [
[[package]] [[package]]
name = "collab-folder" name = "collab-folder"
version = "0.2.0" version = "0.2.0"
source = "git+https://github.com/AppFlowy-IO/AppFlowy-Collab?rev=e77fd84e0e32b4dc1dcfa271547517e5b0d8e987#e77fd84e0e32b4dc1dcfa271547517e5b0d8e987" source = "git+https://github.com/AppFlowy-IO/AppFlowy-Collab?rev=3493d8ca30850d1ee84f5ba1f6d7b673614b172e#3493d8ca30850d1ee84f5ba1f6d7b673614b172e"
dependencies = [ dependencies = [
"anyhow", "anyhow",
"arc-swap", "arc-swap",
@ -2366,7 +2366,7 @@ dependencies = [
[[package]] [[package]]
name = "collab-importer" name = "collab-importer"
version = "0.1.0" version = "0.1.0"
source = "git+https://github.com/AppFlowy-IO/AppFlowy-Collab?rev=e77fd84e0e32b4dc1dcfa271547517e5b0d8e987#e77fd84e0e32b4dc1dcfa271547517e5b0d8e987" source = "git+https://github.com/AppFlowy-IO/AppFlowy-Collab?rev=3493d8ca30850d1ee84f5ba1f6d7b673614b172e#3493d8ca30850d1ee84f5ba1f6d7b673614b172e"
dependencies = [ dependencies = [
"anyhow", "anyhow",
"async-recursion", "async-recursion",
@ -2379,6 +2379,7 @@ dependencies = [
"collab-document", "collab-document",
"collab-entity", "collab-entity",
"collab-folder", "collab-folder",
"csv",
"fancy-regex 0.13.0", "fancy-regex 0.13.0",
"futures", "futures",
"futures-lite", "futures-lite",
@ -2468,7 +2469,7 @@ dependencies = [
[[package]] [[package]]
name = "collab-user" name = "collab-user"
version = "0.2.0" version = "0.2.0"
source = "git+https://github.com/AppFlowy-IO/AppFlowy-Collab?rev=e77fd84e0e32b4dc1dcfa271547517e5b0d8e987#e77fd84e0e32b4dc1dcfa271547517e5b0d8e987" source = "git+https://github.com/AppFlowy-IO/AppFlowy-Collab?rev=3493d8ca30850d1ee84f5ba1f6d7b673614b172e#3493d8ca30850d1ee84f5ba1f6d7b673614b172e"
dependencies = [ dependencies = [
"anyhow", "anyhow",
"collab", "collab",

14
Cargo.toml
View File

@ -304,13 +304,13 @@ debug = true
[patch.crates-io] [patch.crates-io]
# It's diffcult to resovle different version with the same crate used in AppFlowy Frontend and the Client-API crate. # 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. # So using patch to workaround this issue.
collab = { git = "https://github.com/AppFlowy-IO/AppFlowy-Collab", rev = "e77fd84e0e32b4dc1dcfa271547517e5b0d8e987" } collab = { git = "https://github.com/AppFlowy-IO/AppFlowy-Collab", rev = "3493d8ca30850d1ee84f5ba1f6d7b673614b172e" }
collab-entity = { git = "https://github.com/AppFlowy-IO/AppFlowy-Collab", rev = "e77fd84e0e32b4dc1dcfa271547517e5b0d8e987" } collab-entity = { git = "https://github.com/AppFlowy-IO/AppFlowy-Collab", rev = "3493d8ca30850d1ee84f5ba1f6d7b673614b172e" }
collab-folder = { git = "https://github.com/AppFlowy-IO/AppFlowy-Collab", rev = "e77fd84e0e32b4dc1dcfa271547517e5b0d8e987" } collab-folder = { git = "https://github.com/AppFlowy-IO/AppFlowy-Collab", rev = "3493d8ca30850d1ee84f5ba1f6d7b673614b172e" }
collab-document = { git = "https://github.com/AppFlowy-IO/AppFlowy-Collab", rev = "e77fd84e0e32b4dc1dcfa271547517e5b0d8e987" } collab-document = { git = "https://github.com/AppFlowy-IO/AppFlowy-Collab", rev = "3493d8ca30850d1ee84f5ba1f6d7b673614b172e" }
collab-user = { git = "https://github.com/AppFlowy-IO/AppFlowy-Collab", rev = "e77fd84e0e32b4dc1dcfa271547517e5b0d8e987" } collab-user = { git = "https://github.com/AppFlowy-IO/AppFlowy-Collab", rev = "3493d8ca30850d1ee84f5ba1f6d7b673614b172e" }
collab-database = { git = "https://github.com/AppFlowy-IO/AppFlowy-Collab", rev = "e77fd84e0e32b4dc1dcfa271547517e5b0d8e987" } collab-database = { git = "https://github.com/AppFlowy-IO/AppFlowy-Collab", rev = "3493d8ca30850d1ee84f5ba1f6d7b673614b172e" }
collab-importer = { git = "https://github.com/AppFlowy-IO/AppFlowy-Collab", rev = "e77fd84e0e32b4dc1dcfa271547517e5b0d8e987" } collab-importer = { git = "https://github.com/AppFlowy-IO/AppFlowy-Collab", rev = "3493d8ca30850d1ee84f5ba1f6d7b673614b172e" }
[features] [features]
history = [] history = []

42
services/appflowy-worker/src/import_worker/worker.rs
View File

@ -12,7 +12,7 @@ use collab::core::origin::CollabOrigin;
use collab::entity::EncodedCollab; use collab::entity::EncodedCollab;
use collab_database::workspace_database::WorkspaceDatabase; use collab_database::workspace_database::WorkspaceDatabase;
use collab_entity::CollabType; use collab_entity::CollabType;
use collab_folder::Folder; use collab_folder::{Folder, View, ViewLayout};
use collab_importer::imported_collab::ImportType; use collab_importer::imported_collab::ImportType;
use collab_importer::notion::page::CollabResource; use collab_importer::notion::page::CollabResource;
use collab_importer::notion::NotionImporter; use collab_importer::notion::NotionImporter;
@ -44,7 +44,7 @@ use serde_json::from_str;
use sqlx::types::chrono; use sqlx::types::chrono;
use sqlx::types::chrono::{DateTime, Utc}; use sqlx::types::chrono::{DateTime, Utc};
use sqlx::{PgPool, Pool, Postgres}; use sqlx::{PgPool, Pool, Postgres};
use std::collections::HashMap; use std::collections::{HashMap, HashSet};
use std::env::temp_dir; use std::env::temp_dir;
use std::fmt::Display; use std::fmt::Display;
use std::fs::Permissions; use std::fs::Permissions;
@ -435,7 +435,7 @@ async fn download_and_unzip_file(
&import_task.md5_base64, &import_task.md5_base64,
) )
.await?; .await?;
info!( trace!(
"[Import] {} start unzip file: {:?}", "[Import] {} start unzip file: {:?}",
import_task.workspace_id, import_task.workspace_id,
file.path_buf() file.path_buf()
@ -579,21 +579,22 @@ async fn process_unzip_file(
let mut resources = vec![]; let mut resources = vec![];
let mut collab_params_list = vec![]; let mut collab_params_list = vec![];
let mut database_view_ids_by_database_id: HashMap<String, Vec<String>> = HashMap::new(); let mut database_view_ids_by_database_id: HashMap<String, Vec<String>> = HashMap::new();
let mut orphan_view_ids = HashSet::new();
let mem_cache = CollabMemCache::new(redis_client.clone()); let mem_cache = CollabMemCache::new(redis_client.clone());
let timestamp = chrono::Utc::now().timestamp(); let timestamp = chrono::Utc::now().timestamp();
// 3. Collect all collabs and resources // 3. Collect all collabs and resources
let mut stream = imported.into_collab_stream().await; let mut stream = imported.into_collab_stream().await;
while let Some(imported_collab) = stream.next().await { while let Some(imported_collab_info) = stream.next().await {
trace!( trace!(
"[Import]: {} imported collab: {}", "[Import]: {} imported collab: {}",
import_task.workspace_id, import_task.workspace_id,
imported_collab imported_collab_info
); );
resources.push(imported_collab.resource); resources.extend(imported_collab_info.resources);
collab_params_list.extend( collab_params_list.extend(
imported_collab imported_collab_info
.collabs .imported_collabs
.into_iter() .into_iter()
.map(|imported_collab| CollabParams { .map(|imported_collab| CollabParams {
object_id: imported_collab.object_id, object_id: imported_collab.object_id,
@ -604,12 +605,14 @@ async fn process_unzip_file(
.collect::<Vec<_>>(), .collect::<Vec<_>>(),
); );
match imported_collab.import_type { match imported_collab_info.import_type {
ImportType::Database { ImportType::Database {
database_id, database_id,
view_ids, view_ids,
row_document_ids,
} => { } => {
database_view_ids_by_database_id.insert(database_id, view_ids); database_view_ids_by_database_id.insert(database_id, view_ids);
orphan_view_ids.extend(row_document_ids);
}, },
ImportType::Document => { ImportType::Document => {
// do nothing // do nothing
@ -668,7 +671,18 @@ async fn process_unzip_file(
collab_params_list.push(w_database_collab_params); collab_params_list.push(w_database_collab_params);
} }
// 5. Encode Folder // 5. Insert orphan view to folder
let orphan_views = orphan_view_ids
.into_iter()
.map(|orphan_view_id| {
View::orphan_view(&orphan_view_id, ViewLayout::Document, Some(import_task.uid))
})
.collect::<Vec<_>>();
if !orphan_views.is_empty() {
folder.insert_views(orphan_views);
}
// 6. Encode Folder
let folder_collab = folder let folder_collab = folder
.encode_collab_v1(|collab| CollabType::Folder.validate_require_data(collab)) .encode_collab_v1(|collab| CollabType::Folder.validate_require_data(collab))
.map_err(|err| ImportError::Internal(err.into()))?; .map_err(|err| ImportError::Internal(err.into()))?;
@ -698,7 +712,7 @@ async fn process_unzip_file(
let upload_resources = process_resources(resources).await; let upload_resources = process_resources(resources).await;
// 6. Start a transaction to insert all collabs // 7. Start a transaction to insert all collabs
let mut transaction = pg_pool.begin().await.map_err(|err| { let mut transaction = pg_pool.begin().await.map_err(|err| {
ImportError::Internal(anyhow!( ImportError::Internal(anyhow!(
"Failed to start transaction when importing data: {:?}", "Failed to start transaction when importing data: {:?}",
@ -711,7 +725,7 @@ async fn process_unzip_file(
import_task.workspace_id import_task.workspace_id
); );
// 7. write all collab to disk // 8. write all collab to disk
insert_into_af_collab_bulk_for_user( insert_into_af_collab_bulk_for_user(
&mut transaction, &mut transaction,
&import_task.uid, &import_task.uid,
@ -810,13 +824,13 @@ async fn process_unzip_file(
return result; return result;
} }
// 7. after inserting all collabs, upload all files to S3 // 9. after inserting all collabs, upload all files to S3
trace!("[Import]: {} upload files to s3", import_task.workspace_id,); trace!("[Import]: {} upload files to s3", import_task.workspace_id,);
batch_upload_files_to_s3(&import_task.workspace_id, s3_client, upload_resources) batch_upload_files_to_s3(&import_task.workspace_id, s3_client, upload_resources)
.await .await
.map_err(|err| ImportError::Internal(anyhow!("Failed to upload files to S3: {:?}", err)))?; .map_err(|err| ImportError::Internal(anyhow!("Failed to upload files to S3: {:?}", err)))?;
// 8. delete zip file regardless of success or failure // 10. delete zip file regardless of success or failure
match fs::remove_dir_all(unzip_dir_path).await { match fs::remove_dir_all(unzip_dir_path).await {
Ok(_) => trace!( Ok(_) => trace!(
"[Import]: {} deleted unzip file: {:?}", "[Import]: {} deleted unzip file: {:?}",

7
services/appflowy-worker/src/s3_client.rs
View File

@ -17,7 +17,7 @@ use tokio::fs;
use tokio::fs::OpenOptions; use tokio::fs::OpenOptions;
use tokio::io::AsyncWriteExt; use tokio::io::AsyncWriteExt;
use tokio_util::compat::TokioAsyncReadCompatExt; use tokio_util::compat::TokioAsyncReadCompatExt;
use tracing::{error, info, trace}; use tracing::{error, trace};
use uuid::Uuid; use uuid::Uuid;
#[async_trait] #[async_trait]
@ -203,9 +203,10 @@ pub async fn download_file(
zip_file_path zip_file_path
); );
write_stream_to_file(&zip_file_path, expected_md5_base64, stream).await?; write_stream_to_file(&zip_file_path, expected_md5_base64, stream).await?;
info!( trace!(
"[Import] {} finish writing stream to file: {:?}", "[Import] {} finish writing stream to file: {:?}",
workspace_id, zip_file_path workspace_id,
zip_file_path
); );
Ok(AutoRemoveDownloadedFile { Ok(AutoRemoveDownloadedFile {
zip_file_path, zip_file_path,