FAIR dataset deletion

operators/src/error.rs

@@ -500,6 +500,10 @@ pub enum Error {
     Bb8Postgres {
         source: bb8::RunError<tokio_postgres::Error>,
     },
+
+    DatasetDeleted {
+        id: String,
+    },
 }
 
 impl From<crate::adapters::SparseTilesFillAdapterError> for Error {

services/src/api/handlers/upload.rs

@@ -1,19 +1,16 @@
 use crate::api::model::responses::IdResponse;
 use crate::contexts::{ApplicationContext, SessionContext};
-use crate::datasets::upload::{FileId, FileUpload, Upload, UploadDb, UploadId, UploadRootPath};
+use crate::datasets::upload::{create_upload, Upload, UploadDb, UploadId, UploadRootPath};
+use crate::error::Error;
 use crate::error::Result;
-use crate::error::{self, Error};
 use crate::util::path_with_base_path;
 use actix_multipart::Multipart;
 use actix_web::{web, FromRequest, Responder};
-use futures::StreamExt;
 use gdal::vector::LayerAccess;
-use geoengine_datatypes::util::Identifier;
 use geoengine_operators::util::gdal::gdal_open_dataset;
 use serde::{Deserialize, Serialize};
-use snafu::ResultExt;
 use std::path::Path;
-use tokio::{fs, io::AsyncWriteExt};
+use tokio::fs;
 use utoipa::{ToResponse, ToSchema};
 
 pub(crate) fn init_upload_routes<C>(cfg: &mut web::ServiceConfig)
@@ -70,42 +67,9 @@ impl<'a> ToSchema<'a> for FileUploadRequest {
 async fn upload_handler<C: ApplicationContext>(
     session: C::Session,
     app_ctx: web::Data<C>,
-    mut body: Multipart,
+    body: Multipart,
 ) -> Result<web::Json<IdResponse<UploadId>>> {
-    let upload_id = UploadId::new();
-
-    let root = upload_id.root_path()?;
-
-    fs::create_dir_all(&root).await.context(error::Io)?;
-
-    let mut files: Vec<FileUpload> = vec![];
-    while let Some(item) = body.next().await {
-        let mut field = item?;
-        let file_name = field
-            .content_disposition()
-            .get_filename()
-            .ok_or(error::Error::UploadFieldMissingFileName)?
-            .to_owned();
-
-        let file_id = FileId::new();
-        let mut file = fs::File::create(root.join(&file_name))
-            .await
-            .context(error::Io)?;
-
-        let mut byte_size = 0_u64;
-        while let Some(chunk) = field.next().await {
-            let bytes = chunk?;
-            file.write_all(&bytes).await.context(error::Io)?;
-            byte_size += bytes.len() as u64;
-        }
-        file.flush().await.context(error::Io)?;
-
-        files.push(FileUpload {
-            id: file_id,
-            name: file_name,
-            byte_size,
-        });
-    }
+    let (upload_id, files) = create_upload(body).await?;
 
     app_ctx
         .session_context(session)

services/src/contexts/migrations/migration_0010_delete_uploaded_datasets.rs

@@ -0,0 +1,24 @@
+use async_trait::async_trait;
+use tokio_postgres::Transaction;
+
+use crate::error::Result;
+
+use super::database_migration::{DatabaseVersion, Migration};
+
+/// This migration adds new delete options for uploaded user datasets
+pub struct Migration0010DeleteUploadedDatasets;
+
+#[async_trait]
+impl Migration for Migration0010DeleteUploadedDatasets {
+    fn prev_version(&self) -> Option<DatabaseVersion> {
+        Some("0009_oidc_tokens".into())
+    }
+
+    fn version(&self) -> DatabaseVersion {
+        "0010_delete_uploaded_datasets".into()
+    }
+
+    async fn migrate(&self, _tx: &Transaction<'_>) -> Result<()> {
+        Ok(())
+    }
+}

services/src/contexts/migrations/mod.rs

@@ -9,6 +9,7 @@ pub use crate::contexts::migrations::{
     migration_0007_owner_role::Migration0007OwnerRole,
     migration_0008_band_names::Migration0008BandNames,
     migration_0009_oidc_tokens::Migration0009OidcTokens,
+    migration_0010_delete_uploaded_datasets::Migration0010DeleteUploadedDatasets,
 };
 pub use database_migration::{
     initialize_database, migrate_database, DatabaseVersion, Migration, MigrationResult,
@@ -26,6 +27,7 @@ mod migration_0006_ebv_provider;
 pub mod migration_0007_owner_role;
 pub mod migration_0008_band_names;
 pub mod migration_0009_oidc_tokens;
+pub mod migration_0010_delete_uploaded_datasets;
 
 #[cfg(test)]
 mod schema_info;
@@ -49,6 +51,7 @@ pub fn all_migrations() -> Vec<Box<dyn Migration>> {
         Box::new(Migration0007OwnerRole),
         Box::new(Migration0008BandNames),
         Box::new(Migration0009OidcTokens),
+        Box::new(Migration0010DeleteUploadedDatasets),
     ]
 }
 

services/src/contexts/mod.rs

@@ -29,7 +29,7 @@ pub use migrations::{
     Migration0002DatasetListingProvider, Migration0003GbifConfig,
     Migration0004DatasetListingProviderPrio, Migration0005GbifColumnSelection,
     Migration0006EbvProvider, Migration0007OwnerRole, Migration0008BandNames,
-    Migration0009OidcTokens, MigrationResult,
+    Migration0009OidcTokens, Migration0010DeleteUploadedDatasets, MigrationResult,
 };
 pub use postgres::{PostgresContext, PostgresDb, PostgresSessionContext};
 pub use session::{MockableSession, Session, SessionId, SimpleSession};

services/src/datasets/upload.rs

@@ -1,3 +1,4 @@
+use actix_multipart::Multipart;
 use std::fmt::{Display, Formatter};
 use std::path::{Path, PathBuf};
 
@@ -9,7 +10,12 @@ use crate::{
     util::config::{self, get_config_element},
 };
 use async_trait::async_trait;
+use futures_util::StreamExt;
+use geoengine_datatypes::util::Identifier;
 use serde::{Deserialize, Deserializer, Serialize};
+use snafu::ResultExt;
+use tokio::fs;
+use tokio::io::AsyncWriteExt;
 use utoipa::ToSchema;
 
 identifier!(UploadId);
@@ -119,6 +125,52 @@ pub struct UploadListing {
     pub num_files: usize,
 }
 
+pub async fn create_upload(mut body: Multipart) -> Result<(UploadId, Vec<FileUpload>)> {
+    let upload_id = UploadId::new();
+
+    let root = upload_id.root_path()?;
+
+    fs::create_dir_all(&root).await.context(error::Io)?;
+
+    let mut files: Vec<FileUpload> = vec![];
+    while let Some(item) = body.next().await {
+        let mut field = item?;
+        let file_name = field
+            .content_disposition()
+            .get_filename()
+            .ok_or(error::Error::UploadFieldMissingFileName)?
+            .to_owned();
+
+        let file_id = FileId::new();
+        let mut file = fs::File::create(root.join(&file_name))
+            .await
+            .context(error::Io)?;
+
+        let mut byte_size = 0_u64;
+        while let Some(chunk) = field.next().await {
+            let bytes = chunk?;
+            file.write_all(&bytes).await.context(error::Io)?;
+            byte_size += bytes.len() as u64;
+        }
+        file.flush().await.context(error::Io)?;
+
+        files.push(FileUpload {
+            id: file_id,
+            name: file_name,
+            byte_size,
+        });
+    }
+
+    Ok((upload_id, files))
+}
+
+pub async fn delete_upload(upload_id: UploadId) -> Result<()> {
+    let root = upload_id.root_path()?;
+    log::debug!("Deleting {upload_id}");
+    fs::remove_dir_all(&root).await.context(error::Io)?;
+    Ok(())
+}
+
 #[async_trait]
 pub trait UploadDb {
     async fn load_upload(&self, upload: UploadId) -> Result<Upload>;

services/src/error.rs

@@ -491,6 +491,14 @@ pub enum Error {
         resource_type: String,
         resource_id: String,
     },
+
+    ExpirationTimestampInPast,
+    IllegalExpirationUpdate {
+        reason: String,
+    },
+    IllegalDatasetStatus {
+        status: String,
+    },
 }
 
 impl actix_web::error::ResponseError for Error {

services/src/pro/api/apidoc.rs

@@ -54,6 +54,8 @@ use crate::layers::listing::{
 };
 use crate::pro;
 use crate::pro::api::handlers::users::{Quota, UpdateQuota};
+use crate::pro::datasets::Expiration;
+use crate::pro::datasets::ExpirationChange;
 use crate::pro::permissions::{
     Permission, PermissionListing, ResourceId, Role, RoleDescription, RoleId,
 };
@@ -157,7 +159,9 @@ use utoipa::{Modify, OpenApi};
         handlers::upload::upload_handler,
         pro::api::handlers::permissions::add_permission_handler,
         pro::api::handlers::permissions::remove_permission_handler,
-        pro::api::handlers::permissions::get_resource_permissions_handler
+        pro::api::handlers::permissions::get_resource_permissions_handler,
+        pro::api::handlers::datasets::set_dataset_expiration,
+        pro::api::handlers::datasets::gc_expired_datasets
     ),
     components(
         responses(
@@ -370,6 +374,8 @@ use utoipa::{Modify, OpenApi};
             Volume,
             VolumeName,
             DataPath,
+            Expiration,
+            ExpirationChange,
 
             PlotOutputFormat,
             WrappedPlotOutput,