diff --git a/core/src/main/java/com/opencontext/common/CommonResponse.java b/core/src/main/java/com/opencontext/common/CommonResponse.java index ce57a1f..fa5558f 100644 --- a/core/src/main/java/com/opencontext/common/CommonResponse.java +++ b/core/src/main/java/com/opencontext/common/CommonResponse.java @@ -83,6 +83,19 @@ public static CommonResponse error(String message, String errorCode) { ); } + /** + * Static factory method for creating error responses with message only. + */ + public static CommonResponse error(String message) { + return new CommonResponse<>( + false, + null, + message, + "INTERNAL_ERROR", + LocalDateTime.now() + ); + } + /** * Static factory method for creating error responses from ErrorCode enum. */ diff --git a/core/src/main/java/com/opencontext/config/WebConfig.java b/core/src/main/java/com/opencontext/config/WebConfig.java index 8e03e5e..cf6f946 100644 --- a/core/src/main/java/com/opencontext/config/WebConfig.java +++ b/core/src/main/java/com/opencontext/config/WebConfig.java @@ -1,6 +1,8 @@ package com.opencontext.config; +import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; +import org.springframework.web.client.RestTemplate; import org.springframework.web.servlet.config.annotation.CorsRegistry; import org.springframework.web.servlet.config.annotation.WebMvcConfigurer; @@ -15,6 +17,14 @@ public class WebConfig implements WebMvcConfigurer { * CORS configuration. * Allows CORS for communication with frontend in development environment. */ + /** + * RestTemplate bean for HTTP client operations. + */ + @Bean + public RestTemplate restTemplate() { + return new RestTemplate(); + } + @Override public void addCorsMappings(CorsRegistry registry) { registry.addMapping("/api/**") diff --git a/core/src/main/java/com/opencontext/controller/SourceController/DocsSourceController.java b/core/src/main/java/com/opencontext/controller/DocsSourceController.java similarity index 99% rename from core/src/main/java/com/opencontext/controller/SourceController/DocsSourceController.java rename to core/src/main/java/com/opencontext/controller/DocsSourceController.java index 0cacb81..02e1532 100644 --- a/core/src/main/java/com/opencontext/controller/SourceController/DocsSourceController.java +++ b/core/src/main/java/com/opencontext/controller/DocsSourceController.java @@ -1,4 +1,4 @@ -package com.opencontext.controller.SourceController; +package com.opencontext.controller; import com.opencontext.common.CommonResponse; import com.opencontext.common.PageResponse; diff --git a/core/src/main/java/com/opencontext/controller/SourceController.java b/core/src/main/java/com/opencontext/controller/SourceController.java new file mode 100644 index 0000000..c4028b9 --- /dev/null +++ b/core/src/main/java/com/opencontext/controller/SourceController.java @@ -0,0 +1,444 @@ +package com.opencontext.controller; + +import com.opencontext.common.CommonResponse; +import com.opencontext.common.PageResponse; +import com.opencontext.dto.SourceDocumentDto; +import com.opencontext.dto.SourceUploadResponse; +import com.opencontext.entity.SourceDocument; +import com.opencontext.enums.ErrorCode; +import com.opencontext.enums.IngestionStatus; +import com.opencontext.exception.BusinessException; +import com.opencontext.repository.DocumentChunkRepository; +import com.opencontext.repository.SourceDocumentRepository; +import com.opencontext.service.ChunkingService; +import com.opencontext.service.DocumentParsingService; +import com.opencontext.service.EmbeddingService; +import com.opencontext.service.FileStorageService; +import com.opencontext.service.IndexingService; + +import lombok.RequiredArgsConstructor; +import lombok.extern.slf4j.Slf4j; +import org.springframework.beans.factory.annotation.Value; +import org.springframework.data.domain.Page; +import org.springframework.data.domain.PageRequest; +import org.springframework.data.domain.Pageable; +import org.springframework.data.domain.Sort; +import org.springframework.http.*; +import org.springframework.scheduling.annotation.Async; +import org.springframework.transaction.annotation.Transactional; +import org.springframework.web.bind.annotation.*; +import org.springframework.web.client.RestTemplate; +import org.springframework.web.multipart.MultipartFile; + +import java.util.List; +import java.util.UUID; + +/** + * 소스 문서 관리를 위한 REST API 컨트롤러. + * + * REST API controller for managing source documents. + * + * Provides functionalities such as file upload, ingestion pipeline management, and document listing. + */ +@Slf4j +@RestController +@RequestMapping("/api/v1/sources") +@RequiredArgsConstructor +public class SourceController implements DocsSourceController{ + + private final SourceDocumentRepository sourceDocumentRepository; + private final DocumentChunkRepository documentChunkRepository; + private final DocumentParsingService documentParsingService; + private final ChunkingService chunkingService; + private final EmbeddingService embeddingService; + private final IndexingService indexingService; + private final FileStorageService fileStorageService; + private final RestTemplate restTemplate; + + @Value("${app.elasticsearch.url:http://localhost:9200}") + private String elasticsearchUrl; + + @Value("${app.elasticsearch.index:document-chunks}") + private String indexName; + + /** + * 파일 업로드 및 수집 파이프라인 시작 + * + * @param file 업로드할 파일 (multipart/form-data) + * @return 업로드 결과 및 문서 정보 + */ + @Override + @PostMapping("/upload") + public ResponseEntity> uploadFile( + @RequestParam("file") MultipartFile file) { + log.info("File upload requested: filename={}, size={}", file.getOriginalFilename(), file.getSize()); + + try { + // 파일 저장 및 문서 생성 (FileStorageService에서 검증 수행) + SourceDocument sourceDocument = fileStorageService.uploadFileWithMetadata(file); + log.info("File saved successfully: documentId={}, filename={}", + sourceDocument.getId(), sourceDocument.getOriginalFilename()); + + // 비동기 수집 파이프라인 시작 + processIngestionPipeline(sourceDocument.getId()); + + // 응답 생성 + SourceUploadResponse response = SourceUploadResponse.success( + sourceDocument.getId().toString(), + sourceDocument.getOriginalFilename() + ); + + return ResponseEntity.status(HttpStatus.ACCEPTED) + .body(CommonResponse.success(response)); + + } catch (BusinessException e) { + log.error("File upload failed: {}", e.getMessage()); + return ResponseEntity.status(getHttpStatusFromErrorCode(e.getErrorCode())) + .body(CommonResponse.error(e.getErrorCode(), e.getMessage())); + } catch (Exception e) { + log.error("Unexpected error during file upload", e); + return ResponseEntity.internalServerError() + .body(CommonResponse.error("파일 업로드 중 오류가 발생했습니다: " + e.getMessage())); + } + } + + /** + * 업로드된 모든 문서의 최신 상태 목록 조회 + * + * @param page 페이지 번호 (0부터 시작) + * @param size 페이지 크기 + * @param sort 정렬 조건 + * @return 페이지네이션된 문서 목록 + */ + @Override + @GetMapping + public ResponseEntity>> getAllSourceDocuments( + @RequestParam(defaultValue = "0") int page, + @RequestParam(defaultValue = "20") int size, + @RequestParam(defaultValue = "createdAt,desc") String sort) { + log.debug("Getting source documents: page={}, size={}, sort={}", page, size, sort); + + try { + // 정렬 조건 파싱 + Sort sortObj = parseSort(sort); + Pageable pageable = PageRequest.of(page, size, sortObj); + + // 문서 목록 조회 + Page documentPage = sourceDocumentRepository.findAll(pageable); + + // DTO 변환 + List documentDtos = documentPage.getContent().stream() + .map(this::convertToDto) + .toList(); + + PageResponse pageResponse = PageResponse.builder() + .content(documentDtos) + .page(documentPage.getNumber()) + .size(documentPage.getSize()) + .totalElements(documentPage.getTotalElements()) + .totalPages(documentPage.getTotalPages()) + .first(documentPage.isFirst()) + .last(documentPage.isLast()) + .hasNext(documentPage.hasNext()) + .hasPrevious(documentPage.hasPrevious()) + .build(); + + return ResponseEntity.ok(CommonResponse.success(pageResponse)); + + } catch (Exception e) { + log.error("Failed to get source documents", e); + return ResponseEntity.internalServerError() + .body(CommonResponse.error("문서 목록 조회 중 오류가 발생했습니다: " + e.getMessage())); + } + } + + /** + * 특정 문서를 강제로 재수집 + * + * @param sourceId 재수집할 문서 ID + * @return 재수집 시작 응답 + */ + @Override + @PostMapping("/{sourceId}/resync") + public ResponseEntity> resyncSourceDocument(@PathVariable UUID sourceId) { + log.info("Document resync requested: sourceId={}", sourceId); + + try { + // 문서 존재 확인 + SourceDocument sourceDocument = findSourceDocumentById(sourceId); + + // 처리 중인 문서인지 확인 + if (sourceDocument.isProcessing()) { + throw new BusinessException(ErrorCode.RESOURCE_IS_BEING_PROCESSED, + "문서가 이미 처리 중입니다."); + } + + // 상태를 PENDING으로 초기화 + sourceDocument.updateIngestionStatus(IngestionStatus.PENDING); + sourceDocument.clearErrorMessage(); + sourceDocumentRepository.save(sourceDocument); + + // 비동기 수집 파이프라인 시작 + processIngestionPipeline(sourceId); + + log.info("Document resync started successfully: sourceId={}", sourceId); + + return ResponseEntity.status(HttpStatus.ACCEPTED) + .body(CommonResponse.success("문서 재수집이 시작되었습니다.")); + + } catch (BusinessException e) { + log.error("Document resync failed: sourceId={}, error={}", sourceId, e.getMessage()); + return ResponseEntity.status(getHttpStatusFromErrorCode(e.getErrorCode())) + .body(CommonResponse.error(e.getErrorCode(), e.getMessage())); + } catch (Exception e) { + log.error("Unexpected error during document resync: sourceId={}", sourceId, e); + return ResponseEntity.internalServerError() + .body(CommonResponse.error("문서 재수집 중 오류가 발생했습니다: " + e.getMessage())); + } + } + + /** + * 특정 문서를 시스템에서 영구적으로 삭제 + * + * @param sourceId 삭제할 문서 ID + * @return 삭제 시작 응답 + */ + @Override + @DeleteMapping("/{sourceId}") + public ResponseEntity> deleteSourceDocument(@PathVariable UUID sourceId) { + log.info("Document deletion requested: sourceId={}", sourceId); + + try { + // 문서 존재 확인 + SourceDocument sourceDocument = findSourceDocumentById(sourceId); + + // 처리 중인 문서인지 확인 + if (sourceDocument.isProcessing()) { + throw new BusinessException(ErrorCode.RESOURCE_IS_BEING_PROCESSED, + "처리 중인 문서는 삭제할 수 없습니다."); + } + + // 삭제 상태로 변경 + sourceDocument.updateIngestionStatus(IngestionStatus.DELETING); + sourceDocumentRepository.save(sourceDocument); + + // 비동기 삭제 파이프라인 시작 + processDeletionPipeline(sourceId); + + log.info("Document deletion started successfully: sourceId={}", sourceId); + + return ResponseEntity.status(HttpStatus.ACCEPTED) + .body(CommonResponse.success("문서 삭제가 시작되었습니다.")); + + } catch (BusinessException e) { + log.error("Document deletion failed: sourceId={}, error={}", sourceId, e.getMessage()); + return ResponseEntity.status(getHttpStatusFromErrorCode(e.getErrorCode())) + .body(CommonResponse.error(e.getErrorCode(), e.getMessage())); + } catch (Exception e) { + log.error("Unexpected error during document deletion: sourceId={}", sourceId, e); + return ResponseEntity.internalServerError() + .body(CommonResponse.error("문서 삭제 중 오류가 발생했습니다: " + e.getMessage())); + } + } + + /** + * 문서 수집 파이프라인을 비동기적으로 실행합니다. + */ + @Async + @Transactional + public void processIngestionPipeline(UUID documentId) { + log.info("Starting ingestion pipeline processing: documentId={}", documentId); + + try { + // 1. Update status to PARSING + fileStorageService.updateDocumentStatus(documentId, IngestionStatus.PARSING); + + // 2. Parse document using Unstructured API + var parsedElements = documentParsingService.parseDocument(documentId); + log.info("Document parsing completed: id={}, elements={}", documentId, parsedElements.size()); + + // 3. Update status to CHUNKING + fileStorageService.updateDocumentStatus(documentId, IngestionStatus.CHUNKING); + + // 4. Split into chunks + var structuredChunks = chunkingService.createChunks(documentId, parsedElements); + log.info("Document chunking completed: id={}, chunks={}", documentId, structuredChunks.size()); + + // 5. Update status to EMBEDDING + fileStorageService.updateDocumentStatus(documentId, IngestionStatus.EMBEDDING); + + // 6. Generate embeddings + var embeddedChunks = embeddingService.generateEmbeddings(documentId, structuredChunks); + log.info("Embedding generation completed: id={}, embedded_chunks={}", documentId, embeddedChunks.size()); + + // 7. Update status to INDEXING + fileStorageService.updateDocumentStatus(documentId, IngestionStatus.INDEXING); + + // 8. Store in Elasticsearch and PostgreSQL + indexingService.indexChunks(documentId, embeddedChunks); + log.info("Document indexing completed: id={}", documentId); + + // 9. Update status to COMPLETED + fileStorageService.updateDocumentStatusToCompleted(documentId); + + log.info("Ingestion pipeline completed successfully: documentId={}", documentId); + + } catch (Exception e) { + log.error("Ingestion pipeline failed: documentId={}", documentId, e); + fileStorageService.updateDocumentStatusToError(documentId, e.getMessage()); + throw new BusinessException(ErrorCode.INGESTION_PIPELINE_FAILED, + "Ingestion pipeline failed: " + e.getMessage()); + } + } + + /** + * 문서 삭제 파이프라인을 비동기적으로 실행합니다. + */ + @Async + @Transactional + public void processDeletionPipeline(UUID documentId) { + log.info("Starting deletion pipeline processing: documentId={}", documentId); + + try { + // Get document storage path before deletion + String fileStoragePath = fileStorageService.getDocumentStoragePath(documentId); + + // 1. Delete from Elasticsearch + deleteFromElasticsearch(documentId); + log.info("Deleted document from Elasticsearch: id={}", documentId); + + // 2. Delete chunks from PostgreSQL + deleteChunksFromPostgreSQL(documentId); + log.info("Deleted chunks from PostgreSQL: id={}", documentId); + + // 3. Delete document and file using FileStorageService + fileStorageService.deleteDocument(documentId); + log.info("Deleted document and file: id={}, path={}", documentId, fileStoragePath); + + log.info("Deletion pipeline completed successfully: documentId={}", documentId); + + } catch (Exception e) { + log.error("Deletion pipeline failed: documentId={}", documentId, e); + throw new BusinessException(ErrorCode.DELETION_PIPELINE_FAILED, + "Deletion pipeline failed: " + e.getMessage()); + } + } + + + + /** + * Elasticsearch에서 문서 관련 청크들을 삭제합니다. + */ + private void deleteFromElasticsearch(UUID documentId) { + try { + // 문서 ID로 쿼리하여 관련 청크들을 삭제 + String query = String.format(""" + { + "query": { + "term": { + "document_id": "%s" + } + } + } + """, documentId.toString()); + + HttpHeaders headers = new HttpHeaders(); + headers.setContentType(MediaType.APPLICATION_JSON); + HttpEntity requestEntity = new HttpEntity<>(query, headers); + + ResponseEntity response = restTemplate.exchange( + elasticsearchUrl + "/" + indexName + "/_delete_by_query", + HttpMethod.POST, + requestEntity, + String.class + ); + + log.debug("Elasticsearch deletion response: {}", response.getBody()); + + } catch (Exception e) { + log.error("Failed to delete from Elasticsearch: documentId={}", documentId, e); + throw new BusinessException(ErrorCode.EXTERNAL_API_ERROR, + "Failed to delete from Elasticsearch: " + e.getMessage()); + } + } + + /** + * PostgreSQL에서 문서 청크들을 삭제합니다. + */ + private void deleteChunksFromPostgreSQL(UUID documentId) { + try { + int deletedCount = documentChunkRepository.deleteBySourceDocumentId(documentId); + log.debug("Deleted {} chunks from PostgreSQL for document: {}", deletedCount, documentId); + + } catch (Exception e) { + log.error("Failed to delete chunks from PostgreSQL: documentId={}", documentId, e); + throw new BusinessException(ErrorCode.DATABASE_ERROR, + "Failed to delete chunks from PostgreSQL: " + e.getMessage()); + } + } + + // ===== 헬퍼 메소드들 ===== + + /** + * 정렬 조건 파싱 + */ + private Sort parseSort(String sortParam) { + try { + String[] parts = sortParam.split(","); + if (parts.length == 2) { + String property = parts[0].trim(); + String direction = parts[1].trim(); + return "desc".equalsIgnoreCase(direction) + ? Sort.by(property).descending() + : Sort.by(property).ascending(); + } + return Sort.by(parts[0].trim()).descending(); + } catch (Exception e) { + log.warn("Invalid sort parameter: {}, using default", sortParam); + return Sort.by("createdAt").descending(); + } + } + + /** + * SourceDocument를 DTO로 변환 + */ + private SourceDocumentDto convertToDto(SourceDocument document) { + return SourceDocumentDto.builder() + .id(document.getId().toString()) + .originalFilename(document.getOriginalFilename()) + .fileType(document.getFileType()) + .fileSize(document.getFileSize()) + .ingestionStatus(document.getIngestionStatus().name()) + .errorMessage(document.getErrorMessage()) + .lastIngestedAt(document.getLastIngestedAt()) + .createdAt(document.getCreatedAt()) + .updatedAt(document.getUpdatedAt()) + .build(); + } + + /** + * ID로 SourceDocument 조회 (없으면 예외 발생) + */ + private SourceDocument findSourceDocumentById(UUID sourceId) { + return sourceDocumentRepository.findById(sourceId) + .orElseThrow(() -> new BusinessException(ErrorCode.SOURCE_DOCUMENT_NOT_FOUND, + "해당 ID의 문서를 찾을 수 없습니다: " + sourceId)); + } + + /** + * ErrorCode에 따른 HTTP 상태 코드 반환 + */ + private HttpStatus getHttpStatusFromErrorCode(ErrorCode errorCode) { + return switch (errorCode) { + case VALIDATION_FAILED -> HttpStatus.BAD_REQUEST; + case INSUFFICIENT_PERMISSION -> HttpStatus.FORBIDDEN; + case SOURCE_DOCUMENT_NOT_FOUND -> HttpStatus.NOT_FOUND; + case DUPLICATE_FILE_UPLOADED -> HttpStatus.CONFLICT; + case RESOURCE_IS_BEING_PROCESSED -> HttpStatus.CONFLICT; + case PAYLOAD_TOO_LARGE -> HttpStatus.PAYLOAD_TOO_LARGE; + case UNSUPPORTED_MEDIA_TYPE -> HttpStatus.UNSUPPORTED_MEDIA_TYPE; + default -> HttpStatus.INTERNAL_SERVER_ERROR; + }; + } + +} diff --git a/core/src/main/java/com/opencontext/controller/SourceController/SourceController.java b/core/src/main/java/com/opencontext/controller/SourceController/SourceController.java deleted file mode 100644 index 4746cf0..0000000 --- a/core/src/main/java/com/opencontext/controller/SourceController/SourceController.java +++ /dev/null @@ -1,96 +0,0 @@ -package com.opencontext.controller.SourceController; - -import com.opencontext.common.CommonResponse; -import com.opencontext.common.PageResponse; -import com.opencontext.dto.SourceDocumentDto; -import com.opencontext.dto.SourceUploadResponse; -import com.opencontext.service.SourceDocumentService; -import jakarta.validation.constraints.NotNull; -import lombok.RequiredArgsConstructor; -import lombok.extern.slf4j.Slf4j; -import org.springframework.data.domain.Page; -import org.springframework.data.domain.PageRequest; -import org.springframework.data.domain.Pageable; -import org.springframework.data.domain.Sort; -import org.springframework.http.HttpStatus; -import org.springframework.http.ResponseEntity; -import org.springframework.web.bind.annotation.RequestMapping; -import org.springframework.web.bind.annotation.RestController; -import org.springframework.web.multipart.MultipartFile; - -import java.util.UUID; - -/** - * REST controller implementing source document management APIs. - * - * This controller implements the DocsSourceController interface and provides - * clean, Lombok-optimized code for document ingestion pipeline management. - */ -@Slf4j -@RestController -@RequestMapping("/api/v1/sources") -@RequiredArgsConstructor -public class SourceController implements DocsSourceController { - - private final SourceDocumentService sourceDocumentService; - - @Override - public ResponseEntity> uploadFile(@NotNull MultipartFile file) { - log.info("Source document upload request: filename={}, size={}, contentType={}", - file.getOriginalFilename(), file.getSize(), file.getContentType()); - - SourceUploadResponse response = sourceDocumentService.uploadSourceDocument(file); - - log.info("Source document upload accepted: id={}, filename={}", - response.getSourceDocumentId(), response.getOriginalFilename()); - - return ResponseEntity.status(HttpStatus.ACCEPTED) - .body(CommonResponse.success(response, "요청이 성공적으로 처리되었습니다.")); - } - - @Override - public ResponseEntity>> getAllSourceDocuments( - int page, int size, String sort) { - - log.debug("Source documents list request: page={}, size={}, sort={}", page, size, sort); - - // Parse sort parameter - String[] sortParts = sort.split(","); - String sortProperty = sortParts[0]; - Sort.Direction direction = sortParts.length > 1 && "desc".equals(sortParts[1]) - ? Sort.Direction.DESC : Sort.Direction.ASC; - - Pageable pageable = PageRequest.of(page, size, Sort.by(direction, sortProperty)); - - Page documents = sourceDocumentService.getAllSourceDocuments(pageable); - PageResponse pageResponse = PageResponse.from(documents); - - return ResponseEntity.ok(CommonResponse.success(pageResponse, "요청이 성공적으로 처리되었습니다.")); - } - - @Override - public ResponseEntity> resyncSourceDocument(UUID sourceId) { - log.info("Source document resync request: id={}", sourceId); - - sourceDocumentService.resyncSourceDocument(sourceId); - - return ResponseEntity.status(HttpStatus.ACCEPTED) - .body(CommonResponse.success( - "Document re-ingestion has been queued successfully.", - "요청이 성공적으로 처리되었습니다." - )); - } - - @Override - public ResponseEntity> deleteSourceDocument(UUID sourceId) { - log.info("Source document deletion request: id={}", sourceId); - - sourceDocumentService.deleteSourceDocument(sourceId); - - return ResponseEntity.status(HttpStatus.ACCEPTED) - .body(CommonResponse.success( - "Document deletion has been queued successfully.", - "요청이 성공적으로 처리되었습니다." - )); - } -} \ No newline at end of file diff --git a/core/src/main/java/com/opencontext/dto/StructuredChunk.java b/core/src/main/java/com/opencontext/dto/StructuredChunk.java index 6a2f4e1..24a81eb 100644 --- a/core/src/main/java/com/opencontext/dto/StructuredChunk.java +++ b/core/src/main/java/com/opencontext/dto/StructuredChunk.java @@ -2,14 +2,14 @@ import io.swagger.v3.oas.annotations.media.Schema; import jakarta.validation.constraints.NotBlank; -import jakarta.validation.constraints.NotNull; import lombok.AccessLevel; import lombok.AllArgsConstructor; import lombok.Builder; import lombok.Getter; import lombok.NoArgsConstructor; -import java.util.UUID; +import java.util.List; +import java.util.Map; /** * Structured document chunk for Elasticsearch indexing and storage. @@ -17,55 +17,66 @@ */ @Schema(description = "Elasticsearch document structure for structured chunks") @Getter -@Builder +@Builder(toBuilder = true) @NoArgsConstructor(access = AccessLevel.PROTECTED) @AllArgsConstructor public class StructuredChunk { /** - * Unique identifier for the chunk, identical to PostgreSQL document_chunks.id. - * This maintains data consistency between relational and search stores. + * Unique identifier for the chunk. */ @Schema(description = "Unique chunk identifier", requiredMode = Schema.RequiredMode.REQUIRED) - @NotNull - private UUID chunkId; + @NotBlank + private String chunkId; /** * UUID of the source document this chunk belongs to. - * Used for filtering and document-level operations. */ @Schema(description = "Source document identifier", requiredMode = Schema.RequiredMode.REQUIRED) - @NotNull - private UUID sourceDocumentId; - - /** - * Original filename of the source document. - * Provides context for users to identify document source. - */ - @Schema(description = "Original filename of source document", example = "spring-security-guide.pdf") - private String originalFilename; + @NotBlank + private String documentId; /** * The actual text content of this chunk. - * This is the primary searchable content processed by Korean Nori analyzer. */ @Schema(description = "Text content of the chunk", requiredMode = Schema.RequiredMode.REQUIRED) @NotBlank private String content; /** - * 1024-dimensional embedding vector for semantic search. - * Generated by Qwen3-Embedding-0.6B model via Ollama. - * Uses float[] for optimal performance with Elasticsearch dense_vector type. + * Title or heading of the section this chunk belongs to. + */ + @Schema(description = "Section title or heading") + private String title; + + /** + * Hierarchy level of this chunk (1 = root, 2 = first level, etc.). + */ + @Schema(description = "Hierarchy level in document structure") + private Integer hierarchyLevel; + + /** + * ID of the parent chunk if this is a sub-chunk. + */ + @Schema(description = "Parent chunk identifier for hierarchical structure") + private String parentChunkId; + + /** + * Type of the original document element (Title, Header, NarrativeText, etc.). + */ + @Schema(description = "Original element type from document parsing") + private String elementType; + + /** + * Embedding vector for semantic search. + * Generated by embedding model via Ollama. */ - @Schema(description = "Embedding vector for semantic search (1024 dimensions)") - private float[] embedding; + @Schema(description = "Embedding vector for semantic search") + private List embedding; /** - * Metadata containing hierarchical and contextual information. - * Separated into its own class following PRD coding standards. + * Additional metadata from document parsing. */ - @Schema(description = "Chunk metadata including hierarchy and context") - @NotNull - private ChunkMetadata metadata; + @Schema(description = "Additional metadata from document parsing") + private Map metadata; } \ No newline at end of file diff --git a/core/src/main/java/com/opencontext/entity/DocumentChunk.java b/core/src/main/java/com/opencontext/entity/DocumentChunk.java index bba3173..db66106 100644 --- a/core/src/main/java/com/opencontext/entity/DocumentChunk.java +++ b/core/src/main/java/com/opencontext/entity/DocumentChunk.java @@ -35,6 +35,12 @@ public class DocumentChunk { @Column(name = "id") private UUID id; + /** + * Chunk ID as string for consistency with Elasticsearch. + */ + @Column(name = "chunk_id", nullable = false, unique = true, length = 255) + private String chunkId; + /** * Foreign key to the source document this chunk belongs to. * When parent document is deleted, related chunks are also deleted (CASCADE). @@ -44,14 +50,44 @@ public class DocumentChunk { private SourceDocument sourceDocument; /** - * Foreign key to the parent chunk ID for hierarchical structure. + * Source document ID as UUID for direct access. + */ + @Column(name = "source_document_uuid", nullable = false) + private UUID sourceDocumentId; + + /** + * Parent chunk ID as string for consistency with Elasticsearch. + */ + @Column(name = "parent_chunk_id") + private UUID parentChunkId; + + /** + * Foreign key to the parent chunk for hierarchical structure. * This enables reconstruction of document's tree structure. * Root chunks have null parent_chunk_id. */ @ManyToOne(fetch = FetchType.LAZY) - @JoinColumn(name = "parent_chunk_id", foreignKey = @ForeignKey(name = "fk_chunk_parent")) + @JoinColumn(name = "parent_chunk_uuid", foreignKey = @ForeignKey(name = "fk_chunk_parent")) private DocumentChunk parentChunk; + /** + * Title or heading of the section this chunk belongs to. + */ + @Column(name = "title", length = 500) + private String title; + + /** + * Hierarchy level of this chunk in the document structure. + */ + @Column(name = "hierarchy_level", nullable = false) + private Integer hierarchyLevel; + + /** + * Type of the original document element (Title, Header, NarrativeText, etc.). + */ + @Column(name = "element_type", length = 50) + private String elementType; + /** * Order sequence among sibling chunks with the same parent_chunk_id. * Used to maintain the original document structure and order. diff --git a/core/src/main/java/com/opencontext/entity/SourceDocument.java b/core/src/main/java/com/opencontext/entity/SourceDocument.java index bfacbd7..ca0232a 100644 --- a/core/src/main/java/com/opencontext/entity/SourceDocument.java +++ b/core/src/main/java/com/opencontext/entity/SourceDocument.java @@ -125,6 +125,31 @@ public void updateIngestionStatusToError(String errorMessage) { this.errorMessage = errorMessage; } + /** + * Updates the last ingested timestamp. + * + * @param timestamp the timestamp to set + */ + public void updateLastIngestedAt(LocalDateTime timestamp) { + this.lastIngestedAt = timestamp; + } + + /** + * Updates the error message. + * + * @param errorMessage the error message to set + */ + public void updateErrorMessage(String errorMessage) { + this.errorMessage = errorMessage; + } + + /** + * Clears the error message. + */ + public void clearErrorMessage() { + this.errorMessage = null; + } + /** * Checks if the document is currently being processed. * diff --git a/core/src/main/java/com/opencontext/enums/ErrorCode.java b/core/src/main/java/com/opencontext/enums/ErrorCode.java index c76cb77..077d29e 100644 --- a/core/src/main/java/com/opencontext/enums/ErrorCode.java +++ b/core/src/main/java/com/opencontext/enums/ErrorCode.java @@ -41,9 +41,15 @@ public enum ErrorCode { // --- INFRASTRUCTURE/EXTERNAL SERVICES --- INGESTION_PIPELINE_FAILED(HttpStatus.INTERNAL_SERVER_ERROR, "INFRA_001", "Internal error occurred during document processing."), - EXTERNAL_SERVICE_UNAVAILABLE(HttpStatus.SERVICE_UNAVAILABLE, "INFRA_002", "External service is not responding."), - DB_CONNECTION_FAILED(HttpStatus.SERVICE_UNAVAILABLE, "INFRA_003", "Database connection failed."), - ELASTICSEARCH_ERROR(HttpStatus.SERVICE_UNAVAILABLE, "INFRA_004", "Search engine error occurred."); + DELETION_PIPELINE_FAILED(HttpStatus.INTERNAL_SERVER_ERROR, "INFRA_002", "Internal error occurred during document deletion."), + DOCUMENT_PARSING_FAILED(HttpStatus.INTERNAL_SERVER_ERROR, "INFRA_003", "Failed to parse document using external API."), + EMBEDDING_GENERATION_FAILED(HttpStatus.INTERNAL_SERVER_ERROR, "INFRA_004", "Failed to generate embeddings for document chunks."), + INDEXING_FAILED(HttpStatus.INTERNAL_SERVER_ERROR, "INFRA_005", "Failed to index document chunks."), + EXTERNAL_API_ERROR(HttpStatus.SERVICE_UNAVAILABLE, "INFRA_006", "External API call failed."), + DATABASE_ERROR(HttpStatus.INTERNAL_SERVER_ERROR, "INFRA_007", "Database operation failed."), + EXTERNAL_SERVICE_UNAVAILABLE(HttpStatus.SERVICE_UNAVAILABLE, "INFRA_008", "External service is not responding."), + DB_CONNECTION_FAILED(HttpStatus.SERVICE_UNAVAILABLE, "INFRA_009", "Database connection failed."), + ELASTICSEARCH_ERROR(HttpStatus.SERVICE_UNAVAILABLE, "INFRA_010", "Search engine error occurred."); private final HttpStatus httpStatus; private final String code; diff --git a/core/src/main/java/com/opencontext/repository/DocumentChunkRepository.java b/core/src/main/java/com/opencontext/repository/DocumentChunkRepository.java index 72e2ccd..a3f9013 100644 --- a/core/src/main/java/com/opencontext/repository/DocumentChunkRepository.java +++ b/core/src/main/java/com/opencontext/repository/DocumentChunkRepository.java @@ -137,4 +137,13 @@ Integer findMaxSequenceForParent(@Param("sourceDocument") SourceDocument sourceD * @param sourceDocument the source document whose chunks should be deleted */ void deleteBySourceDocument(SourceDocument sourceDocument); + + /** + * Deletes all chunks belonging to a specific source document by ID. + * This is used when a source document is being deleted. + * + * @param sourceDocumentId the source document ID whose chunks should be deleted + * @return the number of deleted chunks + */ + int deleteBySourceDocumentId(UUID sourceDocumentId); } \ No newline at end of file diff --git a/core/src/main/java/com/opencontext/service/ChunkingService.java b/core/src/main/java/com/opencontext/service/ChunkingService.java new file mode 100644 index 0000000..17b82bd --- /dev/null +++ b/core/src/main/java/com/opencontext/service/ChunkingService.java @@ -0,0 +1,267 @@ +package com.opencontext.service; + +import com.opencontext.dto.StructuredChunk; +import lombok.RequiredArgsConstructor; +import lombok.extern.slf4j.Slf4j; +import org.springframework.stereotype.Service; + +import java.util.*; + +/** + * 파싱된 문서 요소를 의미 있는 청크로 분할하는 서비스. + * + * Unstructured API의 결과를 기반으로 계층적 구조를 유지하면서 + * 검색에 최적화된 청크를 생성합니다. + */ +@Slf4j +@Service +@RequiredArgsConstructor +public class ChunkingService { + + private static final int MAX_CHUNK_SIZE = 1000; // 최대 청크 크기 (문자 수) + private static final int CHUNK_OVERLAP = 200; // 청크 간 중복 크기 + + /** + * 파싱된 요소들을 구조화된 청크로 변환합니다. + * + * @param documentId 문서 ID + * @param parsedElements Unstructured API에서 파싱된 요소 목록 + * Service for splitting parsed document elements into meaningful chunks. + * + * Generates search-optimized chunks while maintaining hierarchical structure, + * based on the results from the Unstructured API. + */ +@Slf4j +@Service +@RequiredArgsConstructor +public class ChunkingService { + + private static final int MAX_CHUNK_SIZE = 1000; // Maximum chunk size (number of characters) + private static final int CHUNK_OVERLAP = 200; // Overlap size between chunks + + /** + * Converts parsed elements into structured chunks. + * + * @param documentId Document ID + * @param parsedElements List of elements parsed by the Unstructured API + * @return List of structured chunks + */ + public List createChunks(UUID documentId, List> parsedElements) { + long startTime = System.currentTimeMillis(); + int totalElements = parsedElements.size(); + + log.info("🧩 [CHUNKING] Starting chunking process: documentId={}, elements={}", documentId, totalElements); + + List chunks = new ArrayList<>(); + Stack hierarchyStack = new Stack<>(); + int chunkIndex = 0; + int processedElements = 0; + + log.debug("📄 [CHUNKING] Processing {} elements for hierarchical chunking", totalElements); + + for (Map element : parsedElements) { + processedElements++; + String elementType = (String) element.get("type"); + String text = (String) element.get("text"); + + if (processedElements % 50 == 0 || processedElements == totalElements) { + log.debug("📊 [CHUNKING] Progress: {}/{} elements processed", processedElements, totalElements); + } + + if (text == null || text.trim().isEmpty()) { + log.debug("⚠️ [CHUNKING] Skipping empty element: type={}", elementType); + continue; + } + + // 요소 타입에 따른 처리 + switch (elementType) { + case "Title" -> { + // 새로운 섹션 시작 - 계층 구조 업데이트 + ChunkContext titleContext = new ChunkContext(text, 1, null); + hierarchyStack.clear(); + hierarchyStack.push(titleContext); + + // 제목을 별도 청크로 생성 + chunks.add(createChunk(documentId, chunkIndex++, text, titleContext, element)); + } + case "Header" -> { + // 헤더 레벨 결정 (metadata에서 추출 또는 기본값 사용) + int level = determineHeaderLevel(element); + + // 계층 구조 조정 + adjustHierarchyStack(hierarchyStack, level); + + ChunkContext headerContext = new ChunkContext(text, level, + hierarchyStack.isEmpty() ? null : hierarchyStack.peek()); + hierarchyStack.push(headerContext); + + // 헤더를 별도 청크로 생성 + chunks.add(createChunk(documentId, chunkIndex++, text, headerContext, element)); + } + case "NarrativeText", "ListItem", "Table" -> { + // 현재 계층 구조에서 내용 청크 생성 + ChunkContext currentContext = hierarchyStack.isEmpty() ? null : hierarchyStack.peek(); + + // 긴 텍스트는 여러 청크로 분할 + List textChunks = splitLongText(text); + for (String textChunk : textChunks) { + chunks.add(createChunk(documentId, chunkIndex++, textChunk, currentContext, element)); + } + } + default -> { + // 기타 요소들은 현재 컨텍스트에서 처리 + ChunkContext currentContext = hierarchyStack.isEmpty() ? null : hierarchyStack.peek(); + chunks.add(createChunk(documentId, chunkIndex++, text, currentContext, element)); + } + } + } + + long duration = System.currentTimeMillis() - startTime; + int finalChunkCount = chunks.size(); + + log.info("🎉 [CHUNKING] Chunking completed successfully: documentId={}, elements={}, chunks={}, duration={}ms", + documentId, totalElements, finalChunkCount, duration); + + // 청크 통계 로깅 + if (finalChunkCount > 0) { + double avgChunkLength = chunks.stream() + .mapToInt(c -> c.getContent().length()) + .average() + .orElse(0.0); + + int maxHierarchyLevel = chunks.stream() + .mapToInt(c -> c.getHierarchyLevel() != null ? c.getHierarchyLevel() : 0) + .max() + .orElse(0); + + log.debug("📊 [CHUNKING] Statistics: avgChunkLength={}, maxHierarchyLevel={}, maxChunkSize={}", + Math.round(avgChunkLength), maxHierarchyLevel, MAX_CHUNK_SIZE); + } + + return chunks; + } + + /** + * StructuredChunk 객체를 생성합니다. + */ + private StructuredChunk createChunk(UUID documentId, int chunkIndex, String text, + ChunkContext context, Map element) { + return StructuredChunk.builder() + .documentId(documentId.toString()) + .chunkId(generateChunkId(documentId, chunkIndex)) + .content(text) + .title(context != null ? context.title : null) + .hierarchyLevel(context != null ? context.level : 0) + .parentChunkId(context != null && context.parent != null ? + generateChunkId(documentId, context.parent.chunkIndex) : null) + .elementType((String) element.get("type")) + .metadata(extractMetadata(element)) + .build(); + } + + /** + * 청크 ID를 생성합니다. + */ + private String generateChunkId(UUID documentId, int chunkIndex) { + return documentId.toString() + "-chunk-" + chunkIndex; + } + + /** + * 헤더의 레벨을 결정합니다. + */ + private int determineHeaderLevel(Map element) { + // metadata에서 레벨 정보 추출 시도 + @SuppressWarnings("unchecked") + Map metadata = (Map) element.get("metadata"); + if (metadata != null && metadata.containsKey("category_depth")) { + try { + return ((Number) metadata.get("category_depth")).intValue(); + } catch (Exception e) { + log.debug("Failed to extract header level from metadata", e); + } + } + + // 기본값 반환 + return 2; + } + + /** + * 계층 구조 스택을 조정합니다. + */ + private void adjustHierarchyStack(Stack stack, int newLevel) { + // 새로운 레벨보다 깊거나 같은 레벨의 요소들을 제거 + while (!stack.isEmpty() && stack.peek().level >= newLevel) { + stack.pop(); + } + } + + /** + * 긴 텍스트를 여러 청크로 분할합니다. + */ + private List splitLongText(String text) { + if (text.length() <= MAX_CHUNK_SIZE) { + return List.of(text); + } + + List chunks = new ArrayList<>(); + int start = 0; + + while (start < text.length()) { + int end = Math.min(start + MAX_CHUNK_SIZE, text.length()); + + // 단어 경계에서 자르기 + if (end < text.length()) { + int lastSpace = text.lastIndexOf(' ', end); + if (lastSpace > start) { + end = lastSpace; + } + } + + chunks.add(text.substring(start, end).trim()); + start = Math.max(start + 1, end - CHUNK_OVERLAP); + } + + return chunks; + } + + + + /** + * 요소에서 메타데이터를 추출합니다. + */ + private Map extractMetadata(Map element) { + Map metadata = new HashMap<>(); + + // 기본 메타데이터 복사 + if (element.containsKey("metadata")) { + @SuppressWarnings("unchecked") + Map originalMetadata = (Map) element.get("metadata"); + metadata.putAll(originalMetadata); + } + + // 좌표 정보 추가 + if (element.containsKey("coordinates")) { + metadata.put("coordinates", element.get("coordinates")); + } + + return metadata; + } + + /** + * 청킹 컨텍스트를 관리하는 내부 클래스 + */ + private static class ChunkContext { + final String title; + final int level; + final ChunkContext parent; + final int chunkIndex; + private static int globalChunkIndex = 0; + + ChunkContext(String title, int level, ChunkContext parent) { + this.title = title; + this.level = level; + this.parent = parent; + this.chunkIndex = globalChunkIndex++; + } + } +} diff --git a/core/src/main/java/com/opencontext/service/DocumentParsingService.java b/core/src/main/java/com/opencontext/service/DocumentParsingService.java new file mode 100644 index 0000000..9fdde72 --- /dev/null +++ b/core/src/main/java/com/opencontext/service/DocumentParsingService.java @@ -0,0 +1,183 @@ +package com.opencontext.service; + +import com.opencontext.entity.SourceDocument; +import com.opencontext.enums.ErrorCode; +import com.opencontext.exception.BusinessException; + +import lombok.RequiredArgsConstructor; +import lombok.extern.slf4j.Slf4j; +import org.springframework.beans.factory.annotation.Value; +import org.springframework.http.*; +import org.springframework.stereotype.Service; +import org.springframework.util.LinkedMultiValueMap; +import org.springframework.util.MultiValueMap; +import org.springframework.web.client.RestTemplate; + +import java.io.InputStream; +import java.util.List; +import java.util.Map; +import java.util.UUID; + +/** + * Unstructured API를 사용하여 문서를 파싱하는 서비스. + * + * Service for parsing documents using the Unstructured API. + * + * Converts PDF, Markdown, and text files into structured elements. + */ +@Slf4j +@Service +@RequiredArgsConstructor +public class DocumentParsingService { + + private final FileStorageService fileStorageService; + private final RestTemplate restTemplate; + + @Value("${app.unstructured.api.url:http://localhost:8000}") + private String unstructuredApiUrl; + + /** + * 문서를 파싱하여 구조화된 요소 목록을 반환합니다. + * + * @param documentId 파싱할 문서 ID + * @return 파싱된 요소 목록 (Unstructured API 응답) + */ + public List> parseDocument(UUID documentId) { + long startTime = System.currentTimeMillis(); + log.info("📝 [PARSING] Starting document parsing: documentId={}", documentId); + + log.debug("📖 [PARSING] Step 1/3: Retrieving document metadata: documentId={}", documentId); + SourceDocument document = fileStorageService.getDocument(documentId); + String filename = document.getOriginalFilename(); + String fileType = document.getFileType(); + long fileSize = document.getFileSize(); + + log.info("✅ [PARSING] Document metadata retrieved: filename={}, fileType={}, size={} bytes", + filename, fileType, fileSize); + + try { + // Step 2: MinIO에서 파일 다운로드 + log.debug("☁️ [PARSING] Step 2/3: Downloading file from MinIO: path={}", document.getFileStoragePath()); + InputStream fileStream = fileStorageService.downloadFile(document.getFileStoragePath()); + log.info("✅ [PARSING] File downloaded from storage: filename={}, path={}", + filename, document.getFileStoragePath()); + + // Step 3: Unstructured API 호출 + log.debug("🤖 [PARSING] Step 3/3: Calling Unstructured API: filename={}, fileType={}", filename, fileType); + List> parsedElements = callUnstructuredApi( + fileStream, + filename, + fileType + ); + + long duration = System.currentTimeMillis() - startTime; + log.info("🎉 [PARSING] Document parsing completed successfully: documentId={}, filename={}, elements={}, duration={}ms", + documentId, filename, parsedElements.size(), duration); + + return parsedElements; + + } catch (Exception e) { + long duration = System.currentTimeMillis() - startTime; + log.error("❌ [PARSING] Document parsing failed: documentId={}, filename={}, duration={}ms, error={}", + documentId, filename, duration, e.getMessage(), e); + throw new BusinessException(ErrorCode.DOCUMENT_PARSING_FAILED, + "Document parsing failed: " + e.getMessage()); + } + } + + /** + * Unstructured API를 호출하여 문서를 파싱합니다. + */ + @SuppressWarnings("unchecked") + private List> callUnstructuredApi(InputStream fileStream, + String filename, + String fileType) { + long apiStartTime = System.currentTimeMillis(); + log.debug("🤖 [UNSTRUCTURED-API] Starting API call: filename={}, fileType={}, url={}", + filename, fileType, unstructuredApiUrl); + + try { + // HTTP 헤더 설정 + HttpHeaders headers = new HttpHeaders(); + headers.setContentType(MediaType.MULTIPART_FORM_DATA); + + // 멀티파트 요청 생성 + log.debug("📦 [UNSTRUCTURED-API] Preparing multipart request: filename={}", filename); + MultiValueMap body = new LinkedMultiValueMap<>(); + body.add("files", new InputStreamResource(fileStream, filename)); + + // 파싱 옵션 설정 + body.add("strategy", "auto"); + body.add("coordinates", "true"); + body.add("extract_images_in_pdf", "false"); + body.add("infer_table_structure", "true"); + + log.debug("⚙️ [UNSTRUCTURED-API] Request parameters: strategy=auto, coordinates=true, extract_images_in_pdf=false, infer_table_structure=true"); + + HttpEntity> requestEntity = new HttpEntity<>(body, headers); + + // API 호출 + log.debug("🚀 [UNSTRUCTURED-API] Sending HTTP request: {}/general/v0/general", unstructuredApiUrl); + ResponseEntity>> response = restTemplate.exchange( + unstructuredApiUrl + "/general/v0/general", + HttpMethod.POST, + requestEntity, + (Class>>) (Class) List.class + ); + + long apiDuration = System.currentTimeMillis() - apiStartTime; + + if (response.getStatusCode() != HttpStatus.OK || response.getBody() == null) { + log.error("❌ [UNSTRUCTURED-API] API returned unexpected response: filename={}, status={}, duration={}ms", + filename, response.getStatusCode(), apiDuration); + throw new BusinessException(ErrorCode.EXTERNAL_API_ERROR, + "Unstructured API returned unexpected response"); + } + + List> elements = response.getBody(); + int elementCount = elements.size(); + + log.info("✅ [UNSTRUCTURED-API] API call successful: filename={}, elements={}, duration={}ms, status={}", + filename, elementCount, apiDuration, response.getStatusCode()); + + if (elementCount > 0) { + log.debug("📊 [UNSTRUCTURED-API] Element types found: {}", + elements.stream() + .map(e -> e.get("type")) + .distinct() + .collect(java.util.stream.Collectors.toList())); + } + + return elements; + + } catch (Exception e) { + long apiDuration = System.currentTimeMillis() - apiStartTime; + log.error("❌ [UNSTRUCTURED-API] API call failed: filename={}, duration={}ms, error={}", + filename, apiDuration, e.getMessage(), e); + throw new BusinessException(ErrorCode.EXTERNAL_API_ERROR, + "Failed to call Unstructured API: " + e.getMessage()); + } + } + + /** + * InputStream을 Spring의 Resource로 래핑하는 헬퍼 클래스 + */ + private static class InputStreamResource extends org.springframework.core.io.InputStreamResource { + private final String filename; + + public InputStreamResource(InputStream inputStream, String filename) { + super(inputStream); + this.filename = filename; + } + + @Override + public String getFilename() { + return this.filename; + } + + @Override + public long contentLength() { + return -1; // 알 수 없음을 나타냄 + } + } +} diff --git a/core/src/main/java/com/opencontext/service/EmbeddingService.java b/core/src/main/java/com/opencontext/service/EmbeddingService.java new file mode 100644 index 0000000..62c2f4a --- /dev/null +++ b/core/src/main/java/com/opencontext/service/EmbeddingService.java @@ -0,0 +1,201 @@ +package com.opencontext.service; + +import com.opencontext.dto.StructuredChunk; +import com.opencontext.enums.ErrorCode; +import com.opencontext.exception.BusinessException; +import dev.langchain4j.data.embedding.Embedding; +import dev.langchain4j.data.segment.TextSegment; +import dev.langchain4j.model.embedding.EmbeddingModel; +import dev.langchain4j.model.ollama.OllamaEmbeddingModel; +import dev.langchain4j.model.output.Response; +import lombok.extern.slf4j.Slf4j; +import org.springframework.beans.factory.annotation.Value; +import org.springframework.stereotype.Service; + +import java.util.*; + +/** + * LangChain4j와 Ollama를 사용하여 텍스트 청크의 임베딩 벡터를 생성하는 서비스. + * + * Service for generating embedding vectors for text chunks using LangChain4j and Ollama. + * + * Converts the semantic representation of each chunk into a vector to enable semantic search. + */ +@Slf4j +@Service +public class EmbeddingService { + + private final EmbeddingModel embeddingModel; + + @Value("${app.embedding.batch-size:10}") + private int batchSize; + + public EmbeddingService( + @Value("${app.ollama.api.url:http://localhost:11434}") String ollamaApiUrl, + @Value("${app.ollama.embedding.model:nomic-embed-text}") String modelName) { + + this.embeddingModel = OllamaEmbeddingModel.builder() + .baseUrl(ollamaApiUrl) + .modelName(modelName) + .build(); + + log.info("Initialized LangChain4j OllamaEmbeddingModel with baseUrl: {}, model: {}", + ollamaApiUrl, modelName); + } + + /** + * 구조화된 청크들의 임베딩 벡터를 생성합니다. + * + * @param documentId 문서 ID + * @param structuredChunks 임베딩을 생성할 청크 목록 + * @return 임베딩이 포함된 청크 목록 + */ + public List generateEmbeddings(UUID documentId, List structuredChunks) { + long startTime = System.currentTimeMillis(); + int totalChunks = structuredChunks.size(); + + log.info("🤖 [EMBEDDING] Starting embedding generation: documentId={}, chunks={}", documentId, totalChunks); + + List embeddedChunks = new ArrayList<>(); + int processedChunks = 0; + int batchCount = (int) Math.ceil((double) totalChunks / batchSize); + + log.debug("📦 [EMBEDDING] Processing {} chunks in {} batches (batchSize={})", + totalChunks, batchCount, batchSize); + + // 배치 단위로 임베딩 생성 + for (int i = 0; i < structuredChunks.size(); i += batchSize) { + int endIndex = Math.min(i + batchSize, structuredChunks.size()); + List batch = structuredChunks.subList(i, endIndex); + int currentBatch = (i / batchSize) + 1; + + log.debug("📦 [EMBEDDING] Processing batch {}/{}: chunks {}-{}", + currentBatch, batchCount, i + 1, endIndex); + + long batchStartTime = System.currentTimeMillis(); + List batchResult = processBatchWithLangChain4j(batch); + embeddedChunks.addAll(batchResult); + + processedChunks += batch.size(); + long batchDuration = System.currentTimeMillis() - batchStartTime; + + log.info("✅ [EMBEDDING] Batch {}/{} completed: processed={}/{}, duration={}ms", + currentBatch, batchCount, processedChunks, totalChunks, batchDuration); + } + + long duration = System.currentTimeMillis() - startTime; + int finalEmbeddedCount = embeddedChunks.size(); + + log.info("🎉 [EMBEDDING] Embedding generation completed successfully: documentId={}, chunks={}, duration={}ms", + documentId, finalEmbeddedCount, duration); + + // 임베딩 통계 로깅 + if (finalEmbeddedCount > 0) { + long avgEmbeddingTime = duration / finalEmbeddedCount; + log.debug("📊 [EMBEDDING] Statistics: avgTimePerChunk={}ms, batchSize={}, totalBatches={}", + avgEmbeddingTime, batchSize, batchCount); + } + + return embeddedChunks; + } + + /** + * LangChain4j를 사용하여 청크 배치의 임베딩을 생성합니다. + */ + private List processBatchWithLangChain4j(List batch) { + long batchStartTime = System.currentTimeMillis(); + log.debug("🤖 [LANGCHAIN4J] Processing batch with LangChain4j: chunks={}", batch.size()); + + List result = new ArrayList<>(); + + try { + // 배치 처리를 위한 TextSegment 목록 생성 + List textSegments = new ArrayList<>(); + for (StructuredChunk chunk : batch) { + String textForEmbedding = prepareTextForEmbedding(chunk); + TextSegment segment = TextSegment.from(textForEmbedding); + textSegments.add(segment); + } + + // LangChain4j를 사용한 배치 임베딩 생성 + log.debug("🚀 [LANGCHAIN4J] Calling Ollama embedding model: segments={}", textSegments.size()); + long embeddingStartTime = System.currentTimeMillis(); + Response> response = embeddingModel.embedAll(textSegments); + List embeddings = response.content(); + long embeddingDuration = System.currentTimeMillis() - embeddingStartTime; + + log.info("✅ [LANGCHAIN4J] Ollama embedding completed: segments={}, duration={}ms, vectorDimension={}", + textSegments.size(), embeddingDuration, + embeddings.size() > 0 ? embeddings.get(0).dimension() : 0); + + // 결과 처리 + for (int i = 0; i < batch.size(); i++) { + StructuredChunk chunk = batch.get(i); + Embedding embedding = embeddings.get(i); + + // float[] 벡터를 List로 변환 + List embeddingVector = new ArrayList<>(); + float[] vector = embedding.vector(); + for (float value : vector) { + embeddingVector.add((double) value); + } + + // 임베딩이 포함된 새로운 청크 생성 + StructuredChunk embeddedChunk = StructuredChunk.builder() + .chunkId(chunk.getChunkId()) + .documentId(chunk.getDocumentId()) + .content(chunk.getContent()) + .title(chunk.getTitle()) + .hierarchyLevel(chunk.getHierarchyLevel()) + .parentChunkId(chunk.getParentChunkId()) + .elementType(chunk.getElementType()) + .metadata(chunk.getMetadata()) + .embedding(embeddingVector) + .build(); + + result.add(embeddedChunk); + + log.debug("✅ [LANGCHAIN4J] Embedding processed for chunk: id={}, vectorSize={}, textLength={}", + chunk.getChunkId(), embedding.dimension(), chunk.getContent().length()); + } + + long batchDuration = System.currentTimeMillis() - batchStartTime; + log.info("✅ [LANGCHAIN4J] Batch processing completed: processed={}, duration={}ms, avgPerChunk={}ms", + result.size(), batchDuration, batchDuration / batch.size()); + + } catch (Exception e) { + long batchDuration = System.currentTimeMillis() - batchStartTime; + log.error("❌ [LANGCHAIN4J] Batch processing failed: chunks={}, duration={}ms, error={}", + batch.size(), batchDuration, e.getMessage(), e); + throw new BusinessException(ErrorCode.EMBEDDING_GENERATION_FAILED, + "Failed to generate embeddings: " + e.getMessage()); + } + + return result; + } + + /** + * 임베딩 생성을 위한 텍스트를 준비합니다. + * 제목과 내용을 결합하여 더 풍부한 컨텍스트를 제공합니다. + */ + private String prepareTextForEmbedding(StructuredChunk chunk) { + StringBuilder text = new StringBuilder(); + + log.debug("📝 [EMBEDDING] Preparing text for embedding: chunkId={}", chunk.getChunkId()); + + // 제목이 있으면 추가 + if (chunk.getTitle() != null && !chunk.getTitle().trim().isEmpty()) { + text.append("Title: ").append(chunk.getTitle()).append("\n"); + } + + // 내용 추가 + text.append(chunk.getContent()); + + String finalText = text.toString().trim(); + + log.debug("✅ [EMBEDDING] Text prepared: chunkId={}, finalLength={}, hasTitle={}", + chunk.getChunkId(), finalText.length(), chunk.getTitle() != null); + + return finalText; + } +} diff --git a/core/src/main/java/com/opencontext/service/FileStorageService.java b/core/src/main/java/com/opencontext/service/FileStorageService.java index d73573d..4d17bf2 100644 --- a/core/src/main/java/com/opencontext/service/FileStorageService.java +++ b/core/src/main/java/com/opencontext/service/FileStorageService.java @@ -1,30 +1,137 @@ package com.opencontext.service; import com.opencontext.config.MinIOConfig; +import com.opencontext.dto.SourceDocumentDto; +import com.opencontext.entity.SourceDocument; import com.opencontext.enums.ErrorCode; +import com.opencontext.enums.IngestionStatus; import com.opencontext.exception.BusinessException; +import com.opencontext.repository.DocumentChunkRepository; +import com.opencontext.repository.SourceDocumentRepository; import io.minio.*; import lombok.RequiredArgsConstructor; import lombok.extern.slf4j.Slf4j; +import org.springframework.beans.factory.annotation.Value; +import org.springframework.data.domain.Page; +import org.springframework.data.domain.Pageable; +import org.springframework.http.*; import org.springframework.stereotype.Service; +import org.springframework.transaction.annotation.Transactional; +import org.springframework.web.client.RestTemplate; import org.springframework.web.multipart.MultipartFile; +import java.io.IOException; +import java.io.InputStream; +import java.security.MessageDigest; +import java.security.NoSuchAlgorithmException; import java.time.LocalDateTime; +import java.util.List; import java.util.UUID; /** - * Service for handling basic file storage operations with MinIO. + * Service for handling file storage operations with MinIO and document metadata management. * - * This service provides essential methods for storing and managing - * files in MinIO object storage for the document ingestion pipeline. + * This service provides comprehensive methods for storing and managing + * files in MinIO object storage along with their metadata in PostgreSQL + * for the document ingestion pipeline. */ @Slf4j @Service @RequiredArgsConstructor +@Transactional public class FileStorageService { private final MinioClient minioClient; private final MinIOConfig minioConfig; + private final SourceDocumentRepository sourceDocumentRepository; + private final DocumentChunkRepository documentChunkRepository; + private final RestTemplate restTemplate; + + @Value("${app.elasticsearch.url:http://localhost:9200}") + private String elasticsearchUrl; + + @Value("${app.elasticsearch.index:document-chunks}") + private String indexName; + + // Supported file types for document processing + private static final List SUPPORTED_CONTENT_TYPES = List.of( + "application/pdf", + "text/markdown", + "text/plain" + ); + + /** + * Uploads a file to MinIO storage, creates document metadata, and returns the document. + * + * @param file the multipart file to upload + * @return the created SourceDocument entity + */ + public SourceDocument uploadFileWithMetadata(MultipartFile file) { + String filename = file.getOriginalFilename(); + long fileSize = file.getSize(); + String contentType = file.getContentType(); + + log.info("📤 [UPLOAD] Starting file upload with metadata: filename={}, size={} bytes, contentType={}", + filename, fileSize, contentType); + + long startTime = System.currentTimeMillis(); + + // Validate file + log.debug("📋 [UPLOAD] Step 1/5: Validating file: {}", filename); + validateFile(file); + log.info("✅ [UPLOAD] File validation completed successfully: {}", filename); + + // Calculate file checksum to prevent duplicates + log.debug("🔍 [UPLOAD] Step 2/5: Calculating file checksum: {}", filename); + String fileChecksum = calculateFileChecksum(file); + log.info("✅ [UPLOAD] File checksum calculated: {} -> {}", filename, fileChecksum); + + // Check for duplicate files + log.debug("🔍 [UPLOAD] Step 3/5: Checking for duplicate files: {}", filename); + if (sourceDocumentRepository.existsByFileChecksum(fileChecksum)) { + log.warn("❌ [UPLOAD] Duplicate file upload attempt: filename={}, checksum={}", filename, fileChecksum); + throw new BusinessException(ErrorCode.DUPLICATE_FILE_UPLOADED, + "A file with identical content already exists."); + } + log.info("✅ [UPLOAD] No duplicate files found: {}", filename); + + try { + // Upload file to MinIO + log.debug("☁️ [UPLOAD] Step 4/5: Uploading file to MinIO: {}", filename); + String objectKey = uploadFile(file); + log.info("✅ [UPLOAD] File uploaded to MinIO successfully: {} -> {}", filename, objectKey); + + // Create SourceDocument entity + log.debug("💾 [UPLOAD] Step 5/5: Creating database record: {}", filename); + SourceDocument sourceDocument = SourceDocument.builder() + .originalFilename(file.getOriginalFilename()) + .fileStoragePath(objectKey) + .fileType(determineFileType(file.getContentType())) + .fileSize(file.getSize()) + .fileChecksum(fileChecksum) + .ingestionStatus(IngestionStatus.PENDING) + .build(); + + // Save to database + SourceDocument savedDocument = sourceDocumentRepository.save(sourceDocument); + + long duration = System.currentTimeMillis() - startTime; + log.info("🎉 [UPLOAD] File upload completed successfully: id={}, filename={}, duration={}ms", + savedDocument.getId(), savedDocument.getOriginalFilename(), duration); + + return savedDocument; + + } catch (Exception e) { + long duration = System.currentTimeMillis() - startTime; + log.error("❌ [UPLOAD] File upload failed: filename={}, duration={}ms, error={}", + filename, duration, e.getMessage(), e); + if (e instanceof BusinessException) { + throw e; + } + throw new BusinessException(ErrorCode.FILE_UPLOAD_FAILED, + "Unexpected error during file upload: " + e.getMessage()); + } + } /** * Uploads a file to MinIO storage and returns the object key. @@ -50,18 +157,43 @@ public String uploadFile(MultipartFile file) { minioClient.putObject(putObjectArgs); - log.info("Successfully uploaded file: {} to bucket: {} with key: {}", + log.debug("☁️ [MINIO] File uploaded to MinIO: {} -> bucket:{}, key:{}", file.getOriginalFilename(), minioConfig.getBucketName(), objectKey); return objectKey; } catch (Exception e) { - log.error("Failed to upload file: {}", file.getOriginalFilename(), e); + log.error("❌ [MINIO] MinIO upload failed: filename={}, error={}", + file.getOriginalFilename(), e.getMessage(), e); throw new BusinessException(ErrorCode.FILE_UPLOAD_FAILED, "Failed to upload file to storage: " + e.getMessage()); } } + /** + * Downloads a file from MinIO storage. + * + * @param objectKey the object key of the file to download + * @return InputStream of the file content + */ + public InputStream downloadFile(String objectKey) { + try { + GetObjectArgs getObjectArgs = GetObjectArgs.builder() + .bucket(minioConfig.getBucketName()) + .object(objectKey) + .build(); + + InputStream stream = minioClient.getObject(getObjectArgs); + log.debug("☁️ [MINIO] File downloaded successfully: key={}", objectKey); + return stream; + + } catch (Exception e) { + log.error("❌ [MINIO] File download failed: key={}, error={}", objectKey, e.getMessage(), e); + throw new BusinessException(ErrorCode.FILE_NOT_FOUND, + "Failed to download file: " + e.getMessage()); + } + } + /** * Deletes a file from MinIO storage. * @@ -75,10 +207,10 @@ public void deleteFile(String objectKey) { .build(); minioClient.removeObject(removeObjectArgs); - log.info("Successfully deleted file with key: {}", objectKey); + log.debug("🗑️ [MINIO] File deleted successfully: key={}", objectKey); } catch (Exception e) { - log.error("Failed to delete file with key: {}", objectKey, e); + log.error("❌ [MINIO] File deletion failed: key={}, error={}", objectKey, e.getMessage(), e); throw new BusinessException(ErrorCode.FILE_DELETE_FAILED, "Failed to delete file: " + e.getMessage()); } @@ -122,11 +254,12 @@ private void ensureBucketExists() { .build(); minioClient.makeBucket(makeBucketArgs); - log.info("Created MinIO bucket: {}", minioConfig.getBucketName()); + log.info("☁️ [MINIO] Created MinIO bucket: {}", minioConfig.getBucketName()); } } catch (Exception e) { - log.error("Failed to ensure bucket exists: {}", minioConfig.getBucketName(), e); + log.error("❌ [MINIO] Failed to ensure bucket exists: bucket={}, error={}", + minioConfig.getBucketName(), e.getMessage(), e); throw new BusinessException(ErrorCode.STORAGE_ERROR, "Failed to ensure bucket exists: " + e.getMessage()); } @@ -148,4 +281,351 @@ private String generateObjectKey(String originalFilename) { return String.format("documents/%d/%02d/%02d/%s", now.getYear(), now.getMonthValue(), now.getDayOfMonth(), filename); } + + // ========== Document Metadata Management Methods ========== + + /** + * Retrieves a source document by ID. + * + * @param documentId the document ID + * @return SourceDocument entity + */ + @Transactional(readOnly = true) + public SourceDocument getDocument(UUID documentId) { + log.debug("📖 [QUERY] Retrieving document: id={}", documentId); + + return sourceDocumentRepository.findById(documentId) + .orElseThrow(() -> new BusinessException(ErrorCode.SOURCE_DOCUMENT_NOT_FOUND, + "Document not found: " + documentId)); + } + + /** + * Retrieves all source documents with pagination. + * + * @param pageable pagination parameters + * @return Page of SourceDocumentDto + */ + @Transactional(readOnly = true) + public Page getAllDocuments(Pageable pageable) { + log.debug("📖 [QUERY] Retrieving documents with pagination: page={}, size={}", + pageable.getPageNumber(), pageable.getPageSize()); + + return sourceDocumentRepository.findAllByOrderByCreatedAtDesc(pageable) + .map(this::convertToDto); + } + + /** + * Updates document status. + * + * @param documentId the document ID + * @param status the new ingestion status + */ + public void updateDocumentStatus(UUID documentId, IngestionStatus status) { + sourceDocumentRepository.findById(documentId) + .ifPresent(document -> { + document.updateIngestionStatus(status); + sourceDocumentRepository.save(document); + log.info("📝 [STATUS] Document status updated: id={}, status={}", documentId, status); + }); + } + + /** + * Updates document status to COMPLETED and sets completion timestamp. + * + * @param documentId the document ID + */ + public void updateDocumentStatusToCompleted(UUID documentId) { + sourceDocumentRepository.findById(documentId) + .ifPresent(document -> { + document.updateIngestionStatus(IngestionStatus.COMPLETED); + sourceDocumentRepository.save(document); + log.info("🎉 [STATUS] Document status updated to COMPLETED: id={}", documentId); + }); + } + + /** + * Updates document status to ERROR with error message. + * + * @param documentId the document ID + * @param errorMessage the error message + */ + public void updateDocumentStatusToError(UUID documentId, String errorMessage) { + try { + sourceDocumentRepository.findById(documentId) + .ifPresent(document -> { + document.updateIngestionStatusToError(errorMessage); + sourceDocumentRepository.save(document); + log.error("❌ [STATUS] Document status updated to ERROR: id={}, errorMessage={}", documentId, errorMessage); + }); + } catch (Exception e) { + log.error("❌ [STATUS] Failed to update document status to ERROR: id={}, error={}", + documentId, e.getMessage(), e); + } + } + + /** + * Deletes a document and all its associated data from all storage systems. + * This includes MinIO files, PostgreSQL records, and Elasticsearch indices. + * + * @param documentId the document ID to delete + */ + public void deleteDocument(UUID documentId) { + long startTime = System.currentTimeMillis(); + log.info("🗑️ [DELETE] Starting comprehensive document deletion: id={}", documentId); + + SourceDocument document = getDocument(documentId); + String filename = document.getOriginalFilename(); + String status = document.getIngestionStatus().name(); + + log.info("📝 [DELETE] Document details: filename={}, status={}, size={} bytes", + filename, status, document.getFileSize()); + + // Check if document is currently being processed + if (document.isProcessing()) { + log.warn("⚠️ [DELETE] Cannot delete document in processing state: id={}, status={}", + documentId, status); + throw new BusinessException(ErrorCode.RESOURCE_IS_BEING_PROCESSED, + "Document is currently being processed and cannot be deleted."); + } + + // Update status to DELETING + log.debug("📝 [DELETE] Step 1/4: Updating status to DELETING: {}", filename); + document.updateIngestionStatus(IngestionStatus.DELETING); + sourceDocumentRepository.save(document); + log.info("✅ [DELETE] Status updated to DELETING: {}", filename); + + try { + // Step 2: Delete from Elasticsearch (if exists) + log.debug("🔍 [DELETE] Step 2/4: Deleting from Elasticsearch: {}", filename); + deleteFromElasticsearch(documentId); + log.info("✅ [DELETE] Elasticsearch deletion completed: {}", filename); + + // Step 3: Delete chunks from PostgreSQL + log.debug("💾 [DELETE] Step 3/4: Deleting chunks from PostgreSQL: {}", filename); + int deletedChunks = deleteChunksFromPostgreSQL(documentId); + log.info("✅ [DELETE] PostgreSQL chunks deleted: {} chunks removed for {}", deletedChunks, filename); + + // Step 4: Delete file from MinIO + log.debug("☁️ [DELETE] Step 4/4: Deleting file from MinIO: {}", filename); + deleteFile(document.getFileStoragePath()); + log.info("✅ [DELETE] MinIO file deleted: {} -> {}", filename, document.getFileStoragePath()); + + // Final step: Delete SourceDocument record + log.debug("💾 [DELETE] Final step: Deleting source document record: {}", filename); + sourceDocumentRepository.delete(document); + log.info("✅ [DELETE] Source document record deleted: {}", filename); + + long duration = System.currentTimeMillis() - startTime; + log.info("🎉 [DELETE] Document deletion completed successfully: id={}, filename={}, duration={}ms", + documentId, filename, duration); + + } catch (Exception e) { + long duration = System.currentTimeMillis() - startTime; + log.error("❌ [DELETE] Document deletion failed: id={}, filename={}, duration={}ms, error={}", + documentId, filename, duration, e.getMessage(), e); + + // Try to revert status if possible + try { + SourceDocument updatedDoc = sourceDocumentRepository.findById(documentId).orElse(null); + if (updatedDoc != null) { + updatedDoc.updateIngestionStatusToError("Deletion failed: " + e.getMessage()); + sourceDocumentRepository.save(updatedDoc); + log.info("🔄 [DELETE] Status reverted to ERROR after deletion failure: {}", filename); + } + } catch (Exception revertEx) { + log.error("❌ [DELETE] Failed to revert status after deletion failure: {}", filename, revertEx); + } + + throw new BusinessException(ErrorCode.DELETION_PIPELINE_FAILED, + "Failed to delete document: " + e.getMessage()); + } + } + + /** + * Checks if a document is currently being processed. + * + * @param documentId the document ID + * @return true if document is in processing state + */ + @Transactional(readOnly = true) + public boolean isDocumentProcessing(UUID documentId) { + return sourceDocumentRepository.findById(documentId) + .map(SourceDocument::isProcessing) + .orElse(false); + } + + /** + * Gets the file storage path for a document. + * + * @param documentId the document ID + * @return the file storage path + */ + @Transactional(readOnly = true) + public String getDocumentStoragePath(UUID documentId) { + return sourceDocumentRepository.findById(documentId) + .map(SourceDocument::getFileStoragePath) + .orElse(null); + } + + // ========== Deletion Helper Methods ========== + + /** + * Deletes all chunks associated with a document from Elasticsearch. + * Uses delete-by-query API to remove all chunks with matching document_id. + * + * @param documentId the document ID whose chunks should be deleted + */ + private void deleteFromElasticsearch(UUID documentId) { + try { + log.debug("🔍 [ELASTICSEARCH] Starting deletion for document: {}", documentId); + + // Create delete-by-query request + String deleteQuery = String.format( + "{\"query\": {\"term\": {\"document_id\": \"%s\"}}}", + documentId.toString() + ); + + log.debug("🔍 [ELASTICSEARCH] Delete query: {}", deleteQuery); + + HttpHeaders headers = new HttpHeaders(); + headers.setContentType(MediaType.APPLICATION_JSON); + HttpEntity requestEntity = new HttpEntity<>(deleteQuery, headers); + + String deleteUrl = String.format("%s/%s/_delete_by_query", elasticsearchUrl, indexName); + log.debug("🔍 [ELASTICSEARCH] Delete URL: {}", deleteUrl); + + ResponseEntity response = restTemplate.exchange( + deleteUrl, + HttpMethod.POST, + requestEntity, + String.class + ); + + if (response.getStatusCode().is2xxSuccessful()) { + String responseBody = response.getBody(); + log.info("✅ [ELASTICSEARCH] Document chunks deleted successfully: documentId={}, response={}", + documentId, responseBody); + } else { + log.warn("⚠️ [ELASTICSEARCH] Delete operation returned non-2xx status: documentId={}, status={}, response={}", + documentId, response.getStatusCode(), response.getBody()); + } + + } catch (Exception e) { + // Log the error but don't fail the entire deletion process + // This handles cases where the document was never indexed (ERROR status) + if (e.getMessage() != null && e.getMessage().contains("index_not_found_exception")) { + log.info("📝 [ELASTICSEARCH] Index not found - document was likely never indexed: documentId={}", documentId); + } else { + log.warn("⚠️ [ELASTICSEARCH] Failed to delete from Elasticsearch (continuing deletion): documentId={}, error={}", + documentId, e.getMessage(), e); + } + } + } + + /** + * Deletes all chunks associated with a document from PostgreSQL. + * + * @param documentId the document ID whose chunks should be deleted + * @return the number of deleted chunks + */ + private int deleteChunksFromPostgreSQL(UUID documentId) { + try { + log.debug("💾 [POSTGRESQL] Starting chunk deletion for document: {}", documentId); + + int deletedChunks = documentChunkRepository.deleteBySourceDocumentId(documentId); + + log.info("✅ [POSTGRESQL] Chunks deleted successfully: documentId={}, deletedCount={}", + documentId, deletedChunks); + + return deletedChunks; + + } catch (Exception e) { + log.error("❌ [POSTGRESQL] Failed to delete chunks: documentId={}, error={}", + documentId, e.getMessage(), e); + throw new BusinessException(ErrorCode.DATABASE_ERROR, + "Failed to delete document chunks: " + e.getMessage()); + } + } + + // ========== Private Helper Methods ========== + + /** + * Validates the uploaded file. + */ + private void validateFile(MultipartFile file) { + if (file.isEmpty()) { + throw new BusinessException(ErrorCode.INVALID_REQUEST, "File cannot be empty"); + } + + if (file.getSize() > 100 * 1024 * 1024) { // 100MB + throw new BusinessException(ErrorCode.PAYLOAD_TOO_LARGE, + "File size exceeds maximum limit of 100MB"); + } + + String contentType = file.getContentType(); + if (contentType == null || !SUPPORTED_CONTENT_TYPES.contains(contentType)) { + throw new BusinessException(ErrorCode.UNSUPPORTED_MEDIA_TYPE, + "Unsupported file type. Supported types: PDF, Markdown, and plain text files."); + } + + String filename = file.getOriginalFilename(); + if (filename == null || filename.trim().isEmpty()) { + throw new BusinessException(ErrorCode.INVALID_REQUEST, "Filename cannot be empty"); + } + + log.debug("✅ [UPLOAD] File validation passed: filename={}", filename); + } + + /** + * Calculates SHA-256 checksum of the file content. + */ + private String calculateFileChecksum(MultipartFile file) { + try { + MessageDigest digest = MessageDigest.getInstance("SHA-256"); + byte[] fileBytes = file.getBytes(); + byte[] hashBytes = digest.digest(fileBytes); + + StringBuilder sb = new StringBuilder(); + for (byte b : hashBytes) { + sb.append(String.format("%02x", b)); + } + + return sb.toString(); + + } catch (NoSuchAlgorithmException | IOException e) { + log.error("❌ [UPLOAD] Failed to calculate file checksum: filename={}, error={}", + file.getOriginalFilename(), e.getMessage(), e); + throw new BusinessException(ErrorCode.FILE_UPLOAD_FAILED, + "Failed to calculate file checksum: " + e.getMessage()); + } + } + + /** + * Determines file type from content type. + */ + private String determineFileType(String contentType) { + return switch (contentType) { + case "application/pdf" -> "PDF"; + case "text/markdown" -> "MARKDOWN"; + case "text/plain" -> "TEXT"; + default -> "UNKNOWN"; + }; + } + + /** + * Converts SourceDocument entity to DTO. + */ + private SourceDocumentDto convertToDto(SourceDocument document) { + return SourceDocumentDto.builder() + .id(document.getId().toString()) + .originalFilename(document.getOriginalFilename()) + .fileType(document.getFileType()) + .fileSize(document.getFileSize()) + .ingestionStatus(document.getIngestionStatus().name()) + .errorMessage(document.getErrorMessage()) + .lastIngestedAt(document.getLastIngestedAt()) + .createdAt(document.getCreatedAt()) + .updatedAt(document.getUpdatedAt()) + .build(); + } } \ No newline at end of file diff --git a/core/src/main/java/com/opencontext/service/IndexingService.java b/core/src/main/java/com/opencontext/service/IndexingService.java new file mode 100644 index 0000000..5453479 --- /dev/null +++ b/core/src/main/java/com/opencontext/service/IndexingService.java @@ -0,0 +1,286 @@ +package com.opencontext.service; + +import com.opencontext.dto.StructuredChunk; +import com.opencontext.entity.DocumentChunk; +import com.opencontext.entity.SourceDocument; +import com.opencontext.enums.ErrorCode; +import com.opencontext.exception.BusinessException; +import com.opencontext.repository.DocumentChunkRepository; +import com.opencontext.repository.SourceDocumentRepository; +import lombok.RequiredArgsConstructor; +import lombok.extern.slf4j.Slf4j; +import org.springframework.beans.factory.annotation.Value; +import org.springframework.http.*; +import org.springframework.stereotype.Service; +import org.springframework.transaction.annotation.Transactional; +import org.springframework.web.client.RestTemplate; + +import java.util.*; + +/** + * 임베딩된 청크를 Elasticsearch와 PostgreSQL에 저장하는 서비스. + * + * Elasticsearch에는 검색을 위한 벡터와 메타데이터를, + * Service for storing embedded chunks in Elasticsearch and PostgreSQL. + * + * Stores vectors and metadata in Elasticsearch for search, + * and hierarchical structure information in PostgreSQL. + */ +@Slf4j +@Service +@RequiredArgsConstructor +@Transactional +public class IndexingService { + + private final DocumentChunkRepository documentChunkRepository; + private final SourceDocumentRepository sourceDocumentRepository; + private final RestTemplate restTemplate; + + @Value("${app.elasticsearch.url:http://localhost:9200}") + private String elasticsearchUrl; + + @Value("${app.elasticsearch.index:document-chunks}") + private String indexName; + + /** + * 임베딩된 청크들을 Elasticsearch와 PostgreSQL에 저장합니다. + * + * @param documentId 문서 ID + * @param embeddedChunks 저장할 임베딩된 청크 목록 + */ + public void indexChunks(UUID documentId, List embeddedChunks) { + long startTime = System.currentTimeMillis(); + int totalChunks = embeddedChunks.size(); + + log.info("📎 [INDEXING] Starting chunk indexing: documentId={}, chunks={}", documentId, totalChunks); + + try { + // Step 1: Elasticsearch에 벡터 데이터 저장 + log.debug("🔍 [INDEXING] Step 1/2: Indexing to Elasticsearch: chunks={}", totalChunks); + bulkIndexToElasticsearch(embeddedChunks); + log.info("✅ [INDEXING] Elasticsearch indexing completed: documentId={}, chunks={}", documentId, totalChunks); + + // Step 2: PostgreSQL에 계층 구조 정보 저장 + log.debug("💾 [INDEXING] Step 2/2: Saving hierarchy to PostgreSQL: chunks={}", totalChunks); + int savedChunks = saveChunkHierarchyToPostgreSQL(documentId, embeddedChunks); + log.info("✅ [INDEXING] PostgreSQL hierarchy saved: documentId={}, savedChunks={}", documentId, savedChunks); + + long duration = System.currentTimeMillis() - startTime; + log.info("🎉 [INDEXING] Chunk indexing completed successfully: documentId={}, chunks={}, duration={}ms", + documentId, totalChunks, duration); + + } catch (Exception e) { + long duration = System.currentTimeMillis() - startTime; + log.error("❌ [INDEXING] Chunk indexing failed: documentId={}, chunks={}, duration={}ms, error={}", + documentId, totalChunks, duration, e.getMessage(), e); + throw new BusinessException(ErrorCode.INDEXING_FAILED, + "Failed to index chunks: " + e.getMessage()); + } + } + + /** + * Elasticsearch에 청크들을 벌크 인덱싱합니다. + */ + private void bulkIndexToElasticsearch(List chunks) { + long esStartTime = System.currentTimeMillis(); + int totalChunks = chunks.size(); + + log.debug("🔍 [ELASTICSEARCH] Starting bulk indexing: chunks={}, index={}", totalChunks, indexName); + + try { + // 벌크 요청 바디 생성 + log.debug("📦 [ELASTICSEARCH] Building bulk request body: chunks={}", totalChunks); + StringBuilder bulkBody = new StringBuilder(); + + for (StructuredChunk chunk : chunks) { + // 인덱스 메타데이터 + Map indexMeta = Map.of( + "index", Map.of("_id", chunk.getChunkId()) + ); + bulkBody.append(toJsonString(indexMeta)).append("\n"); + + // 문서 데이터 + Map doc = createElasticsearchDocument(chunk); + bulkBody.append(toJsonString(doc)).append("\n"); + } + + // HTTP 헤더 설정 + HttpHeaders headers = new HttpHeaders(); + headers.setContentType(MediaType.valueOf("application/x-ndjson")); + + HttpEntity requestEntity = new HttpEntity<>(bulkBody.toString(), headers); + + // 벌크 API 호출 + ResponseEntity> response = restTemplate.exchange( + elasticsearchUrl + "/" + indexName + "/_bulk", + HttpMethod.POST, + requestEntity, + (Class>) (Class) Map.class + ); + + if (response.getStatusCode() != HttpStatus.OK) { + throw new BusinessException(ErrorCode.EXTERNAL_API_ERROR, + "Elasticsearch bulk indexing failed"); + } + + // 응답에서 오류 확인 + Map responseBody = response.getBody(); + if (responseBody != null && Boolean.TRUE.equals(responseBody.get("errors"))) { + log.warn("Some documents failed to index in Elasticsearch: {}", responseBody); + } + + long totalDuration = System.currentTimeMillis() - esStartTime; + log.info("✅ [ELASTICSEARCH] Bulk indexing completed successfully: chunks={}, duration={}ms", + totalChunks, totalDuration); + + } catch (Exception e) { + long totalDuration = System.currentTimeMillis() - esStartTime; + log.error("❌ [ELASTICSEARCH] Bulk indexing failed: chunks={}, duration={}ms, error={}", + totalChunks, totalDuration, e.getMessage(), e); + throw new BusinessException(ErrorCode.EXTERNAL_API_ERROR, + "Failed to index to Elasticsearch: " + e.getMessage()); + } + } + + /** + * PostgreSQL에 청크 계층 구조 정보를 저장합니다. + */ + private int saveChunkHierarchyToPostgreSQL(UUID documentId, List chunks) { + long pgStartTime = System.currentTimeMillis(); + int totalChunks = chunks.size(); + + log.debug("💾 [POSTGRESQL] Starting to save chunk hierarchy: documentId={}, chunks={}", + documentId, totalChunks); + + try { + // ✅ SourceDocument 엔티티를 가져와서 JPA 관계 설정 + SourceDocument sourceDocument = sourceDocumentRepository.findById(documentId) + .orElseThrow(() -> new BusinessException(ErrorCode.DATABASE_ERROR, + "SourceDocument not found: " + documentId)); + + List documentChunks = new ArrayList<>(); + + for (StructuredChunk chunk : chunks) { + UUID parentChunkUuid = null; + if (chunk.getParentChunkId() != null && !chunk.getParentChunkId().trim().isEmpty()) { + try { + parentChunkUuid = UUID.fromString(chunk.getParentChunkId()); + } catch (IllegalArgumentException e) { + log.warn("Invalid parent_chunk_id format: {}, skipping parent relationship for chunk: {}", + chunk.getParentChunkId(), chunk.getChunkId()); + parentChunkUuid = null; + } + } + + DocumentChunk documentChunk = DocumentChunk.builder() + .chunkId(chunk.getChunkId()) + .sourceDocument(sourceDocument) + .sourceDocumentId(documentId) + .title(chunk.getTitle()) + .hierarchyLevel(chunk.getHierarchyLevel()) + .parentChunkId(parentChunkUuid) + .elementType(chunk.getElementType()) + .sequenceInDocument(0) // 기본값 설정, 실제로는 순서에 따라 설정해야 함 + .build(); + + documentChunks.add(documentChunk); + } + + // 배치 저장 + long saveStartTime = System.currentTimeMillis(); + List savedChunks = documentChunkRepository.saveAll(documentChunks); + long saveDuration = System.currentTimeMillis() - saveStartTime; + + long totalDuration = System.currentTimeMillis() - pgStartTime; + log.info("✅ [POSTGRESQL] Chunk hierarchy saved successfully: documentId={}, chunks={}, duration={}ms, saveTime={}ms", + documentId, savedChunks.size(), totalDuration, saveDuration); + + return savedChunks.size(); + + } catch (Exception e) { + long totalDuration = System.currentTimeMillis() - pgStartTime; + log.error("❌ [POSTGRESQL] Failed to save chunk hierarchy: documentId={}, chunks={}, duration={}ms, error={}", + documentId, totalChunks, totalDuration, e.getMessage(), e); + throw new BusinessException(ErrorCode.DATABASE_ERROR, + "Failed to save chunk hierarchy: " + e.getMessage()); + } + } + + /** + * Elasticsearch 문서를 생성합니다. + */ + private Map createElasticsearchDocument(StructuredChunk chunk) { + Map doc = new HashMap<>(); + + doc.put("document_id", chunk.getDocumentId()); + doc.put("chunk_id", chunk.getChunkId()); + + // ✅ content 필드 정리하여 저장 (JSON 파싱 에러 방지) + String cleanContent = sanitizeContent(chunk.getContent()); + doc.put("content", cleanContent); + + doc.put("title", chunk.getTitle()); + doc.put("hierarchy_level", chunk.getHierarchyLevel()); + doc.put("parent_chunk_id", chunk.getParentChunkId()); + doc.put("element_type", chunk.getElementType()); + doc.put("embedding", chunk.getEmbedding()); + doc.put("metadata", chunk.getMetadata()); + doc.put("indexed_at", new Date()); + + return doc; + } + + /** + * content 필드의 Java 코드나 특수문자를 정리하여 JSON 파싱 에러를 방지합니다. + */ + private String sanitizeContent(String content) { + if (content == null || content.trim().isEmpty()) { + return ""; + } + + return content + // Java 코드 관련 정리 + .replaceAll("\\{[^}]*\\}", "") // 중괄호 내용 제거 + .replaceAll("\\[.*?\\]", "") // 대괄호 내용 제거 + .replaceAll("\\b(int|String|void|public|private|static|final|class|interface|extends|implements)\\b", "") // Java 키워드 제거 + .replaceAll("\\b(if|else|for|while|switch|case|break|continue|return|throw|try|catch|finally)\\b", "") // Java 제어문 제거 + .replaceAll("\\b(new|this|super|null|true|false)\\b", "") // Java 리터럴 제거 + + // 특수문자 정리 + .replaceAll("\\s+", " ") // 연속 공백을 단일 공백으로 + .replaceAll("[\\r\\n\\t]+", " ") // 줄바꿈, 탭을 공백으로 + .replaceAll("\\s*[;=+\\-*/%<>!&|^~]\\s*", " ") // 연산자 주변 공백 정리 + + // 기타 정리 + .replaceAll("\\s+", " ") // 다시 연속 공백 정리 + .trim(); + } + + /** + * 객체를 JSON 문자열로 변환합니다. + * 실제 구현에서는 Jackson ObjectMapper를 사용해야 합니다. + */ + private String toJsonString(Object obj) { + // 간단한 구현 - 실제로는 ObjectMapper를 사용해야 함 + if (obj instanceof Map) { + Map map = (Map) obj; + StringBuilder json = new StringBuilder("{"); + boolean first = true; + for (Map.Entry entry : map.entrySet()) { + if (!first) json.append(","); + json.append("\"").append(entry.getKey()).append("\":"); + if (entry.getValue() instanceof String) { + json.append("\"").append(entry.getValue()).append("\""); + } else if (entry.getValue() instanceof Map) { + json.append(toJsonString(entry.getValue())); + } else { + json.append(entry.getValue()); + } + first = false; + } + json.append("}"); + return json.toString(); + } + return obj.toString(); + } +} diff --git a/core/src/main/java/com/opencontext/service/SourceDocumentService.java b/core/src/main/java/com/opencontext/service/SourceDocumentService.java deleted file mode 100644 index f0b019f..0000000 --- a/core/src/main/java/com/opencontext/service/SourceDocumentService.java +++ /dev/null @@ -1,328 +0,0 @@ -package com.opencontext.service; - -import com.opencontext.dto.SourceDocumentDto; -import com.opencontext.dto.SourceUploadResponse; -import com.opencontext.entity.SourceDocument; -import com.opencontext.enums.ErrorCode; -import com.opencontext.enums.IngestionStatus; -import com.opencontext.exception.BusinessException; -import com.opencontext.repository.SourceDocumentRepository; -import lombok.RequiredArgsConstructor; -import lombok.extern.slf4j.Slf4j; -import org.springframework.data.domain.Page; -import org.springframework.data.domain.Pageable; -import org.springframework.scheduling.annotation.Async; -import org.springframework.stereotype.Service; -import org.springframework.transaction.annotation.Transactional; -import org.springframework.web.multipart.MultipartFile; - -import java.io.IOException; -import java.security.MessageDigest; -import java.security.NoSuchAlgorithmException; -import java.util.Arrays; -import java.util.List; -import java.util.UUID; - -/** - * Service for managing source document ingestion and lifecycle. - * - * This service handles the complete document ingestion pipeline from upload - * to indexing, integrating with MinIO storage and the async processing pipeline. - */ -@Slf4j -@Service -@RequiredArgsConstructor -@Transactional -public class SourceDocumentService { - - private final SourceDocumentRepository sourceDocumentRepository; - private final FileStorageService fileStorageService; - - // Supported file types for document processing - private static final List SUPPORTED_CONTENT_TYPES = List.of( - "application/pdf", - "text/markdown", - "text/plain" - ); - - /** - * Uploads a source document and starts the ingestion pipeline. - * - * @param file the multipart file to upload - * @return SourceUploadResponse containing the created document information - */ - public SourceUploadResponse uploadSourceDocument(MultipartFile file) { - log.info("Starting source document upload: filename={}, size={}, contentType={}", - file.getOriginalFilename(), file.getSize(), file.getContentType()); - - // Validate file - validateFile(file); - - // Calculate file checksum to prevent duplicates - String fileChecksum = calculateFileChecksum(file); - - // Check for duplicate files - if (sourceDocumentRepository.existsByFileChecksum(fileChecksum)) { - log.warn("Duplicate file upload attempt: checksum={}", fileChecksum); - throw new BusinessException(ErrorCode.DUPLICATE_FILE_UPLOADED, - "A file with identical content already exists."); - } - - try { - // Upload file to MinIO - String objectKey = fileStorageService.uploadFile(file); - - // Create SourceDocument entity - SourceDocument sourceDocument = SourceDocument.builder() - .originalFilename(file.getOriginalFilename()) - .fileStoragePath(objectKey) - .fileType(determineFileType(file.getContentType())) - .fileSize(file.getSize()) - .fileChecksum(fileChecksum) - .ingestionStatus(IngestionStatus.PENDING) - .build(); - - // Save to database - SourceDocument savedDocument = sourceDocumentRepository.save(sourceDocument); - - log.info("Source document created successfully: id={}, filename={}", - savedDocument.getId(), savedDocument.getOriginalFilename()); - - // Start async ingestion pipeline - startIngestionPipeline(savedDocument.getId()); - - // Return response - return SourceUploadResponse.success( - savedDocument.getId().toString(), - savedDocument.getOriginalFilename() - ); - - } catch (Exception e) { - log.error("Failed to upload source document: {}", file.getOriginalFilename(), e); - if (e instanceof BusinessException) { - throw e; - } - throw new BusinessException(ErrorCode.FILE_UPLOAD_FAILED, - "Unexpected error during file upload: " + e.getMessage()); - } - } - - /** - * Retrieves all source documents with pagination. - * - * @param pageable pagination parameters - * @return Page of SourceDocumentDto - */ - @Transactional(readOnly = true) - public Page getAllSourceDocuments(Pageable pageable) { - log.debug("Retrieving source documents with pagination: page={}, size={}", - pageable.getPageNumber(), pageable.getPageSize()); - - return sourceDocumentRepository.findAllByOrderByCreatedAtDesc(pageable) - .map(this::convertToDto); - } - - /** - * Retrieves a specific source document by ID. - * - * @param documentId the document ID - * @return SourceDocumentDto - */ - @Transactional(readOnly = true) - public SourceDocumentDto getSourceDocument(UUID documentId) { - log.debug("Retrieving source document: id={}", documentId); - - SourceDocument document = sourceDocumentRepository.findById(documentId) - .orElseThrow(() -> new BusinessException(ErrorCode.SOURCE_DOCUMENT_NOT_FOUND, - "Document not found: " + documentId)); - - return convertToDto(document); - } - - /** - * Triggers re-ingestion of a source document. - * - * @param documentId the document ID to re-ingest - */ - public void resyncSourceDocument(UUID documentId) { - log.info("Starting source document resync: id={}", documentId); - - SourceDocument document = sourceDocumentRepository.findById(documentId) - .orElseThrow(() -> new BusinessException(ErrorCode.SOURCE_DOCUMENT_NOT_FOUND, - "Document not found: " + documentId)); - - // Check if document is currently being processed - if (document.isProcessing()) { - throw new BusinessException(ErrorCode.RESOURCE_IS_BEING_PROCESSED, - "Document is currently being processed and cannot be resynced."); - } - - // Reset status to PENDING and clear error message - document.updateIngestionStatus(IngestionStatus.PENDING); - sourceDocumentRepository.save(document); - - // Start async ingestion pipeline - startIngestionPipeline(documentId); - - log.info("Source document resync initiated: id={}", documentId); - } - - /** - * Deletes a source document and all associated data. - * - * @param documentId the document ID to delete - */ - public void deleteSourceDocument(UUID documentId) { - log.info("Starting source document deletion: id={}", documentId); - - SourceDocument document = sourceDocumentRepository.findById(documentId) - .orElseThrow(() -> new BusinessException(ErrorCode.SOURCE_DOCUMENT_NOT_FOUND, - "Document not found: " + documentId)); - - // Check if document is currently being processed - if (document.isProcessing()) { - throw new BusinessException(ErrorCode.RESOURCE_IS_BEING_PROCESSED, - "Document is currently being processed and cannot be deleted."); - } - - try { - // Mark document as being deleted - document.updateIngestionStatus(IngestionStatus.DELETING); - sourceDocumentRepository.save(document); - - // Start async deletion process - startDeletionPipeline(documentId); - - log.info("Source document deletion initiated: id={}", documentId); - - } catch (Exception e) { - log.error("Failed to initiate document deletion: id={}", documentId, e); - throw new BusinessException(ErrorCode.FILE_DELETE_FAILED, - "Failed to initiate document deletion: " + e.getMessage()); - } - } - - /** - * Starts the async ingestion pipeline for a document. - * - * @param documentId the document ID to process - */ - @Async - public void startIngestionPipeline(UUID documentId) { - log.info("Starting ingestion pipeline for document: id={}", documentId); - - // TODO: Implement actual ingestion pipeline - // This would typically involve: - // 1. Update status to PARSING - // 2. Parse document using Unstructured API - // 3. Update status to CHUNKING - // 4. Split into chunks - // 5. Update status to EMBEDDING - // 6. Generate embeddings - // 7. Update status to INDEXING - // 8. Store in Elasticsearch - // 9. Update status to COMPLETED - - // For now, just log the pipeline start - log.info("Ingestion pipeline queued for document: id={}", documentId); - } - - /** - * Starts the async deletion pipeline for a document. - * - * @param documentId the document ID to delete - */ - @Async - public void startDeletionPipeline(UUID documentId) { - log.info("Starting deletion pipeline for document: id={}", documentId); - - // TODO: Implement actual deletion pipeline - // This would typically involve: - // 1. Delete from Elasticsearch - // 2. Delete chunks from PostgreSQL - // 3. Delete file from MinIO - // 4. Delete SourceDocument record - - // For now, just log the pipeline start - log.info("Deletion pipeline queued for document: id={}", documentId); - } - - /** - * Validates the uploaded file. - */ - private void validateFile(MultipartFile file) { - if (file.isEmpty()) { - throw new BusinessException(ErrorCode.INVALID_REQUEST, "File cannot be empty"); - } - - if (file.getSize() > 100 * 1024 * 1024) { // 100MB - throw new BusinessException(ErrorCode.PAYLOAD_TOO_LARGE, - "File size exceeds maximum limit of 100MB"); - } - - String contentType = file.getContentType(); - if (contentType == null || !SUPPORTED_CONTENT_TYPES.contains(contentType)) { - throw new BusinessException(ErrorCode.UNSUPPORTED_MEDIA_TYPE, - "Unsupported file type. Supported types: PDF, Markdown, and plain text files."); - } - - String filename = file.getOriginalFilename(); - if (filename == null || filename.trim().isEmpty()) { - throw new BusinessException(ErrorCode.INVALID_REQUEST, "Filename cannot be empty"); - } - - log.debug("File validation passed: filename={}", filename); - } - - /** - * Calculates SHA-256 checksum of the file content. - */ - private String calculateFileChecksum(MultipartFile file) { - try { - MessageDigest digest = MessageDigest.getInstance("SHA-256"); - byte[] fileBytes = file.getBytes(); - byte[] hashBytes = digest.digest(fileBytes); - - StringBuilder sb = new StringBuilder(); - for (byte b : hashBytes) { - sb.append(String.format("%02x", b)); - } - - return sb.toString(); - - } catch (NoSuchAlgorithmException | IOException e) { - log.error("Failed to calculate file checksum", e); - throw new BusinessException(ErrorCode.FILE_UPLOAD_FAILED, - "Failed to calculate file checksum: " + e.getMessage()); - } - } - - /** - * Determines file type from content type. - */ - private String determineFileType(String contentType) { - return switch (contentType) { - case "application/pdf" -> "PDF"; - case "text/markdown" -> "MARKDOWN"; - case "text/plain" -> "TEXT"; - default -> "UNKNOWN"; - }; - } - - /** - * Converts SourceDocument entity to DTO. - */ - private SourceDocumentDto convertToDto(SourceDocument document) { - return SourceDocumentDto.builder() - .id(document.getId().toString()) - .originalFilename(document.getOriginalFilename()) - .fileType(document.getFileType()) - .fileSize(document.getFileSize()) - .ingestionStatus(document.getIngestionStatus().name()) - .errorMessage(document.getErrorMessage()) - .lastIngestedAt(document.getLastIngestedAt()) - .createdAt(document.getCreatedAt()) - .updatedAt(document.getUpdatedAt()) - .build(); - } -}