Merge pull request #355 from AppFlowy-IO/delete-objects-after-delete-workspace

feat: delete all files when workspace is deleted

.github/workflows/integration_test.yml

@@ -40,14 +40,14 @@ jobs:
           sed -i 's/GOTRUE_EXTERNAL_GOOGLE_ENABLED=.*/GOTRUE_EXTERNAL_GOOGLE_ENABLED=true/' .env
           sed -i 's/GOTRUE_MAILER_AUTOCONFIRM=.*/GOTRUE_MAILER_AUTOCONFIRM=false/' .env
           sed -i 's/API_EXTERNAL_URL=http:\/\/your-host/API_EXTERNAL_URL=http:\/\/localhost/' .env
-          # expose port for sqlx tests
+          # expose port for tests (postgres, minio)
           sed -i '38s/$/\n    ports:\n      - 5432:5432/' docker-compose.yml
+          sed -i '26s/$/\n    ports:\n      - 9000:9000/' docker-compose.yml
 
       - name: Update Nginx Configuration
         run: |
           # the wasm-pack headless tests will run on random ports, so we need to allow all origins
           sed -i 's/http:\/\/127\.0\.0\.1:8000/http:\/\/127.0.0.1/g' nginx/nginx.conf
-        
 
       - name: Disable appflowyinc images
         run: |
@@ -71,7 +71,6 @@ jobs:
 
       - name: Run WASM tests
         working-directory: ./libs/wasm-test
-        run: | 
+        run: |
           cargo install wasm-pack
           wasm-pack test --headless --firefox
-          

src/api/workspace.rs

@@ -164,8 +164,10 @@ async fn delete_workspace_handler(
   workspace_id: web::Path<Uuid>,
   state: Data<AppState>,
 ) -> Result<Json<AppResponse<()>>> {
+  let bucket_storage = &state.bucket_storage;
+
   // TODO: add permission for workspace deletion
-  workspace::ops::delete_workspace_for_user(&state.pg_pool, &workspace_id).await?;
+  workspace::ops::delete_workspace_for_user(&state.pg_pool, &workspace_id, bucket_storage).await?;
   Ok(AppResponse::Ok().into())
 }
 

src/biz/workspace/ops.rs

@@ -1,7 +1,10 @@
 use anyhow::Context;
 use app_error::AppError;
 use database::collab::upsert_collab_member_with_txn;
+use database::file::bucket_s3_impl::BucketClientS3Impl;
+use database::file::BucketStorage;
 use database::pg_row::{AFWorkspaceMemberRow, AFWorkspaceRow};
