[Feat] 텍스트 파싱 및 Chunk 저장

.codex/hooks.json

@@ -6,7 +6,7 @@
         "hooks": [
           {
             "type": "command",
-            "command": "bash '/Users/giminkim/IdeaProjects/backend/.codex/hooks/pre-bash.sh'"
+            "command": "bash '.codex/hooks/pre-bash.sh'"
           }
         ]
       }

.env.example

@@ -16,6 +16,11 @@ MINIO_ACCESS_KEY=minioadmin
 MINIO_SECRET_KEY=minioadmin1234
 MINIO_BUCKET=docgrid
 
+# 문서 Chunk 정책 - 기본값을 쓰면 설정하지 않아도 됩니다.
+# 같은 배포군의 Worker는 결정성을 위해 반드시 같은 값을 사용해야 합니다.
+# DOCUMENT_CHUNK_SIZE=1000
+# DOCUMENT_CHUNK_OVERLAP=200
+
 # 임베딩 서버 - application.yml에 이미 기본값(http://localhost:8000)이 있어 docker-compose 기본 포트를 쓰면 설정 불필요.
 # 기본값과 다른 포트/호스트를 쓸 때만 주석 해제
 # EMBEDDING_SERVER_URL=http://localhost:8000

src/main/java/com/opensource/docgrid/domain/document/config/DocumentChunkingProperties.java

@@ -0,0 +1,37 @@
+package com.opensource.docgrid.domain.document.config;
+
+import org.springframework.boot.context.properties.ConfigurationProperties;
+import org.springframework.stereotype.Component;
+import org.springframework.validation.annotation.Validated;
+
+import jakarta.validation.constraints.AssertTrue;
+import jakarta.validation.constraints.Positive;
+import lombok.Getter;
+import lombok.Setter;
+
+/**
+ * 문서 텍스트를 고정 크기 Chunk로 나눌 때 사용하는 크기와 중첩 범위를 제공한다.
+ *
+ * <p>{@code document.chunking} 설정을 바인딩하고 애플리케이션 시작 시 Chunk 크기보다 작은
+ * 0 이상의 Overlap만 허용해 무한 반복이나 잘못된 범위 계산을 방지한다.
+ */
+@Getter
+@Setter
+@Validated
+@Component
+@ConfigurationProperties(prefix = "document.chunking")
+public class DocumentChunkingProperties {
+
+    @Positive
+    private int chunkSize = 1000;
+
+    private int overlap = 200;
+
+    /**
+     * 다음 Chunk 시작 위치가 반드시 앞으로 이동하도록 Overlap 조합을 검증한다.
+     */
+    @AssertTrue(message = "Chunk Overlap은 0 이상이며 Chunk 크기보다 작아야 합니다.")
+    public boolean isOverlapValid() {
+        return overlap >= 0 && overlap < chunkSize;
+    }
+}

src/main/java/com/opensource/docgrid/domain/document/entity/DocumentVersion.java

@@ -119,10 +119,18 @@ public DocumentVersion(Document document, FileObject fileObject, int versionNo,
     }
 
     public void markParsing() {
+        // UPLOADED에서 시작한 최초 파싱만 허용하고 재개 여부 판단은 Command Service가 담당한다.
+        if (status != DocumentVersionStatus.UPLOADED) {
+            throw new IllegalStateException("UPLOADED 상태의 문서 버전만 PARSING으로 전환할 수 있습니다.");
+        }
         this.status = DocumentVersionStatus.PARSING;
     }
 
     public void markChunked() {
+        // Chunk Set 저장과 같은 Transaction에서 PARSING Version만 완료 상태로 전환한다.
+        if (status != DocumentVersionStatus.PARSING) {
+            throw new IllegalStateException("PARSING 상태의 문서 버전만 CHUNKED로 전환할 수 있습니다.");
+        }
         this.status = DocumentVersionStatus.CHUNKED;
     }
 

src/main/java/com/opensource/docgrid/domain/document/repository/DocumentChunkRepository.java

