feat: S3 delete file and upload new (#2909)

backend/windmill-api/openapi.yaml

@@ -6813,6 +6813,85 @@ paths:
               schema:
                 $ref: "#/components/schemas/WindmillFilePreview"
 
+  /w/{workspace}/job_helpers/delete_s3_file:
+    delete:
+      summary: Permanently delete file from S3
+      operationId: deleteS3File
+      tags:
+        - helpers
+      parameters:
+        - $ref: "#/components/parameters/WorkspaceId"
+        - name: file_key
+          in: query
+          required: true
+          schema:
+            type: string
+      responses:
+        "200":
+          description: Confirmation
+          content:
+            application/json:
+              schema: {}
+
+  /w/{workspace}/job_helpers/multipart_upload_s3_file:
+    post:
+      summary: Upload file to S3 bucket using multipart upload
+      operationId: multipartFileUpload
+      tags:
+        - helpers
+      parameters:
+        - $ref: "#/components/parameters/WorkspaceId"
+      requestBody:
+        description: Query args for a multipart file upload to S3
+        required: true
+        content:
+          application/json:
+            schema:
+              type: object
+              properties:
+                file_key:
+                  type: string
+                part_content:
+                  type: array
+                  items:
+                    type: integer
+                upload_id:
+                  type: string
+                parts:
+                  type: array
+                  items:
+                    $ref: "#/components/schemas/UploadFilePart"
+                is_final:
+                  type: boolean
+                cancel_upload:
+                  type: boolean
+              required:
+                - file_key
+                - part_content_base64
+                - parts
+                - is_final
+                - cancel_upload
+      responses:
+        "200":
+          description: Chunk upload status
+          content:
+            application/json:
+              schema:
+                type: object
+                properties:
+                  upload_id:
+                    type: string
+                  parts:
+                    type: array
+                    items:
+                      $ref: "#/components/schemas/UploadFilePart"
+                  is_done:
+                    type: boolean
+                required:
+                  - upload_id
+                  - parts
+                  - is_done
+
 components:
   securitySchemes:
     bearerAuth:
@@ -8941,3 +9020,14 @@ components:
       required:
         - script_path
         - git_repo_resource_path
+
+    UploadFilePart:
+      type: object
+      properties:
+        part_number:
+          type: integer
+        tag:
+          type: string
+      required:
+        - part_number
+        - tag

backend/windmill-api/src/job_helpers.rs

@@ -4,16 +4,20 @@ use crate::{
     db::DB, resources::get_resource_value_interpolated_internal, users::Tokened,
     workspaces::LargeFileStorage,
 };
+use anyhow::Context;
 use aws_sdk_s3::{
     config::{BehaviorVersion, Credentials, Region},
     presigning::PresigningConfig,
+    primitives::ByteStream,
+    types::{CompletedMultipartUpload, CompletedPart},
 };
 use axum::{
     extract::{Path, Query},
-    routing::{get, post},
+    routing::{delete, get, post},
     Extension, Json, Router,
 };
 use hyper::http;
+use itertools::Itertools;
 use object_store::ClientConfigKey;
 use polars::{
     io::{
@@ -72,6 +76,14 @@ pub fn workspaced_service() -> Router {
             "/load_file_preview",
             get(load_file_preview).layer(cors.clone()),
         )
+        .route(
+            "/delete_s3_file",
+            delete(delete_s3_file).layer(cors.clone()),
+        )
+        .route(
+            "/multipart_upload_s3_file",
+            post(multipart_upload_s3_file).layer(cors.clone()),
+        )
 }
 
 #[derive(Debug, Deserialize, Serialize, Clone)]
@@ -663,6 +675,193 @@ async fn load_file_preview(
     return Ok(Json(response));
 }
 
+#[derive(Deserialize)]
+struct DeleteS3FileQuery {
+    pub file_key: String,
+}
+
+async fn delete_s3_file(
+    authed: ApiAuthed,
+    Extension(user_db): Extension<UserDB>,
+    Extension(db): Extension<DB>,
+    Tokened { token }: Tokened,
+    Path(w_id): Path<String>,
+    Query(query): Query<DeleteS3FileQuery>,
+) -> error::JsonResult<()> {
+    let file_key = query.file_key.clone();
+    let s3_resource_opt = get_workspace_s3_resource(&authed, &user_db, &db, &token, &w_id).await?;
+
+    let s3_resource = s3_resource_opt.ok_or(error::Error::InternalErr(
+        "No files storage resource defined at the workspace level".to_string(),
+    ))?;
+    let s3_client = build_s3_client(&s3_resource);
+
+    let s3_bucket = s3_resource.bucket.clone();
+    s3_client
+        .delete_object()
+        .bucket(&s3_bucket)
+        .key(&file_key)
+        .send()
+        .await
+        .map_err(|err| {
+            tracing::error!("{:?}", err);
+            error::Error::InternalErr(err.to_string())
+        })?;
+    return Ok(Json(()));
+}
+
+#[derive(Deserialize)]
+struct UploadFileQuery {
+    pub file_key: String,
+    pub part_content: Vec<u8>,
+    pub upload_id: Option<String>, // should be None for the first call to initiate the upload
+
+    pub parts: Vec<UploadFilePart>, // parts already uploaded, with their part_number and the tag associated
+
+    pub is_final: bool,      // whether it's the final chunk
+    pub cancel_upload: bool, // whether the upload should be cancelled. upload_id should be set. subsequent calls with this upload_id will fail
+}
+
+#[derive(Deserialize, Serialize, Clone)]
+struct UploadFilePart {
+    pub part_number: u16,
+    pub tag: String,
+}
+
+#[derive(Serialize)]
+struct UploadFileResponse {
+    pub upload_id: String,
+    pub parts: Vec<UploadFilePart>, // parts already uploaded, with their part_number and the tag associated
+    pub is_done: bool, // whether the transfer is finished, either b/c it got cancelled or because the last chunk was uploaded
+}
+
+async fn multipart_upload_s3_file(
+    authed: ApiAuthed,
+    Extension(user_db): Extension<UserDB>,
+    Extension(db): Extension<DB>,
+    Tokened { token }: Tokened,
+    Path(w_id): Path<String>,
+    Json(query): Json<UploadFileQuery>,
+) -> error::JsonResult<UploadFileResponse> {
+    tracing::debug!(
+        "Multi part file upload: part number: {} - is_final: {}",
+        query.parts.len(),
+        query.is_final
+    );
+    let file_key = query.file_key.clone();
+    let s3_resource_opt = get_workspace_s3_resource(&authed, &user_db, &db, &token, &w_id).await?;
+
+    let s3_resource = s3_resource_opt.ok_or(error::Error::InternalErr(
+        "No files storage resource defined at the workspace level".to_string(),
+    ))?;
+    let bucket = s3_resource.bucket.clone();
+    let s3_client = build_s3_client(&s3_resource);
+
+    if query.cancel_upload && query.upload_id.clone().is_some() {
+        let upload_id = query.upload_id.unwrap();
+        s3_client
+            .abort_multipart_upload()
+            .bucket(&bucket)
+            .key(&file_key)
+            .upload_id(&upload_id)
+            .send()
+            .await
+            .map_err(|err| {
+                tracing::error!("{:?}", err);
+                error::Error::InternalErr(err.to_string())
+            })?;
+        return Ok(Json(UploadFileResponse {
+            upload_id: upload_id,
+            parts: vec![], // empty parts as the transfer has been cancelled
+            is_done: true,
+        }));
+    }
+
+    let (upload_id, part_number) = match query {
+        UploadFileQuery { upload_id: Some(upload_id), ref parts, .. } if parts.len() > 0 => {
+            (upload_id, parts.len() + 1)
+        }
+        UploadFileQuery { upload_id: None, ref parts, .. } if parts.len() == 0 => {
+            let multipart_upload_res = s3_client
+                .create_multipart_upload()
+                .bucket(&bucket)
+                .key(&file_key)
+                .send()
+                .await
+                .map_err(|err| {
+                    tracing::error!("{:?}", err);
+                    error::Error::InternalErr(err.to_string())
+                })?;
+            let upload_id = multipart_upload_res
+                .upload_id
+                .context("Upload ID is missing in the response")?;
+            (upload_id, 1)
+        }
+        _ => {
+            return Err(error::Error::BadRequest(
+                "parts should be empty when upload_id is not provided, as a new upload will be created."
+                    .to_string(),
+            ))
+        }
+    };
+
+    let chunk_content = query.part_content.clone();
+    let chunk_content_stream = ByteStream::from(chunk_content);
+    let multipart_upload_res = s3_client
+        .upload_part()
+        .bucket(&bucket)
+        .key(&file_key)
+        .upload_id(&upload_id)
+        .body(chunk_content_stream)
+        .part_number(part_number as i32)
+        .send()
+        .await
+        .map_err(|err| {
+            tracing::error!("{:?}", err);
+            error::Error::InternalErr(err.to_string())
+        })?;
+
+    let mut new_parts = query.parts.clone();
+    new_parts.push(UploadFilePart {
+        part_number: part_number as u16,
+        tag: multipart_upload_res.e_tag.unwrap_or_default(),
+    });
+
+    if query.is_final {
+        let completed_parts = new_parts
+            .iter()
+            .map(|part| {
+                CompletedPart::builder()
+                    .e_tag(&part.tag)
+                    .part_number(part.part_number as i32)
+                    .build()
+            })
+            .collect_vec();
+        let _complete_multipart_upload_res = s3_client
+            .complete_multipart_upload()
+            .bucket(&s3_resource.bucket)
+            .key(&query.file_key)
+            .upload_id(&upload_id)
+            .multipart_upload(
+                CompletedMultipartUpload::builder()
+                    .set_parts(Some(completed_parts))
+                    .build(),
+            )
+            .send()
+            .await
+            .map_err(|err| {
+                tracing::error!("{:?}", err);
+                error::Error::InternalErr(err.to_string())
+            })?;
+    }
+
+    return Ok(Json(UploadFileResponse {
+        upload_id: upload_id,
+        parts: new_parts,
+        is_done: query.is_final,
+    }));
+}
+
 async fn get_workspace_s3_resource<'c>(
     authed: &ApiAuthed,
     user_db: &UserDB,

frontend/src/lib/components/S3FilePicker.svelte

@@ -1,14 +1,35 @@
 <script lang="ts">
-	import { File, FolderClosed, FolderOpen, RotateCw, Loader2, Download } from 'lucide-svelte'
+	import {
+		File as FileIcon,
+		FolderClosed,
+		FolderOpen,
+		RotateCw,
+		Loader2,
+		Download,
+		Trash,
+		FileUp
+	} from 'lucide-svelte'
 	import { workspaceStore } from '$lib/stores'
-	import { HelpersService } from '$lib/gen'
-	import { displayDate, displaySize, emptyString } from '$lib/utils'
+	import { HelpersService, type UploadFilePart } from '$lib/gen'
+	import { displayDate, displaySize, emptyString, sendUserToast } from '$lib/utils'
 	import { Alert, Button, Drawer } from './common'
 	import DrawerContent from './common/drawer/DrawerContent.svelte'
 	import Section from './Section.svelte'
 	import { createEventDispatcher } from 'svelte'
 	import VirtualList from 'svelte-tiny-virtual-list'
 	import TableSimple from './TableSimple.svelte'
+	import ConfirmationModal from './common/confirmationModal/ConfirmationModal.svelte'
+	import FileUploadModal from './common/fileUpload/FileUploadModal.svelte'
+
+	let deletionModalOpen = false
+	let fileDeletionInProgress = false
+
+	let uploadModalOpen = false
+	let fileToUpload: File | undefined = undefined
+	let fileToUploadKey: string | undefined = undefined
+	let fileUploadProgress: number | undefined = undefined
+	let fileUploadCancelled: boolean = false
+	let fileUploadErrorMsg: string | undefined = undefined
 
 	let workspaceSettingsInitialized = true
 
@@ -177,6 +198,103 @@
 		filePreviewLoading = false
 	}
 
+	async function deleteFileFromS3(fileKey: string | undefined) {
+		fileDeletionInProgress = true
+		if (fileKey === undefined) {
+			return
+		}
+		try {
+			await HelpersService.deleteS3File({
+				workspace: $workspaceStore!,
+				fileKey: fileKey
+			})
+		} finally {
+			fileDeletionInProgress = false
+		}
+		sendUserToast(`${fileKey} deleted from S3 bucket`)
+		const idx = displayedFileKeys.indexOf(fileKey)
+		if (idx >= 0) {
+			displayedFileKeys.splice(idx, 1)
+			displayedFileKeys = [...displayedFileKeys]
+		}
+		delete allFilesByKey[fileKey]
+	}
+
+	async function uploadFileToS3() {
+		fileUploadErrorMsg = undefined
+		if (fileToUpload === undefined || fileToUploadKey === undefined) {
+			return
+		}
+		if (allFilesByKey[fileToUploadKey] !== undefined) {
+			fileUploadErrorMsg =
+				'A file with this name already exists in the S3 bucket. If you want to replace it, delete it first.'
+			return
+		}
+
+		let upload_id: string | undefined = undefined
+		let parts: UploadFilePart[] = []
+
+		let reader = fileToUpload?.stream().getReader()
+		let { value: chunk, done: readerDone } = await reader.read()
+		if (chunk === undefined || readerDone) {
+			sendUserToast('Error reading file, no data read', true)
+			return
+		}
+
+		fileUploadProgress = 0
+		while (true) {
+			let { value: chunk_2, done: readerDone } = await reader.read()
+			if (!readerDone && chunk_2 !== undefined && chunk.length <= 5 * 1024 * 1024) {
+				// AWS enforces part to be bigger than 5MB, so we accumulate bytes until we reach that limit before triggering the request to the BE
+				chunk = new Uint8Array([...chunk, ...chunk_2])
+				continue
+			}
+			fileUploadProgress += (chunk.length * 100) / fileToUpload.size
+			let response = await HelpersService.multipartFileUpload({
+				workspace: $workspaceStore!,
+				requestBody: {
+					file_key: fileToUploadKey,
+					part_content: Array.from(chunk),
+					upload_id: upload_id,
+					parts: parts,
+					is_final: readerDone,
+					cancel_upload: fileUploadCancelled
+				}
+			})
+			upload_id = response.upload_id
+			parts = response.parts
+			if (response.is_done) {
+				if (fileUploadCancelled) {
+					sendUserToast('File upload cancelled!')
+				} else {
+					sendUserToast('File upload finished!')
+				}
+				break
+			}
+			if (chunk_2 === undefined) {
+				sendUserToast(
+					'File upload is not finished, yet there is no more data to stream. This is unexpected',
+					true
+				)
+				return
+			}
+			chunk = chunk_2
+		}
+		uploadModalOpen = false
+
+		if (!fileUploadCancelled) {
+			selectedFileKey = { s3: fileToUploadKey }
+			await loadFiles()
+			await loadFileMetadataPlusPreviewAsync(selectedFileKey['s3'])
+		}
+
+		fileToUpload = undefined
+		fileToUploadKey = undefined
+		fileUploadProgress = undefined
+		fileUploadCancelled = false
+		fileUploadErrorMsg = undefined
+	}
+
 	export async function open(preSelectedFileKey: { s3: string } | undefined = undefined) {
 		if (preSelectedFileKey !== undefined) {
 			initialFileKey = { ...preSelectedFileKey }
@@ -185,6 +303,8 @@
 		displayedFileKeys = []
 		allFilesByKey = {}
 		paginationMarker = undefined
+		fileMetadata = undefined
+		filePreview = undefined
 		reloadContent()
 		drawer.openDrawer?.()
 	}
@@ -334,7 +454,7 @@
 												{file_info.display_name}
 											</div>
 										{:else}
-											<File size={16} />
+											<FileIcon size={16} />
 											<div class="truncate text-ellipsis w-56">
 												{file_info.display_name}
 											</div>
@@ -367,7 +487,7 @@
 					{:else}
 						<div class="p-4 gap-2">
 							<Section label={fileMetadata.fileKey}>
-								<div slot="action">
+								<div slot="action" class="flex gap-2">
 									{#if filePreview !== undefined}
 										<Button
 											title="Download file from S3"
@@ -377,6 +497,18 @@
 											startIcon={{ icon: Download }}
 											iconOnly={true}
 										/>
+										<Button
+											title="Delete file from S3"
+											variant="border"
+											color="red"
+											on:click={() => {
+												deletionModalOpen = true
+											}}
+											startIcon={fileDeletionInProgress
+												? { icon: Loader2, classes: 'animate-spin' }
+												: { icon: Trash }}
+											iconOnly={true}
+										/>
 									{/if}
 								</div>
 							</Section>
@@ -449,6 +581,15 @@
 
 		<div slot="actions" class="flex gap-1">
 			{#if !readOnlyMode}
+				<Button
+					variant="border"
+					color="light"
+					disabled={workspaceSettingsInitialized === false}
+					startIcon={{ icon: FileUp }}
+					on:click={() => {
+						uploadModalOpen = true
+					}}>Upload New</Button
+				>
 				<Button
 					disable={selectedFileKey === undefined || emptyString(selectedFileKey.s3)}
 					on:click={selectAndClose}>Select</Button
@@ -457,3 +598,43 @@
 		</div>
 	</DrawerContent>
 </Drawer>
+
+<ConfirmationModal
+	open={deletionModalOpen}
+	title="Permanently delete file"
+	confirmationText="Delete permanently"
+	on:canceled={() => {
+		deletionModalOpen = false
+	}}
+	on:confirmed={() => {
+		deleteFileFromS3(fileMetadata?.fileKey)
+		deletionModalOpen = false
+	}}
+>
+	<div class="flex flex-col w-full space-y-4">
+		<span
+			>Are you sure you want to permanently delete {fileMetadata?.fileKey} from the S3 bucket?</span
+		>
+	</div>
+</ConfirmationModal>
+
+<FileUploadModal
+	open={uploadModalOpen}
+	title="Upload file to S3 bucket"
+	bind:fileToUpload
+	bind:fileKey={fileToUploadKey}
+	on:canceled={() => {
+		if (fileUploadProgress !== undefined) {
+			fileUploadCancelled = true
+			fileUploadErrorMsg = 'Cancelling in progress, it might take a few seconds...'
+		} else {
+			fileUploadErrorMsg = undefined
+			uploadModalOpen = false
+		}
+	}}
+	on:confirmed={() => {
+		uploadFileToS3()
+	}}
+	bind:progressPct={fileUploadProgress}
+	bind:errorMsg={fileUploadErrorMsg}
+/>

frontend/src/lib/components/common/fileUpload/FileUploadModal.svelte

@@ -0,0 +1,113 @@
+<script lang="ts">
+	import { classNames } from '$lib/utils'
+	import { createEventDispatcher } from 'svelte'
+	import { fade } from 'svelte/transition'
+	import Button from '../button/Button.svelte'
+	import { FileUp, Loader2 } from 'lucide-svelte'
+
+	export let title: string
+	export let open: boolean = false
+	export let progressPct: number | undefined = undefined
+	export let errorMsg: string | undefined = undefined
+
+	export let fileKey: string | undefined = undefined
+	export let fileToUpload: File | undefined = undefined
+
+	const dispatch = createEventDispatcher()
+
+	function fadeFast(node: HTMLElement) {
+		return fade(node, { duration: 100 })
+	}
+</script>
+
+{#if open}
+	<div
+		transition:fadeFast|local
+		class={'absolute top-0 bottom-0 left-0 right-0 z-[5000]'}
+		role="dialog"
+	>
+		<div
+			class={classNames(
+				'fixed inset-0 bg-gray-500 bg-opacity-75 transition-opacity',
+				open ? 'ease-out duration-300 opacity-100' : 'ease-in duration-200 opacity-0'
+			)}
+		/>
+
+		<div class="fixed inset-0 z-10 overflow-y-auto">
+			<div class="flex min-h-full items-center justify-center p-4">
+				<div
+					class={classNames(
+						'relative transform overflow-hidden rounded-lg bg-surface px-4 pt-5 pb-4 text-left shadow-xl transition-all sm:my-8 sm:w-full sm:max-w-lg sm:p-6',
+						open
+							? 'ease-out duration-300 opacity-100 translate-y-0 sm:scale-100'
+							: 'ease-in duration-200 opacity-0 translate-y-4 sm:translate-y-0 sm:scale-95'
+					)}
+				>
+					<div class="flex flex-col gap-2">
+						<h3 class="text-lg font-medium text-primary">
+							{title}
+						</h3>
+						<div class="flex items-center gap-2">
+							<span>Key: </span>
+							<input
+								type="text"
+								placeholder="folder/nested/file.txt"
+								bind:value={fileKey}
+								class="text-2xl grow"
+							/>
+						</div>
+
+						<div class="w-full h-full">
+							<input
+								type="file"
+								title={fileToUpload ? `${fileToUpload.name}` : 'No file chosen'}
+								on:change={({ currentTarget }) => {
+									if (
+										currentTarget.files === undefined ||
+										currentTarget.files === null ||
+										currentTarget.files.length === 0
+									) {
+										fileToUpload = undefined
+									} else {
+										fileToUpload = currentTarget.files[0]
+										if (fileKey === undefined || fileKey === '') {
+											fileKey = fileToUpload.name
+										}
+									}
+								}}
+								accept="*"
+								multiple={false}
+							/>
+						</div>
+
+						<div class="flex w-full bg-gray-200 rounded-full h-4 overflow-hidden">
+							<div class="h-full bg-blue-400" style="width: {progressPct ?? 0}%" />
+						</div>
+
+						{#if errorMsg !== undefined}
+							<div class="text-red-500 dark:text-red-400 text-sm">
+								{errorMsg}
+							</div>
+						{/if}
+					</div>
+					<div class="flex items-center space-x-2 flex-row-reverse space-x-reverse mt-4">
+						<Button
+							disabled={progressPct !== undefined}
+							on:click={() => dispatch('confirmed')}
+							color="blue"
+							size="sm"
+							startIcon={progressPct !== undefined
+								? { icon: Loader2, classes: 'animate-spin' }
+								: { icon: FileUp }}
+						>
+							<span>Upload</span>
+						</Button>
+						<Button on:click={() => dispatch('canceled')} color="light" size="sm">
+							<span>Cancel</span>
+						</Button>
+					</div>
+				</div>
+			</div>
+		</div>
+	</div>
+{/if}