diff --git a/easyflow-api/easyflow-api-admin/src/main/java/tech/easyflow/admin/controller/ai/DocumentChunkController.java b/easyflow-api/easyflow-api-admin/src/main/java/tech/easyflow/admin/controller/ai/DocumentChunkController.java index 92142b73..5cec5021 100644 --- a/easyflow-api/easyflow-api-admin/src/main/java/tech/easyflow/admin/controller/ai/DocumentChunkController.java +++ b/easyflow-api/easyflow-api-admin/src/main/java/tech/easyflow/admin/controller/ai/DocumentChunkController.java @@ -1,23 +1,18 @@ package tech.easyflow.admin.controller.ai; import cn.dev33.satoken.annotation.SaCheckPermission; -import com.easyagents.core.model.embedding.EmbeddingModel; import com.mybatisflex.core.paginate.Page; +import com.mybatisflex.core.query.QueryWrapper; +import tech.easyflow.ai.dto.DocumentChunkContentUpdateRequest; +import tech.easyflow.ai.dto.DocumentChunkSyncRetryRequest; +import tech.easyflow.ai.dto.DocumentChunkSyncStatus; +import tech.easyflow.ai.dto.DocumentChunkSyncStatusRequest; import tech.easyflow.ai.entity.DocumentChunk; -import tech.easyflow.ai.entity.DocumentCollection; -import tech.easyflow.ai.entity.Model; import tech.easyflow.ai.service.DocumentChunkService; -import tech.easyflow.ai.service.DocumentCollectionService; -import tech.easyflow.ai.service.ModelService; -import tech.easyflow.ai.support.DocumentStoreLifecycleSupport; import tech.easyflow.common.annotation.UsePermission; import tech.easyflow.common.domain.Result; -import tech.easyflow.common.web.controller.BaseCurdController; +import tech.easyflow.common.web.controller.BaseController; import tech.easyflow.common.web.jsonbody.JsonBody; -import com.easyagents.core.document.Document; -import com.easyagents.core.store.DocumentStore; -import com.easyagents.core.store.StoreOptions; -import com.easyagents.core.store.StoreResult; import jakarta.servlet.http.HttpServletRequest; import org.springframework.web.bind.annotation.PostMapping; import org.springframework.web.bind.annotation.GetMapping; @@ -28,12 +23,8 @@ import tech.easyflow.system.enums.ResourceAction; import tech.easyflow.system.enums.ResourceLookup; import tech.easyflow.system.permission.resource.RequireResourceAccess; -import javax.annotation.Resource; import java.math.BigInteger; -import java.util.ArrayList; -import java.util.HashMap; import java.util.List; -import java.util.Map; /** * 控制层。 @@ -44,19 +35,12 @@ import java.util.Map; @RestController @RequestMapping("/api/v1/documentChunk") @UsePermission(moduleName = "/api/v1/documentCollection") -public class DocumentChunkController extends BaseCurdController { +public class DocumentChunkController extends BaseController { - @Resource - DocumentCollectionService documentCollectionService; - - @Resource - ModelService modelService; - - @Resource - DocumentChunkService documentChunkService; + private final DocumentChunkService documentChunkService; public DocumentChunkController(DocumentChunkService service) { - super(service); + this.documentChunkService = service; } @GetMapping("page") @@ -68,9 +52,30 @@ public class DocumentChunkController extends BaseCurdController> page(HttpServletRequest request, String sortKey, String sortType, Long pageNumber, Long pageSize) { - return super.page(request, sortKey, sortType, pageNumber, pageSize); + public Result> page( + HttpServletRequest request, + Long pageNumber, + Long pageSize + ) { + String documentIdValue = request.getParameter("documentId"); + if (documentIdValue == null || documentIdValue.isBlank()) { + return Result.>fail("documentId不能为空", null); + } + BigInteger documentId; + try { + documentId = new BigInteger(documentIdValue); + } catch (NumberFormatException e) { + return Result.>fail("documentId格式不正确", null); + } + long normalizedPageNumber = pageNumber == null || pageNumber < 1 ? 1 : pageNumber; + long normalizedPageSize = pageSize == null || pageSize < 1 ? 10 : pageSize; + QueryWrapper query = QueryWrapper.create() + .eq(DocumentChunk::getDocumentId, documentId) + .orderBy("sorting asc"); + return Result.ok(documentChunkService.page( + new Page<>(normalizedPageNumber, normalizedPageSize), + query + )); } @PostMapping("update") @@ -79,43 +84,23 @@ public class DocumentChunkController extends BaseCurdController update(@JsonBody DocumentChunk documentChunk) { - boolean success = service.updateById(documentChunk); - if (success){ - DocumentChunk record = documentChunkService.getById(documentChunk.getId()); - DocumentCollection knowledge = documentCollectionService.getById(record.getDocumentCollectionId()); - if (knowledge == null) { - return Result.fail(1, "知识库不存在"); - } - DocumentStore documentStore = knowledge.toDocumentStore(); - if (documentStore == null) { - return Result.fail(2, "知识库没有配置向量库"); - } - try { - // 设置向量模型 - Model model = modelService.getModelInstance(knowledge.getVectorEmbedModelId()); - if (model == null) { - return Result.fail(3, "知识库没有配置向量模型"); - } - EmbeddingModel embeddingModel = model.toEmbeddingModel(); - documentStore.setEmbeddingModel(embeddingModel); - StoreOptions options = StoreOptions.ofCollectionName(knowledge.getVectorStoreCollection()); - Document document = Document.of(documentChunk.getContent()); - document.setId(documentChunk.getId()); - Map metadata = new HashMap<>(); - metadata.put("keywords", documentChunk.getMetadataKeyWords()); - metadata.put("questions", documentChunk.getMetadataQuestions()); - document.setMetadataMap(metadata); - StoreResult result = documentStore.update(document, options); // 更新已有记录 - return Result.ok(result); - } finally { - DocumentStoreLifecycleSupport.closeQuietly(documentStore); - } + public Result update( + @JsonBody(required = true, skipConvertError = false) + DocumentChunkContentUpdateRequest request + ) { + DocumentChunk current = documentChunkService.getById(request.getId()); + if (current == null) { + return Result.fail(1, "记录不存在"); } - return Result.ok(false); + DocumentChunk updated = documentChunkService.updateContent( + current.getDocumentCollectionId(), + current.getId(), + request.getContent() + ); + return Result.ok(updated); } @PostMapping("removeChunk") @@ -127,36 +112,58 @@ public class DocumentChunkController extends BaseCurdController remove(@JsonBody(value = "id", required = true) BigInteger chunkId) { - DocumentChunk docChunk = documentChunkService.getById(chunkId); + public Result removeChunk(@JsonBody(value = "id", required = true) BigInteger chunkId) { + DocumentChunk docChunk = documentChunkService.getById(chunkId); if (docChunk == null) { return Result.fail(1, "记录不存在"); } - DocumentCollection knowledge = documentCollectionService.getById(docChunk.getDocumentCollectionId()); - if (knowledge == null) { - return Result.fail(2, "知识库不存在"); - } - DocumentStore documentStore = knowledge.toDocumentStore(); - if (documentStore == null) { - return Result.fail(3, "知识库没有配置向量库"); - } - try { - // 设置向量模型 - Model model = modelService.getModelInstance(knowledge.getVectorEmbedModelId()); - if (model == null) { - return Result.fail(4, "知识库没有配置向量模型"); - } - EmbeddingModel embeddingModel = model.toEmbeddingModel(); - documentStore.setEmbeddingModel(embeddingModel); - StoreOptions options = StoreOptions.ofCollectionName(knowledge.getVectorStoreCollection()); - List deleteList = new ArrayList<>(); - deleteList.add(chunkId); - documentStore.delete(deleteList, options); - documentChunkService.removeChunk(knowledge, chunkId); + return Result.ok(documentChunkService.deleteChunk( + docChunk.getDocumentCollectionId(), + chunkId + )); + } - return super.remove(chunkId); - } finally { - DocumentStoreLifecycleSupport.closeQuietly(documentStore); + @PostMapping("syncStatus") + @SaCheckPermission("/api/v1/documentCollection/query") + @RequireResourceAccess( + resource = CategoryResourceType.KNOWLEDGE, + action = ResourceAction.READ, + lookup = ResourceLookup.DOCUMENT_ID, + idExpr = "#request.documentId", + denyMessage = "无权限访问知识库" + ) + public Result> syncStatus( + @JsonBody(required = true, skipConvertError = false) + DocumentChunkSyncStatusRequest request + ) { + return Result.ok(documentChunkService.listIndexSyncStatus( + null, + request.getDocumentId(), + request.getIds() + )); + } + + @PostMapping("retrySync") + @SaCheckPermission("/api/v1/documentCollection/save") + @RequireResourceAccess( + resource = CategoryResourceType.KNOWLEDGE, + action = ResourceAction.MANAGE, + lookup = ResourceLookup.DOCUMENT_CHUNK_ID, + idExpr = "#request.id", + denyMessage = "无权限管理知识库" + ) + public Result retrySync( + @JsonBody(required = true, skipConvertError = false) + DocumentChunkSyncRetryRequest request + ) { + DocumentChunk current = documentChunkService.getById(request.getId()); + if (current == null || request.getIndexSyncVersion() == null) { + return Result.fail(1, "记录不存在或同步版本缺失"); } + return Result.ok(documentChunkService.retryIndexSync( + current.getDocumentCollectionId(), + current.getId(), + request.getIndexSyncVersion() + )); } } diff --git a/easyflow-api/easyflow-api-admin/src/main/java/tech/easyflow/admin/controller/ai/DocumentController.java b/easyflow-api/easyflow-api-admin/src/main/java/tech/easyflow/admin/controller/ai/DocumentController.java index ad92f16b..19f7cc0b 100644 --- a/easyflow-api/easyflow-api-admin/src/main/java/tech/easyflow/admin/controller/ai/DocumentController.java +++ b/easyflow-api/easyflow-api-admin/src/main/java/tech/easyflow/admin/controller/ai/DocumentController.java @@ -118,12 +118,10 @@ public class DocumentController extends BaseCurdController ids = Collections.singletonList(id); Result result = onRemoveBefore(ids); if (result != null) return result; - boolean isSuccess = documentService.removeDoc(id); - if (!isSuccess){ - return Result.ok(false); + boolean success = documentService.removeDoc(id); + if (success) { + onRemoveAfter(ids); } - boolean success = service.removeById(id); - onRemoveAfter(ids); return Result.ok(success); } diff --git a/easyflow-api/easyflow-api-admin/src/main/java/tech/easyflow/admin/controller/ai/ShareKnowledgeController.java b/easyflow-api/easyflow-api-admin/src/main/java/tech/easyflow/admin/controller/ai/ShareKnowledgeController.java index f3d2f0ec..ad1a1331 100644 --- a/easyflow-api/easyflow-api-admin/src/main/java/tech/easyflow/admin/controller/ai/ShareKnowledgeController.java +++ b/easyflow-api/easyflow-api-admin/src/main/java/tech/easyflow/admin/controller/ai/ShareKnowledgeController.java @@ -1,10 +1,6 @@ package tech.easyflow.admin.controller.ai; import cn.hutool.core.io.IoUtil; -import com.easyagents.core.model.embedding.EmbeddingModel; -import com.easyagents.core.store.DocumentStore; -import com.easyagents.core.store.StoreOptions; -import com.easyagents.core.store.StoreResult; import com.mybatisflex.core.paginate.Page; import com.mybatisflex.core.query.QueryColumn; import com.mybatisflex.core.query.QueryWrapper; @@ -22,6 +18,11 @@ import org.springframework.web.servlet.mvc.method.annotation.SseEmitter; import org.springframework.web.multipart.MultipartFile; import tech.easyflow.ai.documentimport.DocumentImportDtos; import tech.easyflow.ai.documentimport.task.DocumentImportTaskStatusStreamService; +import tech.easyflow.ai.dto.DocumentChunkContentUpdateRequest; +import tech.easyflow.ai.dto.DocumentChunkDeleteResult; +import tech.easyflow.ai.dto.DocumentChunkSyncRetryRequest; +import tech.easyflow.ai.dto.DocumentChunkSyncStatus; +import tech.easyflow.ai.dto.DocumentChunkSyncStatusRequest; import tech.easyflow.ai.dto.KnowledgeShareLimitedConfigRequest; import tech.easyflow.ai.dto.KnowledgeSearchResultItem; import tech.easyflow.ai.entity.Document; @@ -43,7 +44,6 @@ import tech.easyflow.ai.service.KnowledgeEmbeddingService; import tech.easyflow.ai.service.KnowledgeShareAuditService; import tech.easyflow.ai.service.KnowledgeShareService; import tech.easyflow.ai.service.ModelService; -import tech.easyflow.ai.support.DocumentStoreLifecycleSupport; import tech.easyflow.ai.vo.FaqImportResultVo; import tech.easyflow.ai.vo.KnowledgeShareAuthContext; import tech.easyflow.ai.vo.KnowledgeShareViewDetail; @@ -62,7 +62,6 @@ import java.net.URLEncoder; import java.nio.charset.StandardCharsets; import java.text.SimpleDateFormat; import java.util.Arrays; -import java.util.Collections; import java.util.Date; import java.util.HashMap; import java.util.HashSet; @@ -505,43 +504,27 @@ public class ShareKnowledgeController { @PostMapping("/documentChunk/update") public Result updateDocumentChunk( @RequestParam String shareKey, - @JsonBody DocumentChunk documentChunk + @JsonBody(required = true, skipConvertError = false) + DocumentChunkContentUpdateRequest request ) { KnowledgeShareAuthContext context = knowledgeShareService.assertUrlShareAccess( shareKey, null, KnowledgeShareActionScope.CONTENT_UPDATE.name() ); - DocumentChunk current = documentChunkService.getById(documentChunk.getId()); + DocumentChunk current = documentChunkService.getById(request.getId()); if (current == null || current.getDocumentCollectionId() == null || current.getDocumentCollectionId().compareTo(context.getKnowledge().getId()) != 0) { throw new BusinessException("记录不存在"); } - boolean success = documentChunkService.updateById(documentChunk); - if (success) { - DocumentStore documentStore = context.getKnowledge().toDocumentStore(); - if (documentStore == null) { - return Result.fail(2, "知识库没有配置向量库"); - } - try { - Model model = modelService.getModelInstance(context.getKnowledge().getVectorEmbedModelId()); - if (model == null) { - return Result.fail(3, "知识库没有配置向量模型"); - } - EmbeddingModel embeddingModel = model.toEmbeddingModel(); - documentStore.setEmbeddingModel(embeddingModel); - StoreOptions options = StoreOptions.ofCollectionName(context.getKnowledge().getVectorStoreCollection()); - com.easyagents.core.document.Document doc = com.easyagents.core.document.Document.of(documentChunk.getContent()); - doc.setId(documentChunk.getId()); - StoreResult result = documentStore.update(doc, options); - audit(context, "更新分享文档 Chunk", "KNOWLEDGE_SHARE_URL_WRITE", true, - auditDetail("knowledgeId", context.getKnowledge().getId(), "chunkId", documentChunk.getId())); - return Result.ok(result); - } finally { - DocumentStoreLifecycleSupport.closeQuietly(documentStore); - } - } - return Result.ok(false); + DocumentChunk updated = documentChunkService.updateContent( + context.getKnowledge().getId(), + current.getId(), + request.getContent() + ); + audit(context, "更新分享文档 Chunk", "KNOWLEDGE_SHARE_URL_WRITE", true, + auditDetail("knowledgeId", context.getKnowledge().getId(), "chunkId", request.getId())); + return Result.ok(updated); } /** @@ -562,25 +545,50 @@ public class ShareKnowledgeController { || current.getDocumentCollectionId().compareTo(context.getKnowledge().getId()) != 0) { return Result.fail(1, "记录不存在"); } - DocumentStore documentStore = context.getKnowledge().toDocumentStore(); - if (documentStore == null) { - return Result.fail(2, "知识库没有配置向量库"); + DocumentChunkDeleteResult removed = documentChunkService.deleteChunk( + context.getKnowledge().getId(), + chunkId + ); + audit(context, "删除分享文档 Chunk", "KNOWLEDGE_SHARE_URL_WRITE", true, + auditDetail("knowledgeId", context.getKnowledge().getId(), "chunkId", chunkId)); + return Result.ok(removed); + } + + @PostMapping("/documentChunk/syncStatus") + public Result> documentChunkSyncStatus( + @RequestParam String shareKey, + @JsonBody DocumentChunkSyncStatusRequest request + ) { + KnowledgeShareAuthContext context = knowledgeShareService.assertUrlShareAccess( + shareKey, null, KnowledgeShareActionScope.VIEW.name() + ); + Document document = documentService.getById(request.getDocumentId()); + if (document == null || document.getCollectionId() == null + || document.getCollectionId().compareTo(context.getKnowledge().getId()) != 0) { + throw new BusinessException("文档不存在"); } - try { - Model model = modelService.getModelInstance(context.getKnowledge().getVectorEmbedModelId()); - if (model == null) { - return Result.fail(3, "知识库没有配置向量模型"); - } - documentStore.setEmbeddingModel(model.toEmbeddingModel()); - StoreOptions options = StoreOptions.ofCollectionName(context.getKnowledge().getVectorStoreCollection()); - documentStore.delete(Collections.singletonList(chunkId), options); - documentChunkService.removeById(chunkId); - audit(context, "删除分享文档 Chunk", "KNOWLEDGE_SHARE_URL_WRITE", true, - auditDetail("knowledgeId", context.getKnowledge().getId(), "chunkId", chunkId)); - return Result.ok(true); - } finally { - DocumentStoreLifecycleSupport.closeQuietly(documentStore); + return Result.ok(documentChunkService.listIndexSyncStatus( + context.getKnowledge().getId(), request.getDocumentId(), request.getIds() + )); + } + + @PostMapping("/documentChunk/retrySync") + public Result retryDocumentChunkSync( + @RequestParam String shareKey, + @JsonBody DocumentChunkSyncRetryRequest request + ) { + KnowledgeShareAuthContext context = knowledgeShareService.assertUrlShareAccess( + shareKey, null, KnowledgeShareActionScope.CONTENT_UPDATE.name() + ); + if (request.getIndexSyncVersion() == null) { + throw new BusinessException("同步版本不能为空"); } + DocumentChunk chunk = documentChunkService.retryIndexSync( + context.getKnowledge().getId(), request.getId(), request.getIndexSyncVersion() + ); + audit(context, "重试分享文档 Chunk 索引同步", "KNOWLEDGE_SHARE_URL_WRITE", true, + auditDetail("knowledgeId", context.getKnowledge().getId(), "chunkId", request.getId())); + return Result.ok(chunk); } /** diff --git a/easyflow-api/easyflow-api-admin/src/test/java/tech/easyflow/admin/controller/ai/DocumentChunkControllerContractTest.java b/easyflow-api/easyflow-api-admin/src/test/java/tech/easyflow/admin/controller/ai/DocumentChunkControllerContractTest.java new file mode 100644 index 00000000..9dc4d547 --- /dev/null +++ b/easyflow-api/easyflow-api-admin/src/test/java/tech/easyflow/admin/controller/ai/DocumentChunkControllerContractTest.java @@ -0,0 +1,120 @@ +package tech.easyflow.admin.controller.ai; + +import org.testng.Assert; +import org.testng.annotations.Test; +import org.springframework.web.bind.annotation.PostMapping; +import tech.easyflow.ai.dto.DocumentChunkContentUpdateRequest; +import tech.easyflow.ai.entity.DocumentChunk; +import tech.easyflow.ai.service.DocumentChunkService; +import tech.easyflow.common.web.controller.BaseController; +import tech.easyflow.common.web.jsonbody.JsonBody; +import tech.easyflow.system.permission.resource.RequireResourceAccess; + +import java.lang.reflect.Field; +import java.lang.reflect.Method; +import java.lang.reflect.Parameter; +import java.math.BigInteger; +import java.util.Arrays; +import java.util.Set; +import java.util.stream.Collectors; + +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.never; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; + +/** + * 文档分块维护接口契约测试。 + */ +public class DocumentChunkControllerContractTest { + + @Test + public void updateEndpointsShouldOnlyAcceptChunkIdAndContent() throws Exception { + Method adminUpdate = DocumentChunkController.class.getDeclaredMethod( + "update", + DocumentChunkContentUpdateRequest.class + ); + Method shareUpdate = ShareKnowledgeController.class.getDeclaredMethod( + "updateDocumentChunk", + String.class, + DocumentChunkContentUpdateRequest.class + ); + + Assert.assertNotNull(adminUpdate); + Assert.assertNotNull(shareUpdate); + assertStrictJsonBody(adminUpdate.getParameters()[0]); + assertStrictJsonBody(shareUpdate.getParameters()[1]); + Assert.assertEquals( + adminUpdate.getAnnotation(RequireResourceAccess.class).idExpr(), + "#request.id" + ); + Set fields = Arrays.stream( + DocumentChunkContentUpdateRequest.class.getDeclaredFields() + ) + .map(Field::getName) + .collect(Collectors.toSet()); + Assert.assertEquals(Set.of("id", "content"), fields); + Assert.assertEquals( + DocumentChunkController.class.getMethod( + "update", + DocumentChunkContentUpdateRequest.class + ).getDeclaringClass(), + DocumentChunkController.class + ); + Assert.assertEquals( + Arrays.stream(DocumentChunkController.class.getDeclaredMethods()) + .filter(method -> method.getName().equals("update")) + .filter(method -> !method.isBridge() && !method.isSynthetic()) + .count(), + 1L + ); + } + + @Test + public void controllerShouldNotExposeGenericWriteEndpoints() { + Assert.assertEquals(DocumentChunkController.class.getSuperclass(), BaseController.class); + Set postMappings = Arrays.stream(DocumentChunkController.class.getMethods()) + .map(method -> method.getAnnotation(PostMapping.class)) + .filter(annotation -> annotation != null) + .flatMap(annotation -> Arrays.stream(annotation.value())) + .collect(Collectors.toSet()); + + Assert.assertEquals( + postMappings, + Set.of("update", "removeChunk", "syncStatus", "retrySync") + ); + Assert.assertFalse(postMappings.contains("save")); + Assert.assertFalse(postMappings.contains("remove")); + Assert.assertFalse(postMappings.contains("removeBatch")); + } + + @Test + public void adminUpdateAndDeleteShouldUseUnifiedMaintenanceService() { + DocumentChunkService service = mock(DocumentChunkService.class); + DocumentChunkController controller = new DocumentChunkController(service); + + DocumentChunk current = new DocumentChunk(); + current.setId(BigInteger.ONE); + current.setDocumentCollectionId(BigInteger.TWO); + when(service.getById(BigInteger.ONE)).thenReturn(current); + + DocumentChunkContentUpdateRequest request = new DocumentChunkContentUpdateRequest(); + request.setId(BigInteger.ONE); + request.setContent("updated"); + controller.update(request); + controller.removeChunk(BigInteger.ONE); + + verify(service).updateContent(BigInteger.TWO, BigInteger.ONE, "updated"); + verify(service).deleteChunk(BigInteger.TWO, BigInteger.ONE); + verify(service, never()).updateById(any(DocumentChunk.class)); + verify(service, never()).removeById(any()); + } + + private static void assertStrictJsonBody(Parameter parameter) { + JsonBody jsonBody = parameter.getAnnotation(JsonBody.class); + Assert.assertNotNull(jsonBody); + Assert.assertTrue(jsonBody.required()); + Assert.assertFalse(jsonBody.skipConvertError()); + } +} diff --git a/easyflow-api/easyflow-api-public/src/main/java/tech/easyflow/publicapi/controller/PublicKnowledgeShareController.java b/easyflow-api/easyflow-api-public/src/main/java/tech/easyflow/publicapi/controller/PublicKnowledgeShareController.java index a459034e..df65ce03 100644 --- a/easyflow-api/easyflow-api-public/src/main/java/tech/easyflow/publicapi/controller/PublicKnowledgeShareController.java +++ b/easyflow-api/easyflow-api-public/src/main/java/tech/easyflow/publicapi/controller/PublicKnowledgeShareController.java @@ -1,10 +1,6 @@ package tech.easyflow.publicapi.controller; import cn.hutool.core.io.IoUtil; -import com.easyagents.core.model.embedding.EmbeddingModel; -import com.easyagents.core.store.DocumentStore; -import com.easyagents.core.store.StoreOptions; -import com.easyagents.core.store.StoreResult; import com.mybatisflex.core.paginate.Page; import com.mybatisflex.core.query.QueryWrapper; import jakarta.servlet.http.HttpServletRequest; @@ -18,11 +14,15 @@ import org.springframework.web.bind.annotation.RequestParam; import org.springframework.web.bind.annotation.RestController; import org.springframework.web.multipart.MultipartFile; import tech.easyflow.ai.documentimport.DocumentImportDtos; +import tech.easyflow.ai.dto.DocumentChunkDeleteResult; +import tech.easyflow.ai.dto.DocumentChunkAsyncUpdateResult; +import tech.easyflow.ai.dto.DocumentChunkSyncRetryRequest; +import tech.easyflow.ai.dto.DocumentChunkSyncStatus; +import tech.easyflow.ai.dto.DocumentChunkSyncStatusRequest; import tech.easyflow.ai.entity.Document; import tech.easyflow.ai.entity.DocumentChunk; import tech.easyflow.ai.entity.DocumentCollection; import tech.easyflow.ai.entity.FaqItem; -import tech.easyflow.ai.entity.Model; import tech.easyflow.ai.enums.KnowledgeApiPermissionScope; import tech.easyflow.ai.rag.KnowledgeRetrievalModes; import tech.easyflow.ai.rag.KnowledgeRetrievalRequest; @@ -33,9 +33,7 @@ import tech.easyflow.ai.service.FaqCategoryService; import tech.easyflow.ai.service.FaqItemService; import tech.easyflow.ai.service.KnowledgeShareAuditService; import tech.easyflow.ai.service.KnowledgeSharePermissionService; -import tech.easyflow.ai.service.ModelService; import tech.easyflow.ai.service.impl.KnowledgeSharePermissionServiceImpl; -import tech.easyflow.ai.support.DocumentStoreLifecycleSupport; import tech.easyflow.ai.vo.FaqImportResultVo; import tech.easyflow.common.domain.Result; import tech.easyflow.common.filestorage.FileStorageService; @@ -81,8 +79,6 @@ public class PublicKnowledgeShareController { private FaqItemService faqItemService; @Resource private FaqCategoryService faqCategoryService; - @Resource - private ModelService modelService; @Resource(name = "default") private FileStorageService fileStorageService; @@ -351,37 +347,20 @@ public class PublicKnowledgeShareController { public Result updateDocumentChunk( @RequestHeader("ApiKey") String apiKey, @JsonBody(value = "knowledgeId", required = true) BigInteger knowledgeId, - @JsonBody DocumentChunk documentChunk, + @JsonBody(required = true, skipConvertError = false) + DocumentChunk documentChunk, HttpServletRequest request ) { assertApiShare(apiKey, request.getRequestURI(), knowledgeId, KnowledgeApiPermissionScope.KNOWLEDGE_MAINTENANCE.name()); requireDocumentKnowledge(knowledgeId); DocumentChunk current = requireDocumentChunk(documentChunk.getId(), knowledgeId); - boolean success = documentChunkService.updateById(documentChunk); - if (success) { - DocumentCollection knowledge = documentCollectionService.getById(knowledgeId); - DocumentStore documentStore = knowledge.toDocumentStore(); - if (documentStore == null) { - return Result.fail(2, "知识库没有配置向量库"); - } - try { - Model model = modelService.getModelInstance(knowledge.getVectorEmbedModelId()); - if (model == null) { - return Result.fail(3, "知识库没有配置向量模型"); - } - EmbeddingModel embeddingModel = model.toEmbeddingModel(); - documentStore.setEmbeddingModel(embeddingModel); - StoreOptions options = StoreOptions.ofCollectionName(knowledge.getVectorStoreCollection()); - com.easyagents.core.document.Document doc = com.easyagents.core.document.Document.of(documentChunk.getContent()); - doc.setId(current.getId()); - StoreResult result = documentStore.update(doc, options); - audit(apiKey, "API更新文档 Chunk", "KNOWLEDGE_API_SHARE_WRITE", request.getRequestURI(), Map.of("knowledgeId", knowledgeId, "chunkId", documentChunk.getId())); - return Result.ok(result); - } finally { - DocumentStoreLifecycleSupport.closeQuietly(documentStore); - } - } - return Result.ok(false); + DocumentChunk updated = documentChunkService.updateContent( + knowledgeId, + current.getId(), + documentChunk.getContent() + ); + audit(apiKey, "API更新文档 Chunk", "KNOWLEDGE_API_SHARE_WRITE", request.getRequestURI(), Map.of("knowledgeId", knowledgeId, "chunkId", documentChunk.getId())); + return Result.ok(DocumentChunkAsyncUpdateResult.from(updated)); } /** @@ -397,25 +376,51 @@ public class PublicKnowledgeShareController { assertApiShare(apiKey, request.getRequestURI(), knowledgeId, KnowledgeApiPermissionScope.KNOWLEDGE_MAINTENANCE.name()); requireDocumentKnowledge(knowledgeId); requireDocumentChunk(chunkId, knowledgeId); - DocumentCollection knowledge = documentCollectionService.getById(knowledgeId); - DocumentStore documentStore = knowledge.toDocumentStore(); - if (documentStore == null) { - return Result.fail(2, "知识库没有配置向量库"); - } - try { - Model model = modelService.getModelInstance(knowledge.getVectorEmbedModelId()); - if (model == null) { - return Result.fail(3, "知识库没有配置向量模型"); - } - documentStore.setEmbeddingModel(model.toEmbeddingModel()); - StoreOptions options = StoreOptions.ofCollectionName(knowledge.getVectorStoreCollection()); - documentStore.delete(Collections.singletonList(chunkId), options); - documentChunkService.removeById(chunkId); - audit(apiKey, "API删除文档 Chunk", "KNOWLEDGE_API_SHARE_WRITE", request.getRequestURI(), Map.of("knowledgeId", knowledgeId, "chunkId", chunkId)); - return Result.ok(true); - } finally { - DocumentStoreLifecycleSupport.closeQuietly(documentStore); + DocumentChunkDeleteResult removed = documentChunkService.deleteChunk( + knowledgeId, + chunkId + ); + audit(apiKey, "API删除文档 Chunk", "KNOWLEDGE_API_SHARE_WRITE", request.getRequestURI(), Map.of("knowledgeId", knowledgeId, "chunkId", chunkId)); + return Result.ok(removed != null); + } + + @PostMapping("/documentChunk/syncStatus") + public Result> documentChunkSyncStatus( + @RequestHeader("ApiKey") String apiKey, + @JsonBody(value = "knowledgeId", required = true) BigInteger knowledgeId, + @JsonBody(required = true, skipConvertError = false) + DocumentChunkSyncStatusRequest statusRequest, + HttpServletRequest request + ) { + assertApiShare(apiKey, request.getRequestURI(), knowledgeId, + KnowledgeApiPermissionScope.KNOWLEDGE_READ.name()); + requireDocumentKnowledge(knowledgeId); + requireDocument(statusRequest.getDocumentId(), knowledgeId); + return Result.ok(documentChunkService.listIndexSyncStatus( + knowledgeId, statusRequest.getDocumentId(), statusRequest.getIds() + )); + } + + @PostMapping("/documentChunk/retrySync") + public Result retryDocumentChunkSync( + @RequestHeader("ApiKey") String apiKey, + @JsonBody(value = "knowledgeId", required = true) BigInteger knowledgeId, + @JsonBody(required = true, skipConvertError = false) + DocumentChunkSyncRetryRequest retryRequest, + HttpServletRequest request + ) { + assertApiShare(apiKey, request.getRequestURI(), knowledgeId, + KnowledgeApiPermissionScope.KNOWLEDGE_MAINTENANCE.name()); + requireDocumentKnowledge(knowledgeId); + if (retryRequest.getIndexSyncVersion() == null) { + throw new BusinessException("同步版本不能为空"); } + DocumentChunk chunk = documentChunkService.retryIndexSync( + knowledgeId, retryRequest.getId(), retryRequest.getIndexSyncVersion() + ); + audit(apiKey, "API重试文档 Chunk 索引同步", "KNOWLEDGE_API_SHARE_WRITE", + request.getRequestURI(), Map.of("knowledgeId", knowledgeId, "chunkId", retryRequest.getId())); + return Result.ok(chunk); } /** diff --git a/easyflow-api/easyflow-api-public/src/test/java/tech/easyflow/publicapi/controller/PublicKnowledgeShareControllerContractTest.java b/easyflow-api/easyflow-api-public/src/test/java/tech/easyflow/publicapi/controller/PublicKnowledgeShareControllerContractTest.java index 245aefa3..bd4a3011 100644 --- a/easyflow-api/easyflow-api-public/src/test/java/tech/easyflow/publicapi/controller/PublicKnowledgeShareControllerContractTest.java +++ b/easyflow-api/easyflow-api-public/src/test/java/tech/easyflow/publicapi/controller/PublicKnowledgeShareControllerContractTest.java @@ -66,6 +66,8 @@ public class PublicKnowledgeShareControllerContractTest { JsonBody chunkBody = chunk.getAnnotation(JsonBody.class); Assert.assertNotNull(chunkBody); Assert.assertEquals("", chunkBody.value()); + Assert.assertTrue(chunkBody.required()); + Assert.assertFalse(chunkBody.skipConvertError()); } /** diff --git a/easyflow-modules/easyflow-module-ai/pom.xml b/easyflow-modules/easyflow-module-ai/pom.xml index 47efcee1..3402de6b 100644 --- a/easyflow-modules/easyflow-module-ai/pom.xml +++ b/easyflow-modules/easyflow-module-ai/pom.xml @@ -53,6 +53,10 @@ com.easyagents easy-agents-spring-boot-starter + + io.milvus + milvus-sdk-java + com.google.re2j re2j diff --git a/easyflow-modules/easyflow-module-ai/src/main/java/tech/easyflow/ai/config/AiMilvusClientManager.java b/easyflow-modules/easyflow-module-ai/src/main/java/tech/easyflow/ai/config/AiMilvusClientManager.java new file mode 100644 index 00000000..42a78dd2 --- /dev/null +++ b/easyflow-modules/easyflow-module-ai/src/main/java/tech/easyflow/ai/config/AiMilvusClientManager.java @@ -0,0 +1,23 @@ +package tech.easyflow.ai.config; + +import com.easyagents.store.milvus.MilvusClientManager; +import jakarta.annotation.PreDestroy; +import org.springframework.stereotype.Component; +import org.springframework.context.annotation.Lazy; + +/** + * 应用级 Milvus 客户端池。 + */ +@Component +@Lazy +public class AiMilvusClientManager extends MilvusClientManager { + + public AiMilvusClientManager(AiMilvusConfig config) { + super(config); + } + + @PreDestroy + public void destroy() { + close(); + } +} diff --git a/easyflow-modules/easyflow-module-ai/src/main/java/tech/easyflow/ai/config/AiMilvusConfig.java b/easyflow-modules/easyflow-module-ai/src/main/java/tech/easyflow/ai/config/AiMilvusConfig.java index 99211062..e17d77d5 100644 --- a/easyflow-modules/easyflow-module-ai/src/main/java/tech/easyflow/ai/config/AiMilvusConfig.java +++ b/easyflow-modules/easyflow-module-ai/src/main/java/tech/easyflow/ai/config/AiMilvusConfig.java @@ -17,6 +17,13 @@ public class AiMilvusConfig extends MilvusVectorStoreConfig { config.setPassword(getPassword()); config.setAutoCreateCollection(isAutoCreateCollection()); config.setDefaultCollectionName(collectionName); + config.setPoolMaxTotal(getPoolMaxTotal()); + config.setPoolMaxTotalPerKey(getPoolMaxTotalPerKey()); + config.setPoolMaxIdlePerKey(getPoolMaxIdlePerKey()); + config.setPoolMinIdlePerKey(getPoolMinIdlePerKey()); + config.setPoolMaxWaitMillis(getPoolMaxWaitMillis()); + config.setPoolEvictionIntervalMillis(getPoolEvictionIntervalMillis()); + config.setPoolMinEvictableIdleMillis(getPoolMinEvictableIdleMillis()); return config; } } diff --git a/easyflow-modules/easyflow-module-ai/src/main/java/tech/easyflow/ai/config/RagHealthIndicator.java b/easyflow-modules/easyflow-module-ai/src/main/java/tech/easyflow/ai/config/RagHealthIndicator.java index 2bbd356a..d6bc93c4 100644 --- a/easyflow-modules/easyflow-module-ai/src/main/java/tech/easyflow/ai/config/RagHealthIndicator.java +++ b/easyflow-modules/easyflow-module-ai/src/main/java/tech/easyflow/ai/config/RagHealthIndicator.java @@ -25,16 +25,23 @@ public class RagHealthIndicator { public static class RagMilvusHealthIndicator extends CachedHealthIndicatorSupport implements HealthIndicator { private final AiMilvusConfig aiMilvusConfig; + private final AiMilvusClientManager milvusClientManager; /** * 创建 Milvus 健康检查器。 * * @param aiMilvusConfig Milvus 配置 + * @param milvusClientManager 应用级 Milvus 客户端池 * @param healthProperties RAG 健康检查配置 */ - public RagMilvusHealthIndicator(AiMilvusConfig aiMilvusConfig, RagHealthProperties healthProperties) { + public RagMilvusHealthIndicator( + AiMilvusConfig aiMilvusConfig, + AiMilvusClientManager milvusClientManager, + RagHealthProperties healthProperties + ) { super(healthProperties); this.aiMilvusConfig = aiMilvusConfig; + this.milvusClientManager = milvusClientManager; } /** @@ -51,8 +58,10 @@ public class RagHealthIndicator { protected Health doHealthCheck() { MilvusVectorStore vectorStore = null; try { + milvusClientManager.reconfigureIfNeeded(aiMilvusConfig); vectorStore = new MilvusVectorStore( - aiMilvusConfig.copyForCollection("__rag_health_probe__") + aiMilvusConfig.copyForCollection("__rag_health_probe__"), + milvusClientManager ); if (vectorStore.checkAvailable()) { return Health.up().withDetail("uri", aiMilvusConfig.getUri()).build(); diff --git a/easyflow-modules/easyflow-module-ai/src/main/java/tech/easyflow/ai/documentchunk/DocumentChunkSyncState.java b/easyflow-modules/easyflow-module-ai/src/main/java/tech/easyflow/ai/documentchunk/DocumentChunkSyncState.java new file mode 100644 index 00000000..9fb24969 --- /dev/null +++ b/easyflow-modules/easyflow-module-ai/src/main/java/tech/easyflow/ai/documentchunk/DocumentChunkSyncState.java @@ -0,0 +1,34 @@ +package tech.easyflow.ai.documentchunk; + +/** + * 分块检索索引同步状态常量。 + */ +public final class DocumentChunkSyncState { + + private static final String PARENT_LOCK_PREFIX = + "easyflow:lock:document-chunk-parent:"; + private static final String SYNC_LOCK_PREFIX = + "easyflow:lock:document-chunk-sync:"; + + public static final String SYNCED = "SYNCED"; + public static final String PENDING = "PENDING"; + public static final String FAILED = "FAILED"; + + public static final String TASK_RUNNING = "RUNNING"; + public static final String TASK_SUCCEEDED = "SUCCEEDED"; + public static final String TASK_SUPERSEDED = "SUPERSEDED"; + + public static final String OPERATION_UPSERT = "UPSERT"; + public static final String OPERATION_DELETE = "DELETE"; + + public static String parentLockKey(Object documentId) { + return PARENT_LOCK_PREFIX + documentId; + } + + public static String syncLockKey(Object chunkId) { + return SYNC_LOCK_PREFIX + chunkId; + } + + private DocumentChunkSyncState() { + } +} diff --git a/easyflow-modules/easyflow-module-ai/src/main/java/tech/easyflow/ai/documentchunk/DocumentChunkSyncTaskAppService.java b/easyflow-modules/easyflow-module-ai/src/main/java/tech/easyflow/ai/documentchunk/DocumentChunkSyncTaskAppService.java new file mode 100644 index 00000000..d97d2f12 --- /dev/null +++ b/easyflow-modules/easyflow-module-ai/src/main/java/tech/easyflow/ai/documentchunk/DocumentChunkSyncTaskAppService.java @@ -0,0 +1,414 @@ +package tech.easyflow.ai.documentchunk; + +import com.easyagents.core.model.embedding.EmbeddingModel; +import com.easyagents.core.model.embedding.EmbeddingOptions; +import com.easyagents.core.store.DocumentStore; +import com.easyagents.core.store.StoreOptions; +import com.easyagents.core.store.StoreResult; +import com.easyagents.search.engine.service.DocumentSearcher; +import com.easyagents.search.engine.service.KeywordSearchMetadataKeys; +import com.easyagents.store.milvus.MilvusVectorStore; +import com.easyagents.store.milvus.MilvusVectorStoreConfig; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import org.springframework.stereotype.Service; +import org.springframework.beans.factory.ObjectProvider; +import org.springframework.transaction.PlatformTransactionManager; +import org.springframework.transaction.support.TransactionTemplate; +import tech.easyflow.ai.config.AiMilvusClientManager; +import tech.easyflow.ai.config.AiMilvusConfig; +import tech.easyflow.ai.config.SearcherFactory; +import tech.easyflow.ai.entity.DocumentChunk; +import tech.easyflow.ai.entity.DocumentChunkSyncTask; +import tech.easyflow.ai.entity.DocumentCollection; +import tech.easyflow.ai.entity.Model; +import tech.easyflow.ai.mapper.DocumentChunkMapper; +import tech.easyflow.ai.mapper.DocumentChunkSyncTaskMapper; +import tech.easyflow.ai.service.DocumentCollectionService; +import tech.easyflow.ai.service.ModelService; +import tech.easyflow.ai.support.DocumentStoreLifecycleSupport; +import tech.easyflow.common.cache.RedisLockExecutor; + +import java.math.BigInteger; +import java.time.Duration; +import java.util.Collections; +import java.util.Date; +import java.util.UUID; + +/** + * 持久化分块索引同步任务的投递、执行和恢复。 + */ +@Service +public class DocumentChunkSyncTaskAppService { + + private static final Logger LOG = LoggerFactory.getLogger(DocumentChunkSyncTaskAppService.class); + private static final int MAX_ATTEMPTS = 5; + private static final int DISPATCH_LIMIT = 100; + private static final long REDISPATCH_MILLIS = 5_000L; + private static final long LEASE_MILLIS = 300_000L; + + private final DocumentChunkSyncTaskMapper taskMapper; + private final DocumentChunkMapper chunkMapper; + private final DocumentCollectionService collectionService; + private final ModelService modelService; + private final SearcherFactory searcherFactory; + private final DocumentChunkSyncTaskProducer producer; + private final PlatformTransactionManager transactionManager; + private final RedisLockExecutor redisLockExecutor; + private final AiMilvusConfig milvusConfig; + private final ObjectProvider milvusClientManagerProvider; + + public DocumentChunkSyncTaskAppService( + DocumentChunkSyncTaskMapper taskMapper, + DocumentChunkMapper chunkMapper, + DocumentCollectionService collectionService, + ModelService modelService, + SearcherFactory searcherFactory, + DocumentChunkSyncTaskProducer producer, + PlatformTransactionManager transactionManager, + RedisLockExecutor redisLockExecutor, + AiMilvusConfig milvusConfig, + ObjectProvider milvusClientManagerProvider + ) { + this.taskMapper = taskMapper; + this.chunkMapper = chunkMapper; + this.collectionService = collectionService; + this.modelService = modelService; + this.searcherFactory = searcherFactory; + this.producer = producer; + this.transactionManager = transactionManager; + this.redisLockExecutor = redisLockExecutor; + this.milvusConfig = milvusConfig; + this.milvusClientManagerProvider = milvusClientManagerProvider; + } + + public DocumentChunkSyncTask createTask( + DocumentChunk chunk, + DocumentCollection collection, + String operation, + long version, + Date now + ) { + taskMapper.supersedeOlder(chunk.getId(), version, now); + DocumentChunkSyncTask task = new DocumentChunkSyncTask(); + task.setChunkId(chunk.getId()); + task.setDocumentId(chunk.getDocumentId()); + task.setDocumentCollectionId(chunk.getDocumentCollectionId()); + task.setVectorCollection(collection.getVectorStoreCollection()); + task.setOperation(operation); + task.setSyncVersion(version); + task.setStatus(DocumentChunkSyncState.PENDING); + task.setAttemptCount(0); + task.setNextRetryAt(now); + task.setCreated(now); + task.setModified(now); + if (taskMapper.insert(task) <= 0) { + throw new IllegalStateException("创建分块索引同步任务失败"); + } + return task; + } + + public void dispatchBestEffort(BigInteger taskId) { + try { + producer.send(taskId); + } catch (RuntimeException exception) { + LOG.warn("分块索引同步消息投递失败，等待数据库补投: taskId={}", taskId, exception); + } + } + + public void dispatchPendingTasks() { + Date now = new Date(); + taskMapper.recoverExpired(now); + Date redispatchBefore = new Date(now.getTime() - REDISPATCH_MILLIS); + for (DocumentChunkSyncTask task : taskMapper.selectPendingDue( + now, + redispatchBefore, + DISPATCH_LIMIT + )) { + if (taskMapper.markDispatched(task.getId(), now, redispatchBefore) <= 0) { + continue; + } + try { + producer.send(task.getId()); + } catch (RuntimeException exception) { + LOG.warn("补投分块索引同步消息失败: taskId={}", task.getId(), exception); + } + } + } + + public void handleTask(BigInteger taskId) { + DocumentChunkSyncTask snapshot = taskMapper.selectOneById(taskId); + if (snapshot == null || !DocumentChunkSyncState.PENDING.equals(snapshot.getStatus())) { + return; + } + redisLockExecutor.executeWithRenewingLock( + DocumentChunkSyncState.syncLockKey(snapshot.getChunkId()), + Duration.ofSeconds(3), + Duration.ofMinutes(5), + () -> { + claimAndExecute(taskId, snapshot.getChunkId()); + return null; + } + ); + } + + private void claimAndExecute(BigInteger taskId, BigInteger chunkId) { + Date now = new Date(); + String token = UUID.randomUUID().toString(); + if (taskMapper.claim( + taskId, + token, + new Date(now.getTime() + LEASE_MILLIS), + now + ) <= 0) { + return; + } + new TransactionTemplate(transactionManager).executeWithoutResult(status -> + executeOwnedTask(taskId, chunkId, token) + ); + } + + private void executeOwnedTask( + BigInteger taskId, + BigInteger chunkId, + String token + ) { + // FOR UPDATE 必须是本事务的第一次读取。MySQL REPEATABLE READ 下， + // 若先做普通查询会建立旧快照，使删除或新版本建单已经提交后仍读取旧分块。 + taskMapper.lockChunkTasks(chunkId); + DocumentChunkSyncTask task = taskMapper.selectOneById(taskId); + if (task == null + || !DocumentChunkSyncState.TASK_RUNNING.equals(task.getStatus()) + || !token.equals(task.getExecutionToken())) { + return; + } + if (isSuperseded(task)) { + finishTask(task, token, DocumentChunkSyncState.TASK_SUPERSEDED, null, null); + return; + } + try { + synchronizeIndexes(task); + finishSuccess(task, token); + } catch (IndexSyncException exception) { + LOG.warn("分块索引同步失败: taskId={}, chunkId={}, code={}", + task.getId(), task.getChunkId(), exception.code, exception.getCause()); + finishFailure(task, token, exception.code, exception.getMessage()); + } catch (RuntimeException exception) { + LOG.error("分块索引同步发生未分类异常: taskId={}, chunkId={}", + task.getId(), task.getChunkId(), exception); + finishFailure(task, token, "INDEX_SYNC_FAILED", "检索索引同步失败"); + } + } + + private boolean isSuperseded(DocumentChunkSyncTask task) { + DocumentChunk chunk = chunkMapper.selectOneById(task.getChunkId()); + if (DocumentChunkSyncState.OPERATION_DELETE.equals(task.getOperation())) { + return chunk != null; + } + return chunk == null + || chunk.getIndexSyncVersion() == null + || chunk.getIndexSyncVersion().longValue() != task.getSyncVersion().longValue() + || !DocumentChunkSyncState.PENDING.equals(chunk.getIndexSyncStatus()); + } + + private void synchronizeIndexes(DocumentChunkSyncTask task) { + if (DocumentChunkSyncState.OPERATION_DELETE.equals(task.getOperation())) { + synchronizeDelete(task); + return; + } + synchronizeUpsert(task); + } + + private void synchronizeUpsert(DocumentChunkSyncTask task) { + DocumentChunk chunk = chunkMapper.selectOneById(task.getChunkId()); + DocumentCollection collection = collectionService.getById(task.getDocumentCollectionId()); + if (chunk == null || collection == null) { + throw new IndexSyncException("INDEX_SOURCE_MISSING", "分块或知识库不存在", null); + } + StoreContext context = prepareUpsertContext(collection, task.getVectorCollection()); + try { + com.easyagents.core.document.Document document = toSearchDocument(chunk, task.getDocumentCollectionId()); + StoreResult vectorResult = context.documentStore.update( + Collections.singletonList(document), + context.storeOptions + ); + if (vectorResult == null || !vectorResult.isSuccess()) { + throw new IndexSyncException("VECTOR_UPSERT_FAILED", "向量索引更新失败", null); + } + if (context.searcher != null + && !context.searcher.addDocuments(Collections.singletonList(document))) { + throw new IndexSyncException("KEYWORD_UPSERT_FAILED", "关键词索引更新失败", null); + } + } catch (IndexSyncException exception) { + throw exception; + } catch (RuntimeException exception) { + throw new IndexSyncException("INDEX_UPSERT_FAILED", "检索索引更新失败", exception); + } finally { + DocumentStoreLifecycleSupport.closeQuietly(context.documentStore); + } + } + + private void synchronizeDelete(DocumentChunkSyncTask task) { + MilvusVectorStoreConfig storeConfig = milvusConfig.copyForCollection(task.getVectorCollection()); + AiMilvusClientManager clientManager = milvusClientManagerProvider.getObject(); + clientManager.reconfigureIfNeeded(milvusConfig); + DocumentStore documentStore = new MilvusVectorStore( + storeConfig, + clientManager + ); + StoreOptions storeOptions = StoreOptions.ofCollectionName(task.getVectorCollection()); + DocumentSearcher searcher = searcherFactory.getSearcher(); + try { + StoreResult result = documentStore.delete( + Collections.singletonList(task.getChunkId().toString()), + storeOptions + ); + if (result == null || !result.isSuccess()) { + throw new IndexSyncException("VECTOR_DELETE_FAILED", "向量索引删除失败", null); + } + if (searcher != null && !searcher.deleteDocument(task.getChunkId())) { + throw new IndexSyncException("KEYWORD_DELETE_FAILED", "关键词索引删除失败", null); + } + } catch (IndexSyncException exception) { + throw exception; + } catch (RuntimeException exception) { + throw new IndexSyncException("INDEX_DELETE_FAILED", "检索索引删除失败", exception); + } finally { + DocumentStoreLifecycleSupport.closeQuietly(documentStore); + } + } + + private StoreContext prepareUpsertContext( + DocumentCollection collection, + String vectorCollection + ) { + DocumentStore documentStore = collection.toDocumentStore(); + if (documentStore == null) { + throw new IndexSyncException("VECTOR_STORE_MISSING", "知识库没有配置向量库", null); + } + try { + Model model = modelService.getModelInstance(collection.getVectorEmbedModelId()); + if (model == null) { + throw new IndexSyncException("EMBEDDING_MODEL_MISSING", "知识库没有配置向量模型", null); + } + EmbeddingModel embeddingModel = model.toEmbeddingModel(); + documentStore.setEmbeddingModel(embeddingModel); + StoreOptions options = StoreOptions.ofCollectionName(vectorCollection); + EmbeddingOptions embeddingOptions = new EmbeddingOptions(); + embeddingOptions.setModel(model.getModelName()); + embeddingOptions.setDimensions(collection.getDimensionOfVectorModel()); + options.setEmbeddingOptions(embeddingOptions); + options.setIndexName(vectorCollection); + return new StoreContext(documentStore, options, searcherFactory.getSearcher()); + } catch (RuntimeException exception) { + DocumentStoreLifecycleSupport.closeQuietly(documentStore); + throw exception; + } + } + + private com.easyagents.core.document.Document toSearchDocument( + DocumentChunk chunk, + BigInteger knowledgeId + ) { + com.easyagents.core.document.Document document = + com.easyagents.core.document.Document.of(chunk.getContent()); + document.setId(chunk.getId()); + document.addMetadata(KeywordSearchMetadataKeys.KNOWLEDGE_ID, knowledgeId.toString()); + return document; + } + + private void finishSuccess(DocumentChunkSyncTask task, String token) { + new TransactionTemplate(transactionManager).executeWithoutResult(status -> { + if (taskMapper.finishOwned( + task.getId(), token, DocumentChunkSyncState.TASK_SUCCEEDED, null, null, new Date() + ) > 0 && DocumentChunkSyncState.OPERATION_UPSERT.equals(task.getOperation())) { + chunkMapper.updateSyncState( + task.getChunkId(), task.getSyncVersion(), DocumentChunkSyncState.SYNCED, null, null + ); + } + }); + } + + private void finishTask( + DocumentChunkSyncTask task, + String token, + String status, + String errorCode, + String errorMessage + ) { + taskMapper.finishOwned(task.getId(), token, status, errorCode, errorMessage, new Date()); + } + + private void finishFailure( + DocumentChunkSyncTask task, + String token, + String errorCode, + String errorMessage + ) { + int attempts = task.getAttemptCount() == null ? 1 : task.getAttemptCount(); + boolean deleteOperation = DocumentChunkSyncState.OPERATION_DELETE.equals( + task.getOperation() + ); + // 删除后已没有页面实体承载手动重试入口，因此清理任务必须保留为持久化 tombstone。 + boolean exhausted = !deleteOperation && attempts >= MAX_ATTEMPTS; + String nextStatus = exhausted ? DocumentChunkSyncState.FAILED : DocumentChunkSyncState.PENDING; + long delaySeconds = Math.min(1L << Math.min(attempts, 6), 60L); + Date now = new Date(); + Date nextRetryAt = exhausted ? now : new Date(now.getTime() + delaySeconds * 1_000L); + new TransactionTemplate(transactionManager).executeWithoutResult(status -> { + if (taskMapper.failOrRetryOwned( + task.getId(), token, nextStatus, nextRetryAt, errorCode, errorMessage, now + ) > 0 && exhausted && DocumentChunkSyncState.OPERATION_UPSERT.equals(task.getOperation())) { + chunkMapper.updateSyncState( + task.getChunkId(), task.getSyncVersion(), DocumentChunkSyncState.FAILED, + errorCode, errorMessage + ); + } + }); + } + + public DocumentChunk retry(BigInteger chunkId, long syncVersion) { + Date now = new Date(); + DocumentChunk chunk = chunkMapper.selectOneById(chunkId); + if (chunk == null || chunk.getIndexSyncVersion() == null + || chunk.getIndexSyncVersion().longValue() != syncVersion + || !DocumentChunkSyncState.FAILED.equals(chunk.getIndexSyncStatus())) { + throw new IllegalStateException("分块同步状态已变化，请刷新后重试"); + } + DocumentChunkSyncTask task = taskMapper.selectFailed(chunkId, syncVersion); + if (task == null) { + throw new IllegalStateException("未找到可重试的索引同步任务"); + } + new TransactionTemplate(transactionManager).executeWithoutResult(status -> { + if (taskMapper.retryFailed(task.getId(), chunkId, syncVersion, now) <= 0) { + throw new IllegalStateException("索引同步任务状态已变化"); + } + if (chunkMapper.updateSyncState( + chunkId, syncVersion, DocumentChunkSyncState.PENDING, null, null + ) <= 0) { + throw new IllegalStateException("分块同步状态已变化"); + } + }); + dispatchBestEffort(task.getId()); + chunk.setIndexSyncStatus(DocumentChunkSyncState.PENDING); + chunk.setIndexSyncErrorCode(null); + chunk.setIndexSyncErrorMessage(null); + return chunk; + } + + private record StoreContext( + DocumentStore documentStore, + StoreOptions storeOptions, + DocumentSearcher searcher + ) { + } + + private static final class IndexSyncException extends RuntimeException { + private final String code; + + private IndexSyncException(String code, String message, Throwable cause) { + super(message, cause); + this.code = code; + } + } +} diff --git a/easyflow-modules/easyflow-module-ai/src/main/java/tech/easyflow/ai/documentchunk/DocumentChunkSyncTaskConsumer.java b/easyflow-modules/easyflow-module-ai/src/main/java/tech/easyflow/ai/documentchunk/DocumentChunkSyncTaskConsumer.java new file mode 100644 index 00000000..16395120 --- /dev/null +++ b/easyflow-modules/easyflow-module-ai/src/main/java/tech/easyflow/ai/documentchunk/DocumentChunkSyncTaskConsumer.java @@ -0,0 +1,66 @@ +package tech.easyflow.ai.documentchunk; + +import com.alibaba.fastjson2.JSON; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import org.springframework.stereotype.Component; +import tech.easyflow.common.mq.config.MQProperties; +import tech.easyflow.common.mq.core.MQConsumerHandler; +import tech.easyflow.common.mq.core.MQDeferException; +import tech.easyflow.common.mq.core.MQMessage; +import tech.easyflow.common.mq.core.MQSubscription; + +import java.util.List; + +/** + * 分块索引同步消息消费者。 + */ +@Component +public class DocumentChunkSyncTaskConsumer implements MQConsumerHandler { + + private static final Logger LOG = LoggerFactory.getLogger(DocumentChunkSyncTaskConsumer.class); + + private final DocumentChunkSyncTaskAppService appService; + private final MQProperties mqProperties; + + public DocumentChunkSyncTaskConsumer( + DocumentChunkSyncTaskAppService appService, + MQProperties mqProperties + ) { + this.appService = appService; + this.mqProperties = mqProperties; + } + + @Override + public MQSubscription subscription() { + MQSubscription subscription = new MQSubscription(); + subscription.setTopic(DocumentChunkSyncTaskProducer.TOPIC); + subscription.setConsumerGroup(DocumentChunkSyncTaskProducer.GROUP); + subscription.setShardCount(Math.max( + mqProperties.getRedis().getChatPersistShardCount(), + 1 + )); + subscription.setBatchEnabled(false); + return subscription; + } + + @Override + public void handle(List messages) { + for (MQMessage message : messages) { + DocumentChunkSyncTaskMessage event = JSON.parseObject( + message.getBody(), + DocumentChunkSyncTaskMessage.class + ); + if (event == null || event.getTaskId() == null) { + LOG.warn("忽略非法分块索引同步消息: messageId={}", message.getMessageId()); + continue; + } + try { + appService.handleTask(event.getTaskId()); + } catch (RuntimeException exception) { + LOG.error("分块索引同步任务状态处理失败: taskId={}", event.getTaskId(), exception); + throw new MQDeferException("分块索引同步任务暂时无法处理", exception); + } + } + } +} diff --git a/easyflow-modules/easyflow-module-ai/src/main/java/tech/easyflow/ai/documentchunk/DocumentChunkSyncTaskMessage.java b/easyflow-modules/easyflow-module-ai/src/main/java/tech/easyflow/ai/documentchunk/DocumentChunkSyncTaskMessage.java new file mode 100644 index 00000000..86d44b77 --- /dev/null +++ b/easyflow-modules/easyflow-module-ai/src/main/java/tech/easyflow/ai/documentchunk/DocumentChunkSyncTaskMessage.java @@ -0,0 +1,18 @@ +package tech.easyflow.ai.documentchunk; + +import java.io.Serializable; +import java.math.BigInteger; +import java.util.Date; + +/** + * 分块索引同步消息。 + */ +public class DocumentChunkSyncTaskMessage implements Serializable { + private BigInteger taskId; + private Date occurredAt; + + public BigInteger getTaskId() { return taskId; } + public void setTaskId(BigInteger taskId) { this.taskId = taskId; } + public Date getOccurredAt() { return occurredAt; } + public void setOccurredAt(Date occurredAt) { this.occurredAt = occurredAt; } +} diff --git a/easyflow-modules/easyflow-module-ai/src/main/java/tech/easyflow/ai/documentchunk/DocumentChunkSyncTaskMonitor.java b/easyflow-modules/easyflow-module-ai/src/main/java/tech/easyflow/ai/documentchunk/DocumentChunkSyncTaskMonitor.java new file mode 100644 index 00000000..c9b43aff --- /dev/null +++ b/easyflow-modules/easyflow-module-ai/src/main/java/tech/easyflow/ai/documentchunk/DocumentChunkSyncTaskMonitor.java @@ -0,0 +1,33 @@ +package tech.easyflow.ai.documentchunk; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import org.springframework.scheduling.annotation.Scheduled; +import org.springframework.stereotype.Component; +import tech.easyflow.common.cache.DistributedScheduledLock; + +/** + * 补投待同步任务并回收过期租约。 + */ +@Component +public class DocumentChunkSyncTaskMonitor { + + private static final Logger LOG = LoggerFactory.getLogger(DocumentChunkSyncTaskMonitor.class); + + private final DocumentChunkSyncTaskAppService appService; + + public DocumentChunkSyncTaskMonitor(DocumentChunkSyncTaskAppService appService) { + this.appService = appService; + } + + @Scheduled(fixedDelayString = "${easyflow.ai.document-chunk-sync.dispatch-interval:2s}", + initialDelayString = "${easyflow.ai.document-chunk-sync.dispatch-interval:2s}") + @DistributedScheduledLock(key = "easyflow:schedule:document-chunk-index-sync", leaseSeconds = 2L) + public void dispatchPendingTasks() { + try { + appService.dispatchPendingTasks(); + } catch (RuntimeException exception) { + LOG.error("分块索引同步补投失败", exception); + } + } +} diff --git a/easyflow-modules/easyflow-module-ai/src/main/java/tech/easyflow/ai/documentchunk/DocumentChunkSyncTaskProducer.java b/easyflow-modules/easyflow-module-ai/src/main/java/tech/easyflow/ai/documentchunk/DocumentChunkSyncTaskProducer.java new file mode 100644 index 00000000..609c912f --- /dev/null +++ b/easyflow-modules/easyflow-module-ai/src/main/java/tech/easyflow/ai/documentchunk/DocumentChunkSyncTaskProducer.java @@ -0,0 +1,39 @@ +package tech.easyflow.ai.documentchunk; + +import com.alibaba.fastjson2.JSON; +import org.springframework.stereotype.Service; +import tech.easyflow.common.mq.core.MQMessage; +import tech.easyflow.common.mq.core.MQProducer; + +import java.math.BigInteger; +import java.util.Date; + +/** + * 分块索引同步消息生产者。 + */ +@Service +public class DocumentChunkSyncTaskProducer { + + static final String TOPIC = "document-chunk-index-sync"; + static final String GROUP = "document-chunk-index-sync-group"; + + private final MQProducer mqProducer; + + public DocumentChunkSyncTaskProducer(MQProducer mqProducer) { + this.mqProducer = mqProducer; + } + + public void send(BigInteger taskId) { + Date now = new Date(); + DocumentChunkSyncTaskMessage event = new DocumentChunkSyncTaskMessage(); + event.setTaskId(taskId); + event.setOccurredAt(now); + MQMessage message = new MQMessage(); + message.setMessageId("chunk-sync-" + taskId); + message.setTopic(TOPIC); + message.setKey(String.valueOf(taskId)); + message.setCreatedAt(now); + message.setBody(JSON.toJSONString(event)); + mqProducer.send(message); + } +} diff --git a/easyflow-modules/easyflow-module-ai/src/main/java/tech/easyflow/ai/dto/DocumentChunkAsyncUpdateResult.java b/easyflow-modules/easyflow-module-ai/src/main/java/tech/easyflow/ai/dto/DocumentChunkAsyncUpdateResult.java new file mode 100644 index 00000000..02ffb088 --- /dev/null +++ b/easyflow-modules/easyflow-module-ai/src/main/java/tech/easyflow/ai/dto/DocumentChunkAsyncUpdateResult.java @@ -0,0 +1,24 @@ +package tech.easyflow.ai.dto; + +import tech.easyflow.ai.entity.DocumentChunk; + +import java.math.BigInteger; + +/** + * 兼容 success 字段的异步分块更新结果。 + */ +public record DocumentChunkAsyncUpdateResult( + boolean success, + BigInteger id, + String indexSyncStatus, + Long indexSyncVersion +) { + public static DocumentChunkAsyncUpdateResult from(DocumentChunk chunk) { + return new DocumentChunkAsyncUpdateResult( + true, + chunk.getId(), + chunk.getIndexSyncStatus(), + chunk.getIndexSyncVersion() + ); + } +} diff --git a/easyflow-modules/easyflow-module-ai/src/main/java/tech/easyflow/ai/dto/DocumentChunkContentUpdateRequest.java b/easyflow-modules/easyflow-module-ai/src/main/java/tech/easyflow/ai/dto/DocumentChunkContentUpdateRequest.java new file mode 100644 index 00000000..06809d51 --- /dev/null +++ b/easyflow-modules/easyflow-module-ai/src/main/java/tech/easyflow/ai/dto/DocumentChunkContentUpdateRequest.java @@ -0,0 +1,29 @@ +package tech.easyflow.ai.dto; + +import java.io.Serializable; +import java.math.BigInteger; + +/** + * 分块正文更新请求，只允许修改 Markdown 内容。 + */ +public class DocumentChunkContentUpdateRequest implements Serializable { + + private BigInteger id; + private String content; + + public BigInteger getId() { + return id; + } + + public void setId(BigInteger id) { + this.id = id; + } + + public String getContent() { + return content; + } + + public void setContent(String content) { + this.content = content; + } +} diff --git a/easyflow-modules/easyflow-module-ai/src/main/java/tech/easyflow/ai/dto/DocumentChunkDeleteResult.java b/easyflow-modules/easyflow-module-ai/src/main/java/tech/easyflow/ai/dto/DocumentChunkDeleteResult.java new file mode 100644 index 00000000..fa542b2a --- /dev/null +++ b/easyflow-modules/easyflow-module-ai/src/main/java/tech/easyflow/ai/dto/DocumentChunkDeleteResult.java @@ -0,0 +1,14 @@ +package tech.easyflow.ai.dto; + +import java.io.Serializable; +import java.math.BigInteger; + +/** + * 分块删除结果。 + */ +public record DocumentChunkDeleteResult( + BigInteger id, + BigInteger documentId, + long remainingChunkCount +) implements Serializable { +} diff --git a/easyflow-modules/easyflow-module-ai/src/main/java/tech/easyflow/ai/dto/DocumentChunkSyncRetryRequest.java b/easyflow-modules/easyflow-module-ai/src/main/java/tech/easyflow/ai/dto/DocumentChunkSyncRetryRequest.java new file mode 100644 index 00000000..106d13f0 --- /dev/null +++ b/easyflow-modules/easyflow-module-ai/src/main/java/tech/easyflow/ai/dto/DocumentChunkSyncRetryRequest.java @@ -0,0 +1,16 @@ +package tech.easyflow.ai.dto; + +import java.math.BigInteger; + +/** + * 重试分块索引同步请求。 + */ +public class DocumentChunkSyncRetryRequest { + private BigInteger id; + private Long indexSyncVersion; + + public BigInteger getId() { return id; } + public void setId(BigInteger id) { this.id = id; } + public Long getIndexSyncVersion() { return indexSyncVersion; } + public void setIndexSyncVersion(Long indexSyncVersion) { this.indexSyncVersion = indexSyncVersion; } +} diff --git a/easyflow-modules/easyflow-module-ai/src/main/java/tech/easyflow/ai/dto/DocumentChunkSyncStatus.java b/easyflow-modules/easyflow-module-ai/src/main/java/tech/easyflow/ai/dto/DocumentChunkSyncStatus.java new file mode 100644 index 00000000..d1f01347 --- /dev/null +++ b/easyflow-modules/easyflow-module-ai/src/main/java/tech/easyflow/ai/dto/DocumentChunkSyncStatus.java @@ -0,0 +1,15 @@ +package tech.easyflow.ai.dto; + +import java.math.BigInteger; + +/** + * 分块索引同步状态。 + */ +public record DocumentChunkSyncStatus( + BigInteger id, + String indexSyncStatus, + Long indexSyncVersion, + String indexSyncErrorCode, + String indexSyncErrorMessage +) { +} diff --git a/easyflow-modules/easyflow-module-ai/src/main/java/tech/easyflow/ai/dto/DocumentChunkSyncStatusRequest.java b/easyflow-modules/easyflow-module-ai/src/main/java/tech/easyflow/ai/dto/DocumentChunkSyncStatusRequest.java new file mode 100644 index 00000000..1e1231b9 --- /dev/null +++ b/easyflow-modules/easyflow-module-ai/src/main/java/tech/easyflow/ai/dto/DocumentChunkSyncStatusRequest.java @@ -0,0 +1,17 @@ +package tech.easyflow.ai.dto; + +import java.math.BigInteger; +import java.util.List; + +/** + * 批量查询分块索引同步状态。 + */ +public class DocumentChunkSyncStatusRequest { + private BigInteger documentId; + private List ids; + + public BigInteger getDocumentId() { return documentId; } + public void setDocumentId(BigInteger documentId) { this.documentId = documentId; } + public List getIds() { return ids; } + public void setIds(List ids) { this.ids = ids; } +} diff --git a/easyflow-modules/easyflow-module-ai/src/main/java/tech/easyflow/ai/entity/DocumentChunkSyncTask.java b/easyflow-modules/easyflow-module-ai/src/main/java/tech/easyflow/ai/entity/DocumentChunkSyncTask.java new file mode 100644 index 00000000..c3a506b1 --- /dev/null +++ b/easyflow-modules/easyflow-module-ai/src/main/java/tech/easyflow/ai/entity/DocumentChunkSyncTask.java @@ -0,0 +1,78 @@ +package tech.easyflow.ai.entity; + +import com.mybatisflex.annotation.Column; +import com.mybatisflex.annotation.Id; +import com.mybatisflex.annotation.KeyType; +import com.mybatisflex.annotation.Table; +import tech.easyflow.common.entity.DateEntity; + +import java.io.Serializable; +import java.math.BigInteger; +import java.util.Date; + +/** + * 文档分块检索索引同步任务。 + */ +@Table("tb_document_chunk_sync_task") +public class DocumentChunkSyncTask extends DateEntity implements Serializable { + + @Id(keyType = KeyType.Generator, value = "snowFlakeId") + private BigInteger id; + private BigInteger chunkId; + private BigInteger documentId; + private BigInteger documentCollectionId; + private String vectorCollection; + private String operation; + private Long syncVersion; + private String status; + private Integer attemptCount; + private Date nextRetryAt; + private Date lastDispatchedAt; + private String executionToken; + private Date leaseUntil; + private String errorCode; + private String errorMessage; + private Date created; + private BigInteger createdBy; + private Date modified; + private BigInteger modifiedBy; + + public BigInteger getId() { return id; } + public void setId(BigInteger id) { this.id = id; } + public BigInteger getChunkId() { return chunkId; } + public void setChunkId(BigInteger chunkId) { this.chunkId = chunkId; } + public BigInteger getDocumentId() { return documentId; } + public void setDocumentId(BigInteger documentId) { this.documentId = documentId; } + public BigInteger getDocumentCollectionId() { return documentCollectionId; } + public void setDocumentCollectionId(BigInteger documentCollectionId) { this.documentCollectionId = documentCollectionId; } + public String getVectorCollection() { return vectorCollection; } + public void setVectorCollection(String vectorCollection) { this.vectorCollection = vectorCollection; } + public String getOperation() { return operation; } + public void setOperation(String operation) { this.operation = operation; } + public Long getSyncVersion() { return syncVersion; } + public void setSyncVersion(Long syncVersion) { this.syncVersion = syncVersion; } + public String getStatus() { return status; } + public void setStatus(String status) { this.status = status; } + public Integer getAttemptCount() { return attemptCount; } + public void setAttemptCount(Integer attemptCount) { this.attemptCount = attemptCount; } + public Date getNextRetryAt() { return nextRetryAt; } + public void setNextRetryAt(Date nextRetryAt) { this.nextRetryAt = nextRetryAt; } + public Date getLastDispatchedAt() { return lastDispatchedAt; } + public void setLastDispatchedAt(Date lastDispatchedAt) { this.lastDispatchedAt = lastDispatchedAt; } + public String getExecutionToken() { return executionToken; } + public void setExecutionToken(String executionToken) { this.executionToken = executionToken; } + public Date getLeaseUntil() { return leaseUntil; } + public void setLeaseUntil(Date leaseUntil) { this.leaseUntil = leaseUntil; } + public String getErrorCode() { return errorCode; } + public void setErrorCode(String errorCode) { this.errorCode = errorCode; } + public String getErrorMessage() { return errorMessage; } + public void setErrorMessage(String errorMessage) { this.errorMessage = errorMessage; } + @Override public Date getCreated() { return created; } + @Override public void setCreated(Date created) { this.created = created; } + public BigInteger getCreatedBy() { return createdBy; } + public void setCreatedBy(BigInteger createdBy) { this.createdBy = createdBy; } + @Override public Date getModified() { return modified; } + @Override public void setModified(Date modified) { this.modified = modified; } + public BigInteger getModifiedBy() { return modifiedBy; } + public void setModifiedBy(BigInteger modifiedBy) { this.modifiedBy = modifiedBy; } +} diff --git a/easyflow-modules/easyflow-module-ai/src/main/java/tech/easyflow/ai/entity/DocumentCollection.java b/easyflow-modules/easyflow-module-ai/src/main/java/tech/easyflow/ai/entity/DocumentCollection.java index a9bdfaa7..d2fc2d8f 100644 --- a/easyflow-modules/easyflow-module-ai/src/main/java/tech/easyflow/ai/entity/DocumentCollection.java +++ b/easyflow-modules/easyflow-module-ai/src/main/java/tech/easyflow/ai/entity/DocumentCollection.java @@ -8,6 +8,7 @@ import com.easyagents.store.milvus.MilvusVectorStore; import com.easyagents.store.milvus.MilvusVectorStoreConfig; import com.mybatisflex.annotation.Table; import tech.easyflow.ai.config.AiMilvusConfig; +import tech.easyflow.ai.config.AiMilvusClientManager; import tech.easyflow.ai.chattime.availability.ChatTimeToolAvailabilityContext; import tech.easyflow.ai.easyagents.tool.DocumentCollectionTool; import tech.easyflow.ai.entity.base.DocumentCollectionBase; @@ -103,8 +104,10 @@ public class DocumentCollection extends DocumentCollectionBase implements Visibi private DocumentStore milvusStore() { AiMilvusConfig aiMilvusConfig = SpringContextUtil.getBean(AiMilvusConfig.class); + AiMilvusClientManager clientManager = SpringContextUtil.getBean(AiMilvusClientManager.class); + clientManager.reconfigureIfNeeded(aiMilvusConfig); MilvusVectorStoreConfig milvusVectorStoreConfig = aiMilvusConfig.copyForCollection(this.getVectorStoreCollection()); - return new MilvusVectorStore(milvusVectorStoreConfig); + return new MilvusVectorStore(milvusVectorStoreConfig, clientManager); } public Tool toFunction(boolean needEnglishName) { diff --git a/easyflow-modules/easyflow-module-ai/src/main/java/tech/easyflow/ai/entity/base/DocumentChunkBase.java b/easyflow-modules/easyflow-module-ai/src/main/java/tech/easyflow/ai/entity/base/DocumentChunkBase.java index 47fe1ba2..39646f2b 100644 --- a/easyflow-modules/easyflow-module-ai/src/main/java/tech/easyflow/ai/entity/base/DocumentChunkBase.java +++ b/easyflow-modules/easyflow-module-ai/src/main/java/tech/easyflow/ai/entity/base/DocumentChunkBase.java @@ -46,6 +46,18 @@ public class DocumentChunkBase implements Serializable { @Column(typeHandler = FastjsonTypeHandler.class, comment = "扩展元信息") private Map options; + @Column(comment = "检索索引同步状态") + private String indexSyncStatus; + + @Column(comment = "检索索引同步版本") + private Long indexSyncVersion; + + @Column(comment = "脱敏同步错误码") + private String indexSyncErrorCode; + + @Column(comment = "脱敏同步错误摘要") + private String indexSyncErrorMessage; + public BigInteger getId() { return id; } @@ -94,4 +106,36 @@ public class DocumentChunkBase implements Serializable { this.options = options; } + public String getIndexSyncStatus() { + return indexSyncStatus; + } + + public void setIndexSyncStatus(String indexSyncStatus) { + this.indexSyncStatus = indexSyncStatus; + } + + public Long getIndexSyncVersion() { + return indexSyncVersion; + } + + public void setIndexSyncVersion(Long indexSyncVersion) { + this.indexSyncVersion = indexSyncVersion; + } + + public String getIndexSyncErrorCode() { + return indexSyncErrorCode; + } + + public void setIndexSyncErrorCode(String indexSyncErrorCode) { + this.indexSyncErrorCode = indexSyncErrorCode; + } + + public String getIndexSyncErrorMessage() { + return indexSyncErrorMessage; + } + + public void setIndexSyncErrorMessage(String indexSyncErrorMessage) { + this.indexSyncErrorMessage = indexSyncErrorMessage; + } + } diff --git a/easyflow-modules/easyflow-module-ai/src/main/java/tech/easyflow/ai/mapper/DocumentChunkMapper.java b/easyflow-modules/easyflow-module-ai/src/main/java/tech/easyflow/ai/mapper/DocumentChunkMapper.java index f289dcf7..13de9417 100644 --- a/easyflow-modules/easyflow-module-ai/src/main/java/tech/easyflow/ai/mapper/DocumentChunkMapper.java +++ b/easyflow-modules/easyflow-module-ai/src/main/java/tech/easyflow/ai/mapper/DocumentChunkMapper.java @@ -2,6 +2,12 @@ package tech.easyflow.ai.mapper; import tech.easyflow.ai.entity.DocumentChunk; import com.mybatisflex.core.BaseMapper; +import org.apache.ibatis.annotations.Param; +import org.apache.ibatis.annotations.Select; +import org.apache.ibatis.annotations.Update; + +import java.math.BigInteger; +import java.util.List; /** * 映射层。 @@ -11,4 +17,23 @@ import com.mybatisflex.core.BaseMapper; */ public interface DocumentChunkMapper extends BaseMapper { + @Update("UPDATE tb_document_chunk SET index_sync_status=#{status}, " + + "index_sync_error_code=#{errorCode}, index_sync_error_message=#{errorMessage} " + + "WHERE id=#{id} AND index_sync_version=#{version}") + int updateSyncState(@Param("id") BigInteger id, + @Param("version") long version, + @Param("status") String status, + @Param("errorCode") String errorCode, + @Param("errorMessage") String errorMessage); + + @Select("") + List selectSyncStates(@Param("documentId") BigInteger documentId, + @Param("ids") List ids); } diff --git a/easyflow-modules/easyflow-module-ai/src/main/java/tech/easyflow/ai/mapper/DocumentChunkSyncTaskMapper.java b/easyflow-modules/easyflow-module-ai/src/main/java/tech/easyflow/ai/mapper/DocumentChunkSyncTaskMapper.java new file mode 100644 index 00000000..82d1ec03 --- /dev/null +++ b/easyflow-modules/easyflow-module-ai/src/main/java/tech/easyflow/ai/mapper/DocumentChunkSyncTaskMapper.java @@ -0,0 +1,115 @@ +package tech.easyflow.ai.mapper; + +import com.mybatisflex.core.BaseMapper; +import org.apache.ibatis.annotations.Param; +import org.apache.ibatis.annotations.Select; +import org.apache.ibatis.annotations.Update; +import tech.easyflow.ai.entity.DocumentChunkSyncTask; + +import java.math.BigInteger; +import java.util.Date; +import java.util.List; + +/** + * 文档分块索引同步任务映射层。 + */ +public interface DocumentChunkSyncTaskMapper extends BaseMapper { + + String SELECT_COLUMNS = "id, chunk_id AS chunkId, document_id AS documentId, " + + "document_collection_id AS documentCollectionId, " + + "vector_collection AS vectorCollection, operation, sync_version AS syncVersion, " + + "status, attempt_count AS attemptCount, next_retry_at AS nextRetryAt, " + + "last_dispatched_at AS lastDispatchedAt, execution_token AS executionToken, " + + "lease_until AS leaseUntil, error_code AS errorCode, error_message AS errorMessage, " + + "created, created_by AS createdBy, modified, modified_by AS modifiedBy"; + + @Update("UPDATE tb_document_chunk_sync_task SET status='SUPERSEDED', " + + "execution_token=NULL, lease_until=NULL, modified=#{now} " + + "WHERE chunk_id=#{chunkId} AND sync_version < #{syncVersion} " + + "AND status IN ('PENDING','RUNNING','FAILED')") + int supersedeOlder(@Param("chunkId") BigInteger chunkId, + @Param("syncVersion") long syncVersion, + @Param("now") Date now); + + @Update("UPDATE tb_document_chunk_sync_task SET status='SUPERSEDED', " + + "execution_token=NULL, lease_until=NULL, modified=#{now} " + + "WHERE chunk_id=#{chunkId} AND status IN ('PENDING','RUNNING','FAILED')") + int supersedeChunk(@Param("chunkId") BigInteger chunkId, + @Param("now") Date now); + + /** + * 锁定同一分块的任务版本范围，保证外部索引写入与新版本建单按版本串行。 + */ + @Select("SELECT id FROM tb_document_chunk_sync_task WHERE chunk_id=#{chunkId} " + + "ORDER BY sync_version, id FOR UPDATE") + List lockChunkTasks(@Param("chunkId") BigInteger chunkId); + + @Select("SELECT " + SELECT_COLUMNS + " FROM tb_document_chunk_sync_task " + + "WHERE status='PENDING' AND next_retry_at <= #{now} " + + "AND (last_dispatched_at IS NULL OR last_dispatched_at <= #{redispatchBefore}) " + + "ORDER BY next_retry_at, id LIMIT #{limit}") + List selectPendingDue(@Param("now") Date now, + @Param("redispatchBefore") Date redispatchBefore, + @Param("limit") int limit); + + @Update("UPDATE tb_document_chunk_sync_task SET last_dispatched_at=#{now}, modified=#{now} " + + "WHERE id=#{id} AND status='PENDING' AND next_retry_at <= #{now} " + + "AND (last_dispatched_at IS NULL OR last_dispatched_at <= #{redispatchBefore})") + int markDispatched(@Param("id") BigInteger id, + @Param("now") Date now, + @Param("redispatchBefore") Date redispatchBefore); + + @Update("UPDATE tb_document_chunk_sync_task SET status='PENDING', " + + "execution_token=NULL, lease_until=NULL, next_retry_at=#{now}, " + + "last_dispatched_at=NULL, modified=#{now} " + + "WHERE status='RUNNING' AND lease_until <= #{now}") + int recoverExpired(@Param("now") Date now); + + @Update("UPDATE tb_document_chunk_sync_task SET status='RUNNING', " + + "attempt_count=attempt_count + 1, execution_token=#{token}, " + + "lease_until=#{leaseUntil}, error_code=NULL, error_message=NULL, modified=#{now} " + + "WHERE id=#{id} AND status='PENDING' AND next_retry_at <= #{now}") + int claim(@Param("id") BigInteger id, + @Param("token") String token, + @Param("leaseUntil") Date leaseUntil, + @Param("now") Date now); + + @Update("UPDATE tb_document_chunk_sync_task SET status=#{status}, " + + "execution_token=NULL, lease_until=NULL, error_code=#{errorCode}, " + + "error_message=#{errorMessage}, modified=#{now} " + + "WHERE id=#{id} AND status='RUNNING' AND execution_token=#{token}") + int finishOwned(@Param("id") BigInteger id, + @Param("token") String token, + @Param("status") String status, + @Param("errorCode") String errorCode, + @Param("errorMessage") String errorMessage, + @Param("now") Date now); + + @Update("UPDATE tb_document_chunk_sync_task SET status=#{status}, " + + "execution_token=NULL, lease_until=NULL, next_retry_at=#{nextRetryAt}, " + + "last_dispatched_at=NULL, error_code=#{errorCode}, " + + "error_message=#{errorMessage}, modified=#{now} " + + "WHERE id=#{id} AND status='RUNNING' AND execution_token=#{token}") + int failOrRetryOwned(@Param("id") BigInteger id, + @Param("token") String token, + @Param("status") String status, + @Param("nextRetryAt") Date nextRetryAt, + @Param("errorCode") String errorCode, + @Param("errorMessage") String errorMessage, + @Param("now") Date now); + + @Update("UPDATE tb_document_chunk_sync_task SET status='PENDING', " + + "attempt_count=0, next_retry_at=#{now}, last_dispatched_at=NULL, " + + "execution_token=NULL, lease_until=NULL, error_code=NULL, error_message=NULL, modified=#{now} " + + "WHERE id=#{id} AND chunk_id=#{chunkId} AND sync_version=#{syncVersion} " + + "AND status='FAILED'") + int retryFailed(@Param("id") BigInteger id, + @Param("chunkId") BigInteger chunkId, + @Param("syncVersion") long syncVersion, + @Param("now") Date now); + + @Select("SELECT " + SELECT_COLUMNS + " FROM tb_document_chunk_sync_task WHERE chunk_id=#{chunkId} " + + "AND sync_version=#{syncVersion} AND status='FAILED' ORDER BY id DESC LIMIT 1") + DocumentChunkSyncTask selectFailed(@Param("chunkId") BigInteger chunkId, + @Param("syncVersion") long syncVersion); +} diff --git a/easyflow-modules/easyflow-module-ai/src/main/java/tech/easyflow/ai/service/DocumentChunkService.java b/easyflow-modules/easyflow-module-ai/src/main/java/tech/easyflow/ai/service/DocumentChunkService.java index a70d1a1b..450a59f4 100644 --- a/easyflow-modules/easyflow-module-ai/src/main/java/tech/easyflow/ai/service/DocumentChunkService.java +++ b/easyflow-modules/easyflow-module-ai/src/main/java/tech/easyflow/ai/service/DocumentChunkService.java @@ -1,10 +1,12 @@ package tech.easyflow.ai.service; -import tech.easyflow.ai.entity.DocumentChunk; import com.mybatisflex.core.service.IService; -import tech.easyflow.ai.entity.DocumentCollection; +import tech.easyflow.ai.dto.DocumentChunkDeleteResult; +import tech.easyflow.ai.dto.DocumentChunkSyncStatus; +import tech.easyflow.ai.entity.DocumentChunk; import java.math.BigInteger; +import java.util.List; /** * 服务层。 @@ -14,5 +16,36 @@ import java.math.BigInteger; */ public interface DocumentChunkService extends IService { - boolean removeChunk(DocumentCollection knowledge, BigInteger chunkId); + /** + * 更新分块的 Markdown 正文，并创建后台检索索引同步任务。 + * + * @param knowledgeId 知识库 ID + * @param chunkId 分块 ID + * @param markdown Markdown 正文 + * @return 更新后的分块 + */ + DocumentChunk updateContent(BigInteger knowledgeId, BigInteger chunkId, String markdown); + + /** + * 删除分块，并创建后台检索索引清理任务。 + * + * @param knowledgeId 知识库 ID + * @param chunkId 分块 ID + * @return 删除结果 + */ + DocumentChunkDeleteResult deleteChunk(BigInteger knowledgeId, BigInteger chunkId); + + /** + * 手动重试当前版本的检索索引同步。 + */ + DocumentChunk retryIndexSync(BigInteger knowledgeId, BigInteger chunkId, long syncVersion); + + /** + * 批量读取文档内分块的索引同步状态。 + */ + List listIndexSyncStatus( + BigInteger knowledgeId, + BigInteger documentId, + List chunkIds + ); } diff --git a/easyflow-modules/easyflow-module-ai/src/main/java/tech/easyflow/ai/service/impl/DocumentChunkServiceImpl.java b/easyflow-modules/easyflow-module-ai/src/main/java/tech/easyflow/ai/service/impl/DocumentChunkServiceImpl.java index 263aa3f9..7b50d2c8 100644 --- a/easyflow-modules/easyflow-module-ai/src/main/java/tech/easyflow/ai/service/impl/DocumentChunkServiceImpl.java +++ b/easyflow-modules/easyflow-module-ai/src/main/java/tech/easyflow/ai/service/impl/DocumentChunkServiceImpl.java @@ -1,36 +1,365 @@ package tech.easyflow.ai.service.impl; -import com.easyagents.search.engine.service.DocumentSearcher; -import org.springframework.beans.factory.annotation.Autowired; -import tech.easyflow.ai.config.SearcherFactory; -import tech.easyflow.ai.entity.DocumentChunk; -import tech.easyflow.ai.entity.DocumentCollection; -import tech.easyflow.ai.mapper.DocumentChunkMapper; -import tech.easyflow.ai.service.DocumentChunkService; +import com.easyagents.rag.core.BgeM3ChunkSafety; +import com.easyagents.rag.core.RagDefaults; +import com.mybatisflex.core.query.QueryWrapper; import com.mybatisflex.spring.service.impl.ServiceImpl; import org.springframework.stereotype.Service; +import org.springframework.transaction.PlatformTransactionManager; +import org.springframework.transaction.support.TransactionTemplate; +import tech.easyflow.ai.documentchunk.DocumentChunkSyncState; +import tech.easyflow.ai.documentchunk.DocumentChunkSyncTaskAppService; +import tech.easyflow.ai.documentimport.DocumentImportKeys; +import tech.easyflow.ai.dto.DocumentChunkDeleteResult; +import tech.easyflow.ai.dto.DocumentChunkSyncStatus; +import tech.easyflow.ai.entity.Document; +import tech.easyflow.ai.entity.DocumentChunk; +import tech.easyflow.ai.entity.DocumentChunkSyncTask; +import tech.easyflow.ai.entity.DocumentCollection; +import tech.easyflow.ai.mapper.DocumentChunkMapper; +import tech.easyflow.ai.mapper.DocumentMapper; +import tech.easyflow.ai.service.DocumentChunkService; +import tech.easyflow.ai.service.DocumentCollectionService; +import tech.easyflow.common.cache.RedisLockExecutor; +import tech.easyflow.common.web.exceptions.BusinessException; import java.math.BigInteger; +import java.time.Duration; +import java.util.Date; +import java.util.HashMap; +import java.util.Map; +import java.util.Collections; +import java.util.List; +import java.util.regex.Matcher; +import java.util.regex.Pattern; +import java.util.function.Supplier; /** - * 服务层实现。 - * - * @author michael - * @since 2024-08-23 + * 分块服务层实现。 */ @Service -public class DocumentChunkServiceImpl extends ServiceImpl implements DocumentChunkService { +public class DocumentChunkServiceImpl + extends ServiceImpl + implements DocumentChunkService { - @Autowired - private SearcherFactory searcherFactory; + public static final int DOCUMENT_CHUNK_EMPTY_REQUIRES_DELETE = 42901; + public static final int DOCUMENT_CHUNK_LOCK_UNAVAILABLE = 42905; + + private static final Duration LOCK_WAIT_TIMEOUT = Duration.ofSeconds(5); + private static final Duration LOCK_LEASE_TIMEOUT = Duration.ofSeconds(30); + private static final Pattern MARKDOWN_IMAGE_PATTERN = Pattern.compile( + "!\\[([^\\]]*)\\]\\((?:[^()\\r\\n]|\\([^()\\r\\n]*\\))*\\)" + ); + private static final Pattern HTML_IMAGE_PATTERN = Pattern.compile( + "]*>", + Pattern.CASE_INSENSITIVE + ); + private static final Pattern HTML_ALT_PATTERN = Pattern.compile( + "\\balt\\s*=\\s*(?:\"([^\"]*)\"|'([^']*)')", + Pattern.CASE_INSENSITIVE + ); + + private final DocumentChunkMapper documentChunkMapper; + private final DocumentMapper documentMapper; + private final DocumentCollectionService documentCollectionService; + private final DocumentChunkSyncTaskAppService syncTaskAppService; + private final TransactionTemplate transactionTemplate; + private final RedisLockExecutor redisLockExecutor; + + public DocumentChunkServiceImpl( + DocumentChunkMapper documentChunkMapper, + DocumentMapper documentMapper, + DocumentCollectionService documentCollectionService, + DocumentChunkSyncTaskAppService syncTaskAppService, + PlatformTransactionManager transactionManager, + RedisLockExecutor redisLockExecutor + ) { + this.documentChunkMapper = documentChunkMapper; + this.documentMapper = documentMapper; + this.documentCollectionService = documentCollectionService; + this.syncTaskAppService = syncTaskAppService; + this.transactionTemplate = new TransactionTemplate(transactionManager); + this.redisLockExecutor = redisLockExecutor; + } @Override - public boolean removeChunk(DocumentCollection knowledge, BigInteger chunkId) { - DocumentSearcher searcher = searcherFactory.getSearcher(); - // 删除搜索引擎中的数据 - if (searcher == null){ - return true; + public DocumentChunk updateContent( + BigInteger knowledgeId, + BigInteger chunkId, + String markdown + ) { + if (markdown == null || markdown.trim().isEmpty()) { + throw new BusinessException( + 422, + DOCUMENT_CHUNK_EMPTY_REQUIRES_DELETE, + "分块内容为空，请删除该分块" + ); } - return searcher.deleteDocument(chunkId); + DocumentChunk snapshot = requireChunk(knowledgeId, chunkId); + UpdateOutcome outcome = withDocumentLock( + snapshot.getDocumentId(), + () -> updateContentLocked(knowledgeId, chunkId, markdown) + ); + syncTaskAppService.dispatchBestEffort(outcome.taskId()); + return outcome.chunk(); + } + + private UpdateOutcome updateContentLocked( + BigInteger knowledgeId, + BigInteger chunkId, + String markdown + ) { + DocumentChunk current = requireChunk(knowledgeId, chunkId); + DocumentCollection collection = requireKnowledge(knowledgeId); + String searchableContent = toSearchableContent(markdown); + assertWithinEmbeddingLimit(chunkId, searchableContent); + Map options = current.getOptions() == null + ? new HashMap<>() + : new HashMap<>(current.getOptions()); + options.put(DocumentImportKeys.KEY_DOCUMENT_RENDER_MARKDOWN, markdown); + long nextVersion = current.getIndexSyncVersion() == null + ? 1L + : current.getIndexSyncVersion() + 1L; + + UpdateOutcome outcome = transactionTemplate.execute(status -> { + DocumentChunkSyncTask task = syncTaskAppService.createTask( + current, + collection, + DocumentChunkSyncState.OPERATION_UPSERT, + nextVersion, + new Date() + ); + DocumentChunk update = new DocumentChunk(); + update.setId(chunkId); + update.setContent(searchableContent); + update.setOptions(options); + update.setIndexSyncStatus(DocumentChunkSyncState.PENDING); + update.setIndexSyncVersion(nextVersion); + update.setIndexSyncErrorCode(null); + update.setIndexSyncErrorMessage(null); + if (documentChunkMapper.update(update) <= 0) { + throw new BusinessException("分块更新失败"); + } + touchDocument(current.getDocumentId(), null); + current.setContent(searchableContent); + current.setOptions(options); + current.setIndexSyncStatus(DocumentChunkSyncState.PENDING); + current.setIndexSyncVersion(nextVersion); + current.setIndexSyncErrorCode(null); + current.setIndexSyncErrorMessage(null); + return new UpdateOutcome(current, task.getId()); + }); + if (outcome == null) { + throw new BusinessException("分块更新失败"); + } + return outcome; + } + + @Override + public DocumentChunkDeleteResult deleteChunk(BigInteger knowledgeId, BigInteger chunkId) { + DocumentChunk snapshot = requireChunk(knowledgeId, chunkId); + DeleteOutcome outcome = withDocumentLock( + snapshot.getDocumentId(), + () -> deleteChunkLocked(knowledgeId, chunkId) + ); + syncTaskAppService.dispatchBestEffort(outcome.taskId()); + return outcome.result(); + } + + private T withDocumentLock(BigInteger documentId, Supplier action) { + RedisLockExecutor.LockHandle handle; + try { + handle = redisLockExecutor.tryAcquire( + DocumentChunkSyncState.parentLockKey(documentId), + LOCK_WAIT_TIMEOUT, + LOCK_LEASE_TIMEOUT + ); + } catch (RuntimeException exception) { + throw new BusinessException( + 503, + DOCUMENT_CHUNK_LOCK_UNAVAILABLE, + "分块正在处理中，请稍后重试", + exception + ); + } + if (handle == null) { + throw new BusinessException( + 503, + DOCUMENT_CHUNK_LOCK_UNAVAILABLE, + "分块正在处理中，请稍后重试" + ); + } + try (handle) { + return action.get(); + } + } + + private DeleteOutcome deleteChunkLocked(BigInteger knowledgeId, BigInteger chunkId) { + DocumentChunk current = requireChunk(knowledgeId, chunkId); + DocumentCollection collection = requireKnowledge(knowledgeId); + long nextVersion = current.getIndexSyncVersion() == null + ? 1L + : current.getIndexSyncVersion() + 1L; + DeleteOutcome outcome = transactionTemplate.execute(status -> { + DocumentChunkSyncTask task = syncTaskAppService.createTask( + current, + collection, + DocumentChunkSyncState.OPERATION_DELETE, + nextVersion, + new Date() + ); + if (documentChunkMapper.deleteById(chunkId) <= 0) { + throw new BusinessException("分块删除失败"); + } + long remaining = documentChunkMapper.selectCountByQuery( + QueryWrapper.create().eq(DocumentChunk::getDocumentId, current.getDocumentId()) + ); + touchDocument(current.getDocumentId(), remaining); + return new DeleteOutcome( + new DocumentChunkDeleteResult(chunkId, current.getDocumentId(), remaining), + task.getId() + ); + }); + if (outcome == null) { + throw new BusinessException("分块删除失败"); + } + return outcome; + } + + @Override + public DocumentChunk retryIndexSync( + BigInteger knowledgeId, + BigInteger chunkId, + long syncVersion + ) { + requireChunk(knowledgeId, chunkId); + return syncTaskAppService.retry(chunkId, syncVersion); + } + + @Override + public List listIndexSyncStatus( + BigInteger knowledgeId, + BigInteger documentId, + List chunkIds + ) { + if (documentId == null || chunkIds == null || chunkIds.isEmpty()) { + return Collections.emptyList(); + } + BigInteger effectiveKnowledgeId = knowledgeId; + if (effectiveKnowledgeId == null) { + Document document = documentMapper.selectOneById(documentId); + if (document == null || document.getCollectionId() == null) { + throw new BusinessException("文档不存在"); + } + effectiveKnowledgeId = document.getCollectionId(); + } + if (chunkIds.size() > 200) { + throw new BusinessException("单次最多查询 200 个分块状态"); + } + List chunks = documentChunkMapper.selectSyncStates(documentId, chunkIds); + for (DocumentChunk chunk : chunks) { + if (chunk.getDocumentCollectionId() == null + || chunk.getDocumentCollectionId().compareTo(effectiveKnowledgeId) != 0) { + throw new BusinessException("分块不存在"); + } + } + return chunks.stream().map(chunk -> new DocumentChunkSyncStatus( + chunk.getId(), + chunk.getIndexSyncStatus(), + chunk.getIndexSyncVersion(), + chunk.getIndexSyncErrorCode(), + chunk.getIndexSyncErrorMessage() + )).toList(); + } + + private DocumentChunk requireChunk(BigInteger knowledgeId, BigInteger chunkId) { + if (knowledgeId == null || chunkId == null) { + throw new BusinessException("分块不存在"); + } + DocumentChunk chunk = documentChunkMapper.selectOneById(chunkId); + if (chunk == null || chunk.getDocumentCollectionId() == null + || chunk.getDocumentCollectionId().compareTo(knowledgeId) != 0) { + throw new BusinessException("分块不存在"); + } + return chunk; + } + + private DocumentCollection requireKnowledge(BigInteger knowledgeId) { + DocumentCollection knowledge = documentCollectionService.getById(knowledgeId); + if (knowledge == null) { + throw new BusinessException("知识库不存在"); + } + return knowledge; + } + + private void touchDocument(BigInteger documentId, Long chunkCount) { + if (documentId == null) { + return; + } + Document update = new Document(); + update.setId(documentId); + update.setModified(new Date()); + if (chunkCount != null) { + int count = chunkCount > Integer.MAX_VALUE + ? Integer.MAX_VALUE + : chunkCount.intValue(); + update.setTotalChunks(count); + update.setCompletedChunks(count); + update.setFailedChunks(0); + update.setProgressPercent(100); + } + if (documentMapper.update(update) <= 0) { + throw new BusinessException("文档状态更新失败"); + } + } + + private void assertWithinEmbeddingLimit(BigInteger chunkId, String content) { + int tokenEstimate = BgeM3ChunkSafety.estimateContentTokens(content); + if (tokenEstimate > RagDefaults.BGE_M3_HARD_CHUNK_TOKEN_LIMIT) { + throw new BusinessException( + "分块内容超过向量模型上下文上限，请缩短后保存：chunkId=" + chunkId + ); + } + } + + private String toSearchableContent(String markdown) { + return replaceHtmlImages(replaceMarkdownImages(markdown)).trim(); + } + + private String replaceMarkdownImages(String markdown) { + Matcher matcher = MARKDOWN_IMAGE_PATTERN.matcher(markdown); + StringBuffer output = new StringBuffer(); + while (matcher.find()) { + matcher.appendReplacement( + output, + Matcher.quoteReplacement(imageSearchText(matcher.group(1))) + ); + } + matcher.appendTail(output); + return output.toString(); + } + + private String replaceHtmlImages(String content) { + Matcher matcher = HTML_IMAGE_PATTERN.matcher(content); + StringBuffer output = new StringBuffer(); + while (matcher.find()) { + Matcher altMatcher = HTML_ALT_PATTERN.matcher(matcher.group()); + String alt = altMatcher.find() + ? (altMatcher.group(1) == null ? altMatcher.group(2) : altMatcher.group(1)) + : null; + matcher.appendReplacement(output, Matcher.quoteReplacement(imageSearchText(alt))); + } + matcher.appendTail(output); + return output.toString(); + } + + private String imageSearchText(String alt) { + return alt == null || alt.trim().isEmpty() ? "图片" : alt.trim(); + } + + private record UpdateOutcome(DocumentChunk chunk, BigInteger taskId) { + } + + private record DeleteOutcome(DocumentChunkDeleteResult result, BigInteger taskId) { } } diff --git a/easyflow-modules/easyflow-module-ai/src/main/java/tech/easyflow/ai/service/impl/DocumentCollectionServiceImpl.java b/easyflow-modules/easyflow-module-ai/src/main/java/tech/easyflow/ai/service/impl/DocumentCollectionServiceImpl.java index 1488886b..73351559 100644 --- a/easyflow-modules/easyflow-module-ai/src/main/java/tech/easyflow/ai/service/impl/DocumentCollectionServiceImpl.java +++ b/easyflow-modules/easyflow-module-ai/src/main/java/tech/easyflow/ai/service/impl/DocumentCollectionServiceImpl.java @@ -648,6 +648,7 @@ public class DocumentCollectionServiceImpl extends ServiceImpl chunkMap = documentChunkMapper.selectListByQuery(chunkWrapper).stream() .collect(Collectors.toMap(item -> item.getId().toString(), item -> item, (a, b) -> a)); if (chunkMap.isEmpty()) { diff --git a/easyflow-modules/easyflow-module-ai/src/main/java/tech/easyflow/ai/service/impl/DocumentServiceImpl.java b/easyflow-modules/easyflow-module-ai/src/main/java/tech/easyflow/ai/service/impl/DocumentServiceImpl.java index 0627e3e4..c9de273c 100644 --- a/easyflow-modules/easyflow-module-ai/src/main/java/tech/easyflow/ai/service/impl/DocumentServiceImpl.java +++ b/easyflow-modules/easyflow-module-ai/src/main/java/tech/easyflow/ai/service/impl/DocumentServiceImpl.java @@ -31,6 +31,7 @@ import org.springframework.beans.factory.annotation.Autowired; import org.springframework.stereotype.Service; import org.springframework.transaction.annotation.Transactional; import tech.easyflow.ai.config.SearcherFactory; +import tech.easyflow.ai.documentchunk.DocumentChunkSyncState; import tech.easyflow.common.util.SearchKeywordUtil; import tech.easyflow.ai.documentimport.DocumentImportDtos; import tech.easyflow.ai.documentimport.DocumentImportKeys; @@ -40,6 +41,7 @@ import tech.easyflow.ai.documentimport.task.KnowledgeDocumentImportTaskAppServic import tech.easyflow.ai.entity.*; import tech.easyflow.ai.enums.DocumentProcessStatus; import tech.easyflow.ai.mapper.DocumentChunkMapper; +import tech.easyflow.ai.mapper.DocumentChunkSyncTaskMapper; import tech.easyflow.ai.mapper.DocumentMapper; import tech.easyflow.ai.service.DocumentChunkService; import tech.easyflow.ai.service.DocumentCollectionService; @@ -47,6 +49,7 @@ import tech.easyflow.ai.service.DocumentService; import tech.easyflow.ai.service.ModelService; import tech.easyflow.ai.support.DocumentStoreLifecycleSupport; import tech.easyflow.common.ai.rag.ExcelDocumentSplitter; +import tech.easyflow.common.cache.RedisLockExecutor; import tech.easyflow.common.domain.Result; import tech.easyflow.common.filestorage.FileStorageService; import tech.easyflow.common.util.FileUtil; @@ -58,8 +61,11 @@ import java.io.IOException; import java.io.InputStream; import java.math.BigInteger; import java.math.BigDecimal; +import java.time.Duration; import java.util.*; +import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicInteger; +import java.util.concurrent.atomic.AtomicReference; import static tech.easyflow.ai.entity.DocumentCollection.KEY_CAN_UPDATE_EMBEDDING_MODEL; import static tech.easyflow.ai.entity.table.DocumentChunkTableDef.DOCUMENT_CHUNK; @@ -83,6 +89,9 @@ public class DocumentServiceImpl extends ServiceImpl i @Resource private DocumentChunkMapper documentChunkMapper; + @Resource + private DocumentChunkSyncTaskMapper documentChunkSyncTaskMapper; + @Resource private DocumentCollectionService knowledgeService; @@ -110,6 +119,9 @@ public class DocumentServiceImpl extends ServiceImpl i @Autowired private KnowledgeDocumentImportTaskAppService importTaskAppService; + @Autowired + private RedisLockExecutor redisLockExecutor; + @Override public Page getDocumentList(String knowledgeId, int pageSize, int pageNum, String fileName) { return queryDocumentList(knowledgeId, pageSize, pageNum, fileName, null, false); @@ -202,6 +214,49 @@ public class DocumentServiceImpl extends ServiceImpl i @Override @Transactional public boolean removeDoc(String id) { + AtomicBoolean removalCompleted = new AtomicBoolean(); + AtomicReference removalResult = new AtomicReference<>(); + try { + return redisLockExecutor.executeWithRenewingLock( + DocumentChunkSyncState.parentLockKey(id), + Duration.ofSeconds(5), + Duration.ofSeconds(30), + () -> { + try { + boolean result = removeDocLocked(id); + removalResult.set(result); + removalCompleted.set(true); + return result; + } catch (BusinessException exception) { + throw exception; + } catch (RuntimeException exception) { + throw new DocumentRemovalExecutionException(exception); + } + } + ); + } catch (BusinessException exception) { + throw exception; + } catch (DocumentRemovalExecutionException exception) { + throw exception.getOriginalCause(); + } catch (RuntimeException exception) { + if (removalCompleted.get()) { + Log.warn( + "文档删除已完成但分布式锁在收尾阶段失效，继续提交数据库事务: documentId={}", + id, + exception + ); + return Boolean.TRUE.equals(removalResult.get()); + } + throw new BusinessException( + 503, + DocumentChunkServiceImpl.DOCUMENT_CHUNK_LOCK_UNAVAILABLE, + "文档正在处理中，请稍后重试", + exception + ); + } + } + + private boolean removeDocLocked(String id) { // 查询该文档对应哪些分割的字段，先删除 QueryWrapper queryWrapperDocument = QueryWrapper.create().eq(Document::getId, id); Document oneByQuery = documentMapper.selectOneByQuery(queryWrapperDocument); @@ -218,23 +273,38 @@ public class DocumentServiceImpl extends ServiceImpl i QueryWrapper queryWrapper = QueryWrapper.create() .select(DOCUMENT_CHUNK.ID).eq(DocumentChunk::getDocumentId, id); - List chunkIds = documentChunkMapper.selectListByQueryAs( - queryWrapper, - BigInteger.class + List chunkIds = new ArrayList<>( + documentChunkMapper.selectListByQueryAs(queryWrapper, BigInteger.class) ); + chunkIds.sort(Comparator.naturalOrder()); DocumentStore documentStore = null; try { + Model model = null; if (!chunkIds.isEmpty()) { documentStore = knowledge.toDocumentStore(); if (documentStore == null) { - return false; + throw new BusinessException("文档向量存储不可用"); } - Model model = modelService.getById( + model = modelService.getById( knowledge.getVectorEmbedModelId() ); if (model == null) { - return false; + throw new BusinessException("文档向量模型不存在"); } + } + Date supersededAt = new Date(); + for (BigInteger chunkId : chunkIds) { + redisLockExecutor.executeWithRenewingLock( + DocumentChunkSyncState.syncLockKey(chunkId), + Duration.ofSeconds(5), + Duration.ofSeconds(30), + () -> { + documentChunkSyncTaskMapper.supersedeChunk(chunkId, supersededAt); + return null; + } + ); + } + if (!chunkIds.isEmpty()) { StoreOptions options = StoreOptions.ofCollectionName( knowledge.getVectorStoreCollection() ); @@ -253,7 +323,16 @@ public class DocumentServiceImpl extends ServiceImpl i // 删除搜索引擎中的数据 DocumentSearcher searcher = searcherFactory.getSearcher(); if (searcher != null) { - chunkIds.forEach(searcher::deleteDocument); + for (BigInteger chunkId : chunkIds) { + if (!searcher.deleteDocument(chunkId)) { + Log.error( + "删除文档关键词索引失败: documentId={}, chunkId={}", + id, + chunkId + ); + throw new BusinessException("文档关键词索引删除失败"); + } + } } int ck = documentChunkMapper.deleteByQuery(QueryWrapper.create().eq(DocumentChunk::getDocumentId, id)); if (ck < 0) { @@ -1099,6 +1178,17 @@ public class DocumentServiceImpl extends ServiceImpl i } } + private static final class DocumentRemovalExecutionException extends RuntimeException { + + private DocumentRemovalExecutionException(RuntimeException cause) { + super(cause); + } + + private RuntimeException getOriginalCause() { + return (RuntimeException) getCause(); + } + } + public DocumentSplitter getDocumentSplitter(DocumentCollectionSplitParams params) { String splitterName = params.getSplitterName(); int chunkSize = params.getChunkSize(); diff --git a/easyflow-modules/easyflow-module-ai/src/test/java/tech/easyflow/ai/documentchunk/DocumentChunkSyncTaskAppServiceTest.java b/easyflow-modules/easyflow-module-ai/src/test/java/tech/easyflow/ai/documentchunk/DocumentChunkSyncTaskAppServiceTest.java new file mode 100644 index 00000000..72f7489b --- /dev/null +++ b/easyflow-modules/easyflow-module-ai/src/test/java/tech/easyflow/ai/documentchunk/DocumentChunkSyncTaskAppServiceTest.java @@ -0,0 +1,233 @@ +package tech.easyflow.ai.documentchunk; + +import org.junit.Test; +import org.mockito.InOrder; +import org.mockito.Mockito; +import org.springframework.transaction.PlatformTransactionManager; +import org.springframework.transaction.support.SimpleTransactionStatus; +import org.springframework.beans.factory.ObjectProvider; +import tech.easyflow.ai.config.AiMilvusConfig; +import tech.easyflow.ai.config.SearcherFactory; +import tech.easyflow.ai.entity.DocumentChunk; +import tech.easyflow.ai.entity.DocumentChunkSyncTask; +import tech.easyflow.ai.mapper.DocumentChunkMapper; +import tech.easyflow.ai.mapper.DocumentChunkSyncTaskMapper; +import tech.easyflow.ai.service.DocumentCollectionService; +import tech.easyflow.ai.service.ModelService; +import tech.easyflow.common.cache.RedisLockExecutor; + +import java.math.BigInteger; +import java.util.List; +import java.util.function.Supplier; + +/** + * 分块索引同步任务恢复与版本语义测试。 + */ +public class DocumentChunkSyncTaskAppServiceTest { + + @Test + public void dispatchPendingShouldRecoverExpiredTasksAndSendClaimedRows() { + Fixture fixture = fixture(1); + Mockito.when(fixture.taskMapper.selectPendingDue( + Mockito.any(), Mockito.any(), Mockito.anyInt() + )).thenReturn(List.of(fixture.task)); + Mockito.when(fixture.taskMapper.markDispatched( + Mockito.eq(fixture.taskId), Mockito.any(), Mockito.any() + )).thenReturn(1); + + fixture.service.dispatchPendingTasks(); + + Mockito.verify(fixture.taskMapper).recoverExpired(Mockito.any()); + Mockito.verify(fixture.producer).send(fixture.taskId); + } + + @Test + public void outdatedTaskShouldBeSupersededWithoutTouchingIndexes() { + Fixture fixture = fixture(1); + fixture.chunk.setIndexSyncVersion(2L); + + fixture.service.handleTask(fixture.taskId); + + Mockito.verify(fixture.taskMapper).finishOwned( + Mockito.eq(fixture.taskId), Mockito.anyString(), + Mockito.eq(DocumentChunkSyncState.TASK_SUPERSEDED), + Mockito.isNull(), Mockito.isNull(), Mockito.any() + ); + Mockito.verifyNoInteractions(fixture.collectionService); + } + + @Test + public void transientFailureShouldReturnTaskToPending() { + Fixture fixture = fixture(1); + + fixture.service.handleTask(fixture.taskId); + + InOrder order = Mockito.inOrder( + fixture.taskMapper, + fixture.collectionService + ); + order.verify(fixture.taskMapper).lockChunkTasks(fixture.chunkId); + order.verify(fixture.collectionService).getById(Mockito.any()); + Mockito.verify(fixture.taskMapper).failOrRetryOwned( + Mockito.eq(fixture.taskId), Mockito.anyString(), + Mockito.eq(DocumentChunkSyncState.PENDING), Mockito.any(), + Mockito.eq("INDEX_SOURCE_MISSING"), Mockito.anyString(), Mockito.any() + ); + Mockito.verify(fixture.chunkMapper, Mockito.never()).updateSyncState( + Mockito.any(), Mockito.anyLong(), Mockito.anyString(), Mockito.any(), Mockito.any() + ); + } + + @Test + public void workerShouldLockChunkTaskRangeBeforeReadingOwnedTask() { + Fixture fixture = fixture(1); + + fixture.service.handleTask(fixture.taskId); + + InOrder order = Mockito.inOrder(fixture.taskMapper); + order.verify(fixture.taskMapper).selectOneById(fixture.taskId); + order.verify(fixture.taskMapper).claim( + Mockito.eq(fixture.taskId), Mockito.anyString(), Mockito.any(), Mockito.any() + ); + order.verify(fixture.taskMapper).lockChunkTasks(fixture.chunkId); + order.verify(fixture.taskMapper).selectOneById(fixture.taskId); + } + + @Test + public void exhaustedFailureShouldExposeFailedStateOnCurrentVersion() { + Fixture fixture = fixture(5); + + fixture.service.handleTask(fixture.taskId); + + Mockito.verify(fixture.taskMapper).failOrRetryOwned( + Mockito.eq(fixture.taskId), Mockito.anyString(), + Mockito.eq(DocumentChunkSyncState.FAILED), Mockito.any(), + Mockito.eq("INDEX_SOURCE_MISSING"), Mockito.anyString(), Mockito.any() + ); + Mockito.verify(fixture.chunkMapper).updateSyncState( + fixture.chunkId, 1L, DocumentChunkSyncState.FAILED, + "INDEX_SOURCE_MISSING", "分块或知识库不存在" + ); + } + + @Test + public void exhaustedDeleteFailureShouldRemainPendingForDurableCleanup() { + Fixture fixture = fixture(5); + fixture.task.setOperation(DocumentChunkSyncState.OPERATION_DELETE); + Mockito.when(fixture.chunkMapper.selectOneById(fixture.chunkId)).thenReturn(null); + + fixture.service.handleTask(fixture.taskId); + + Mockito.verify(fixture.taskMapper).failOrRetryOwned( + Mockito.eq(fixture.taskId), Mockito.anyString(), + Mockito.eq(DocumentChunkSyncState.PENDING), Mockito.any(), + Mockito.eq("INDEX_SYNC_FAILED"), Mockito.anyString(), Mockito.any() + ); + Mockito.verify(fixture.chunkMapper, Mockito.never()).updateSyncState( + Mockito.any(), Mockito.anyLong(), Mockito.anyString(), Mockito.any(), Mockito.any() + ); + } + + @Test + public void manualRetryShouldResetTaskAndChunkBeforeDispatch() { + Fixture fixture = fixture(5); + fixture.chunk.setIndexSyncStatus(DocumentChunkSyncState.FAILED); + fixture.task.setStatus(DocumentChunkSyncState.FAILED); + Mockito.when(fixture.taskMapper.selectFailed(fixture.chunkId, 1L)) + .thenReturn(fixture.task); + Mockito.when(fixture.taskMapper.retryFailed( + Mockito.eq(fixture.taskId), Mockito.eq(fixture.chunkId), + Mockito.eq(1L), Mockito.any() + )).thenReturn(1); + Mockito.when(fixture.chunkMapper.updateSyncState( + fixture.chunkId, 1L, DocumentChunkSyncState.PENDING, null, null + )).thenReturn(1); + + fixture.service.retry(fixture.chunkId, 1L); + + Mockito.verify(fixture.producer).send(fixture.taskId); + } + + private static Fixture fixture(int attemptCount) { + BigInteger taskId = BigInteger.valueOf(10); + BigInteger chunkId = BigInteger.valueOf(20); + DocumentChunkSyncTask task = new DocumentChunkSyncTask(); + task.setId(taskId); + task.setChunkId(chunkId); + task.setDocumentId(BigInteger.valueOf(30)); + task.setDocumentCollectionId(BigInteger.valueOf(40)); + task.setVectorCollection("kb-test"); + task.setOperation(DocumentChunkSyncState.OPERATION_UPSERT); + task.setSyncVersion(1L); + task.setStatus(DocumentChunkSyncState.PENDING); + task.setAttemptCount(attemptCount); + + DocumentChunk chunk = new DocumentChunk(); + chunk.setId(chunkId); + chunk.setDocumentCollectionId(task.getDocumentCollectionId()); + chunk.setIndexSyncVersion(1L); + chunk.setIndexSyncStatus(DocumentChunkSyncState.PENDING); + + DocumentChunkSyncTaskMapper taskMapper = Mockito.mock(DocumentChunkSyncTaskMapper.class); + Mockito.when(taskMapper.selectOneById(taskId)).thenReturn(task); + Mockito.when(taskMapper.claim( + Mockito.eq(taskId), Mockito.anyString(), Mockito.any(), Mockito.any() + )).thenAnswer(invocation -> { + task.setStatus(DocumentChunkSyncState.TASK_RUNNING); + task.setExecutionToken(invocation.getArgument(1)); + return 1; + }); + Mockito.when(taskMapper.lockChunkTasks(chunkId)).thenReturn(List.of(taskId)); + Mockito.when(taskMapper.failOrRetryOwned( + Mockito.any(), Mockito.anyString(), Mockito.anyString(), Mockito.any(), + Mockito.any(), Mockito.any(), Mockito.any() + )).thenReturn(1); + Mockito.when(taskMapper.finishOwned( + Mockito.any(), Mockito.anyString(), Mockito.anyString(), Mockito.any(), + Mockito.any(), Mockito.any() + )).thenReturn(1); + + DocumentChunkMapper chunkMapper = Mockito.mock(DocumentChunkMapper.class); + Mockito.when(chunkMapper.selectOneById(chunkId)).thenReturn(chunk); + DocumentCollectionService collectionService = Mockito.mock(DocumentCollectionService.class); + ModelService modelService = Mockito.mock(ModelService.class); + SearcherFactory searcherFactory = Mockito.mock(SearcherFactory.class); + DocumentChunkSyncTaskProducer producer = Mockito.mock(DocumentChunkSyncTaskProducer.class); + PlatformTransactionManager transactionManager = Mockito.mock(PlatformTransactionManager.class); + Mockito.when(transactionManager.getTransaction(Mockito.any())) + .thenReturn(new SimpleTransactionStatus()); + RedisLockExecutor redisLockExecutor = Mockito.mock(RedisLockExecutor.class); + Mockito.doAnswer(invocation -> invocation.>getArgument(3).get()) + .when(redisLockExecutor).executeWithRenewingLock( + Mockito.anyString(), Mockito.any(), Mockito.any(), Mockito.>any() + ); + + DocumentChunkSyncTaskAppService service = new DocumentChunkSyncTaskAppService( + taskMapper, + chunkMapper, + collectionService, + modelService, + searcherFactory, + producer, + transactionManager, + redisLockExecutor, + Mockito.mock(AiMilvusConfig.class), + Mockito.mock(ObjectProvider.class) + ); + return new Fixture(service, taskMapper, chunkMapper, collectionService, + producer, task, chunk, taskId, chunkId); + } + + private record Fixture( + DocumentChunkSyncTaskAppService service, + DocumentChunkSyncTaskMapper taskMapper, + DocumentChunkMapper chunkMapper, + DocumentCollectionService collectionService, + DocumentChunkSyncTaskProducer producer, + DocumentChunkSyncTask task, + DocumentChunk chunk, + BigInteger taskId, + BigInteger chunkId + ) { + } +} diff --git a/easyflow-modules/easyflow-module-ai/src/test/java/tech/easyflow/ai/service/impl/DocumentChunkServiceImplTest.java b/easyflow-modules/easyflow-module-ai/src/test/java/tech/easyflow/ai/service/impl/DocumentChunkServiceImplTest.java new file mode 100644 index 00000000..83d78c55 --- /dev/null +++ b/easyflow-modules/easyflow-module-ai/src/test/java/tech/easyflow/ai/service/impl/DocumentChunkServiceImplTest.java @@ -0,0 +1,259 @@ +package tech.easyflow.ai.service.impl; + +import com.mybatisflex.core.query.QueryWrapper; +import org.junit.Assert; +import org.junit.Test; +import org.mockito.ArgumentCaptor; +import org.mockito.InOrder; +import org.mockito.Mockito; +import org.springframework.transaction.PlatformTransactionManager; +import org.springframework.transaction.support.SimpleTransactionStatus; +import tech.easyflow.ai.documentchunk.DocumentChunkSyncState; +import tech.easyflow.ai.documentchunk.DocumentChunkSyncTaskAppService; +import tech.easyflow.ai.documentimport.DocumentImportKeys; +import tech.easyflow.ai.dto.DocumentChunkDeleteResult; +import tech.easyflow.ai.entity.Document; +import tech.easyflow.ai.entity.DocumentChunk; +import tech.easyflow.ai.entity.DocumentChunkSyncTask; +import tech.easyflow.ai.entity.DocumentCollection; +import tech.easyflow.ai.mapper.DocumentChunkMapper; +import tech.easyflow.ai.mapper.DocumentMapper; +import tech.easyflow.ai.service.DocumentCollectionService; +import tech.easyflow.common.cache.RedisLockExecutor; +import tech.easyflow.common.web.exceptions.BusinessException; + +import java.math.BigInteger; +import java.util.HashMap; +import java.util.List; + +/** + * {@link DocumentChunkServiceImpl} 异步索引任务回归测试。 + */ +public class DocumentChunkServiceImplTest { + + @Test + public void updateContentShouldPersistPendingVersionBeforeDispatch() { + Fixture fixture = fixture(); + + DocumentChunk result = fixture.service.updateContent( + fixture.knowledgeId, + fixture.chunkId, + "# 新标题\n\n![说明](https://example.com/image.png)\n正文" + ); + + ArgumentCaptor update = ArgumentCaptor.forClass(DocumentChunk.class); + Mockito.verify(fixture.chunkMapper).update(update.capture()); + Assert.assertEquals("# 新标题\n\n说明\n正文", update.getValue().getContent()); + Assert.assertEquals(DocumentChunkSyncState.PENDING, update.getValue().getIndexSyncStatus()); + Assert.assertEquals(Long.valueOf(3L), update.getValue().getIndexSyncVersion()); + Assert.assertEquals( + "# 新标题\n\n![说明](https://example.com/image.png)\n正文", + update.getValue().getOptions().get(DocumentImportKeys.KEY_DOCUMENT_RENDER_MARKDOWN) + ); + Assert.assertEquals(DocumentChunkSyncState.PENDING, result.getIndexSyncStatus()); + InOrder order = Mockito.inOrder( + fixture.syncTaskAppService, + fixture.chunkMapper + ); + order.verify(fixture.syncTaskAppService).createTask( + Mockito.eq(result), + Mockito.eq(fixture.collection), + Mockito.eq(DocumentChunkSyncState.OPERATION_UPSERT), + Mockito.eq(3L), + Mockito.any() + ); + order.verify(fixture.chunkMapper).update(Mockito.any(DocumentChunk.class)); + Mockito.verify(fixture.syncTaskAppService).dispatchBestEffort(fixture.taskId); + } + + @Test + public void updateContentShouldRejectBlankWithoutSideEffects() { + Fixture fixture = fixture(); + + try { + fixture.service.updateContent(fixture.knowledgeId, fixture.chunkId, " \n "); + Assert.fail("空内容必须转删除，不能进入更新链路"); + } catch (BusinessException expected) { + Assert.assertEquals( + DocumentChunkServiceImpl.DOCUMENT_CHUNK_EMPTY_REQUIRES_DELETE, + expected.getErrorCode() + ); + } + + Mockito.verifyNoInteractions(fixture.chunkMapper, fixture.documentMapper, + fixture.syncTaskAppService); + } + + @Test + public void deleteChunkShouldPersistCleanupTaskAndUpdateParentCount() { + Fixture fixture = fixture(); + Mockito.when(fixture.chunkMapper.deleteById(fixture.chunkId)).thenReturn(1); + Mockito.when(fixture.chunkMapper.selectCountByQuery(Mockito.any(QueryWrapper.class))) + .thenReturn(0L); + + DocumentChunkDeleteResult result = fixture.service.deleteChunk( + fixture.knowledgeId, + fixture.chunkId + ); + + Assert.assertEquals(0L, result.remainingChunkCount()); + InOrder order = Mockito.inOrder( + fixture.syncTaskAppService, + fixture.chunkMapper, + fixture.documentMapper + ); + order.verify(fixture.syncTaskAppService).createTask( + Mockito.any(), Mockito.eq(fixture.collection), + Mockito.eq(DocumentChunkSyncState.OPERATION_DELETE), + Mockito.eq(3L), Mockito.any() + ); + order.verify(fixture.chunkMapper).deleteById(fixture.chunkId); + order.verify(fixture.documentMapper).update(Mockito.any(Document.class)); + Mockito.verify(fixture.syncTaskAppService).dispatchBestEffort(fixture.taskId); + ArgumentCaptor parent = ArgumentCaptor.forClass(Document.class); + Mockito.verify(fixture.documentMapper).update(parent.capture()); + Assert.assertEquals(Integer.valueOf(0), parent.getValue().getTotalChunks()); + Assert.assertEquals(Integer.valueOf(100), parent.getValue().getProgressPercent()); + } + + @Test + public void listStatusShouldRejectChunkFromAnotherKnowledge() { + Fixture fixture = fixture(); + DocumentChunk foreign = new DocumentChunk(); + foreign.setId(BigInteger.valueOf(999)); + foreign.setDocumentId(fixture.documentId); + foreign.setDocumentCollectionId(BigInteger.valueOf(888)); + Mockito.when(fixture.chunkMapper.selectSyncStates( + fixture.documentId, List.of(foreign.getId()) + )).thenReturn(List.of(foreign)); + + try { + fixture.service.listIndexSyncStatus( + fixture.knowledgeId, fixture.documentId, List.of(foreign.getId()) + ); + Assert.fail("跨知识库状态查询必须被拒绝"); + } catch (BusinessException expected) { + Assert.assertEquals("分块不存在", expected.getMessage()); + } + } + + @Test + public void listStatusShouldResolveKnowledgeFromDocumentForAdminPolling() { + Fixture fixture = fixture(); + Document parent = new Document(); + parent.setId(fixture.documentId); + parent.setCollectionId(fixture.knowledgeId); + Mockito.when(fixture.documentMapper.selectOneById(fixture.documentId)) + .thenReturn(parent); + DocumentChunk state = new DocumentChunk(); + state.setId(fixture.chunkId); + state.setDocumentId(fixture.documentId); + state.setDocumentCollectionId(fixture.knowledgeId); + state.setIndexSyncStatus(DocumentChunkSyncState.PENDING); + state.setIndexSyncVersion(3L); + Mockito.when(fixture.chunkMapper.selectSyncStates( + fixture.documentId, List.of(fixture.chunkId) + )).thenReturn(List.of(state)); + + List result = + fixture.service.listIndexSyncStatus( + null, fixture.documentId, List.of(fixture.chunkId) + ); + + Assert.assertEquals(1, result.size()); + Assert.assertEquals(DocumentChunkSyncState.PENDING, result.get(0).indexSyncStatus()); + } + + @Test + public void updateContentShouldExposeRetryableErrorWhenLockIsUnavailable() { + Fixture fixture = fixture(); + Mockito.when(fixture.redisLockExecutor.tryAcquire( + Mockito.anyString(), Mockito.any(), Mockito.any() + )).thenReturn(null); + + try { + fixture.service.updateContent(fixture.knowledgeId, fixture.chunkId, "更新内容"); + Assert.fail("锁不可用时必须返回可重试错误"); + } catch (BusinessException expected) { + Assert.assertEquals(503, expected.getHttpStatus()); + Assert.assertEquals( + DocumentChunkServiceImpl.DOCUMENT_CHUNK_LOCK_UNAVAILABLE, + expected.getErrorCode() + ); + } + Mockito.verify(fixture.chunkMapper, Mockito.never()).update(Mockito.any()); + } + + private static Fixture fixture() { + BigInteger knowledgeId = BigInteger.valueOf(101); + BigInteger documentId = BigInteger.valueOf(201); + BigInteger chunkId = BigInteger.valueOf(301); + BigInteger taskId = BigInteger.valueOf(401); + + DocumentChunk current = new DocumentChunk(); + current.setId(chunkId); + current.setDocumentId(documentId); + current.setDocumentCollectionId(knowledgeId); + current.setContent("旧内容"); + current.setIndexSyncStatus(DocumentChunkSyncState.SYNCED); + current.setIndexSyncVersion(2L); + HashMap options = new HashMap<>(); + options.put("existing", "保留"); + current.setOptions(options); + + DocumentCollection collection = Mockito.mock(DocumentCollection.class); + Mockito.when(collection.getId()).thenReturn(knowledgeId); + Mockito.when(collection.getVectorStoreCollection()).thenReturn("kb-test"); + DocumentCollectionService collectionService = Mockito.mock(DocumentCollectionService.class); + Mockito.when(collectionService.getById(knowledgeId)).thenReturn(collection); + + DocumentChunkMapper chunkMapper = Mockito.mock(DocumentChunkMapper.class); + Mockito.when(chunkMapper.selectOneById(chunkId)).thenReturn(current); + Mockito.when(chunkMapper.update(Mockito.any(DocumentChunk.class))).thenReturn(1); + DocumentMapper documentMapper = Mockito.mock(DocumentMapper.class); + Mockito.when(documentMapper.update(Mockito.any(Document.class))).thenReturn(1); + + DocumentChunkSyncTask task = new DocumentChunkSyncTask(); + task.setId(taskId); + DocumentChunkSyncTaskAppService syncTaskAppService = + Mockito.mock(DocumentChunkSyncTaskAppService.class); + Mockito.when(syncTaskAppService.createTask( + Mockito.any(), Mockito.any(), Mockito.anyString(), Mockito.anyLong(), Mockito.any() + )).thenReturn(task); + + PlatformTransactionManager transactionManager = Mockito.mock(PlatformTransactionManager.class); + Mockito.when(transactionManager.getTransaction(Mockito.any())) + .thenReturn(new SimpleTransactionStatus()); + RedisLockExecutor redisLockExecutor = Mockito.mock(RedisLockExecutor.class); + RedisLockExecutor.LockHandle lockHandle = Mockito.mock(RedisLockExecutor.LockHandle.class); + Mockito.when(redisLockExecutor.tryAcquire( + Mockito.anyString(), Mockito.any(), Mockito.any() + )).thenReturn(lockHandle); + + DocumentChunkServiceImpl service = new DocumentChunkServiceImpl( + chunkMapper, + documentMapper, + collectionService, + syncTaskAppService, + transactionManager, + redisLockExecutor + ); + return new Fixture(service, knowledgeId, documentId, chunkId, taskId, + chunkMapper, documentMapper, collection, syncTaskAppService, + redisLockExecutor); + } + + private record Fixture( + DocumentChunkServiceImpl service, + BigInteger knowledgeId, + BigInteger documentId, + BigInteger chunkId, + BigInteger taskId, + DocumentChunkMapper chunkMapper, + DocumentMapper documentMapper, + DocumentCollection collection, + DocumentChunkSyncTaskAppService syncTaskAppService, + RedisLockExecutor redisLockExecutor + ) { + } +} diff --git a/easyflow-modules/easyflow-module-ai/src/test/java/tech/easyflow/ai/service/impl/DocumentCollectionServiceImplTest.java b/easyflow-modules/easyflow-module-ai/src/test/java/tech/easyflow/ai/service/impl/DocumentCollectionServiceImplTest.java index 72b76319..eded1214 100644 --- a/easyflow-modules/easyflow-module-ai/src/test/java/tech/easyflow/ai/service/impl/DocumentCollectionServiceImplTest.java +++ b/easyflow-modules/easyflow-module-ai/src/test/java/tech/easyflow/ai/service/impl/DocumentCollectionServiceImplTest.java @@ -3,6 +3,7 @@ package tech.easyflow.ai.service.impl; import com.easyagents.core.document.Document; import com.easyagents.search.engine.service.DocumentSearcher; import com.easyagents.search.engine.service.KeywordSearchRequest; +import com.mybatisflex.core.query.QueryWrapper; import org.junit.Assert; import org.junit.Test; import org.springframework.beans.factory.ObjectProvider; @@ -20,6 +21,7 @@ import java.math.BigInteger; import java.util.HashMap; import java.util.List; import java.util.Map; +import java.util.concurrent.atomic.AtomicReference; import static tech.easyflow.ai.entity.DocumentCollection.KEY_DOC_RECALL_MAX_NUM; import static tech.easyflow.ai.entity.DocumentCollection.KEY_SIMILARITY_THRESHOLD; @@ -100,7 +102,9 @@ public class DocumentCollectionServiceImplTest { DocumentCollectionServiceImpl service = new TestDocumentCollectionService(collection); setField(service, "searcherFactory", new SearcherFactory(new StaticObjectProvider(searcher))); - setField(service, "documentChunkMapper", mockDocumentChunkMapper(completedChunk, indexingChunk)); + AtomicReference chunkQuery = new AtomicReference(); + setField(service, "documentChunkMapper", + mockDocumentChunkMapper(chunkQuery, completedChunk, indexingChunk)); setField(service, "documentMapper", mockDocumentMapper(completedDocument)); tech.easyflow.ai.rag.KnowledgeRetrievalRequest request = new tech.easyflow.ai.rag.KnowledgeRetrievalRequest(); @@ -116,6 +120,10 @@ public class DocumentCollectionServiceImplTest { Assert.assertEquals(completedDocument.getTitle(), result.get(0).getTitle()); Assert.assertEquals("completed chunk", result.get(0).getContent()); Assert.assertEquals(String.valueOf(knowledgeId), searcher.lastKnowledgeId); + Assert.assertTrue( + "检索补齐必须过滤尚未同步的分块", + chunkQuery.get().toSQL().toUpperCase().contains("INDEX_SYNC_STATUS") + ); Assert.assertEquals( tech.easyflow.ai.entity.DocumentCollection.TYPE_DOCUMENT, result.get(0).getMetadata("resultType") @@ -210,7 +218,10 @@ public class DocumentCollectionServiceImplTest { return document; } - private static DocumentChunkMapper mockDocumentChunkMapper(tech.easyflow.ai.entity.DocumentChunk... chunks) { + private static DocumentChunkMapper mockDocumentChunkMapper( + AtomicReference query, + tech.easyflow.ai.entity.DocumentChunk... chunks + ) { Map chunkMap = new HashMap(); for (tech.easyflow.ai.entity.DocumentChunk chunk : chunks) { chunkMap.put(String.valueOf(chunk.getId()), chunk); @@ -220,6 +231,7 @@ public class DocumentCollectionServiceImplTest { new Class[]{DocumentChunkMapper.class}, (proxy, method, args) -> { if ("selectListByQuery".equals(method.getName())) { + query.set((QueryWrapper) args[0]); return List.copyOf(chunkMap.values()); } return defaultValue(method.getReturnType()); diff --git a/easyflow-modules/easyflow-module-ai/src/test/java/tech/easyflow/ai/service/impl/DocumentServiceImplTest.java b/easyflow-modules/easyflow-module-ai/src/test/java/tech/easyflow/ai/service/impl/DocumentServiceImplTest.java index 743d5e4b..635de373 100644 --- a/easyflow-modules/easyflow-module-ai/src/test/java/tech/easyflow/ai/service/impl/DocumentServiceImplTest.java +++ b/easyflow-modules/easyflow-module-ai/src/test/java/tech/easyflow/ai/service/impl/DocumentServiceImplTest.java @@ -3,22 +3,27 @@ package tech.easyflow.ai.service.impl; import com.easyagents.core.store.DocumentStore; import com.easyagents.core.store.StoreResult; import com.easyagents.rag.core.RagDefaults; +import com.easyagents.search.engine.service.DocumentSearcher; import com.mybatisflex.core.query.QueryWrapper; import org.junit.Assert; import org.junit.Test; import org.mockito.ArgumentCaptor; +import org.mockito.InOrder; import org.mockito.Mockito; import tech.easyflow.ai.config.SearcherFactory; +import tech.easyflow.ai.documentchunk.DocumentChunkSyncState; import tech.easyflow.ai.entity.Document; import tech.easyflow.ai.entity.DocumentChunk; import tech.easyflow.ai.entity.DocumentCollection; import tech.easyflow.ai.entity.Model; import tech.easyflow.ai.enums.DocumentProcessStatus; import tech.easyflow.ai.mapper.DocumentChunkMapper; +import tech.easyflow.ai.mapper.DocumentChunkSyncTaskMapper; import tech.easyflow.ai.mapper.DocumentMapper; import tech.easyflow.ai.service.DocumentCollectionService; import tech.easyflow.ai.service.ModelService; import tech.easyflow.common.filestorage.FileStorageService; +import tech.easyflow.common.cache.RedisLockExecutor; import tech.easyflow.common.web.exceptions.BusinessException; import java.lang.reflect.Field; @@ -28,6 +33,7 @@ import java.math.BigInteger; import java.util.List; import java.util.Locale; import java.util.regex.Pattern; +import java.util.function.Supplier; /** * {@link DocumentServiceImpl} 文档维护回归测试。 @@ -155,9 +161,15 @@ public class DocumentServiceImplTest { setField(service, "modelService", modelService); setField(service, "storageService", storageService); setField(service, "searcherFactory", searcherFactory); + DocumentChunkSyncTaskMapper taskMapper = configureRemovalCoordination(service); Assert.assertTrue(service.removeDoc(documentId.toString())); + InOrder order = Mockito.inOrder(taskMapper, documentStore); + order.verify(taskMapper).supersedeChunk( + Mockito.eq(BigInteger.valueOf(201)), Mockito.any() + ); + order.verify(documentStore).delete(Mockito.anyList(), Mockito.any()); Mockito.verify(documentMapper).deleteById(documentId); Mockito.verify(storageService).delete(document.getDocumentPath()); } @@ -206,6 +218,7 @@ public class DocumentServiceImplTest { setField(service, "modelService", modelService); setField(service, "storageService", storageService); setField(service, "searcherFactory", searcherFactory); + configureRemovalCoordination(service); Assert.assertTrue(service.removeDoc(documentId.toString())); @@ -238,6 +251,7 @@ public class DocumentServiceImplTest { setField(service, "documentMapper", documentMapper); setField(service, "documentChunkMapper", chunkMapper); setField(service, "knowledgeService", knowledgeService); + configureRemovalCoordination(service); try { service.removeDoc(documentId.toString()); @@ -295,6 +309,7 @@ public class DocumentServiceImplTest { setField(service, "modelService", modelService); setField(service, "storageService", storageService); setField(service, "searcherFactory", searcherFactory); + configureRemovalCoordination(service); try { service.removeDoc(documentId.toString()); @@ -308,6 +323,176 @@ public class DocumentServiceImplTest { Mockito.verifyNoInteractions(storageService); } + /** + * 验证关键词索引拒绝删除时保留数据库记录，供用户安全重试。 + * + * @throws Exception 反射注入异常 + */ + @Test + public void removeDocShouldStopWhenKeywordDeleteFails() throws Exception { + BigInteger documentId = BigInteger.valueOf(451); + BigInteger knowledgeId = BigInteger.valueOf(452); + BigInteger modelId = BigInteger.valueOf(453); + BigInteger chunkId = BigInteger.valueOf(454); + Document document = new Document(); + document.setId(documentId); + document.setCollectionId(knowledgeId); + document.setProcessStatus(DocumentProcessStatus.COMPLETED.name()); + DocumentStore documentStore = Mockito.mock(DocumentStore.class); + Mockito.when(documentStore.delete(Mockito.anyList(), Mockito.any())) + .thenReturn(StoreResult.success()); + DocumentCollection knowledge = Mockito.mock(DocumentCollection.class); + Mockito.when(knowledge.getVectorEmbedModelId()).thenReturn(modelId); + Mockito.when(knowledge.getVectorStoreCollection()).thenReturn("kb-test"); + Mockito.when(knowledge.toDocumentStore()).thenReturn(documentStore); + Model model = new Model(); + model.setModelName("embedding-test"); + DocumentMapper documentMapper = Mockito.mock(DocumentMapper.class); + Mockito.when(documentMapper.selectOneByQuery(Mockito.any())) + .thenReturn(document); + DocumentChunkMapper chunkMapper = Mockito.mock(DocumentChunkMapper.class); + Mockito.when(chunkMapper.selectListByQueryAs( + Mockito.any(), Mockito.eq(BigInteger.class))) + .thenReturn(List.of(chunkId)); + DocumentCollectionService knowledgeService = Mockito.mock( + DocumentCollectionService.class + ); + Mockito.when(knowledgeService.getById(knowledgeId)).thenReturn(knowledge); + ModelService modelService = Mockito.mock(ModelService.class); + Mockito.when(modelService.getById(modelId)).thenReturn(model); + FileStorageService storageService = Mockito.mock(FileStorageService.class); + DocumentSearcher searcher = Mockito.mock(DocumentSearcher.class); + Mockito.when(searcher.deleteDocument(chunkId)).thenReturn(false); + SearcherFactory searcherFactory = Mockito.mock(SearcherFactory.class); + Mockito.when(searcherFactory.getSearcher()).thenReturn(searcher); + + DocumentServiceImpl service = new DocumentServiceImpl(); + setField(service, "documentMapper", documentMapper); + setField(service, "documentChunkMapper", chunkMapper); + setField(service, "knowledgeService", knowledgeService); + setField(service, "modelService", modelService); + setField(service, "storageService", storageService); + setField(service, "searcherFactory", searcherFactory); + configureRemovalCoordination(service); + + try { + service.removeDoc(documentId.toString()); + Assert.fail("关键词索引删除失败时应停止文档删除"); + } catch (BusinessException expected) { + Assert.assertEquals("文档关键词索引删除失败", expected.getMessage()); + } + + Mockito.verify(chunkMapper, Mockito.never()).deleteByQuery(Mockito.any()); + Mockito.verify(documentMapper, Mockito.never()).deleteById(Mockito.any()); + Mockito.verifyNoInteractions(storageService); + } + + /** + * 验证外部与数据库删除已经执行完成时，父锁收尾阶段丢失不会回滚数据库事务。 + */ + @Test + public void removeDocShouldCommitAfterPostExecutionLockLoss() throws Exception { + BigInteger documentId = BigInteger.valueOf(501); + BigInteger knowledgeId = BigInteger.valueOf(502); + Document document = new Document(); + document.setId(documentId); + document.setCollectionId(knowledgeId); + document.setDocumentPath("storage://lock-lost.txt"); + document.setProcessStatus(DocumentProcessStatus.COMPLETED.name()); + + DocumentMapper documentMapper = Mockito.mock(DocumentMapper.class); + Mockito.when(documentMapper.selectOneByQuery(Mockito.any())) + .thenReturn(document); + Mockito.when(documentMapper.deleteById(documentId)).thenReturn(1); + DocumentChunkMapper chunkMapper = Mockito.mock(DocumentChunkMapper.class); + Mockito.when(chunkMapper.selectListByQueryAs( + Mockito.any(), Mockito.eq(BigInteger.class))).thenReturn(List.of()); + Mockito.when(chunkMapper.deleteByQuery(Mockito.any())).thenReturn(0); + DocumentCollectionService knowledgeService = Mockito.mock( + DocumentCollectionService.class + ); + Mockito.when(knowledgeService.getById(knowledgeId)) + .thenReturn(Mockito.mock(DocumentCollection.class)); + FileStorageService storageService = Mockito.mock(FileStorageService.class); + + DocumentServiceImpl service = new DocumentServiceImpl(); + setField(service, "documentMapper", documentMapper); + setField(service, "documentChunkMapper", chunkMapper); + setField(service, "knowledgeService", knowledgeService); + setField(service, "storageService", storageService); + setField(service, "searcherFactory", Mockito.mock(SearcherFactory.class)); + configureRemovalCoordination(service); + RedisLockExecutor lockExecutor = getField( + service, "redisLockExecutor", RedisLockExecutor.class + ); + Mockito.doAnswer(invocation -> { + String lockKey = invocation.getArgument(0); + Object result = invocation.>getArgument(3).get(); + if (DocumentChunkSyncState.parentLockKey(documentId.toString()) + .equals(lockKey)) { + throw new IllegalStateException("模拟 callback 完成后续租丢失"); + } + return result; + }).when(lockExecutor).executeWithRenewingLock( + Mockito.anyString(), Mockito.any(), Mockito.any(), + Mockito.>any() + ); + + Assert.assertTrue(service.removeDoc(documentId.toString())); + Mockito.verify(documentMapper).deleteById(documentId); + Mockito.verify(storageService).delete(document.getDocumentPath()); + } + + /** + * 验证向量存储配置缺失时不会先废弃仍可恢复的分块同步任务。 + */ + @Test + public void removeDocShouldKeepSyncTasksWhenVectorStoreIsUnavailable() + throws Exception { + BigInteger documentId = BigInteger.valueOf(601); + BigInteger knowledgeId = BigInteger.valueOf(602); + BigInteger chunkId = BigInteger.valueOf(603); + Document document = new Document(); + document.setId(documentId); + document.setCollectionId(knowledgeId); + document.setProcessStatus(DocumentProcessStatus.COMPLETED.name()); + + DocumentCollection knowledge = Mockito.mock(DocumentCollection.class); + Mockito.when(knowledge.toDocumentStore()).thenReturn(null); + DocumentMapper documentMapper = Mockito.mock(DocumentMapper.class); + Mockito.when(documentMapper.selectOneByQuery(Mockito.any())) + .thenReturn(document); + DocumentChunkMapper chunkMapper = Mockito.mock(DocumentChunkMapper.class); + Mockito.when(chunkMapper.selectListByQueryAs( + Mockito.any(), Mockito.eq(BigInteger.class))) + .thenReturn(List.of(chunkId)); + DocumentCollectionService knowledgeService = Mockito.mock( + DocumentCollectionService.class + ); + Mockito.when(knowledgeService.getById(knowledgeId)).thenReturn(knowledge); + + DocumentServiceImpl service = new DocumentServiceImpl(); + setField(service, "documentMapper", documentMapper); + setField(service, "documentChunkMapper", chunkMapper); + setField(service, "knowledgeService", knowledgeService); + setField(service, "modelService", Mockito.mock(ModelService.class)); + setField(service, "storageService", Mockito.mock(FileStorageService.class)); + setField(service, "searcherFactory", Mockito.mock(SearcherFactory.class)); + DocumentChunkSyncTaskMapper taskMapper = configureRemovalCoordination(service); + + try { + service.removeDoc(documentId.toString()); + Assert.fail("向量存储不可用时应停止文档删除"); + } catch (BusinessException expected) { + Assert.assertEquals("文档向量存储不可用", expected.getMessage()); + } + + Mockito.verify(taskMapper, Mockito.never()) + .supersedeChunk(Mockito.any(), Mockito.any()); + Mockito.verify(chunkMapper, Mockito.never()).deleteByQuery(Mockito.any()); + Mockito.verify(documentMapper, Mockito.never()).deleteById(Mockito.any()); + } + /** * 验证旧版向量化入口同样拒绝超过 BGE-M3 上下文预算的分块。 * @@ -351,6 +536,33 @@ public class DocumentServiceImplTest { field.set(target, value); } + private static T getField(Object target, + String fieldName, + Class fieldType) throws Exception { + Field field = DocumentServiceImpl.class.getDeclaredField(fieldName); + field.setAccessible(true); + return fieldType.cast(field.get(target)); + } + + private static DocumentChunkSyncTaskMapper configureRemovalCoordination( + DocumentServiceImpl service + ) throws Exception { + DocumentChunkSyncTaskMapper taskMapper = Mockito.mock( + DocumentChunkSyncTaskMapper.class + ); + RedisLockExecutor lockExecutor = Mockito.mock(RedisLockExecutor.class); + Mockito.doAnswer(invocation -> invocation.>getArgument(3).get()) + .when(lockExecutor).executeWithRenewingLock( + Mockito.anyString(), + Mockito.any(), + Mockito.any(), + Mockito.>any() + ); + setField(service, "documentChunkSyncTaskMapper", taskMapper); + setField(service, "redisLockExecutor", lockExecutor); + return taskMapper; + } + /** * 统一 SQL 文本格式，便于断言查询结构。 * diff --git a/easyflow-starter/easyflow-starter-all/src/main/resources/application.yml b/easyflow-starter/easyflow-starter-all/src/main/resources/application.yml index 2ac313ee..23f5c8e1 100644 --- a/easyflow-starter/easyflow-starter-all/src/main/resources/application.yml +++ b/easyflow-starter/easyflow-starter-all/src/main/resources/application.yml @@ -376,6 +376,13 @@ rag: username: easyflowadmin password: easyflowadmin123 autoCreateCollection: true + poolMaxTotal: 8 + poolMaxTotalPerKey: 8 + poolMaxIdlePerKey: 4 + poolMinIdlePerKey: 1 + poolMaxWaitMillis: 3000 + poolEvictionIntervalMillis: 60000 + poolMinEvictableIdleMillis: 300000 # 搜索引擎配置 searcher: lucene: diff --git a/easyflow-starter/easyflow-starter-all/src/main/resources/db/migration/mysql/V64__mysql_document_chunk_index_sync.sql b/easyflow-starter/easyflow-starter-all/src/main/resources/db/migration/mysql/V64__mysql_document_chunk_index_sync.sql new file mode 100644 index 00000000..e8f4fd76 --- /dev/null +++ b/easyflow-starter/easyflow-starter-all/src/main/resources/db/migration/mysql/V64__mysql_document_chunk_index_sync.sql @@ -0,0 +1,48 @@ +ALTER TABLE `tb_document_chunk` + ADD COLUMN `index_sync_status` varchar(16) NOT NULL DEFAULT 'SYNCED' + COMMENT '检索索引同步状态' AFTER `options`, + ADD COLUMN `index_sync_version` bigint UNSIGNED NOT NULL DEFAULT 0 + COMMENT '检索索引同步版本' AFTER `index_sync_status`, + ADD COLUMN `index_sync_error_code` varchar(64) NULL DEFAULT NULL + COMMENT '脱敏同步错误码' AFTER `index_sync_version`, + ADD COLUMN `index_sync_error_message` varchar(255) NULL DEFAULT NULL + COMMENT '脱敏同步错误摘要' AFTER `index_sync_error_code`, + ADD KEY `idx_document_chunk_sync_filter` + (`document_collection_id`, `index_sync_status`, `id`); + +CREATE TABLE `tb_document_chunk_sync_task` +( + `id` bigint UNSIGNED NOT NULL COMMENT '主键', + `chunk_id` bigint UNSIGNED NOT NULL COMMENT '分块ID', + `document_id` bigint UNSIGNED NOT NULL COMMENT '文档ID', + `document_collection_id` bigint UNSIGNED NOT NULL COMMENT '知识库ID', + `vector_collection` varchar(128) NULL DEFAULT NULL COMMENT '向量集合名', + `operation` varchar(16) NOT NULL COMMENT 'UPSERT/DELETE', + `sync_version` bigint UNSIGNED NOT NULL COMMENT '同步版本', + `status` varchar(16) NOT NULL DEFAULT 'PENDING' + COMMENT 'PENDING/RUNNING/SUCCEEDED/FAILED/SUPERSEDED', + `attempt_count` int NOT NULL DEFAULT 0 COMMENT '已尝试次数', + `next_retry_at` datetime(3) NOT NULL COMMENT '下次执行时间', + `last_dispatched_at` datetime(3) NULL DEFAULT NULL COMMENT '最近投递时间', + `execution_token` varchar(64) NULL DEFAULT NULL COMMENT '执行令牌', + `lease_until` datetime(3) NULL DEFAULT NULL COMMENT '租约截止时间', + `error_code` varchar(64) NULL DEFAULT NULL COMMENT '脱敏错误码', + `error_message` varchar(255) NULL DEFAULT NULL COMMENT '脱敏错误摘要', + `created` datetime(3) NULL DEFAULT NULL COMMENT '创建时间', + `created_by` bigint UNSIGNED NULL DEFAULT NULL COMMENT '创建人', + `modified` datetime(3) NULL DEFAULT NULL COMMENT '修改时间', + `modified_by` bigint UNSIGNED NULL DEFAULT NULL COMMENT '修改人', + PRIMARY KEY (`id`) USING BTREE, + UNIQUE KEY `uk_document_chunk_sync_version` + (`chunk_id`, `sync_version`, `operation`), + KEY `idx_document_chunk_sync_due` + (`status`, `next_retry_at`, `last_dispatched_at`, `id`), + KEY `idx_document_chunk_sync_lease` + (`status`, `lease_until`, `id`), + KEY `idx_document_chunk_sync_chunk` + (`chunk_id`, `sync_version`, `status`) +) ENGINE = InnoDB + CHARACTER SET = utf8mb4 + COLLATE = utf8mb4_0900_ai_ci + COMMENT = '文档分块检索索引同步任务' + ROW_FORMAT = DYNAMIC; diff --git a/pom.xml b/pom.xml index a9c1f899..ad14b0fe 100644 --- a/pom.xml +++ b/pom.xml @@ -52,6 +52,7 @@ 7.9.1 2.7.0 1.8 + 2.3.11 @@ -184,6 +185,11 @@ easy-agents-bom ${easy-agents.version} + + io.milvus + milvus-sdk-java + ${milvus.version} + com.easyagents easy-agents-flow