refac(api): overhaul upload pipeline and fix race conditions

backend/src/endpoints/file/file.controller.ts

@@ -224,6 +224,22 @@ export class FileController {
         });
     }
 
+    @Delete('uploads')
+    @UserOnly() //Push back authentication to the queue to accelerate the request
+    @OutputDto(CancelUploadResponseDto)
+    async cancelUpload(
+        @Body() dto: CancelFileUploadDto,
+        @AddUser() auth: AuthHeader,
+    ): Promise<CancelUploadResponseDto> {
+        logger.debug(`cancelUpload ${JSON.stringify(dto)}`);
+        await this.fileLifecycleService.cancelUpload(
+            dto.uuids,
+            dto.missionUuid,
+            auth.user.uuid,
+        );
+        return { success: true };
+    }
+
     @Delete(':uuid')
     @CanDeleteFile()
     @OutputDto(DeleteFileResponseDto)
@@ -292,25 +308,10 @@ export class FileController {
             auth.user.uuid,
             auth.apiKey?.action,
             source,
+            body.fileSizes,
         );
     }
 
-    @Delete('uploads')
-    @UserOnly() //Push back authentication to the queue to accelerate the request
-    @OutputDto(CancelUploadResponseDto)
-    async cancelUpload(
-        @Body() dto: CancelFileUploadDto,
-        @AddUser() auth: AuthHeader,
-    ): Promise<CancelUploadResponseDto> {
-        logger.debug(`cancelUpload ${JSON.stringify(dto)}`);
-        await this.fileLifecycleService.cancelUpload(
-            dto.uuids,
-            dto.missionUuid,
-            auth.user.uuid,
-        );
-        return { success: true };
-    }
-
     @Delete()
     @CanDeleteMission()
     @ApiOkResponse({

backend/src/services/file-lifecycle.service.ts

@@ -13,16 +13,22 @@ import {
     IStorageBucket,
     StorageCredentials,
 } from '@kleinkram/backend-common/modules/storage/types';
+import { MissionAccessViewEntity } from '@kleinkram/backend-common/viewEntities/mission-access-view.entity';
+import { ProjectAccessViewEntity } from '@kleinkram/backend-common/viewEntities/project-access-view.entity';
 import {
+    AccessGroupRights,
     FileEventType,
     FileOrigin,
     FileState,
     FileType,
     TriggerEvent,
+    UserRole,
 } from '@kleinkram/shared';
 import {
     BadRequestException,
     ConflictException,
+    HttpException,
+    HttpStatus,
     Inject,
     Injectable,
     NotFoundException,
@@ -35,6 +41,7 @@ import {
     DataSource,
     In,
     MoreThan,
+    MoreThanOrEqual,
     QueryFailedError,
     Repository,
 } from 'typeorm';
@@ -329,6 +336,7 @@ export class FileLifecycleService implements OnModuleInit {
         userUUID: string,
         action?: ActionEntity,
         uploadSource = 'Web Interface',
+        fileSizes?: number[],
     ): Promise<TemporaryFileAccessesDto> {
         const mission = await this.missionRepository.findOneOrFail({
             where: { uuid: missionUUID },
@@ -338,6 +346,25 @@ export class FileLifecycleService implements OnModuleInit {
             where: { uuid: userUUID },
         });
 
+        if (
+            fileSizes &&
+            fileSizes.length > 0 &&
+            this.dataStorage.getSystemMetrics
+        ) {
+            const metrics = await this.dataStorage.getSystemMetrics();
+            const freeBytes = metrics.totalBytes - metrics.usedBytes;
+            const totalRequestedSize = fileSizes.reduce(
+                (sum, size) => sum + size,
+                0,
+            );
+            if (totalRequestedSize > freeBytes) {
+                throw new HttpException(
+                    'Insufficient storage space on the server',
+                    HttpStatus.INSUFFICIENT_STORAGE,
+                );
+            }
+        }
+
         return await this.dataSource.transaction(async (manager) => {
             // Deduplicate filenames to avoid self-collisions
             const uniqueFilenames = [...new Set(filenames)];
@@ -361,10 +388,6 @@ export class FileLifecycleService implements OnModuleInit {
                 },
             });
 
-            const existingFilenames = new Set(
-                existingFiles.map((f) => f.filename),
-            );
-
             for (const filename of uniqueFilenames) {
                 const emptyCredentials: {
                     bucket: string | null;
@@ -409,7 +432,13 @@ export class FileLifecycleService implements OnModuleInit {
                 if (fileType === undefined)
                     throw new UnsupportedMediaTypeException();
 
-                if (existingFilenames.has(filename)) {
+                const existingFile = existingFiles.find(
+                    (f) => f.filename === filename,
+                );
+                const isConflict =
+                    existingFile && existingFile.state !== FileState.CANCELED;
+
+                if (isConflict) {
                     invalidFiles.push({
                         filename,
                         error: 'File already exists',
@@ -420,19 +449,33 @@ export class FileLifecycleService implements OnModuleInit {
                 try {
                     // Use a nested transaction (savepoint) for each file
                     await manager.transaction(async (nestedManager) => {
-                        const file = await nestedManager.save(
-                            FileEntity,
-                            nestedManager.create(FileEntity, {
-                                date: new Date(),
-                                size: 0,
-                                filename,
-                                mission,
-                                creator: user,
-                                type: fileType,
-                                state: FileState.UPLOADING,
-                                origin: FileOrigin.UPLOAD,
-                            }),
-                        );
+                        let file: FileEntity;
+                        if (existingFile?.state === FileState.CANCELED) {
+                            existingFile.state = FileState.UPLOADING;
+                            existingFile.creator = user;
+                            existingFile.date = new Date();
+                            existingFile.size = 0;
+                            existingFile.hash = '';
+                            existingFile.origin = FileOrigin.UPLOAD;
+                            file = await nestedManager.save(
+                                FileEntity,
+                                existingFile,
+                            );
+                        } else {
+                            file = await nestedManager.save(
+                                FileEntity,
+                                nestedManager.create(FileEntity, {
+                                    date: new Date(),
+                                    size: 0,
+                                    filename,
+                                    mission,
+                                    creator: user,
+                                    type: fileType,
+                                    state: FileState.UPLOADING,
+                                    origin: FileOrigin.UPLOAD,
+                                }),
+                            );
+                        }
 
                         await this.auditService.log(
                             FileEventType.UPLOAD_STARTED,
@@ -459,9 +502,6 @@ export class FileLifecycleService implements OnModuleInit {
                                     file.uuid,
                                 ),
                         });
-
-                        // Add to local set to catch duplicates in the same batch
-                        existingFilenames.add(filename);
                     });
                 } catch (error: unknown) {
                     if (
@@ -473,8 +513,6 @@ export class FileLifecycleService implements OnModuleInit {
                             filename,
                             error: 'File already exists',
                         });
-                        // Also add to set so we don't try again if it appears again in list (though deduplication handles this)
-                        existingFilenames.add(filename);
                     } else {
                         throw error;
                     }
@@ -508,12 +546,84 @@ export class FileLifecycleService implements OnModuleInit {
         uuids: string[],
         missionUUID: string,
         userUUID: string,
-    ): Promise<Queue.Job> {
-        // Cleanup cannot be done synchronously as this takes too long; the request is sent on unload; delaying the onUnload is difficult and discouraged
-        return this.fileCleanupQueue.add('cancelUpload', {
-            uuids,
-            missionUUID,
+    ): Promise<void> {
+        const canCancelUpload = await this.canCancelUpload(
             userUUID,
+            missionUUID,
+        );
+        if (!canCancelUpload) {
+            logger.debug(`User ${userUUID} can't cancel upload`);
+            return;
+        }
+
+        await Promise.all(
+            uuids.map(async (uuid) => {
+                const file = await this.fileRepository.findOne({
+                    where: { uuid, mission: { uuid: missionUUID } },
+                    relations: ['mission'],
+                });
+                if (!file) {
+                    return;
+                }
+                if (file.state === FileState.OK) {
+                    return;
+                }
+
+                if (file.mission === undefined) {
+                    logger.error(
+                        `Mission of file ${file.uuid} is undefined, skipping`,
+                    );
+                    return;
+                }
+
+                file.state = FileState.CANCELED;
+                await this.fileRepository.save(file);
+                return;
+            }),
+        );
+    }
+
+    private async canCancelUpload(
+        userUUID: string,
+        missionUUID: string,
+    ): Promise<boolean> {
+        const user = await this.userRepository.findOneOrFail({
+            where: { uuid: userUUID },
+        });
+        if (user.role === UserRole.ADMIN) {
+            return true;
+        }
+        const mission = await this.missionRepository.findOneOrFail({
+            where: { uuid: missionUUID },
+            relations: ['project'],
+        });
+
+        if (mission.project === undefined) {
+            logger.error(
+                `Project of mission ${mission.uuid} is undefined, skipping`,
+            );
+            return false;
+        }
+
+        const canAccessProject = await this.dataSource.manager.exists(
+            ProjectAccessViewEntity,
+            {
+                where: {
+                    projectUuid: mission.project.uuid,
+                    userUuid: userUUID,
+                    rights: MoreThanOrEqual(AccessGroupRights.WRITE),
+                },
+            },
+        );
+        if (canAccessProject) {
+            return true;
+        }
+        return await this.dataSource.manager.exists(MissionAccessViewEntity, {
+            where: {
+                missionUuid: missionUUID,
+                userUuid: userUUID,
+                rights: MoreThanOrEqual(AccessGroupRights.WRITE),
+            },
         });
     }
 

backend/src/services/queue.service.ts

@@ -154,33 +154,30 @@ export class QueueService implements OnModuleInit {
         actor: UserEntity,
         source: FileSource | string = FileSource.WEB_INTERFACE,
     ): Promise<void> {
-        let job = await this.queueRepository.findOne({
-            where: { identifier: uuid },
+        const file = await this.fileRepository.findOneOrFail({
+            where: { uuid },
             relations: ['mission', 'mission.project'],
         });
 
-        if (!job) {
-            logger.warn(
-                `confirmUpload: Job missing for file ${uuid}.Recreating...`,
-            );
-
-            const file = await this.fileRepository.findOneOrFail({
-                where: { uuid },
-                relations: ['mission', 'mission.project'],
-            });
+        if (file.state === FileState.CANCELED) {
+            throw new ConflictException('Cannot confirm a canceled upload');
+        }
 
-            job = await this.queueRepository.save(
-                this.queueRepository.create({
-                    identifier: file.uuid,
+        let job = await this.queueRepository.findOne({
+            where: { identifier: uuid },
+            relations: ['mission', 'mission.project'],
+        });
 
-                    displayName: file.filename,
-                    state: QueueState.AWAITING_UPLOAD,
-                    location: FileLocation.S3,
-                    mission: file.mission,
-                    creator: actor,
-                } as IngestionJobEntity),
-            );
-        }
+        job ??= await this.queueRepository.save(
+            this.queueRepository.create({
+                identifier: file.uuid,
+                displayName: file.filename,
+                state: QueueState.AWAITING_UPLOAD,
+                location: FileLocation.S3,
+                mission: file.mission,
+                creator: actor,
+            } as IngestionJobEntity),
+        );
 
         if (
             job.state !== QueueState.AWAITING_UPLOAD &&
@@ -193,11 +190,6 @@ export class QueueService implements OnModuleInit {
             );
         }
 
-        const file = await this.fileRepository.findOneOrFail({
-            where: { uuid: uuid },
-            relations: ['mission', 'mission.project'],
-        });
-
         const fileInfo = await this.dataStorage
             .getFileInfo(file.uuid)
             .catch((error: unknown): void => {