From 997daae6398af616379c4dd50a235b688e4e9347 Mon Sep 17 00:00:00 2001 From: vladp Date: Tue, 16 Dec 2025 03:44:59 +0700 Subject: [PATCH] save chunks --- docker-compose.yaml | 6 +- .../controller/file/FileUploadController.java | 194 ++++-- .../no_copy/dto/file/ChunkStatusResponse.java | 23 + .../entity/file/FileUploadSession.java | 22 + .../FileUploadSessionRepository.java | 21 + .../service/file/FileUploadService.java | 8 +- .../service/file/FileUploadServiceImpl.java | 635 +++++++++++++----- src/main/resources/application.yaml | 11 +- 8 files changed, 703 insertions(+), 217 deletions(-) create mode 100644 src/main/java/ru/soune/no_copy/dto/file/ChunkStatusResponse.java diff --git a/docker-compose.yaml b/docker-compose.yaml index 8817acc..c9e2616 100644 --- a/docker-compose.yaml +++ b/docker-compose.yaml @@ -22,6 +22,9 @@ services: build: . container_name: app-backend environment: + FILE_STORAGE_PATH: /data/uploads + MAX_FILE_SIZE: 1073741824 + FILE_CHUNK_SIZE: 5242880 POSTGRES_DB: no_copy_ POSTGRES_USER: postgres POSTGRES_PASSWORD: postgres @@ -38,7 +41,7 @@ services: - backend - api volumes: - - ./uploads:/data/uploads + - uploads_volume:/data/uploads grafana: image: grafana/grafana:10.3.1 @@ -143,6 +146,7 @@ volumes: loki_chunks: loki_index: loki_rules: + uploads_volume: networks: app-network: diff --git a/src/main/java/ru/soune/no_copy/controller/file/FileUploadController.java b/src/main/java/ru/soune/no_copy/controller/file/FileUploadController.java index 47b7f47..a36728b 100644 --- a/src/main/java/ru/soune/no_copy/controller/file/FileUploadController.java +++ b/src/main/java/ru/soune/no_copy/controller/file/FileUploadController.java @@ -3,23 +3,25 @@ package ru.soune.no_copy.controller.file; import jakarta.validation.Valid; import lombok.RequiredArgsConstructor; import lombok.extern.slf4j.Slf4j; +import org.springframework.http.HttpStatus; import org.springframework.http.ResponseEntity; import org.springframework.web.bind.annotation.*; import org.springframework.web.multipart.MultipartFile; -import ru.soune.no_copy.dto.file.ChunkUploadRequest; import ru.soune.no_copy.dto.file.FileApiResponse; import ru.soune.no_copy.dto.file.InitUploadRequest; import ru.soune.no_copy.dto.file.UploadProgressResponse; import ru.soune.no_copy.entity.AuthToken; import ru.soune.no_copy.entity.file.FileType; import ru.soune.no_copy.entity.file.FileUploadSession; +import ru.soune.no_copy.entity.file.UploadStatus; +import ru.soune.no_copy.exception.FileUploadException; import ru.soune.no_copy.exception.NotFoundAuthToken; +import ru.soune.no_copy.exception.UploadSessionNotFoundException; import ru.soune.no_copy.repository.AuthTokenRepository; +import ru.soune.no_copy.repository.FileUploadSessionRepository; import ru.soune.no_copy.service.file.FileUploadService; -import java.util.Arrays; -import java.util.List; -import java.util.Optional; +import java.util.*; @Slf4j @RestController @@ -31,6 +33,8 @@ public class FileUploadController { private final AuthTokenRepository authTokenRepository; + private final FileUploadSessionRepository fileUploadSessionRepository; + //TODO ADD TEMPLATE JSON, RESPONSE OK,ADD MESSAGE CODE @PostMapping("/init") public ResponseEntity> initUpload(@Valid @RequestBody InitUploadRequest request, @@ -61,17 +65,26 @@ public class FileUploadController { @PostMapping("/chunk") public ResponseEntity> uploadChunk( - @Valid @RequestBody ChunkUploadRequest chunkUploadRequest, - @RequestParam("file") MultipartFile chunk, @RequestHeader("Authorization") String tokenHeader) { - String token = tokenHeader.replace("Bearer ", ""); + @RequestParam("uploadId") String uploadId, + @RequestParam("chunkNumber") Integer chunkNumber, + @RequestParam("chunk") MultipartFile chunk) { - Optional tokenOptional = authTokenRepository.findByToken(token); + log.info("Uploading chunk {} for session {}, file size: {} bytes", + chunkNumber, uploadId, chunk.getSize()); - tokenOptional.orElseThrow(() -> new NotFoundAuthToken("Token: " + token + "not found")); + if (chunk.isEmpty()) { + return ResponseEntity.badRequest() + .body(FileApiResponse.error("Chunk file is empty")); + } + + if (uploadId == null || uploadId.isBlank()) { + return ResponseEntity.badRequest() + .body(FileApiResponse.error("Upload ID is required")); + } try { UploadProgressResponse progress = fileUploadService.uploadChunk( - chunkUploadRequest.getUploadId(), chunkUploadRequest.getChunkNumber(), chunk); + uploadId, chunkNumber, chunk); return ResponseEntity.ok(FileApiResponse.success(progress)); @@ -82,39 +95,130 @@ public class FileUploadController { } } -// @GetMapping("/progress/{uploadId}") -// public ResponseEntity> getProgress( -// @PathVariable String uploadId, -// @RequestHeader(value = "X-User-Id", required = false, defaultValue = "1") Long userId) { -// -// try { -// UploadProgressResponse progress = fileUploadService.getUploadProgress(uploadId); -// return ResponseEntity.ok(FileApiResponse.success(progress)); -// -// } catch (Exception e) { -// log.error("Error getting progress", e); -// return ResponseEntity.badRequest() -// .body(FileApiResponse.error(e.getMessage())); -// } -// } -// -// @PostMapping("/complete/{uploadId}") -// public ResponseEntity> completeUpload( -// @PathVariable String uploadId, -// @RequestHeader(value = "X-User-Id", required = false, defaultValue = "1") Long userId) { -// -// log.info("Completing upload for session {}", uploadId); -// -// try { -// UploadProgressResponse progress = fileUploadService.getUploadProgress(uploadId); -// -// return ResponseEntity.ok(FileApiResponse.success( -// "File assembly in progress", progress)); -// -// } catch (Exception e) { -// log.error("Error completing upload", e); -// return ResponseEntity.badRequest() -// .body(FileApiResponse.error(e.getMessage())); -// } -// } + @GetMapping("/progress/{uploadId}") + public ResponseEntity> getProgress( + @PathVariable String uploadId) { + + try { + UploadProgressResponse progress = fileUploadService.getUploadProgress(uploadId); + return ResponseEntity.ok(FileApiResponse.success(progress)); + + } catch (UploadSessionNotFoundException e) { + return ResponseEntity.status(HttpStatus.NOT_FOUND) + .body(FileApiResponse.error("Upload session not found")); + } catch (Exception e) { + log.error("Error getting progress", e); + return ResponseEntity.badRequest() + .body(FileApiResponse.error("Failed to get progress")); + } + } + + @PostMapping("/{uploadId}/complete") + public ResponseEntity> completeUpload( + @PathVariable String uploadId) { + + log.info("Manual completion requested for session: {}", uploadId); + + try { + UploadProgressResponse progress = fileUploadService.getUploadProgress(uploadId); + + if (progress.getStatus() == UploadStatus.COMPLETED) { + return ResponseEntity.ok(FileApiResponse.success( + "Upload already completed", progress)); + } + + if (progress.getUploadedChunks() < progress.getTotalChunks()) { + return ResponseEntity.badRequest() + .body(FileApiResponse.error( + String.format("Not all chunks uploaded: %d/%d", + progress.getUploadedChunks(), progress.getTotalChunks()))); + } + + return ResponseEntity.ok(FileApiResponse.success( + "File assembly in progress", progress)); + + } catch (UploadSessionNotFoundException e) { + return ResponseEntity.status(HttpStatus.NOT_FOUND) + .body(FileApiResponse.error("Upload session not found")); + } catch (Exception e) { + log.error("Error completing upload", e); + return ResponseEntity.internalServerError() + .body(FileApiResponse.error("Failed to complete upload")); + } + } + + @PostMapping("/{uploadId}/cancel") + public ResponseEntity> cancelUpload( + @PathVariable String uploadId) { + + log.info("Cancelling upload session: {}", uploadId); + + try { + fileUploadService.cancelUpload(uploadId); + return ResponseEntity.ok(FileApiResponse.success( + "Upload cancelled successfully", null)); + + } catch (UploadSessionNotFoundException e) { + return ResponseEntity.status(HttpStatus.NOT_FOUND) + .body(FileApiResponse.error("Upload session not found")); + } catch (FileUploadException e) { + return ResponseEntity.badRequest() + .body(FileApiResponse.error(e.getMessage())); + } catch (Exception e) { + log.error("Error cancelling upload", e); + return ResponseEntity.internalServerError() + .body(FileApiResponse.error("Failed to cancel upload")); + } + } + + @PostMapping("/{uploadId}/retry") + public ResponseEntity> retryUpload( + @PathVariable String uploadId) { + + log.info("Retry requested for failed upload: {}", uploadId); + + try { + fileUploadService.retryFailedUpload(uploadId); + UploadProgressResponse progress = fileUploadService.getUploadProgress(uploadId); + + return ResponseEntity.ok(FileApiResponse.success( + "Upload retry initiated", progress)); + + } catch (UploadSessionNotFoundException e) { + return ResponseEntity.status(HttpStatus.NOT_FOUND) + .body(FileApiResponse.error("Upload session not found")); + } catch (FileUploadException e) { + return ResponseEntity.badRequest() + .body(FileApiResponse.error(e.getMessage())); + } catch (Exception e) { + log.error("Error retrying upload", e); + return ResponseEntity.internalServerError() + .body(FileApiResponse.error("Failed to retry upload")); + } + } + + @GetMapping("/{uploadId}/chunks") + public ResponseEntity>> getChunkStatus( + @PathVariable String uploadId) { + + try { + FileUploadSession session = fileUploadSessionRepository.findById(uploadId) + .orElseThrow(() -> new UploadSessionNotFoundException(uploadId)); + + Map chunkStatus = new HashMap<>(); + for (int i = 0; i < session.getTotalChunks(); i++) { + chunkStatus.put(i, session.getChunkPaths().containsKey(i)); + } + + return ResponseEntity.ok(FileApiResponse.success(chunkStatus)); + + } catch (UploadSessionNotFoundException e) { + return ResponseEntity.status(HttpStatus.NOT_FOUND) + .body(FileApiResponse.error("Upload session not found")); + } catch (Exception e) { + log.error("Error getting chunk status", e); + return ResponseEntity.internalServerError() + .body(FileApiResponse.error("Failed to get chunk status")); + } + } } diff --git a/src/main/java/ru/soune/no_copy/dto/file/ChunkStatusResponse.java b/src/main/java/ru/soune/no_copy/dto/file/ChunkStatusResponse.java new file mode 100644 index 0000000..0a5f269 --- /dev/null +++ b/src/main/java/ru/soune/no_copy/dto/file/ChunkStatusResponse.java @@ -0,0 +1,23 @@ +package ru.soune.no_copy.dto.file; + +import lombok.AllArgsConstructor; +import lombok.Builder; +import lombok.Data; +import lombok.NoArgsConstructor; +import ru.soune.no_copy.entity.file.UploadStatus; + +import java.util.List; +import java.util.Map; + +@Data +@Builder +@NoArgsConstructor +@AllArgsConstructor +public class ChunkStatusResponse { + private String uploadId; + private Integer totalChunks; + private Integer uploadedChunks; + private List missingChunks; + private Map chunkPaths; + private UploadStatus status; +} diff --git a/src/main/java/ru/soune/no_copy/entity/file/FileUploadSession.java b/src/main/java/ru/soune/no_copy/entity/file/FileUploadSession.java index 08f54f3..0eed4a0 100644 --- a/src/main/java/ru/soune/no_copy/entity/file/FileUploadSession.java +++ b/src/main/java/ru/soune/no_copy/entity/file/FileUploadSession.java @@ -53,6 +53,18 @@ public class FileUploadSession { @Column(name = "created_at", updatable = false, nullable = false) private LocalDateTime createdAt; + @Column(name = "expires_at") + private LocalDateTime expiresAt; + + @Version + private Long version; + + @Column(name = "last_error") + private String lastError; + + @Column(name = "retry_count") + private Integer retryCount = 0; + @Column(name = "completed_at") private LocalDateTime completedAt; @@ -64,4 +76,14 @@ public class FileUploadSession { @MapKeyColumn(name = "chunk_number") @Column(name = "chunk_path") private Map chunkPaths = new HashMap<>(); + + @PrePersist + public void prePersist() { + this.createdAt = LocalDateTime.now(); + this.expiresAt = LocalDateTime.now().plusHours(24); + this.chunksUploaded = 0; + if (this.status == null) { + this.status = UploadStatus.INITIATED; + } + } } diff --git a/src/main/java/ru/soune/no_copy/repository/FileUploadSessionRepository.java b/src/main/java/ru/soune/no_copy/repository/FileUploadSessionRepository.java index de7b788..0a9175f 100644 --- a/src/main/java/ru/soune/no_copy/repository/FileUploadSessionRepository.java +++ b/src/main/java/ru/soune/no_copy/repository/FileUploadSessionRepository.java @@ -1,9 +1,30 @@ package ru.soune.no_copy.repository; import org.springframework.data.jpa.repository.JpaRepository; +import org.springframework.data.jpa.repository.Query; +import org.springframework.data.repository.query.Param; import org.springframework.stereotype.Repository; import ru.soune.no_copy.entity.file.FileUploadSession; +import ru.soune.no_copy.entity.file.UploadStatus; + +import java.time.LocalDateTime; +import java.util.List; +import java.util.Set; @Repository public interface FileUploadSessionRepository extends JpaRepository { + + List findByUserId(Long userId); + + List findByStatus(UploadStatus status); + + @Query("SELECT s FROM FileUploadSession s WHERE s.status IN :statuses AND s.createdAt < :threshold") + List findByStatusInAndCreatedAtBefore( + @Param("statuses") Set statuses, + @Param("threshold") LocalDateTime threshold); + + @Query("SELECT s FROM FileUploadSession s WHERE s.expiresAt < :now AND s.status IN :statuses") + List findExpiredSessions( + @Param("now") LocalDateTime now, + @Param("statuses") Set statuses); } diff --git a/src/main/java/ru/soune/no_copy/service/file/FileUploadService.java b/src/main/java/ru/soune/no_copy/service/file/FileUploadService.java index dbf4776..e401240 100644 --- a/src/main/java/ru/soune/no_copy/service/file/FileUploadService.java +++ b/src/main/java/ru/soune/no_copy/service/file/FileUploadService.java @@ -4,14 +4,20 @@ import org.springframework.web.multipart.MultipartFile; import ru.soune.no_copy.dto.file.UploadProgressResponse; import ru.soune.no_copy.entity.file.FileUploadSession; +import java.io.IOException; + public interface FileUploadService { FileUploadSession initUpload(Long userId, String fileName, String fileType, long fileSize); UploadProgressResponse uploadChunk(String uploadId, Integer chunkNumber, - MultipartFile chunkFile); + MultipartFile chunkFile) throws IOException; UploadProgressResponse getUploadProgress(String uploadId); void cleanupExpiredSessions(); + + public void retryFailedUpload(String uploadId); + + public void cancelUpload(String uploadId); } diff --git a/src/main/java/ru/soune/no_copy/service/file/FileUploadServiceImpl.java b/src/main/java/ru/soune/no_copy/service/file/FileUploadServiceImpl.java index 5030223..f70ffd2 100644 --- a/src/main/java/ru/soune/no_copy/service/file/FileUploadServiceImpl.java +++ b/src/main/java/ru/soune/no_copy/service/file/FileUploadServiceImpl.java @@ -17,14 +17,16 @@ import ru.soune.no_copy.exception.UploadSessionNotFoundException; import ru.soune.no_copy.repository.FileUploadSessionRepository; import java.io.*; -import java.nio.file.Files; -import java.nio.file.Path; -import java.nio.file.Paths; -import java.nio.file.StandardOpenOption; +import java.nio.file.*; +import java.nio.file.attribute.PosixFilePermission; import java.security.MessageDigest; import java.security.NoSuchAlgorithmException; import java.time.LocalDateTime; import java.util.Comparator; +import java.util.HashSet; +import java.util.Set; +import java.util.UUID; +import java.util.concurrent.CompletableFuture; import java.util.stream.Stream; @Slf4j @@ -43,24 +45,36 @@ public class FileUploadServiceImpl implements FileUploadService { @Value("${file.storage.max-file-size}") private long maxFileSize; + @Value("${file.storage.max-retry-attempts:3}") + private int maxRetryAttempts; + + @Value("${file.storage.chunk-timeout-ms:300000}") // 5 минут + private long chunkTimeoutMs; + + @Value("${file.storage.session-expiry-hours:24}") + private int sessionExpiryHours; + + private Path storageRoot; + @PostConstruct public void init() { try { - Path storageRoot = Paths.get(basePath).toAbsolutePath().normalize(); + storageRoot = Paths.get(basePath).toAbsolutePath().normalize(); + log.info("Initializing file storage at: {}", storageRoot); - log.info("Storage will be at: {}", storageRoot); + createDirectoryIfNotExists(storageRoot); + createDirectoryIfNotExists(storageRoot.resolve("temp")); + createDirectoryIfNotExists(storageRoot.resolve("uploads")); - if (!Files.exists(storageRoot)) { - Files.createDirectories(storageRoot); - log.info("Created: {}", storageRoot); + if (!System.getProperty("os.name").toLowerCase().contains("win")) { + setDirectoryPermissions(storageRoot); } - Files.createDirectories(storageRoot.resolve("temp")); - Files.createDirectories(storageRoot.resolve("user")); + log.info("File storage initialized successfully"); } catch (IOException e) { - log.error("CANNOT CREATE STORAGE!", e); - throw new RuntimeException("Storage failed", e); + log.error("Failed to initialize file storage at: {}", basePath, e); + throw new RuntimeException("Storage initialization failed", e); } } @@ -68,13 +82,22 @@ public class FileUploadServiceImpl implements FileUploadService { @Transactional public FileUploadSession initUpload(Long userId, String fileName, String fileType, long fileSize) { + log.info("Initializing upload for user {}: {} ({} bytes, type: {})", + userId, fileName, fileSize, fileType); + if (fileSize > maxFileSize) { throw new FileUploadException( String.format("File size %d exceeds maximum allowed size %d", fileSize, maxFileSize)); } + if (fileSize <= 0) { + throw new FileUploadException("File size must be positive"); + } + int totalChunks = (int) Math.ceil((double) fileSize / chunkSize); + log.debug("File will be split into {} chunks (chunk size: {} bytes)", + totalChunks, chunkSize); FileUploadSession session = FileUploadSession.builder() .userId(userId) @@ -84,9 +107,15 @@ public class FileUploadServiceImpl implements FileUploadService { .totalChunks(totalChunks) .chunksUploaded(0) .status(UploadStatus.INITIATED) + .expiresAt(LocalDateTime.now().plusHours(sessionExpiryHours)) + .retryCount(0) .build(); - return sessionRepository.save(session); + FileUploadSession savedSession = sessionRepository.save(session); + log.info("Upload session created: {} for file: {}", + savedSession.getUploadId(), fileName); + + return savedSession; } @Override @@ -94,38 +123,92 @@ public class FileUploadServiceImpl implements FileUploadService { public UploadProgressResponse uploadChunk(String uploadId, Integer chunkNumber, MultipartFile chunkFile) { - log.info("Uploading chunk {} for session {}, chunk size: {} bytes", + log.info("Processing chunk {} for session {}, size: {} bytes", chunkNumber, uploadId, chunkFile.getSize()); FileUploadSession session = sessionRepository.findById(uploadId) .orElseThrow(() -> new UploadSessionNotFoundException(uploadId)); - if (session.getChunkPaths().containsKey(chunkNumber)) { - log.info("Chunk {} already uploaded for session {}", chunkNumber, uploadId); - return UploadProgressResponse.fromSession(session); + validateSession(session); + + if (session.getExpiresAt().isBefore(LocalDateTime.now())) { + handleExpiredSession(session); + throw new FileUploadException("Upload session expired"); + } + + if (chunkNumber < 0 || chunkNumber >= session.getTotalChunks()) { + throw new FileUploadException( + String.format("Invalid chunk number %d. Expected 0-%d", + chunkNumber, session.getTotalChunks() - 1)); } if (chunkFile.getSize() > chunkSize) { throw new ChunkSizeExceededException(chunkFile.getSize(), chunkSize); } - try { - log.debug("Chunk {} actual size: {} bytes", chunkNumber, chunkFile.getSize()); + return processChunk(session, chunkNumber, chunkFile); + } - String chunkPath = saveChunkToDisk(session, chunkNumber, chunkFile); + private void validateSession(FileUploadSession session) { + UploadStatus status = session.getStatus(); + + if (status == UploadStatus.FAILED) { + if (session.getRetryCount() >= maxRetryAttempts) { + throw new FileUploadException( + "Upload failed after maximum retry attempts"); + } + session.setStatus(UploadStatus.UPLOADING); + session.setRetryCount(session.getRetryCount() + 1); + sessionRepository.save(session); + log.info("Retrying failed upload session: {}, attempt: {}", + session.getUploadId(), session.getRetryCount()); + } + + if (status == UploadStatus.COMPLETED) { + throw new FileUploadException("Upload already completed"); + } + + if (status == UploadStatus.CANCELLED) { + throw new FileUploadException("Upload was cancelled"); + } + + if (status == UploadStatus.INITIATED) { + session.setStatus(UploadStatus.UPLOADING); + sessionRepository.save(session); + } + } + + private void handleExpiredSession(FileUploadSession session) { + session.setStatus(UploadStatus.FAILED); + session.setLastError("Upload session expired"); + sessionRepository.save(session); + + CompletableFuture.runAsync(() -> cleanupSessionFiles(session)); + } + + private UploadProgressResponse processChunk(FileUploadSession session, + Integer chunkNumber, + MultipartFile chunkFile) { + String chunkPath = null; + + try { + if (session.getChunkPaths().containsKey(chunkNumber)) { + return handleExistingChunk(session, chunkNumber, chunkFile); + } + + chunkPath = saveChunkWithIntegrityCheck(session, chunkNumber, chunkFile); session.getChunkPaths().put(chunkNumber, chunkPath); session.setChunksUploaded(session.getChunksUploaded() + 1); + session.setStatus(UploadStatus.UPLOADING); - log.info("Chunk {} uploaded successfully. Total uploaded: {}/{}", + log.debug("Chunk {} saved successfully. Uploaded: {}/{}", chunkNumber, session.getChunksUploaded(), session.getTotalChunks()); if (session.getChunksUploaded().equals(session.getTotalChunks())) { - session.setStatus(UploadStatus.UPLOADING); + log.info("All chunks uploaded for session {}. Starting assembly...", + session.getUploadId()); sessionRepository.save(session); - - log.info("All chunks uploaded for session {}. Starting assembly...", uploadId); - assembleFileAsync(session); } else { sessionRepository.save(session); @@ -133,53 +216,99 @@ public class FileUploadServiceImpl implements FileUploadService { return UploadProgressResponse.fromSession(session); - } catch (IOException e) { - log.error("Failed to save chunk {}", chunkNumber, e); + } catch (Exception e) { + if (chunkPath != null) { + cleanupFailedChunk(chunkPath); + } + + log.error("Failed to process chunk {} for session {}: {}", + chunkNumber, session.getUploadId(), e.getMessage(), e); + session.setStatus(UploadStatus.FAILED); + session.setLastError(e.getMessage()); sessionRepository.save(session); - throw new RuntimeException("Failed to save chunk: " + e.getMessage(), e); + + throw new FileUploadException("Failed to upload chunk: " + e.getMessage(), e); } } - private String saveChunkToDisk(FileUploadSession session, - Integer chunkNumber, - MultipartFile chunkFile) throws IOException { + private UploadProgressResponse handleExistingChunk(FileUploadSession session, + Integer chunkNumber, + MultipartFile chunkFile) throws IOException { + String existingPath = session.getChunkPaths().get(chunkNumber); + Path chunkPath = Paths.get(existingPath); + + if (!Files.exists(chunkPath)) { + log.warn("Chunk file missing, re-uploading: {}", chunkPath); + session.getChunkPaths().remove(chunkNumber); + session.setChunksUploaded(session.getChunksUploaded() - 1); + sessionRepository.save(session); + return processChunk(session, chunkNumber, chunkFile); + } + + long existingSize = Files.size(chunkPath); + if (existingSize != chunkFile.getSize()) { + log.warn("Chunk size mismatch, re-uploading: {} != {}", + existingSize, chunkFile.getSize()); + Files.deleteIfExists(chunkPath); + session.getChunkPaths().remove(chunkNumber); + session.setChunksUploaded(session.getChunksUploaded() - 1); + sessionRepository.save(session); + return processChunk(session, chunkNumber, chunkFile); + } + + log.debug("Chunk {} already uploaded and valid", chunkNumber); + return UploadProgressResponse.fromSession(session); + } + + private String saveChunkWithIntegrityCheck(FileUploadSession session, + Integer chunkNumber, + MultipartFile chunkFile) throws IOException { Long userId = session.getUserId(); String uploadId = session.getUploadId(); - Path chunkDir = Paths.get(basePath, "temp", - "user", String.valueOf(userId), uploadId); + Path chunkDir = getChunkDirectory(userId, uploadId); Files.createDirectories(chunkDir); - String chunkFileName = String.format("chunk_%03d.tmp", chunkNumber); - Path chunkPath = chunkDir.resolve(chunkFileName); + String tempFileName = String.format("chunk_%04d.%s.tmp", + chunkNumber, UUID.randomUUID()); + String finalFileName = String.format("chunk_%04d.tmp", chunkNumber); - try (InputStream inputStream = chunkFile.getInputStream(); - OutputStream outputStream = Files.newOutputStream(chunkPath, - StandardOpenOption.CREATE, StandardOpenOption.TRUNCATE_EXISTING)) { + Path tempPath = chunkDir.resolve(tempFileName); + Path finalPath = chunkDir.resolve(finalFileName); - byte[] buffer = new byte[8192]; - int bytesRead; - long totalBytes = 0; + try { + chunkFile.transferTo(tempPath.toFile()); - while ((bytesRead = inputStream.read(buffer)) != -1) { - outputStream.write(buffer, 0, bytesRead); - totalBytes += bytesRead; + long savedSize = Files.size(tempPath); + long uploadedSize = chunkFile.getSize(); + + if (uploadedSize > 0 && savedSize != uploadedSize) { + throw new IOException( + String.format("Size mismatch: saved %d, uploaded %d", + savedSize, uploadedSize)); } - log.debug("Chunk {} saved to {}: {} bytes written", - chunkNumber, chunkPath, totalBytes); + Files.move(tempPath, finalPath, StandardCopyOption.ATOMIC_MOVE); - } catch (Exception e) { - log.error("Error saving chunk {} to {}", chunkNumber, chunkPath, e); - throw e; + log.debug("Chunk {} saved successfully: {} bytes", + chunkNumber, savedSize); + + return finalPath.toString(); + + } finally { + Files.deleteIfExists(tempPath); } + } - long savedSize = Files.size(chunkPath); - log.debug("Chunk {} saved size: {} bytes", chunkNumber, savedSize); - - return chunkPath.toString(); + private void cleanupFailedChunk(String chunkPath) { + try { + Files.deleteIfExists(Paths.get(chunkPath)); + log.debug("Cleaned up failed chunk: {}", chunkPath); + } catch (IOException e) { + log.warn("Failed to cleanup chunk: {}", chunkPath, e); + } } @Async("fileUploadTaskExecutor") @@ -187,130 +316,307 @@ public class FileUploadServiceImpl implements FileUploadService { public void assembleFileAsync(FileUploadSession session) { try { assembleFile(session); - log.info("File assembly completed for session {}", session.getUploadId()); + log.info("File assembly completed successfully for session: {}", + session.getUploadId()); } catch (Exception e) { - log.error("Failed to assemble file for session {}", - session.getUploadId(), e); + log.error("Failed to assemble file for session {}: {}", + session.getUploadId(), e.getMessage(), e); - session.setStatus(UploadStatus.FAILED); - sessionRepository.save(session); + handleAssemblyFailure(session, e); } } private void assembleFile(FileUploadSession session) throws IOException { - Long userId = session.getUserId(); - String uploadId = session.getUploadId(); + log.info("Starting file assembly for session: {} ({})", + session.getUploadId(), session.getFileName()); - log.info("Assembling file for session {}: {} ({} chunks)", - uploadId, session.getFileName(), session.getTotalChunks()); - - Path finalDir = createUserDirectory(userId, session.getFileType()); - - String safeFileName = generateSafeFileName(session.getFileName()); - Path finalFilePath = finalDir.resolve(safeFileName); - - log.debug("Final file path: {}", finalFilePath); - - try (OutputStream outputStream = new BufferedOutputStream( - Files.newOutputStream(finalFilePath, StandardOpenOption.CREATE))) { - - Path chunkDir = getChunkDirectory(userId, uploadId); - long totalAssembled = 0; - - for (int i = 0; i < session.getTotalChunks(); i++) { - Path chunkPath = chunkDir.resolve(String.format("chunk_%03d.tmp", i)); - - if (!Files.exists(chunkPath)) { - throw new IOException("Missing chunk: " + chunkPath); - } - - byte[] chunkData = Files.readAllBytes(chunkPath); - outputStream.write(chunkData); - totalAssembled += chunkData.length; - - log.debug("Added chunk {}: {} bytes (total: {})", - i, chunkData.length, totalAssembled); - } - - log.info("File assembled: {} bytes total", totalAssembled); - } + Path finalFilePath = null; try { - validateFileIntegrity(finalFilePath, session.getFileSize()); - } catch (IOException e) { - log.warn("File size mismatch (temporarily ignored): {}", e.getMessage()); + finalFilePath = prepareFinalFile(session); + + validateAllChunksExist(session); + + mergeChunksToFile(session, finalFilePath); + + validateFinalFile(session, finalFilePath); + + updateSessionOnSuccess(session, finalFilePath); + + cleanupSessionFiles(session); + + log.info("File assembly completed: {} -> {} ({} bytes)", + session.getFileName(), finalFilePath, session.getFileSize()); + + } catch (Exception e) { + if (finalFilePath != null) { + Files.deleteIfExists(finalFilePath); + } + throw e; + } + } + + private Path prepareFinalFile(FileUploadSession session) throws IOException { + Path userUploadsDir = storageRoot.resolve("uploads") + .resolve(String.valueOf(session.getUserId())) + .resolve(extractCategory(session.getFileType())); + + Files.createDirectories(userUploadsDir); + + String safeFileName = generateUniqueFileName(session.getFileName()); + Path finalPath = userUploadsDir.resolve(safeFileName); + + if (Files.exists(finalPath)) { + throw new IOException("File already exists: " + finalPath); } - session.setFilePath(finalFilePath.toString()); - session.setChecksum(calculateChecksum(finalFilePath)); + return finalPath; + } + + private String generateUniqueFileName(String originalName) { + String safeName = originalName.replaceAll("[^a-zA-Z0-9\\.\\-_]", "_"); + + String timestamp = String.valueOf(System.currentTimeMillis()); + String uuid = UUID.randomUUID().toString().substring(0, 8); + + int dotIndex = safeName.lastIndexOf('.'); + if (dotIndex > 0) { + String name = safeName.substring(0, dotIndex); + String extension = safeName.substring(dotIndex); + return String.format("%s_%s_%s%s", name, timestamp, uuid, extension); + } else { + return String.format("%s_%s_%s", safeName, timestamp, uuid); + } + } + + private void validateAllChunksExist(FileUploadSession session) throws IOException { + Path chunkDir = getChunkDirectory(session.getUserId(), session.getUploadId()); + + for (int i = 0; i < session.getTotalChunks(); i++) { + Path chunkPath = chunkDir.resolve(String.format("chunk_%04d.tmp", i)); + if (!Files.exists(chunkPath)) { + throw new IOException("Missing chunk: " + chunkPath.getFileName()); + } + + if (Files.size(chunkPath) == 0) { + throw new IOException("Empty chunk: " + chunkPath.getFileName()); + } + } + } + + private void mergeChunksToFile(FileUploadSession session, Path finalPath) + throws IOException { + + Path chunkDir = getChunkDirectory(session.getUserId(), session.getUploadId()); + byte[] buffer = new byte[8192]; // 8KB буфер + + try (BufferedOutputStream outputStream = new BufferedOutputStream( + Files.newOutputStream(finalPath, StandardOpenOption.CREATE_NEW))) { + + long totalWritten = 0; + + for (int i = 0; i < session.getTotalChunks(); i++) { + Path chunkPath = chunkDir.resolve(String.format("chunk_%04d.tmp", i)); + + try (BufferedInputStream inputStream = new BufferedInputStream( + Files.newInputStream(chunkPath))) { + + int bytesRead; + while ((bytesRead = inputStream.read(buffer)) != -1) { + outputStream.write(buffer, 0, bytesRead); + totalWritten += bytesRead; + } + } + + if (log.isDebugEnabled()) { + log.debug("Processed chunk {}: {} bytes (total: {})", + i, Files.size(chunkPath), totalWritten); + } + } + + outputStream.flush(); + log.info("Total bytes written to final file: {}", totalWritten); + } + } + + private void validateFinalFile(FileUploadSession session, Path finalPath) + throws IOException { + + long actualSize = Files.size(finalPath); + + if (actualSize != session.getFileSize()) { + throw new IOException( + String.format("File size mismatch: expected %d, got %d", + session.getFileSize(), actualSize)); + } + + String checksum = calculateChecksum(finalPath); + log.debug("Final file checksum: {}", checksum); + } + + private void updateSessionOnSuccess(FileUploadSession session, Path finalPath) { + session.setFilePath(finalPath.toString()); + session.setChecksum(calculateChecksum(finalPath)); session.setStatus(UploadStatus.COMPLETED); session.setCompletedAt(LocalDateTime.now()); sessionRepository.save(session); + } - log.info("File assembled successfully: {} -> {}", - session.getFileName(), finalFilePath); - + private void cleanupSessionFiles(FileUploadSession session) { try { - cleanupChunks(getChunkDirectory(userId, uploadId)); + Path chunkDir = getChunkDirectory(session.getUserId(), session.getUploadId()); + if (Files.exists(chunkDir)) { + deleteDirectoryRecursively(chunkDir); + log.info("Cleaned up temp directory for session: {}", + session.getUploadId()); + } } catch (IOException e) { - log.warn("Failed to cleanup chunks: {}", e.getMessage()); + log.warn("Failed to cleanup temp files for session {}: {}", + session.getUploadId(), e.getMessage()); } } - private Path createUserDirectory(Long userId, String fileType) throws IOException { - Path userDir = Paths.get(basePath, "user", String.valueOf(userId), extractCategory(fileType)); + private void deleteDirectoryRecursively(Path directory) throws IOException { + try (Stream walk = Files.walk(directory)) { + walk.sorted(Comparator.reverseOrder()) + .forEach(path -> { + try { + Files.deleteIfExists(path); + } catch (IOException e) { + log.warn("Failed to delete: {}", path, e); + } + }); + } + } - Files.createDirectories(userDir); + private void handleAssemblyFailure(FileUploadSession session, Exception e) { + session.setStatus(UploadStatus.FAILED); + session.setLastError(e.getMessage()); + sessionRepository.save(session); - return userDir; + log.error("File assembly failed for session {}: {}", + session.getUploadId(), e.getMessage()); + } + + @Override + @Transactional + public void cancelUpload(String uploadId) { + FileUploadSession session = sessionRepository.findById(uploadId) + .orElseThrow(() -> new UploadSessionNotFoundException(uploadId)); + + if (session.getStatus() == UploadStatus.COMPLETED) { + throw new FileUploadException("Cannot cancel completed upload"); + } + + session.setStatus(UploadStatus.CANCELLED); + sessionRepository.save(session); + + CompletableFuture.runAsync(() -> cleanupSessionFiles(session)); + + log.info("Upload cancelled: {}", uploadId); + } + + @Override + @Transactional + public void retryFailedUpload(String uploadId) { + FileUploadSession session = sessionRepository.findById(uploadId) + .orElseThrow(() -> new UploadSessionNotFoundException(uploadId)); + + if (session.getStatus() != UploadStatus.FAILED) { + throw new FileUploadException("Only failed uploads can be retried"); + } + + if (session.getRetryCount() >= maxRetryAttempts) { + throw new FileUploadException("Max retry attempts exceeded"); + } + + session.setStatus(UploadStatus.UPLOADING); + session.setRetryCount(session.getRetryCount() + 1); + session.setLastError(null); + sessionRepository.save(session); + + if (session.getChunksUploaded().equals(session.getTotalChunks())) { + log.info("Retrying file assembly for session: {}", uploadId); + assembleFileAsync(session); + } + + log.info("Upload retry initiated for session: {} (attempt {})", + uploadId, session.getRetryCount()); + } + + @Override + public UploadProgressResponse getUploadProgress(String uploadId) { + FileUploadSession session = sessionRepository.findById(uploadId) + .orElseThrow(() -> new UploadSessionNotFoundException(uploadId)); + + return UploadProgressResponse.fromSession(session); + } + + @Override + @Transactional + public void cleanupExpiredSessions() { + log.info("Starting cleanup of expired upload sessions"); + + LocalDateTime expiryThreshold = LocalDateTime.now().minusHours(sessionExpiryHours); + + try { + var expiredSessions = sessionRepository.findByStatusInAndCreatedAtBefore( + Set.of(UploadStatus.INITIATED, UploadStatus.UPLOADING, UploadStatus.FAILED), + expiryThreshold + ); + + int cleanedCount = 0; + for (FileUploadSession session : expiredSessions) { + session.setStatus(UploadStatus.FAILED); + session.setLastError("Session expired during cleanup"); + sessionRepository.save(session); + + CompletableFuture.runAsync(() -> cleanupSessionFiles(session)); + cleanedCount++; + } + + log.info("Cleaned up {} expired upload sessions", cleanedCount); + + } catch (Exception e) { + log.error("Error during session cleanup", e); + } + } + + private Path getChunkDirectory(Long userId, String uploadId) { + return storageRoot.resolve("temp") + .resolve("user") + .resolve(String.valueOf(userId)) + .resolve(uploadId); } private String extractCategory(String fileType) { - if (fileType.startsWith("image/")) { + if (fileType == null) { + return "other"; + } + + String lowerType = fileType.toLowerCase(); + if (lowerType.startsWith("image/")) { return "images"; - } else if (fileType.startsWith("video/")) { + } else if (lowerType.startsWith("video/")) { return "videos"; - } else if (fileType.startsWith("audio/")) { + } else if (lowerType.startsWith("audio/")) { return "audio"; - } else if (fileType.contains("pdf")) { + } else if (lowerType.contains("pdf") || + lowerType.contains("document") || + lowerType.contains("text") || + lowerType.contains("msword") || + lowerType.contains("wordprocessing") || + lowerType.contains("spreadsheet")) { return "documents"; - } else if (fileType.contains("text")) { - return "text"; } else { return "other"; } } - private String generateSafeFileName(String originalName) { - String safeName = originalName.replaceAll("[^a-zA-Z0-9\\.\\-_]", "_"); - - String timestamp = String.valueOf(System.currentTimeMillis()); - int dotIndex = safeName.lastIndexOf('.'); - - if (dotIndex > 0) { - String name = safeName.substring(0, dotIndex); - String extension = safeName.substring(dotIndex); - return name + "_" + timestamp + extension; - } else { - return safeName + "_" + timestamp; - } - } - - private void validateFileIntegrity(Path filePath, long expectedSize) throws IOException { - long actualSize = Files.size(filePath); - log.info("File integrity check: expected={}, actual={}", expectedSize, actualSize); - - if (actualSize != expectedSize) { - throw new IOException( - String.format("File size mismatch: expected %d, got %d", - expectedSize, actualSize)); - } - } - - private String calculateChecksum(Path filePath) throws IOException { + private String calculateChecksum(Path filePath) { try { - MessageDigest md = MessageDigest.getInstance("MD5"); + MessageDigest md = MessageDigest.getInstance("SHA-256"); byte[] buffer = new byte[8192]; try (InputStream inputStream = Files.newInputStream(filePath)) { @@ -328,37 +634,34 @@ public class FileUploadServiceImpl implements FileUploadService { return sb.toString(); - } catch (NoSuchAlgorithmException e) { - log.warn("MD5 algorithm not available, skipping checksum", e); + } catch (NoSuchAlgorithmException | IOException e) { + log.warn("Failed to calculate checksum for file: {}", filePath, e); return "N/A"; } } - private void cleanupChunks(Path chunkDir) throws IOException { - if (Files.exists(chunkDir)) { - try (Stream walk = Files.walk(chunkDir)) { - walk.sorted(Comparator.reverseOrder()) - .map(Path::toFile) - .forEach(File::delete); - log.info("Cleaned up chunk directory: {}", chunkDir); - } + private void createDirectoryIfNotExists(Path directory) throws IOException { + if (!Files.exists(directory)) { + Files.createDirectories(directory); + log.debug("Created directory: {}", directory); } } - private Path getChunkDirectory(Long userId, String uploadId) { - return Paths.get(basePath, "temp", "user", String.valueOf(userId), uploadId); - } + private void setDirectoryPermissions(Path directory) throws IOException { + try { + Set permissions = new HashSet<>(); + permissions.add(PosixFilePermission.OWNER_READ); + permissions.add(PosixFilePermission.OWNER_WRITE); + permissions.add(PosixFilePermission.OWNER_EXECUTE); + permissions.add(PosixFilePermission.GROUP_READ); + permissions.add(PosixFilePermission.GROUP_EXECUTE); + permissions.add(PosixFilePermission.OTHERS_READ); - @Override - public UploadProgressResponse getUploadProgress(String uploadId) { - FileUploadSession session = sessionRepository.findById(uploadId) - .orElseThrow(() -> new UploadSessionNotFoundException(uploadId)); + Files.setPosixFilePermissions(directory, permissions); + log.debug("Set permissions for directory: {}", directory); - return UploadProgressResponse.fromSession(session); + } catch (UnsupportedOperationException e) { + log.debug("Posix permissions not supported on this system"); + } } - - @Override - public void cleanupExpiredSessions() { - log.info("Cleanup not implemented yet"); - } -} +} \ No newline at end of file diff --git a/src/main/resources/application.yaml b/src/main/resources/application.yaml index 4406117..2880f4c 100644 --- a/src/main/resources/application.yaml +++ b/src/main/resources/application.yaml @@ -23,10 +23,13 @@ spring: file: storage: - base-path: ${FILE_STORAGE_PATH:./uploads} - chunk-size: 5242880 # 5MB - temp-ttl-hours: 24 - max-file-size: 10737418240 # 10GB + base-path: ${FILE_STORAGE_PATH:/data/uploads} + chunk-size: ${FILE_CHUNK_SIZE:5242880} # 5MB + max-file-size: ${MAX_FILE_SIZE:10737418240} # 10GB + max-retry-attempts: ${MAX_RETRY_ATTEMPTS:3} + chunk-timeout-ms: ${CHUNK_TIMEOUT_MS:300000} # 5 минут + temp-ttl-hours: ${TEMP_TTL_HOURS:72} # 3 дня + session-expiry-hours: ${SESSION_EXPIRY_HOURS:24} server: port: ${SERVER_PORT:8080}