@@ -0,0 +1,18 @@
+package com.opensource.docgrid.domain.document.repository;
+
+import org.springframework.data.jpa.repository.JpaRepository;
+
+import com.opensource.docgrid.domain.document.entity.DocumentChunk;
+
+/**
+ * 문서 Version별 Chunk 존재 여부와 개수 조회 및 Chunk Set 전체 저장을 담당한다.
+ *
+ * <p>Chunk 생성 Transaction은 기존 결과 확인에 존재·개수 조회를 사용하고, 신규 결과는
+ * {@link JpaRepository#saveAllAndFlush(Iterable)}로 같은 Transaction 안에서 즉시 검증한다.
+ */
+public interface DocumentChunkRepository extends JpaRepository<DocumentChunk, Long> {
+
+    boolean existsByDocumentVersionId(Long documentVersionId);
+
+    long countByDocumentVersionId(Long documentVersionId);
+}

src/main/java/com/opensource/docgrid/domain/document/repository/DocumentVersionRepository.java

@@ -3,14 +3,22 @@
 import java.util.Collection;
 import java.util.Optional;
 
-import org.springframework.data.jpa.repository.Query;
-import org.springframework.data.repository.query.Param;
+import jakarta.persistence.LockModeType;
 
 import org.springframework.data.jpa.repository.JpaRepository;
+import org.springframework.data.jpa.repository.Lock;
+import org.springframework.data.jpa.repository.Query;
+import org.springframework.data.repository.query.Param;
 
 import com.opensource.docgrid.domain.document.entity.DocumentVersion;
 import com.opensource.docgrid.domain.document.enums.DocumentVersionStatus;
 
+/**
+ * Document Version의 일반 조회와 파이프라인 상태 전이를 위한 행 잠금 조회를 제공한다.
+ *
+ * <p>Chunk 생성 흐름은 Embedding Job을 먼저 잠근 뒤 이 Repository로 대상 Version을 잠가
+ * 상태 변경과 동일 Version의 Chunk Set 생성을 직렬화한다.
+ */
 public interface DocumentVersionRepository extends JpaRepository<DocumentVersion, Long> {
 
     boolean existsByDocumentIdAndStatusIn(Long documentId, Collection<DocumentVersionStatus> statuses);
@@ -19,4 +27,14 @@ public interface DocumentVersionRepository extends JpaRepository<DocumentVersion
 
     @Query("SELECT COALESCE(MAX(dv.versionNo), 0) FROM DocumentVersion dv WHERE dv.document.id = :documentId")
     int findMaxVersionNo(@Param("documentId") Long documentId);
+
+    /**
+     * 지정한 Version을 현재 Transaction이 끝날 때까지 쓰기 잠금 상태로 조회한다.
+     *
+     * @param documentVersionId 잠글 Document Version 식별자
+     * @return 잠금을 획득한 Version, 존재하지 않으면 빈 값
+     */
+    @Lock(LockModeType.PESSIMISTIC_WRITE)
+    @Query("SELECT version FROM DocumentVersion version WHERE version.id = :documentVersionId")
+    Optional<DocumentVersion> findByIdForUpdate(@Param("documentVersionId") Long documentVersionId);
 }

src/main/java/com/opensource/docgrid/domain/document/service/DocumentChunkDraft.java

@@ -0,0 +1,20 @@
+package com.opensource.docgrid.domain.document.service;
+
+/**
+ * 영속화 전 단계에서 계산된 문서 Chunk 한 건의 불변 값을 전달한다.
+ *
+ * <p>외부 파일 읽기와 Chunk 계산 구간이 JPA Entity나 Persistence Context에 의존하지 않도록
+ * 저장에 필요한 값만 보관한다. 페이지·섹션·Metadata는 고정 크기 텍스트 Chunking 범위에서 비워 둔다.
+ */
+public record DocumentChunkDraft(
+    int chunkIndex,
+    String chunkText,
+    int tokenCount,
+    int charStart,
+    int charEnd,
+    Integer pageNo,
+    String sectionTitle,
+    String contentHash,
+    String metadataJson
+) {
+}

src/main/java/com/opensource/docgrid/domain/document/service/DocumentParsingService.java

