diff --git a/src/main/java/com/kw/readwith/aws/s3/AmazonS3Manager.java b/src/main/java/com/kw/readwith/aws/s3/AmazonS3Manager.java index 10817e0..c8601c7 100644 --- a/src/main/java/com/kw/readwith/aws/s3/AmazonS3Manager.java +++ b/src/main/java/com/kw/readwith/aws/s3/AmazonS3Manager.java @@ -3,8 +3,11 @@ import com.amazonaws.services.s3.AmazonS3; import com.amazonaws.services.s3.model.GeneratePresignedUrlRequest; import com.amazonaws.HttpMethod; +import com.amazonaws.services.s3.model.ListObjectsV2Request; +import com.amazonaws.services.s3.model.ListObjectsV2Result; import com.amazonaws.services.s3.model.ObjectMetadata; import com.amazonaws.services.s3.model.PutObjectRequest; +import com.amazonaws.services.s3.model.S3Object; import com.kw.readwith.config.AmazonConfig; import lombok.RequiredArgsConstructor; import lombok.extern.slf4j.Slf4j; @@ -15,8 +18,10 @@ import java.io.IOException; import java.io.InputStream; import java.time.Duration; +import java.util.ArrayList; import java.util.Base64; import java.util.Date; +import java.util.List; import java.util.UUID; @Slf4j @@ -58,6 +63,36 @@ public byte[] downloadFile(String keyName) { } } + public T readObject(String keyName, S3ObjectReader reader) throws IOException { + try (S3Object object = amazonS3.getObject(amazonConfig.getBucket(), keyName); + InputStream inputStream = object.getObjectContent()) { + return reader.read(inputStream); + } + } + + public List listKeys(String prefix) { + List keys = new ArrayList<>(); + ListObjectsV2Request request = new ListObjectsV2Request() + .withBucketName(amazonConfig.getBucket()) + .withPrefix(prefix); + + ListObjectsV2Result result; + do { + result = amazonS3.listObjectsV2(request); + result.getObjectSummaries().forEach(summary -> keys.add(summary.getKey())); + request.setContinuationToken(result.getNextContinuationToken()); + } while (result.isTruncated()); + + return keys; + } + + public void deleteKeys(List keyNames) { + if (keyNames == null || keyNames.isEmpty()) { + return; + } + keyNames.forEach(keyName -> amazonS3.deleteObject(amazonConfig.getBucket(), keyName)); + } + public String getObjectUrl(String keyName) { return amazonS3.getUrl(amazonConfig.getBucket(), keyName).toString(); } @@ -118,4 +153,9 @@ private String generateUniqueFileName(String originalFilename) { } return UUID.randomUUID() + ext; } + + @FunctionalInterface + public interface S3ObjectReader { + T read(InputStream inputStream) throws IOException; + } } diff --git a/src/main/java/com/kw/readwith/config/SwaggerConfig.java b/src/main/java/com/kw/readwith/config/SwaggerConfig.java index 60a2232..38ec683 100644 --- a/src/main/java/com/kw/readwith/config/SwaggerConfig.java +++ b/src/main/java/com/kw/readwith/config/SwaggerConfig.java @@ -6,9 +6,12 @@ import io.swagger.v3.oas.models.security.SecurityRequirement; import io.swagger.v3.oas.models.security.SecurityScheme; import io.swagger.v3.oas.models.servers.Server; +import org.springdoc.core.customizers.OpenApiCustomizer; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; +import java.util.List; + @Configuration public class SwaggerConfig { @@ -34,4 +37,21 @@ public OpenAPI melissaServerAPI() { .addSecurityItem(securityRequirement) .components(components); } + + @Bean + public OpenApiCustomizer hideLegacyApiAliases() { + return openApi -> { + if (openApi.getPaths() == null) { + return; + } + + List legacyAliases = openApi.getPaths().keySet().stream() + .filter(path -> path.startsWith("/api/")) + .filter(path -> !path.startsWith("/api/v2/")) + .filter(path -> openApi.getPaths().containsKey("/api/v2" + path.substring("/api".length()))) + .toList(); + + legacyAliases.forEach(openApi.getPaths()::remove); + }; + } } diff --git a/src/main/java/com/kw/readwith/dto/admin/ProcessingJobResponseDTO.java b/src/main/java/com/kw/readwith/dto/admin/ProcessingJobResponseDTO.java new file mode 100644 index 0000000..1530ca7 --- /dev/null +++ b/src/main/java/com/kw/readwith/dto/admin/ProcessingJobResponseDTO.java @@ -0,0 +1,57 @@ +package com.kw.readwith.dto.admin; + +import com.kw.readwith.domain.enums.ProcessingJobStatus; +import com.kw.readwith.domain.enums.ProcessingPipelineType; +import com.kw.readwith.domain.processing.ProcessingJob; +import lombok.AllArgsConstructor; +import lombok.Builder; +import lombok.Getter; +import lombok.NoArgsConstructor; + +import java.time.LocalDateTime; + +@Getter +@Builder +@NoArgsConstructor +@AllArgsConstructor +public class ProcessingJobResponseDTO { + private Long id; + private Long bookId; + private String bookTitle; + private ProcessingPipelineType pipelineType; + private String runId; + private String sourceVersion; + private String artifactPath; + private ProcessingJobStatus status; + private String currentStep; + private String failureCode; + private String failureMessage; + private String triggeredBy; + private String ruleVersion; + private String locatorVersion; + private LocalDateTime createdAt; + private LocalDateTime startedAt; + private LocalDateTime finishedAt; + + public static ProcessingJobResponseDTO from(ProcessingJob job) { + return ProcessingJobResponseDTO.builder() + .id(job.getId()) + .bookId(job.getBook().getId()) + .bookTitle(job.getBook().getTitle()) + .pipelineType(job.getPipelineType()) + .runId(job.getRunId()) + .sourceVersion(job.getSourceVersion()) + .artifactPath(job.getArtifactPath()) + .status(job.getStatus()) + .currentStep(job.getCurrentStep()) + .failureCode(job.getFailureCode()) + .failureMessage(job.getFailureMessage()) + .triggeredBy(job.getTriggeredBy()) + .ruleVersion(job.getRuleVersion()) + .locatorVersion(job.getLocatorVersion()) + .createdAt(job.getCreatedAt()) + .startedAt(job.getStartedAt()) + .finishedAt(job.getFinishedAt()) + .build(); + } +} diff --git a/src/main/java/com/kw/readwith/service/AdminService.java b/src/main/java/com/kw/readwith/service/AdminService.java index c44af50..de59276 100644 --- a/src/main/java/com/kw/readwith/service/AdminService.java +++ b/src/main/java/com/kw/readwith/service/AdminService.java @@ -406,28 +406,11 @@ public int uploadRelationshipDeltas(Long bookId, List files) { try { RelationshipUploadDTO dto = objectMapper.readValue(file.getInputStream(), RelationshipUploadDTO.class); - if (dto == null || dto.getChapterIndex() == null) { - throw new GeneralException(ErrorStatus._BAD_REQUEST, "relationship delta payload chapterIndex is required."); + RelationshipDeltaCandidate candidate = resolveRelationshipDeltaCandidate(bookId, book, dto); + if (candidatesByEventId.containsKey(candidate.eventId())) { + throw new GeneralException(ErrorStatus._BAD_REQUEST, "Duplicate relationship delta eventId in request: " + candidate.eventId()); } - validateRelationshipDeltaContract(dto); - - String eventId = requireText(dto.getEventId(), "relationship.eventId"); - int chapterIdx = dto.getChapterIndex(); - int eventIdx = parseEventIdx(eventId, chapterIdx); - - Event event = eventRepository.findByBookIdAndChapterIdxAndEventIdx(bookId, chapterIdx, eventIdx) - .orElseThrow(() -> new GeneralException( - ErrorStatus.EVENT_NOT_FOUND, - String.format("Event not found for book=%d chapter=%d event=%d", bookId, chapterIdx, eventIdx) - )); - - String normalizedEventId = normalizeEventId(eventId, chapterIdx, eventIdx); - if (candidatesByEventId.containsKey(normalizedEventId)) { - throw new GeneralException(ErrorStatus._BAD_REQUEST, "Duplicate relationship delta eventId in request: " + normalizedEventId); - } - - RelationshipDeltaCandidate candidate = buildRelationshipDeltaCandidate(event, book, dto); - candidatesByEventId.put(normalizedEventId, candidate); + candidatesByEventId.put(candidate.eventId(), candidate); } catch (IOException e) { throw new GeneralException(ErrorStatus.JSON_PARSING_ERROR, "Failed to parse relationship delta JSON: " + file.getOriginalFilename()); } @@ -446,6 +429,45 @@ public int uploadRelationshipDeltas(Long bookId, List files) { } } + @Transactional + public int replaceRelationshipDeltaPayload(Long bookId, RelationshipUploadDTO dto, boolean refreshStatus) { + Book book = bookRepository.findById(bookId) + .orElseThrow(() -> new GeneralException(ErrorStatus.BOOK_NOT_FOUND)); + + try { + RelationshipDeltaCandidate candidate = resolveRelationshipDeltaCandidate(bookId, book, dto); + replaceRelationshipDelta(candidate.event()); + int savedCount = saveRelationshipDelta(candidate); + if (refreshStatus) { + bookAnalysisStatusService.refreshStatus(bookId); + } + return savedCount; + } catch (GeneralException e) { + markAnalysisRejectedIfNeeded(book, e); + throw e; + } + } + + private RelationshipDeltaCandidate resolveRelationshipDeltaCandidate(Long bookId, Book book, RelationshipUploadDTO dto) { + if (dto == null || dto.getChapterIndex() == null) { + throw new GeneralException(ErrorStatus._BAD_REQUEST, "relationship delta payload chapterIndex is required."); + } + validateRelationshipDeltaContract(dto); + + String eventId = requireText(dto.getEventId(), "relationship.eventId"); + int chapterIdx = dto.getChapterIndex(); + int eventIdx = parseEventIdx(eventId, chapterIdx); + + Event event = eventRepository.findByBookIdAndChapterIdxAndEventIdx(bookId, chapterIdx, eventIdx) + .orElseThrow(() -> new GeneralException( + ErrorStatus.EVENT_NOT_FOUND, + String.format("Event not found for book=%d chapter=%d event=%d", bookId, chapterIdx, eventIdx) + )); + + String normalizedEventId = normalizeEventId(eventId, chapterIdx, eventIdx); + return buildRelationshipDeltaCandidate(normalizedEventId, event, book, dto); + } + private void validateRelationshipDeltaContract(RelationshipUploadDTO dto) { String contractVersion = requireText(dto.getContractVersion(), "relationship.contractVersion"); if (!RELATIONSHIP_DELTA_CONTRACT_VERSION.equals(contractVersion)) { @@ -453,10 +475,10 @@ private void validateRelationshipDeltaContract(RelationshipUploadDTO dto) { } } - private RelationshipDeltaCandidate buildRelationshipDeltaCandidate(Event event, Book book, RelationshipUploadDTO dto) { + private RelationshipDeltaCandidate buildRelationshipDeltaCandidate(String eventId, Event event, Book book, RelationshipUploadDTO dto) { List stats = buildRelationshipNodeWeights(event, book, dto.getNodeWeights()); List edges = buildRelationshipDeltaEdges(event, book, dto.getItems()); - return new RelationshipDeltaCandidate(event, stats, edges); + return new RelationshipDeltaCandidate(eventId, event, stats, edges); } private List buildRelationshipNodeWeights(Event event, Book book, Map nodeWeightsMap) { @@ -572,6 +594,7 @@ private int saveRelationshipDelta(RelationshipDeltaCandidate candidate) { } private record RelationshipDeltaCandidate( + String eventId, Event event, List stats, List edges diff --git a/src/main/java/com/kw/readwith/service/RelationshipDeltaImportJobDispatcher.java b/src/main/java/com/kw/readwith/service/RelationshipDeltaImportJobDispatcher.java new file mode 100644 index 0000000..c6d94c1 --- /dev/null +++ b/src/main/java/com/kw/readwith/service/RelationshipDeltaImportJobDispatcher.java @@ -0,0 +1,17 @@ +package com.kw.readwith.service; + +import lombok.RequiredArgsConstructor; +import org.springframework.scheduling.annotation.Async; +import org.springframework.stereotype.Service; + +@Service +@RequiredArgsConstructor +public class RelationshipDeltaImportJobDispatcher { + + private final RelationshipDeltaImportJobService relationshipDeltaImportJobService; + + @Async("normalizationJobExecutor") + public void dispatch(Long jobId) { + relationshipDeltaImportJobService.execute(jobId); + } +} diff --git a/src/main/java/com/kw/readwith/service/RelationshipDeltaImportJobService.java b/src/main/java/com/kw/readwith/service/RelationshipDeltaImportJobService.java new file mode 100644 index 0000000..3624ce1 --- /dev/null +++ b/src/main/java/com/kw/readwith/service/RelationshipDeltaImportJobService.java @@ -0,0 +1,419 @@ +package com.kw.readwith.service; + +import com.kw.readwith.apiPayload.code.status.ErrorStatus; +import com.kw.readwith.apiPayload.exception.GeneralException; +import com.kw.readwith.aws.s3.AmazonS3Manager; +import com.kw.readwith.config.ArtifactStorageProperties; +import com.kw.readwith.domain.Book; +import com.kw.readwith.domain.enums.ProcessingJobLogLevel; +import com.kw.readwith.domain.enums.ProcessingJobStatus; +import com.kw.readwith.domain.enums.ProcessingPipelineType; +import com.kw.readwith.domain.processing.ProcessingJob; +import com.kw.readwith.domain.processing.ProcessingJobLog; +import com.kw.readwith.dto.admin.ProcessingJobLogResponseDTO; +import com.kw.readwith.dto.admin.ProcessingJobResponseDTO; +import com.kw.readwith.dto.admin.RelationshipUploadDTO; +import com.kw.readwith.repository.BookRepository; +import com.kw.readwith.repository.ProcessingJobLogRepository; +import com.kw.readwith.repository.ProcessingJobRepository; +import lombok.RequiredArgsConstructor; +import lombok.extern.slf4j.Slf4j; +import org.springframework.stereotype.Service; +import org.springframework.transaction.PlatformTransactionManager; +import org.springframework.transaction.TransactionDefinition; +import org.springframework.transaction.annotation.Transactional; +import org.springframework.transaction.support.TransactionTemplate; +import org.springframework.web.multipart.MultipartFile; +import tools.jackson.core.JacksonException; +import tools.jackson.databind.ObjectMapper; + +import java.io.IOException; +import java.util.EnumSet; +import java.util.HashSet; +import java.util.List; +import java.util.Map; +import java.util.Set; +import java.util.UUID; +import java.util.regex.Matcher; +import java.util.regex.Pattern; + +@Slf4j +@Service +@RequiredArgsConstructor +public class RelationshipDeltaImportJobService { + + private static final String SOURCE_VERSION = "relationship-delta-v1"; + private static final String TRIGGERED_BY = "RELATIONSHIP_DELTA_UPLOAD"; + private static final Pattern EVENT_ID_PATTERN = Pattern.compile("^ch(\\d+)-e(\\d+)$"); + + private final BookRepository bookRepository; + private final ProcessingJobRepository processingJobRepository; + private final ProcessingJobLogRepository processingJobLogRepository; + private final AdminService adminService; + private final BookAnalysisStatusService bookAnalysisStatusService; + private final AmazonS3Manager amazonS3Manager; + private final ArtifactStorageProperties artifactStorageProperties; + private final ObjectMapper objectMapper; + private final PlatformTransactionManager transactionManager; + + @Transactional + public ProcessingJobResponseDTO queueRelationshipDeltaImport(Long bookId, List files) { + if (files == null || files.isEmpty()) { + throw new GeneralException(ErrorStatus._BAD_REQUEST, "No relationship delta files provided."); + } + + Book book = bookRepository.findById(bookId) + .orElseThrow(() -> new GeneralException(ErrorStatus.BOOK_NOT_FOUND)); + + processingJobRepository.findFirstByBookIdAndPipelineTypeAndStatusInOrderByCreatedAtDesc( + bookId, + ProcessingPipelineType.AI_ANALYSIS, + EnumSet.of(ProcessingJobStatus.QUEUED, ProcessingJobStatus.PROCESSING) + ) + .ifPresent(existing -> { + throw new GeneralException(ErrorStatus._BAD_REQUEST, "An AI analysis job is already active for this book."); + }); + + String runId = "relationship-delta-" + UUID.randomUUID(); + String artifactRoot = buildArtifactRoot(bookId, runId); + S3StagingResult stagingResult = stageFiles(artifactRoot, files); + + try { + ProcessingJob job = ProcessingJob.builder() + .book(book) + .pipelineType(ProcessingPipelineType.AI_ANALYSIS) + .runId(runId) + .sourceVersion(SOURCE_VERSION) + .artifactPath(artifactRoot) + .status(ProcessingJobStatus.QUEUED) + .currentStep("queued") + .triggeredBy(TRIGGERED_BY) + .build(); + + ProcessingJob savedJob = processingJobRepository.save(job); + writeLog(savedJob, ProcessingJobLogLevel.INFO, "queued", "Relationship delta import job has been queued.", Map.of( + "fileCount", stagingResult.fileCount(), + "artifactRoot", artifactRoot, + "inputPrefix", privateKey(inputPrefix(artifactRoot)) + )); + return ProcessingJobResponseDTO.from(savedJob); + } catch (RuntimeException e) { + cleanupStagedFiles(artifactRoot); + throw e; + } + } + + @Transactional(readOnly = true) + public ProcessingJobResponseDTO getRelationshipDeltaImportJob(Long jobId) { + ProcessingJob job = findAnalysisJob(jobId); + return ProcessingJobResponseDTO.from(job); + } + + @Transactional(readOnly = true) + public List getRelationshipDeltaImportJobLogs(Long jobId) { + ProcessingJob job = findAnalysisJob(jobId); + return processingJobLogRepository.findAllByJobIdOrderBySeqAsc(jobId).stream() + .map(logEntry -> ProcessingJobLogResponseDTO.builder() + .id(logEntry.getId()) + .jobId(job.getId()) + .bookTitle(job.getBook().getTitle()) + .seq(logEntry.getSeq()) + .level(logEntry.getLevel()) + .step(logEntry.getStep()) + .message(logEntry.getMessage()) + .payloadJson(logEntry.getPayloadJson()) + .createdAt(logEntry.getCreatedAt()) + .build()) + .toList(); + } + + public void execute(Long jobId) { + try { + RelationshipDeltaExecutionContext context = transitionToProcessing(jobId); + int processedFileCount = processArtifactRoot(jobId, context.bookId(), context.artifactRoot()); + completeSuccess(jobId, processedFileCount); + } catch (Exception e) { + log.error("Relationship delta import job failed. jobId={}", jobId, e); + completeFailure(jobId, e); + } + } + + private S3StagingResult stageFiles(String artifactRoot, List files) { + int storedCount = 0; + + for (int i = 0; i < files.size(); i++) { + MultipartFile file = files.get(i); + if (file == null || file.isEmpty()) { + continue; + } + + String key = privateKey(inputPrefix(artifactRoot) + "/" + safeFileName(file.getOriginalFilename(), i + 1)); + try { + amazonS3Manager.uploadFile(key, file); + } catch (RuntimeException e) { + cleanupStagedFiles(artifactRoot); + throw new GeneralException(ErrorStatus._INTERNAL_SERVER_ERROR, "Failed to stage relationship delta files to S3."); + } + storedCount++; + } + + if (storedCount == 0) { + throw new GeneralException(ErrorStatus._BAD_REQUEST, "No non-empty relationship delta files provided."); + } + return new S3StagingResult(artifactRoot, storedCount); + } + + private int processArtifactRoot(Long jobId, Long bookId, String artifactRoot) { + List keys = listStagedInputKeys(artifactRoot); + if (keys.isEmpty()) { + throw new GeneralException(ErrorStatus._INTERNAL_SERVER_ERROR, "Relationship delta staged input files are missing."); + } + + Set seenEventIds = new HashSet<>(); + int processedFileCount = 0; + + for (String key : keys) { + String fileName = fileNameFromKey(key); + advanceStep(jobId, "processing_file", "Processing relationship delta file.", Map.of( + "fileName", fileName, + "processedFileCount", processedFileCount + )); + + RelationshipUploadDTO dto = readRelationshipDeltaFile(key, fileName); + String eventKey = resolveEventKey(dto, fileName); + if (!seenEventIds.add(eventKey)) { + throw new GeneralException(ErrorStatus._BAD_REQUEST, "Duplicate relationship delta eventId in job: " + eventKey); + } + + int savedCount = adminService.replaceRelationshipDeltaPayload(bookId, dto, false); + processedFileCount++; + writeLogInNewTransaction(jobId, ProcessingJobLogLevel.INFO, "file_processed", "Relationship delta file has been imported.", Map.of( + "fileName", fileName, + "eventId", eventKey, + "savedCount", savedCount, + "processedFileCount", processedFileCount + )); + } + + return processedFileCount; + } + + private RelationshipUploadDTO readRelationshipDeltaFile(String key, String fileName) { + try { + return amazonS3Manager.readObject(key, inputStream -> objectMapper.readValue(inputStream, RelationshipUploadDTO.class)); + } catch (IOException e) { + throw new GeneralException(ErrorStatus.JSON_PARSING_ERROR, "Failed to parse relationship delta JSON: " + fileName); + } catch (RuntimeException e) { + throw new GeneralException(ErrorStatus._INTERNAL_SERVER_ERROR, "Failed to read staged relationship delta file: " + fileName); + } + } + + private RelationshipDeltaExecutionContext transitionToProcessing(Long jobId) { + return writableTransaction().execute(status -> { + ProcessingJob job = findAnalysisJob(jobId); + if (job.getStatus() != ProcessingJobStatus.QUEUED) { + throw new GeneralException(ErrorStatus._BAD_REQUEST, "Relationship delta import job is not queued."); + } + if (job.getArtifactPath() == null || job.getArtifactPath().isBlank()) { + throw new GeneralException(ErrorStatus._INTERNAL_SERVER_ERROR, "Relationship delta import artifact root is missing."); + } + + job.markProcessing("processing_files"); + job.getBook().resetAnalysisStatus(); + writeLog(job, ProcessingJobLogLevel.INFO, "processing_files", "Relationship delta import has started.", Map.of( + "runId", job.getRunId(), + "artifactRoot", job.getArtifactPath(), + "inputPrefix", privateKey(inputPrefix(job.getArtifactPath())) + )); + + return new RelationshipDeltaExecutionContext(job.getBook().getId(), job.getArtifactPath()); + }); + } + + private void advanceStep(Long jobId, String step, String message, Map payload) { + writableTransaction().executeWithoutResult(status -> { + ProcessingJob job = findAnalysisJob(jobId); + job.markProcessing(step); + writeLog(job, ProcessingJobLogLevel.INFO, step, message, payload); + }); + } + + private void completeSuccess(Long jobId, int processedFileCount) { + writableTransaction().executeWithoutResult(status -> { + ProcessingJob job = findAnalysisJob(jobId); + bookAnalysisStatusService.refreshStatus(job.getBook().getId()); + job.markReady(job.getArtifactPath(), "completed"); + writeLog(job, ProcessingJobLogLevel.INFO, "completed", "Relationship delta import job completed.", Map.of( + "runId", job.getRunId(), + "artifactRoot", job.getArtifactPath(), + "processedFileCount", processedFileCount + )); + }); + } + + private void completeFailure(Long jobId, Exception e) { + try { + writableTransaction().executeWithoutResult(status -> { + ProcessingJob job = processingJobRepository.findById(jobId) + .orElseThrow(() -> new GeneralException(ErrorStatus._BAD_REQUEST, "Relationship delta import job not found.")); + String currentStep = job.getCurrentStep() == null ? "failed" : job.getCurrentStep(); + String failureMessage = e.getMessage() == null ? e.getClass().getSimpleName() : e.getMessage(); + + job.markFailed(currentStep, resolveFailureCode(e), failureMessage); + writeLog(job, ProcessingJobLogLevel.ERROR, currentStep, "Relationship delta import job failed.", Map.of( + "runId", job.getRunId(), + "artifactRoot", job.getArtifactPath(), + "failureCode", resolveFailureCode(e), + "error", failureMessage + )); + }); + } catch (Exception nested) { + log.error("Relationship delta import job failure state persistence failed. jobId={}", jobId, nested); + } + } + + private ProcessingJob findAnalysisJob(Long jobId) { + ProcessingJob job = processingJobRepository.findById(jobId) + .orElseThrow(() -> new GeneralException(ErrorStatus._BAD_REQUEST, "Relationship delta import job not found.")); + if (job.getPipelineType() != ProcessingPipelineType.AI_ANALYSIS) { + throw new GeneralException(ErrorStatus._BAD_REQUEST, "Job is not an AI analysis job."); + } + return job; + } + + private void writeLogInNewTransaction(Long jobId, ProcessingJobLogLevel level, String step, String message, Map payload) { + writableTransaction().executeWithoutResult(status -> { + ProcessingJob job = findAnalysisJob(jobId); + writeLog(job, level, step, message, payload); + }); + } + + private void writeLog( + ProcessingJob job, + ProcessingJobLogLevel level, + String step, + String message, + Map payload + ) { + int sequence = (int) processingJobLogRepository.countByJobId(job.getId()) + 1; + processingJobLogRepository.save(ProcessingJobLog.builder() + .job(job) + .seq(sequence) + .level(level) + .step(step) + .message(message) + .payloadJson(serializePayload(payload)) + .build()); + } + + private String serializePayload(Map payload) { + if (payload == null || payload.isEmpty()) { + return null; + } + try { + return objectMapper.writeValueAsString(payload); + } catch (JacksonException e) { + return "{\"serializationError\":true}"; + } + } + + private List listStagedInputKeys(String artifactRoot) { + try { + return amazonS3Manager.listKeys(privateKey(inputPrefix(artifactRoot))).stream() + .filter(key -> !key.endsWith("/")) + .sorted() + .toList(); + } catch (RuntimeException e) { + throw new GeneralException(ErrorStatus._INTERNAL_SERVER_ERROR, "Failed to list staged relationship delta files."); + } + } + + private void cleanupStagedFiles(String artifactRoot) { + try { + amazonS3Manager.deleteKeys(listStagedInputKeys(artifactRoot)); + } catch (RuntimeException e) { + log.warn("Failed to clean staged relationship delta files. artifactRoot={}", artifactRoot, e); + } + } + + private String safeFileName(String originalFilename, int index) { + String name = originalFilename == null || originalFilename.isBlank() + ? "relationship-delta-" + index + ".json" + : originalFilename.replace('\\', '/'); + int slashIndex = name.lastIndexOf('/'); + if (slashIndex >= 0) { + name = name.substring(slashIndex + 1); + } + name = name.replaceAll("[^A-Za-z0-9._-]", "_"); + if (name.isBlank() || ".".equals(name) || "..".equals(name)) { + name = "relationship-delta-" + index + ".json"; + } + return String.format("%04d-%s", index, name); + } + + private String resolveEventKey(RelationshipUploadDTO dto, String fileName) { + if (dto == null || dto.getChapterIndex() == null || dto.getEventId() == null || dto.getEventId().isBlank()) { + return fileName; + } + + String eventId = dto.getEventId().trim(); + Matcher matcher = EVENT_ID_PATTERN.matcher(eventId); + if (matcher.matches()) { + return "ch" + Integer.parseInt(matcher.group(1)) + "-e" + Integer.parseInt(matcher.group(2)); + } + if (eventId.matches("\\d+")) { + return "ch" + dto.getChapterIndex() + "-e" + Integer.parseInt(eventId); + } + return "ch" + dto.getChapterIndex() + "-" + eventId; + } + + private String fileNameFromKey(String key) { + int slashIndex = key.lastIndexOf('/'); + return slashIndex >= 0 ? key.substring(slashIndex + 1) : key; + } + + private String buildArtifactRoot(Long bookId, String runId) { + return "books/" + bookId + "/analysis/relationship-delta-jobs/" + runId; + } + + private String inputPrefix(String artifactRoot) { + return artifactRoot + "/input"; + } + + private String privateKey(String relativePath) { + return trimSlashes(artifactStorageProperties.getPrivatePrefix()) + "/" + trimLeadingSlash(relativePath); + } + + private String trimSlashes(String value) { + if (value == null || value.isBlank()) { + return ""; + } + return value.replaceAll("^/+|/+$", ""); + } + + private String trimLeadingSlash(String value) { + if (value == null || value.isBlank()) { + return ""; + } + return value.replaceAll("^/+", ""); + } + + private String resolveFailureCode(Exception exception) { + if (exception instanceof GeneralException generalException) { + return generalException.getErrorReason().getCode(); + } + return "RELATIONSHIP_DELTA_IMPORT_FAILED"; + } + + private TransactionTemplate writableTransaction() { + TransactionTemplate transactionTemplate = new TransactionTemplate(transactionManager); + transactionTemplate.setPropagationBehavior(TransactionDefinition.PROPAGATION_REQUIRES_NEW); + return transactionTemplate; + } + + private record S3StagingResult(String artifactRoot, int fileCount) { + } + + private record RelationshipDeltaExecutionContext(Long bookId, String artifactRoot) { + } +} diff --git a/src/main/java/com/kw/readwith/web/controller/AdminController.java b/src/main/java/com/kw/readwith/web/controller/AdminController.java index 9fd52b6..8d322ec 100644 --- a/src/main/java/com/kw/readwith/web/controller/AdminController.java +++ b/src/main/java/com/kw/readwith/web/controller/AdminController.java @@ -8,16 +8,20 @@ import com.kw.readwith.dto.admin.CharacterDTO; import com.kw.readwith.dto.admin.NormalizationJobResponseDTO; import com.kw.readwith.dto.admin.ProcessingJobLogResponseDTO; +import com.kw.readwith.dto.admin.ProcessingJobResponseDTO; import com.kw.readwith.dto.admin.UnsummarizedItemDTO; import com.kw.readwith.dto.book.BookSummaryDTO; import com.kw.readwith.service.AdminService; import com.kw.readwith.service.AnalysisInputExportService; +import com.kw.readwith.service.RelationshipDeltaImportJobDispatcher; +import com.kw.readwith.service.RelationshipDeltaImportJobService; import com.kw.readwith.service.normalization.NormalizationJobDispatcher; import com.kw.readwith.service.normalization.NormalizationJobService; import io.swagger.v3.oas.annotations.Operation; import io.swagger.v3.oas.annotations.Parameter; import io.swagger.v3.oas.annotations.tags.Tag; import lombok.RequiredArgsConstructor; +import org.springframework.http.HttpStatus; import org.springframework.http.MediaType; import org.springframework.web.bind.annotation.DeleteMapping; import org.springframework.web.bind.annotation.GetMapping; @@ -25,6 +29,7 @@ import org.springframework.web.bind.annotation.PostMapping; import org.springframework.web.bind.annotation.RequestMapping; import org.springframework.web.bind.annotation.RequestParam; +import org.springframework.web.bind.annotation.ResponseStatus; import org.springframework.web.bind.annotation.RestController; import org.springframework.web.multipart.MultipartFile; @@ -40,6 +45,8 @@ public class AdminController { private final AnalysisInputExportService analysisInputExportService; private final NormalizationJobService normalizationJobService; private final NormalizationJobDispatcher normalizationJobDispatcher; + private final RelationshipDeltaImportJobService relationshipDeltaImportJobService; + private final RelationshipDeltaImportJobDispatcher relationshipDeltaImportJobDispatcher; @Operation(summary = "모든 도서 전체 정보 조회", description = "book 테이블의 모든 행들의 모든 칼럼값들을 조회합니다. (관리자용)") @GetMapping("/books") @@ -151,6 +158,36 @@ public ApiResponse uploadRelationshipDeltas( return ApiResponse.onSuccess("Relationship deltas for the book have been successfully uploaded."); } + @Operation( + summary = "Queue relationship delta import job", + description = "Stages relationship-delta-v1 files and imports them asynchronously one file at a time." + ) + @ResponseStatus(HttpStatus.ACCEPTED) + @PostMapping(value = "/books/{bookId}/relationship-delta-jobs", consumes = MediaType.MULTIPART_FORM_DATA_VALUE) + public ApiResponse queueRelationshipDeltaImportJob( + @Parameter(description = "Target book ID", required = true) @PathVariable Long bookId, + @Parameter(description = "relationship-delta-v1 JSON files", required = true) @RequestParam("files") List files) { + ProcessingJobResponseDTO response = relationshipDeltaImportJobService.queueRelationshipDeltaImport(bookId, files); + relationshipDeltaImportJobDispatcher.dispatch(response.getId()); + return ApiResponse.onSuccess(response); + } + + @Operation(summary = "Get relationship delta import job") + @GetMapping("/relationship-delta-jobs/{jobId}") + public ApiResponse getRelationshipDeltaImportJob( + @Parameter(description = "Job ID", required = true) @PathVariable Long jobId) { + ProcessingJobResponseDTO response = relationshipDeltaImportJobService.getRelationshipDeltaImportJob(jobId); + return ApiResponse.onSuccess(response); + } + + @Operation(summary = "Get relationship delta import job logs") + @GetMapping("/relationship-delta-jobs/{jobId}/logs") + public ApiResponse> getRelationshipDeltaImportJobLogs( + @Parameter(description = "Job ID", required = true) @PathVariable Long jobId) { + List response = relationshipDeltaImportJobService.getRelationshipDeltaImportJobLogs(jobId); + return ApiResponse.onSuccess(response); + } + @Operation(summary = "이벤트 관계 삭제", description = "특정 이벤트에 적재된 관계 edge와 node weight를 함께 삭제합니다.") @DeleteMapping("/books/{bookId}/chapters/{chapterIdx}/events/{eventIdx}/relationships") public ApiResponse deleteRelationships( diff --git a/src/main/resources/application.yml b/src/main/resources/application.yml index eca7c8d..38823ac 100644 --- a/src/main/resources/application.yml +++ b/src/main/resources/application.yml @@ -38,6 +38,7 @@ spring: order_updates: true servlet: multipart: + file-size-threshold: 0B max-file-size: 20MB max-request-size: 20MB security: diff --git a/src/test/java/com/kw/readwith/config/SwaggerConfigTest.java b/src/test/java/com/kw/readwith/config/SwaggerConfigTest.java new file mode 100644 index 0000000..9f46147 --- /dev/null +++ b/src/test/java/com/kw/readwith/config/SwaggerConfigTest.java @@ -0,0 +1,29 @@ +package com.kw.readwith.config; + +import io.swagger.v3.oas.models.OpenAPI; +import io.swagger.v3.oas.models.PathItem; +import io.swagger.v3.oas.models.Paths; +import org.junit.jupiter.api.Test; + +import static org.assertj.core.api.Assertions.assertThat; + +class SwaggerConfigTest { + + @Test + void hideLegacyApiAliasesRemovesOnlyPathsWithV2Counterparts() { + OpenAPI openAPI = new OpenAPI().paths(new Paths() + .addPathItem("/api/books", new PathItem()) + .addPathItem("/api/v2/books", new PathItem()) + .addPathItem("/api/admin/books/{bookId}/relationship-delta-jobs", new PathItem()) + .addPathItem("/api/v2/admin/books/{bookId}/relationship-delta-jobs", new PathItem()) + .addPathItem("/api/auth/login", new PathItem())); + + new SwaggerConfig().hideLegacyApiAliases().customise(openAPI); + + assertThat(openAPI.getPaths()).doesNotContainKey("/api/books"); + assertThat(openAPI.getPaths()).doesNotContainKey("/api/admin/books/{bookId}/relationship-delta-jobs"); + assertThat(openAPI.getPaths()).containsKey("/api/v2/books"); + assertThat(openAPI.getPaths()).containsKey("/api/v2/admin/books/{bookId}/relationship-delta-jobs"); + assertThat(openAPI.getPaths()).containsKey("/api/auth/login"); + } +} diff --git a/src/test/java/com/kw/readwith/service/RelationshipDeltaImportJobServiceTest.java b/src/test/java/com/kw/readwith/service/RelationshipDeltaImportJobServiceTest.java new file mode 100644 index 0000000..64fc023 --- /dev/null +++ b/src/test/java/com/kw/readwith/service/RelationshipDeltaImportJobServiceTest.java @@ -0,0 +1,157 @@ +package com.kw.readwith.service; + +import com.kw.readwith.apiPayload.exception.GeneralException; +import com.kw.readwith.aws.s3.AmazonS3Manager; +import com.kw.readwith.config.ArtifactStorageProperties; +import com.kw.readwith.domain.Book; +import com.kw.readwith.domain.enums.ProcessingJobLogLevel; +import com.kw.readwith.domain.enums.ProcessingJobStatus; +import com.kw.readwith.domain.enums.ProcessingPipelineType; +import com.kw.readwith.domain.processing.ProcessingJob; +import com.kw.readwith.domain.processing.ProcessingJobLog; +import com.kw.readwith.repository.BookRepository; +import com.kw.readwith.repository.ProcessingJobLogRepository; +import com.kw.readwith.repository.ProcessingJobRepository; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.extension.ExtendWith; +import org.mockito.ArgumentCaptor; +import org.mockito.Mock; +import org.mockito.junit.jupiter.MockitoExtension; +import org.springframework.mock.web.MockMultipartFile; +import org.springframework.test.util.ReflectionTestUtils; +import org.springframework.transaction.PlatformTransactionManager; +import tools.jackson.databind.ObjectMapper; + +import java.util.Optional; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.eq; +import static org.mockito.Mockito.never; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; + +@ExtendWith(MockitoExtension.class) +class RelationshipDeltaImportJobServiceTest { + + @Mock + private BookRepository bookRepository; + + @Mock + private ProcessingJobRepository processingJobRepository; + + @Mock + private ProcessingJobLogRepository processingJobLogRepository; + + @Mock + private AdminService adminService; + + @Mock + private BookAnalysisStatusService bookAnalysisStatusService; + + @Mock + private AmazonS3Manager amazonS3Manager; + + @Mock + private PlatformTransactionManager transactionManager; + + @Test + void queueRelationshipDeltaImportStagesFilesToS3AndCreatesAnalysisJob() { + Book book = Book.builder() + .title("The Great Gatsby") + .author("F. Scott Fitzgerald") + .language("en") + .build(); + ReflectionTestUtils.setField(book, "id", 17L); + + when(bookRepository.findById(17L)).thenReturn(Optional.of(book)); + when(processingJobRepository.findFirstByBookIdAndPipelineTypeAndStatusInOrderByCreatedAtDesc( + eq(17L), + eq(ProcessingPipelineType.AI_ANALYSIS), + any() + )).thenReturn(Optional.empty()); + when(processingJobRepository.save(any(ProcessingJob.class))).thenAnswer(invocation -> { + ProcessingJob job = invocation.getArgument(0); + ReflectionTestUtils.setField(job, "id", 137L); + return job; + }); + when(processingJobLogRepository.countByJobId(137L)).thenReturn(0L); + + RelationshipDeltaImportJobService service = newService(); + MockMultipartFile file = new MockMultipartFile( + "files", + "chapter1_event1.json", + "application/json", + "{\"contractVersion\":\"relationship-delta-v1\"}".getBytes() + ); + + var response = service.queueRelationshipDeltaImport(17L, java.util.List.of(file)); + + ArgumentCaptor jobCaptor = ArgumentCaptor.forClass(ProcessingJob.class); + ArgumentCaptor logCaptor = ArgumentCaptor.forClass(ProcessingJobLog.class); + ArgumentCaptor keyCaptor = ArgumentCaptor.forClass(String.class); + verify(processingJobRepository).save(jobCaptor.capture()); + verify(processingJobLogRepository).save(logCaptor.capture()); + verify(amazonS3Manager).uploadFile(keyCaptor.capture(), eq(file)); + + ProcessingJob savedJob = jobCaptor.getValue(); + assertThat(response.getId()).isEqualTo(137L); + assertThat(savedJob.getPipelineType()).isEqualTo(ProcessingPipelineType.AI_ANALYSIS); + assertThat(savedJob.getStatus()).isEqualTo(ProcessingJobStatus.QUEUED); + assertThat(savedJob.getCurrentStep()).isEqualTo("queued"); + assertThat(savedJob.getSourceVersion()).isEqualTo("relationship-delta-v1"); + assertThat(savedJob.getRunId()).startsWith("relationship-delta-"); + assertThat(savedJob.getArtifactPath()) + .isEqualTo("books/17/analysis/relationship-delta-jobs/" + savedJob.getRunId()); + assertThat(keyCaptor.getValue()) + .isEqualTo("private/" + savedJob.getArtifactPath() + "/input/0001-chapter1_event1.json"); + assertThat(logCaptor.getValue().getLevel()).isEqualTo(ProcessingJobLogLevel.INFO); + assertThat(logCaptor.getValue().getStep()).isEqualTo("queued"); + assertThat(logCaptor.getValue().getPayloadJson()).contains(savedJob.getArtifactPath()); + } + + @Test + void queueRelationshipDeltaImportRejectsWhenAnalysisJobIsActive() { + Book book = Book.builder().title("Book").author("Author").language("en").build(); + ReflectionTestUtils.setField(book, "id", 17L); + ProcessingJob activeJob = ProcessingJob.builder() + .id(99L) + .book(book) + .pipelineType(ProcessingPipelineType.AI_ANALYSIS) + .status(ProcessingJobStatus.PROCESSING) + .build(); + + when(bookRepository.findById(17L)).thenReturn(Optional.of(book)); + when(processingJobRepository.findFirstByBookIdAndPipelineTypeAndStatusInOrderByCreatedAtDesc( + eq(17L), + eq(ProcessingPipelineType.AI_ANALYSIS), + any() + )).thenReturn(Optional.of(activeJob)); + + RelationshipDeltaImportJobService service = newService(); + MockMultipartFile file = new MockMultipartFile("files", "delta.json", "application/json", "{}".getBytes()); + + assertThatThrownBy(() -> service.queueRelationshipDeltaImport(17L, java.util.List.of(file))) + .isInstanceOf(GeneralException.class) + .hasMessageContaining("already active"); + verify(processingJobRepository, never()).save(any()); + verify(amazonS3Manager, never()).uploadFile(any(), any()); + } + + private RelationshipDeltaImportJobService newService() { + ArtifactStorageProperties artifactStorageProperties = new ArtifactStorageProperties(); + artifactStorageProperties.setPrivatePrefix("private"); + return new RelationshipDeltaImportJobService( + bookRepository, + processingJobRepository, + processingJobLogRepository, + adminService, + bookAnalysisStatusService, + amazonS3Manager, + artifactStorageProperties, + new ObjectMapper(), + transactionManager + ); + } +}