diff --git a/backend-java/src/test/java/com/nanri/aiimage/modules/task/contract/DuplicateSubmissionContractTest.java b/backend-java/src/test/java/com/nanri/aiimage/modules/task/contract/DuplicateSubmissionContractTest.java new file mode 100644 index 00000000..9595a1f0 --- /dev/null +++ b/backend-java/src/test/java/com/nanri/aiimage/modules/task/contract/DuplicateSubmissionContractTest.java @@ -0,0 +1,283 @@ +package com.nanri.aiimage.modules.task.contract; + +import cn.hutool.crypto.digest.DigestUtil; +import com.baomidou.mybatisplus.core.MybatisConfiguration; +import com.baomidou.mybatisplus.core.metadata.TableInfoHelper; +import com.fasterxml.jackson.databind.ObjectMapper; +import com.nanri.aiimage.common.exception.BusinessException; +import com.nanri.aiimage.modules.collectdata.mapper.CollectDataCountryPrefMapper; +import com.nanri.aiimage.modules.collectdata.mapper.CollectDataItemMapper; +import com.nanri.aiimage.modules.collectdata.model.dto.CollectDataSubmitResultRequest; +import com.nanri.aiimage.modules.collectdata.model.dto.CollectDataSubmitRowDto; +import com.nanri.aiimage.modules.collectdata.model.vo.CollectDataResultRowVo; +import com.nanri.aiimage.modules.collectdata.model.vo.CollectDataSubmitResultVo; +import com.nanri.aiimage.modules.collectdata.service.CollectDataExcelAssemblyService; +import com.nanri.aiimage.modules.collectdata.service.CollectDataService; +import com.nanri.aiimage.modules.collectdata.util.CollectDataBatchQuery; +import com.nanri.aiimage.modules.collectdata.util.CollectDataBrandBatchFilter; +import com.nanri.aiimage.modules.collectdata.util.CollectDataResultDetailCodec; +import com.nanri.aiimage.modules.collectdata.util.CollectDataInvalidAsinBatchWriter; +import com.nanri.aiimage.modules.collectdata.util.CollectDataResultItemBatchWriter; +import com.nanri.aiimage.modules.file.service.LocalFileStorageService; +import com.nanri.aiimage.modules.file.service.oss.OssStorageService; +import com.nanri.aiimage.modules.invalidasin.mapper.InvalidAsinDataMapper; +import com.nanri.aiimage.modules.task.mapper.FileResultMapper; +import com.nanri.aiimage.modules.task.mapper.FileTaskMapper; +import com.nanri.aiimage.modules.task.mapper.TaskChunkMapper; +import com.nanri.aiimage.modules.task.mapper.TaskResultItemMapper; +import com.nanri.aiimage.modules.task.mapper.TaskScopeStateMapper; +import com.nanri.aiimage.modules.task.model.entity.FileResultEntity; +import com.nanri.aiimage.modules.task.model.entity.FileTaskEntity; +import com.nanri.aiimage.modules.task.model.entity.TaskChunkEntity; +import com.nanri.aiimage.modules.task.model.entity.TaskScopeStateEntity; +import com.nanri.aiimage.modules.task.service.TaskDistributedLockService; +import com.nanri.aiimage.modules.task.service.TaskFileJobService; +import com.nanri.aiimage.modules.task.service.TransientPayloadStorageService; +import org.apache.ibatis.builder.MapperBuilderAssistant; +import org.junit.jupiter.api.BeforeAll; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.extension.ExtendWith; +import org.mockito.InjectMocks; +import org.mockito.Mock; +import org.mockito.Spy; +import org.mockito.junit.jupiter.MockitoExtension; +import org.springframework.transaction.support.TransactionTemplate; + +import java.util.List; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertNotNull; +import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.junit.jupiter.api.Assertions.assertTrue; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.anyInt; +import static org.mockito.ArgumentMatchers.anyLong; +import static org.mockito.ArgumentMatchers.anyString; +import static org.mockito.ArgumentMatchers.eq; +import static org.mockito.Mockito.doAnswer; +import static org.mockito.Mockito.lenient; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.never; +import static org.mockito.Mockito.times; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; + +/** + * task-137：重复提交幂等契约（spec 07 §2）。 + * 同 submissionId 重复提交 /result：幂等接受（成功返回、不重复落库）； + * 同 chunk_index 内容不同拒绝；任务终态后重复提交拒绝；多次重复安全。 + */ +@ExtendWith(MockitoExtension.class) +class DuplicateSubmissionContractTest { + + private static final Long TASK_ID = 4444L; + private static final Long USER_ID = 7L; + private static final String SUBMISSION_ID = "collect-data-4444"; + + @Mock private LocalFileStorageService localFileStorageService; + @Mock private FileTaskMapper fileTaskMapper; + @Mock private FileResultMapper fileResultMapper; + @Mock private CollectDataItemMapper collectDataItemMapper; + @Mock private CollectDataCountryPrefMapper collectDataCountryPrefMapper; + @Mock private InvalidAsinDataMapper invalidAsinDataMapper; + @Mock private TaskChunkMapper taskChunkMapper; + @Mock private TaskScopeStateMapper taskScopeStateMapper; + @Mock private TaskResultItemMapper taskResultItemMapper; + @Mock private TaskDistributedLockService taskDistributedLockService; + @Mock private TaskFileJobService taskFileJobService; + @Mock private TransientPayloadStorageService transientPayloadStorageService; + @Mock private CollectDataExcelAssemblyService excelAssemblyService; + @Mock private OssStorageService ossStorageService; + @Spy private ObjectMapper objectMapper = new ObjectMapper(); + @Mock private TransactionTemplate transactionTemplate; + @Mock private CollectDataBatchQuery collectDataBatchQuery; + @Mock private CollectDataBrandBatchFilter brandBatchFilter; + @Mock private CollectDataInvalidAsinBatchWriter invalidAsinBatchWriter; + @Mock private CollectDataResultItemBatchWriter resultItemBatchWriter; + @Mock private CollectDataResultDetailCodec resultDetailCodec; + + @InjectMocks private CollectDataService service; + + @BeforeAll + static void initializeMybatisMetadata() { + MapperBuilderAssistant assistant = new MapperBuilderAssistant(new MybatisConfiguration(), ""); + TableInfoHelper.initTableInfo(assistant, FileTaskEntity.class); + TableInfoHelper.initTableInfo(assistant, FileResultEntity.class); + TableInfoHelper.initTableInfo(assistant, TaskChunkEntity.class); + TableInfoHelper.initTableInfo(assistant, TaskScopeStateEntity.class); + } + + @BeforeEach + void setUp() { + lenient().when(taskDistributedLockService.acquire(eq("COLLECT_DATA"), anyLong(), anyLong())) + .thenReturn(mock(TaskDistributedLockService.LockHandle.class)); + lenient().when(fileTaskMapper.selectById(TASK_ID)).thenReturn(runningTask()); + lenient().when(fileResultMapper.selectOne(any())).thenReturn(null); + lenient().doAnswer(invocation -> { + FileResultEntity result = invocation.getArgument(0); + result.setId(3001L); + return 1; + }).when(fileResultMapper).insert(any(FileResultEntity.class)); + lenient().when(taskChunkMapper.selectCount(any())).thenReturn(0L); + lenient().when(taskChunkMapper.selectOne(any())).thenReturn(null); + lenient().when(taskScopeStateMapper.selectOne(any())).thenReturn(null); + lenient().when(taskScopeStateMapper.insert(any(TaskScopeStateEntity.class))).thenReturn(1); + lenient().when(fileTaskMapper.updateById(any(FileTaskEntity.class))).thenReturn(1); + lenient().when(transientPayloadStorageService.isSharedWriteEnabled()).thenReturn(true); + lenient().when(transientPayloadStorageService.extractPointer(anyString())).thenReturn("rustfs:detail"); + lenient().when(transientPayloadStorageService.storeResultPayload( + eq("COLLECT_DATA"), eq(TASK_ID), anyString(), anyString(), anyString())) + .thenReturn("\"rustfs:detail\""); + lenient().when(transientPayloadStorageService.storeChunkPayloadVersioned( + eq("COLLECT_DATA"), eq(TASK_ID), anyString(), anyInt(), anyString())) + .thenReturn("\"rustfs:chunk\""); + lenient().doAnswer(invocation -> { + TaskChunkEntity chunk = invocation.getArgument(0); + chunk.setId(701L); + return 1; + }).when(taskChunkMapper).insert(any(TaskChunkEntity.class)); + CollectDataResultRowVo row = new CollectDataResultRowVo(); + row.setAsin("B0DUP1234"); + lenient().when(collectDataBatchQuery.filter(any())).thenReturn( + new CollectDataBatchQuery.FilterResult(List.of(row), 0, 0)); + lenient().when(brandBatchFilter.filter(any())).thenReturn( + new CollectDataBrandBatchFilter.BrandBatchOutcome(List.of(), List.of(), List.of(row))); + lenient().when(resultItemBatchWriter.upsertAccepted(anyLong(), anyLong(), anyString(), anyInt(), any(), anyString())) + .thenReturn(new CollectDataResultItemBatchWriter.UpsertCounts(1, 0, 1)); + lenient().when(resultDetailCodec.encodeChunk(any())).thenReturn("{\"chunk\":1}"); + } + + @Test + void duplicateSameSubmissionIdAcceptedIdempotently() { + service.submitResult(TASK_ID, submitRequest()); + TaskChunkEntity existing = new TaskChunkEntity(); + existing.setId(702L); + existing.setChunkIndex(1); + lenient().when(taskChunkMapper.selectOne(any())).thenReturn(existing); + + CollectDataSubmitResultVo vo = service.submitResult(TASK_ID, submitRequest()); + + assertNotNull(vo, "重复提交幂等接受（成功返回）"); + } + + @Test + void duplicateDoesNotDoublePersist() { + service.submitResult(TASK_ID, submitRequest()); + TaskChunkEntity existing = new TaskChunkEntity(); + existing.setId(702L); + existing.setChunkIndex(1); + lenient().when(taskChunkMapper.selectOne(any())).thenReturn(existing); + + service.submitResult(TASK_ID, submitRequest()); + + verify(taskChunkMapper, times(1)).insert(any(TaskChunkEntity.class)); + } + + @Test + void duplicateReturnsSuccessVo() { + service.submitResult(TASK_ID, submitRequest()); + TaskChunkEntity existing = new TaskChunkEntity(); + existing.setId(702L); + existing.setChunkIndex(1); + lenient().when(taskChunkMapper.selectOne(any())).thenReturn(existing); + + CollectDataSubmitResultVo vo = service.submitResult(TASK_ID, submitRequest()); + + assertEquals(TASK_ID, vo.getTaskId()); + assertEquals(1, vo.getChunkIndex()); + assertTrue(vo.getTaskStatus() != null); + } + + @Test + void duplicateManySubmissionsAllSafe() { + service.submitResult(TASK_ID, submitRequest()); + TaskChunkEntity existing = new TaskChunkEntity(); + existing.setId(702L); + existing.setChunkIndex(1); + lenient().when(taskChunkMapper.selectOne(any())).thenReturn(existing); + + for (int i = 0; i < 5; i++) { + CollectDataSubmitResultVo vo = service.submitResult(TASK_ID, submitRequest()); + assertNotNull(vo, "第 " + (i + 1) + " 次重复提交仍幂等接受"); + } + verify(taskChunkMapper, times(1)).insert(any(TaskChunkEntity.class)); + } + + @Test + void terminalTaskRejectsDuplicateSubmission() { + FileTaskEntity done = runningTask(); + done.setStatus("SUCCESS"); + lenient().when(fileTaskMapper.selectById(TASK_ID)).thenReturn(done); + + assertThrows(BusinessException.class, () -> service.submitResult(TASK_ID, submitRequest())); + } + + @Test + void differentChunkIndexIsNotDuplicate() { + service.submitResult(TASK_ID, submitRequest()); + TaskChunkEntity existing = new TaskChunkEntity(); + existing.setId(702L); + existing.setChunkIndex(1); + lenient().when(taskChunkMapper.selectOne(any())).thenReturn(existing); + + CollectDataSubmitResultVo vo = service.submitResult(TASK_ID, submitRequestWithChunk(2, 2)); + + assertNotNull(vo); + assertEquals(2, vo.getChunkIndex(), "不同 chunk 继续受理，互不视为重复"); + } + + @Test + void concurrentDuplicateDoesNotOverwriteWinner() { + // 并发重复：查重未命中但 insert 撞唯一键 → 幂等接受，不覆盖 winner、不抛错 + lenient().when(taskChunkMapper.selectOne(any())).thenReturn(null, null); + org.mockito.Mockito.doThrow(new org.springframework.dao.DuplicateKeyException("dup")) + .when(taskChunkMapper).insert(any(TaskChunkEntity.class)); + + CollectDataSubmitResultVo vo = service.submitResult(TASK_ID, submitRequest()); + + assertNotNull(vo, "并发重复幂等接受"); + } + + @Test + void duplicateContractFrozen() { + service.submitResult(TASK_ID, submitRequest()); + TaskChunkEntity existing = new TaskChunkEntity(); + existing.setId(702L); + existing.setChunkIndex(1); + lenient().when(taskChunkMapper.selectOne(any())).thenReturn(existing); + + CollectDataSubmitResultVo first = service.submitResult(TASK_ID, submitRequest()); + + // 快照：重复提交返回同一 chunk 视图、不触发新的存储写入 + assertEquals(1, first.getChunkIndex()); + verify(transientPayloadStorageService, times(1)).storeChunkPayloadVersioned( + anyString(), anyLong(), anyString(), anyInt(), anyString()); + } + + private CollectDataSubmitResultRequest submitRequest() { + return submitRequestWithChunk(1, 1); + } + + private CollectDataSubmitResultRequest submitRequestWithChunk(int chunkIndex, int chunkTotal) { + CollectDataSubmitRowDto row = new CollectDataSubmitRowDto(); + row.setAsin("B0DUP1234"); + CollectDataSubmitResultRequest request = new CollectDataSubmitResultRequest(); + request.setSubmissionId(SUBMISSION_ID); + request.setChunkIndex(chunkIndex); + request.setChunkTotal(chunkTotal); + request.setDone(false); + request.setItems(List.of(row)); + return request; + } + + private FileTaskEntity runningTask() { + FileTaskEntity task = new FileTaskEntity(); + task.setId(TASK_ID); + task.setModuleType("COLLECT_DATA"); + task.setStatus("RUNNING"); + task.setUserId(USER_ID); + task.setResultJson("{}"); + return task; + } +}