+use database::resource_usage::get_all_workspace_blob_metadata;
 use database::user::select_uid_from_email;
 use database::workspace::{
   change_workspace_icon, delete_from_workspace, delete_workspace_members, insert_user_workspace,
@@ -14,14 +17,34 @@ use shared_entity::response::AppResponseError;
 use sqlx::{types::uuid, PgPool};
 use std::collections::HashMap;
 use std::ops::DerefMut;
+use std::sync::Arc;
 use tracing::instrument;
 use uuid::Uuid;
 
 pub async fn delete_workspace_for_user(
   pg_pool: &PgPool,
   workspace_id: &Uuid,
+  bucket_storage: &Arc<BucketStorage<BucketClientS3Impl>>,
 ) -> Result<(), AppResponseError> {
+  // remove files from s3
+
+  let blob_metadatas = get_all_workspace_blob_metadata(pg_pool, workspace_id)
+    .await
+    .context("Get all workspace blob metadata")?;
+
+  for blob_metadata in blob_metadatas {
+    bucket_storage
+      .delete_blob(workspace_id, blob_metadata.file_id.as_str())
+      .await
+      .context("Delete blob from s3")?;
+  }
+
+  // remove from postgres
   delete_from_workspace(pg_pool, workspace_id).await?;
+
+  // TODO: There can be a rare case where user uploads while workspace is being deleted.
+  // We need some routine job to clean up these orphaned files.
+
   Ok(())
 }
 

tests/workspace/blob/mod.rs

@@ -1,2 +1,82 @@
+use std::borrow::Cow;
+
 mod put_and_get;
 mod usage;
+use lazy_static::lazy_static;
+use tracing::warn;
+
+lazy_static! {
+  pub static ref LOCALHOST_MINIO_URL: Cow<'static, str> =
+    get_env_var("LOCALHOST_MINIO_URL", "http://localhost:9000");
+  pub static ref LOCALHOST_MINIO_ACCESS_KEY: Cow<'static, str> =
+    get_env_var("LOCALHOST_MINIO_ACCESS_KEY", "minioadmin");
+  pub static ref LOCALHOST_MINIO_SECRET_KEY: Cow<'static, str> =
+    get_env_var("LOCALHOST_MINIO_SECRET_KEY", "minioadmin");
+  pub static ref LOCALHOST_MINIO_BUCKET_NAME: Cow<'static, str> =
+    get_env_var("LOCALHOST_MINIO_BUCKET_NAME", "appflowy");
+}
+
+pub struct TestBucket(pub s3::Bucket);
+
+impl TestBucket {
+  pub async fn new() -> Self {
+    let region = s3::Region::Custom {
+      region: "".to_owned(),
+      endpoint: LOCALHOST_MINIO_URL.to_string(),
+    };
+
+    let cred = s3::creds::Credentials {
+      access_key: Some(LOCALHOST_MINIO_ACCESS_KEY.to_string()),
+      secret_key: Some(LOCALHOST_MINIO_SECRET_KEY.to_string()),
+      security_token: None,
+      session_token: None,
+      expiration: None,
+    };
+
+    match s3::Bucket::create_with_path_style(
+      &LOCALHOST_MINIO_BUCKET_NAME,
+      region.clone(),
+      cred.clone(),
+      s3::BucketConfiguration::default(),
+    )
+    .await
+    {
+      Ok(_) => {},
+      Err(e) => match e {
+        s3::error::S3Error::Http(409, _) => {},
+        _ => panic!("could not create bucket: {}", e),
+      },
+    }
+
+    Self(
+      s3::Bucket::new(&LOCALHOST_MINIO_BUCKET_NAME, region.clone(), cred.clone())
+        .unwrap()
+        .with_path_style(),
+    )
+  }
+
+  pub async fn get_object(&self, workspace_id: &str, file_id: &str) -> Option<bytes::Bytes> {
+    let object_key = format!("{}/{}", workspace_id, file_id);
+    match self.0.get_object(&object_key).await {
+      Ok(resp) => {
+        assert!(resp.status_code() == 200);
+        Some(resp.bytes().to_owned())
+      },
+      Err(err) => match err {
+        s3::error::S3Error::Http(404, _) => None,
+        _ => panic!("could not get object: {}", err),
+      },
+    }
+  }
+}
+
+fn get_env_var<'default>(key: &str, default: &'default str) -> Cow<'default, str> {
+  dotenvy::dotenv().ok();
+  match std::env::var(key) {
+    Ok(value) => Cow::Owned(value),
+    Err(_) => {
+      warn!("could not read env var {}: using default: {}", key, default);
+      Cow::Borrowed(default)
+    },
+  }
+}

tests/workspace/blob/put_and_get.rs

@@ -1,3 +1,4 @@
+use super::TestBucket;
 use app_error::ErrorCode;
 use client_api_test_util::{generate_unique_registered_user_client, workspace_id_from_client};
 
@@ -96,3 +97,40 @@ async fn put_delete_get() {
   let err = c1.get_blob(&url).await.unwrap_err();
   assert_eq!(err.code, ErrorCode::RecordNotFound);
 }
+
+#[tokio::test]
+async fn put_and_delete_workspace() {
+  let test_bucket = TestBucket::new().await;
+
+  let (c1, _user1) = generate_unique_registered_user_client().await;
+  let workspace_id = workspace_id_from_client(&c1).await;
+  let file_id = uuid::Uuid::new_v4().to_string();
+  let blob_to_put = "some contents 1";
+  {
+    // put blob
+    let mime = mime::TEXT_PLAIN_UTF_8;
+    let url = c1.get_blob_url(&workspace_id, &file_id);
+    c1.put_blob(&url, blob_to_put, &mime).await.unwrap();
+  }
+
+  {
+    // blob exists in the bucket
+    let raw_data = test_bucket
+      .get_object(&workspace_id, &file_id)
+      .await
+      .unwrap();
+    assert_eq!(blob_to_put, String::from_utf8_lossy(&raw_data));
+  }
+
+  // delete workspace
+  c1.delete_workspace(&workspace_id).await.unwrap();
+
+  {
+    // blob does not exist in the bucket
+    let is_none = test_bucket
+      .get_object(&workspace_id, &file_id)
+      .await
+      .is_none();
+    assert!(is_none);
+  }
+}