@@ -0,0 +1,76 @@
+package com.opensource.docgrid.domain.document.service;
+
+import java.util.List;
+
+import org.springframework.stereotype.Service;
+
+import com.opensource.docgrid.domain.document.service.command.DocumentChunkTransactionService;
+import com.opensource.docgrid.domain.document.service.command.DocumentChunkTransactionService.ChunkResult;
+import com.opensource.docgrid.domain.document.service.command.DocumentChunkTransactionService.PreparationResult;
+import com.opensource.docgrid.domain.document.storage.FileStorageService;
+import com.opensource.docgrid.domain.embedding.dto.request.CreateDocumentChunksRequest;
+import com.opensource.docgrid.global.exception.DocGridException;
+import com.opensource.docgrid.global.exception.ErrorCode;
+
+import lombok.RequiredArgsConstructor;
+
+/**
+ * 두 개의 짧은 DB Transaction 사이에서 원본 읽기, Text 파싱과 Chunk 계산을 조정한다.
+ *
+ * <p>이 Service 자체에는 Transaction을 적용하지 않아 MinIO I/O와 CPU 계산 중 DB 행 잠금이
+ * 유지되지 않게 한다. 외부 구간에는 불변 Snapshot과 Draft만 전달하고 JPA Entity는 전달하지 않는다.
+ */
+@Service
+@RequiredArgsConstructor
+public class DocumentParsingService {
+
+    private final DocumentChunkTransactionService transactionService;
+    private final FileStorageService fileStorageService;
+    private final TextDocumentParser textDocumentParser;
+    private final FixedSizeChunker fixedSizeChunker;
+
+    /**
+     * 현재 Attempt가 소유한 Job의 원본을 Chunking하고 저장하거나 기존 결과를 재생한다.
+     *
+     * @param jobId 현재 Embedding Job 식별자
+     * @param attemptId 현재 실행 Attempt 식별자
+     * @param request Worker와 Claim Token
+     * @return API 응답과 이번 호출의 신규 저장 여부
+     */
+    public ChunkResult createChunks(
+        Long jobId,
+        Long attemptId,
+        CreateDocumentChunksRequest request
+    ) {
+        // 1. 준비 Transaction에서 소유권과 상태를 검증하고 외부 작업용 Snapshot을 만든다.
+        PreparationResult preparation = transactionService.prepare(
+            jobId,
+            attemptId,
+            request.workerId(),
+            request.claimToken()
+        );
+
+        // 2. 이미 CHUNKED인 유효한 재호출은 Object Storage를 다시 읽지 않고 즉시 반환한다.
+        if (preparation.replayResult() != null) {
+            return preparation.replayResult();
+        }
+
+        // 3. Transaction 밖에서 MinIO 읽기, 엄격한 UTF-8 파싱과 결정적 Chunk 계산을 수행한다.
+        byte[] content = fileStorageService.read(preparation.fileSnapshot().storedFile());
+        String canonicalText = textDocumentParser.parse(content);
+        List<DocumentChunkDraft> drafts = fixedSizeChunker.chunk(canonicalText);
+        if (drafts.isEmpty()) {
+            throw new DocGridException(ErrorCode.DOCUMENT_CHUNKS_INCONSISTENT);
+        }
+
+        // 4. 완료 Transaction이 소유권을 다시 검증하고 Chunk Set과 상태·이벤트를 원자적으로 저장한다.
+        return transactionService.complete(
+            jobId,
+            attemptId,
+            request.workerId(),
+            request.claimToken(),
+            preparation.fileSnapshot().documentVersionId(),
+            drafts
+        );
+    }
+}

src/main/java/com/opensource/docgrid/domain/document/service/FixedSizeChunker.java

@@ -0,0 +1,104 @@
+package com.opensource.docgrid.domain.document.service;
+
+import java.nio.charset.StandardCharsets;
+import java.security.MessageDigest;
+import java.security.NoSuchAlgorithmException;
+import java.util.ArrayList;
+import java.util.HexFormat;
+import java.util.List;
+
+import org.springframework.stereotype.Component;
+
+import com.opensource.docgrid.domain.document.config.DocumentChunkingProperties;
+
+import lombok.RequiredArgsConstructor;
+
+/**
+ * Canonical Text를 Unicode Code Point 기준의 고정 크기·중첩 Chunk 목록으로 계산한다.
+ *
+ * <p>DB와 외부 저장소를 사용하지 않는 순수 계산 경계이며, 같은 Text와 설정에 대해 Index,
+ * 반열린 범위, Token 추정치와 SHA-256 Hash가 항상 같은 Draft 목록을 만든다.
+ */
+@Component
+@RequiredArgsConstructor
+public class FixedSizeChunker {
+
+    private static final String HASH_ALGORITHM = "SHA-256";
+
+    private final DocumentChunkingProperties properties;
+
+    /**
+     * Canonical Text를 저장 가능한 Chunk Draft 목록으로 나눈다.
+     *
+     * @param canonicalText 파싱과 정규화가 끝난 Text
+     * @return 0부터 연속된 Index를 가진 Chunk Draft 목록
+     * @throws IllegalArgumentException Chunk 크기와 Overlap 조합이 유효하지 않은 경우
+     */
+    public List<DocumentChunkDraft> chunk(String canonicalText) {
+        int chunkSize = properties.getChunkSize();
+        int overlap = properties.getOverlap();
+        // 1. 설정 객체가 생성 후 변경되더라도 시작 위치가 반드시 앞으로 이동하게 불변식을 재검증한다.
+        if (chunkSize <= 0 || overlap < 0 || overlap >= chunkSize) {
+            throw new IllegalArgumentException("Chunk 크기는 양수이고 Overlap은 0 이상 Chunk 크기 미만이어야 합니다.");
+        }
+
+        int[] codePoints = canonicalText.codePoints().toArray();
+        // 2. 비어 있는 원문은 저장할 Chunk가 없으므로 즉시 빈 결과로 종료한다.
+        if (codePoints.length == 0) {
+            return List.of();
+        }
+
+        int step = chunkSize - overlap;
+        List<DocumentChunkDraft> drafts = new ArrayList<>();
+
+        // 3. 시작 위치를 Code Point 단위로 이동해 Surrogate Pair 중간 분할을 방지한다.
+        for (int start = 0, chunkIndex = 0; start < codePoints.length; start += step, chunkIndex++) {
+            int end = Math.min(start + chunkSize, codePoints.length);
+            String chunkText = new String(codePoints, start, end - start);
+
+            // 4. Entity에 필요한 계산 값만 Draft에 담고 페이지·섹션·Metadata는 이번 범위에서 비워 둔다.
+            drafts.add(new DocumentChunkDraft(
+                chunkIndex,
+                chunkText,
+                estimateTokenCount(chunkText),
+                start,
+                end,
+                null,
+                null,
+                sha256(chunkText),
+                null
+            ));
+
+            // 5. 마지막 Chunk가 원문 끝에 도달하면 Overlap만 남은 추가 Chunk를 만들지 않는다.
+            if (end == codePoints.length) {
+                break;
+            }
+        }
+        return List.copyOf(drafts);
+    }
+
+    private int estimateTokenCount(String text) {
+        int tokenCount = 0;
+        boolean insideToken = false;
+        int[] codePoints = text.codePoints().toArray();
+
+        for (int codePoint : codePoints) {
+            if (Character.isWhitespace(codePoint)) {
+                insideToken = false;
+            } else if (!insideToken) {
+                tokenCount++;
+                insideToken = true;
+            }
+        }
+        return tokenCount;
+    }
+
+    private String sha256(String text) {
+        try {
+            MessageDigest digest = MessageDigest.getInstance(HASH_ALGORITHM);
+            return HexFormat.of().formatHex(digest.digest(text.getBytes(StandardCharsets.UTF_8)));
+        } catch (NoSuchAlgorithmException exception) {
+            throw new IllegalStateException("SHA-256 알고리즘을 사용할 수 없습니다.", exception);
+        }
+    }
+}