From 95e314cef4c2fac16a52e059ccbc2a507c6a4be7 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E9=BB=84=E8=87=AA=E8=BE=BE?= <980324341@qq.com> Date: Sat, 29 Aug 2026 22:36:11 +0800 Subject: [PATCH] task-37 --- .../service/ShopDataCrawlTaskService.java | 23 +- .../ShopDataCrawlDailyFileJobSplitTest.java | 680 ++++++++++++++++++ .../service/ShopDataCrawlScopeMergeTest.java | 30 + 3 files changed, 732 insertions(+), 1 deletion(-) create mode 100644 backend-java/src/test/java/com/nanri/aiimage/modules/shopdatacrawl/service/ShopDataCrawlDailyFileJobSplitTest.java diff --git a/backend-java/src/main/java/com/nanri/aiimage/modules/shopdatacrawl/service/ShopDataCrawlTaskService.java b/backend-java/src/main/java/com/nanri/aiimage/modules/shopdatacrawl/service/ShopDataCrawlTaskService.java index 39af4d17..b3429cb3 100644 --- a/backend-java/src/main/java/com/nanri/aiimage/modules/shopdatacrawl/service/ShopDataCrawlTaskService.java +++ b/backend-java/src/main/java/com/nanri/aiimage/modules/shopdatacrawl/service/ShopDataCrawlTaskService.java @@ -1571,7 +1571,8 @@ public class ShopDataCrawlTaskService { return; } snapshot.setShopName(firstNonBlank(payload.getShopName(), snapshot.getShopName())); - snapshot.setCountryResults(copyCountryResults(payload.getCountryResults())); + // 按国家增量合并：本次未回传的国家保留快照中的旧结果，同一国家按行去重合并 + snapshot.setCountryResults(mergeCountryResults(snapshot.getCountryResults(), payload.getCountryResults())); if (!blank(payload.getError())) { snapshot.setError(payload.getError().trim()); } @@ -2327,6 +2328,26 @@ public class ShopDataCrawlTaskService { return; } deleteTransientResultChunks(job.getTaskId()); + // Task 37：异步文件作业成功钩子——组装作业完成后显式刷新任务状态： + // 仍有未完成组装作业则保持 RUNNING，全部完成才进入终态。 + // （worker 在 markSuccess 后、任务锁内调用本方法，此处刷新与提交是安全的。） + if ("SUCCESS".equals(job.getStatus())) { + refreshTaskStatusAfterAssembleJob(job.getTaskId()); + } + } + + private void refreshTaskStatusAfterAssembleJob(Long taskId) { + FileTaskEntity task = fileTaskMapper.selectById(taskId); + if (task == null || !MODULE_TYPE.equals(task.getModuleType())) { + return; + } + List rows = listTaskRows(taskId); + if (rows.isEmpty()) { + return; + } + updateTaskStatusFromRows(task, rows); + fileTaskMapper.updateById(task); + taskCacheService.deleteTaskCache(taskId); } private void deleteTransientResultChunks(Long taskId) { diff --git a/backend-java/src/test/java/com/nanri/aiimage/modules/shopdatacrawl/service/ShopDataCrawlDailyFileJobSplitTest.java b/backend-java/src/test/java/com/nanri/aiimage/modules/shopdatacrawl/service/ShopDataCrawlDailyFileJobSplitTest.java new file mode 100644 index 00000000..e4eacc94 --- /dev/null +++ b/backend-java/src/test/java/com/nanri/aiimage/modules/shopdatacrawl/service/ShopDataCrawlDailyFileJobSplitTest.java @@ -0,0 +1,680 @@ +package com.nanri.aiimage.modules.shopdatacrawl.service; + +import com.baomidou.mybatisplus.core.conditions.Wrapper; +import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper; +import com.baomidou.mybatisplus.core.metadata.TableInfoHelper; +import com.fasterxml.jackson.databind.ObjectMapper; +import com.nanri.aiimage.common.exception.BusinessException; +import com.nanri.aiimage.config.InstanceMetadata; +import com.nanri.aiimage.config.TaskPressureProperties; +import com.nanri.aiimage.modules.file.service.oss.OssStorageService; +import com.nanri.aiimage.modules.shopdatacrawl.model.entity.ShopDataCrawlDailyFileEntity; +import com.nanri.aiimage.modules.shopdatacrawl.model.entity.ShopDataCrawlDailyMemberEntity; +import com.nanri.aiimage.modules.shopdatacrawl.model.vo.ShopDataCrawlResultItemVo; +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.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.TaskFileJobEntity; +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.TaskProgressSnapshotService; +import com.nanri.aiimage.modules.task.service.TaskResultItemService; +import com.nanri.aiimage.modules.task.service.TransientPayloadStorageService; +import com.nanri.aiimage.modules.ziniao.service.ZiniaoShopSwitchService; +import org.apache.ibatis.builder.MapperBuilderAssistant; +import org.apache.ibatis.session.Configuration; +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.Mock; +import org.mockito.Spy; +import org.mockito.junit.jupiter.MockitoExtension; + +import java.time.LocalDate; +import java.time.LocalDateTime; +import java.util.ArrayList; +import java.util.Comparator; +import java.util.List; +import java.util.Map; +import java.util.Objects; +import java.util.concurrent.atomic.AtomicLong; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertNotNull; +import static org.junit.jupiter.api.Assertions.assertNull; +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.ArgumentMatchers.nullable; +import static org.mockito.Mockito.doThrow; +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 37：拆分每日累计文件组装与任务结果接收，增加异步文件作业状态。 + * 结果接收（submitResult/tryFinalizeTask → finalizeTaskWorkbook）只做“结果落库 + + * 入队文件作业”，不做任何 Excel 组装/OSS 上传；组装完全由异步文件作业 + * （processResultFileJob）承担；文件作业状态显式反映到任务状态 + * （有未完成作业 → 任务保持 RUNNING，作业成功后才进入终态）。 + */ +@ExtendWith(MockitoExtension.class) +class ShopDataCrawlDailyFileJobSplitTest { + + private static final String MODULE_TYPE = "SHOP_DATA_CRAWL"; + private static final String SHOP_NAME = "shop-a"; + private static final LocalDate BUSINESS_DATE = LocalDate.of(2026, 8, 29); + private static final LocalDateTime BUSINESS_TIME = LocalDateTime.of(2026, 8, 29, 12, 0); + + @BeforeAll + static void initializeMybatisMetadata() { + MapperBuilderAssistant assistant = new MapperBuilderAssistant(new Configuration(), ""); + TableInfoHelper.initTableInfo(assistant, FileTaskEntity.class); + TableInfoHelper.initTableInfo(assistant, FileResultEntity.class); + TableInfoHelper.initTableInfo(assistant, TaskChunkEntity.class); + TableInfoHelper.initTableInfo(assistant, TaskScopeStateEntity.class); + TableInfoHelper.initTableInfo(assistant, ShopDataCrawlDailyFileEntity.class); + TableInfoHelper.initTableInfo(assistant, ShopDataCrawlDailyMemberEntity.class); + TableInfoHelper.initTableInfo(assistant, TaskFileJobEntity.class); + } + + @Mock private FileTaskMapper fileTaskMapper; + @Mock private FileResultMapper fileResultMapper; + @Mock private ShopDataCrawlResolveService shopDataCrawlResolveService; + @Mock private ShopDataCrawlExcelAssemblyService excelAssemblyService; + @Mock private ShopDataCrawlTaskCacheService taskCacheService; + @Mock private OssStorageService ossStorageService; + @Mock private ZiniaoShopSwitchService ziniaoShopSwitchService; + @Mock private TaskPressureProperties taskPressureProperties; + @Mock private TaskFileJobService taskFileJobService; + @Mock private TaskResultItemService taskResultItemService; + @Mock private TaskProgressSnapshotService taskProgressSnapshotService; + @Mock private TaskDistributedLockService taskDistributedLockService; + @Mock private TaskChunkMapper taskChunkMapper; + @Mock private TaskScopeStateMapper taskScopeStateMapper; + @Mock private TransientPayloadStorageService transientPayloadStorageService; + @Mock private InstanceMetadata instanceMetadata; + @Mock private ShopDataCrawlDailyFileService dailyFileService; + @Spy private ObjectMapper objectMapper = new ObjectMapper().findAndRegisterModules(); + + private ShopDataCrawlTaskService service; + + private final List dbResultRows = new ArrayList<>(); + private final List dbDailyFiles = new ArrayList<>(); + private final List dbMembers = new ArrayList<>(); + /** 内存 file_job 表：enqueue/查询/状态流转的仿真。 */ + private final List dbFileJobs = new ArrayList<>(); + private final AtomicLong memberIdSeq = new AtomicLong(1000); + private final AtomicLong fileIdSeq = new AtomicLong(5000); + private final AtomicLong jobIdSeq = new AtomicLong(7000); + private long nextResultId = 8100; + private Long lastJobTaskId; + private String lastUploadedObjectKey; + private List enqueuedScopes = new ArrayList<>(); + /** 接收/组装路径每次落库后的任务状态（taskId → status），验证拆分后任务状态流转。 */ + private final java.util.Map lastTaskStatus = new java.util.HashMap<>(); + /** 接收/组装路径每次落库后的任务实体（taskId → 实体），供后续阶段读取。 */ + private final java.util.Map taskStore = new java.util.HashMap<>(); + + @BeforeEach + void configureStorage() { + service = new ShopDataCrawlTaskService( + fileTaskMapper, + fileResultMapper, + shopDataCrawlResolveService, + excelAssemblyService, + taskCacheService, + ossStorageService, + ziniaoShopSwitchService, + objectMapper, + taskPressureProperties, + taskFileJobService, + taskResultItemService, + taskProgressSnapshotService, + taskDistributedLockService, + taskChunkMapper, + taskScopeStateMapper, + transientPayloadStorageService, + instanceMetadata, + dailyFileService, + null); + + dbResultRows.clear(); + dbDailyFiles.clear(); + dbMembers.clear(); + dbFileJobs.clear(); + memberIdSeq.set(1000); + fileIdSeq.set(5000); + jobIdSeq.set(7000); + nextResultId = 8100; + lastJobTaskId = null; + lastUploadedObjectKey = null; + enqueuedScopes = new ArrayList<>(); + lastTaskStatus.clear(); + taskStore.clear(); + + lenient().when(instanceMetadata.getInstanceId()).thenReturn("instance-a"); + lenient().when(taskDistributedLockService.acquire(eq(MODULE_TYPE), anyLong())) + .thenReturn(mock(TaskDistributedLockService.LockHandle.class)); + lenient().when(taskCacheService.getTaskCacheBatch(any())).thenReturn(Map.of()); + lenient().when(taskCacheService.getTaskHeartbeatMillisBatch(any())).thenReturn(Map.of()); + lenient().when(taskCacheService.getAllShopMergedPayload(anyLong())).thenReturn(Map.of()); + lenient().when(taskResultItemService.listResultSnapshots(anyLong(), eq(MODULE_TYPE), any())) + .thenReturn(List.of()); + lenient().when(taskFileJobService.countUnfinishedAssembleJobs(anyLong(), eq(MODULE_TYPE))).thenReturn(0L); + lenient().doNothing().when(taskResultItemService) + .replaceTaskSnapshots(anyLong(), eq(MODULE_TYPE), any(), any()); + lenient().doNothing().when(taskProgressSnapshotService) + .save(anyLong(), any(), any(), anyInt(), anyInt(), anyInt(), any(), any(), any()); + lenient().when(ziniaoShopSwitchService.normalizeShopName(nullable(String.class))).thenAnswer(invocation -> { + String value = invocation.getArgument(0); + return value == null ? "" : value.trim(); + }); + lenient().when(fileTaskMapper.selectById(anyLong())).thenAnswer(invocation -> { + long taskId = invocation.getArgument(0); + FileTaskEntity stored = taskStore.get(taskId); + return stored == null ? null : stored; + }); + lenient().when(fileTaskMapper.updateById(any(FileTaskEntity.class))).thenAnswer(invocation -> { + FileTaskEntity entity = invocation.getArgument(0); + FileTaskEntity copy = copyTask(entity); + taskStore.put(copy.getId(), copy); + return 1; + }); + lenient().when(fileResultMapper.selectBatchIds(any())).thenReturn(List.of()); + lenient().when(fileResultMapper.selectCount(any())).thenReturn(0L); + lenient().when(ossStorageService.uploadResultFile(any(), eq(MODULE_TYPE))).thenAnswer(invocation -> { + String key = "oss/split/" + System.nanoTime() + ".xlsx"; + lastUploadedObjectKey = key; + return key; + }); + lenient().doNothing().when(ossStorageService).deleteObject(anyString()); + lenient().when(excelAssemblyService.countRows(any())).thenAnswer(invocation -> { + List items = invocation.getArgument(0); + return items == null ? 0 : items.size(); + }); + lenient().doAnswer(invocation -> { + List items = invocation.getArgument(1); + return items == null ? 0 : items.size(); + }).when(excelAssemblyService).writeWorkbook(any(), any()); + + lenient().when(dailyFileService.currentBusinessDate()).thenReturn(BUSINESS_DATE); + lenient().when(dailyFileService.currentBusinessDateTime()).thenReturn(BUSINESS_TIME); + lenient().when(dailyFileService.shopKeyHash(anyString())).thenAnswer(invocation -> { + String key = invocation.getArgument(0); + return key == null ? null : "hash:" + key; + }); + lenient().when(dailyFileService.shopKey(any(FileResultEntity.class))).thenAnswer(invocation -> { + FileResultEntity row = invocation.getArgument(0); + return row == null ? null : row.getSourceFilename(); + }); + lenient().when(dailyFileService.acquireLock(anyLong(), anyString())) + .thenReturn(mock(TaskDistributedLockService.LockHandle.class)); + lenient().when(dailyFileService.findForUpdate(anyLong(), anyString(), any())).thenAnswer(invocation -> + copyDailyFile(findDailyFile(invocation.getArgument(0), invocation.getArgument(1)))); + lenient().when(dailyFileService.findOlder(anyLong(), anyString(), any())).thenReturn(List.of()); + lenient().when(dailyFileService.findByLatestResultId(anyLong())).thenReturn(List.of()); + lenient().when(dailyFileService.findById(anyLong())).thenReturn(null); + lenient().when(dailyFileService.countObjectReferences(anyString())).thenReturn(0L); + lenient().doNothing().when(dailyFileService).deleteDailyFile(anyLong()); + lenient().doNothing().when(dailyFileService).reassignMembers(anyLong(), anyLong()); + lenient().doNothing().when(dailyFileService).deleteMembersForResults(any()); + captureDailyFilePersistence(); + captureMemberInserts(); + captureResultUpdates(); + captureFileJobs(); + + lenient().when(fileResultMapper.selectList(any())).thenAnswer(invocation -> { + Wrapper wrapper = invocation.getArgument(0); + if (wrapper instanceof LambdaQueryWrapper query + && query.getSqlSegment() != null && query.getSqlSegment().contains("taskId")) { + List rows = dbResultRows.stream() + .filter(r -> Objects.equals(r.getTaskId(), lastJobTaskId)) + .sorted(Comparator.comparing(FileResultEntity::getId)) + .toList(); + return new ArrayList<>(rows); + } + return List.of(); + }); + } + + @Test + void test_task_037_daily_file_job_normal_default_path() { + // 默认成功路径：接收完成后只入队一个异步组装作业（status=PENDING、携带 owner scope）， + // 不做任何组装；组装作业执行后产生累计文件、成员行、作业状态置为 SUCCESS。 + FileResultEntity row = addResultRow(8101L, 1L, 1, SHOP_NAME, null); + receiveTask(1L, List.of(row), snapshot(8101L)); + + assertEquals(1, dbFileJobs.size(), "接收路径只入队一个组装作业"); + assertEquals("PENDING", dbFileJobs.get(0).getStatus()); + assertEquals("ASSEMBLE_RESULT", dbFileJobs.get(0).getJobType()); + assertTrue(enqueuedScopes.get(0).contains("owner"), "入队作业携带 owner scope"); + verify(excelAssemblyService, never()).writeWorkbook(any(), any()); + assertEquals(0, dbMembers.size(), "接收路径不产生成员行"); + assertEquals(0, dbDailyFiles.size(), "接收路径不创建累计文件"); + + runAssembleJob(1L); + assertEquals(1, dbMembers.size(), "组装作业执行后产生成员行"); + assertEquals(1, dbDailyFiles.size(), "组装作业执行后产生累计文件"); + assertEquals("SUCCESS", dbFileJobs.get(0).getStatus(), "作业状态显式落到 SUCCESS"); + assertEquals("SUCCESS", lastTaskStatus.get(1L), "作业完成后任务进入终态"); + } + + @Test + void test_task_037_daily_file_job_normal_multiple_items() { + // 批量场景：多个店铺结果入队各自独立的组装作业，逐个执行后 + // 累计文件版本随提交递增，成员行不丢失、顺序稳定。 + receiveTask(1L, List.of(addResultRow(8102L, 1L, 1, SHOP_NAME, null)), snapshot(8102L)); + assertEquals(1, dbFileJobs.size(), "首个结果入队一个作业"); + + runAssembleJob(1L); + receiveTask(2L, List.of(addResultRow(8103L, 2L, 1, SHOP_NAME, null)), snapshot(8103L)); + assertEquals(2, dbFileJobs.size(), "第二个结果入队第二个作业"); + runAssembleJob(2L); + + assertEquals(2, dbMembers.size(), "两个结果各产生一个成员行"); + assertEquals(2L, dbDailyFiles.get(0).getVersion(), "版本号随两次归档递增"); + assertEquals(2, dbFileJobs.stream().filter(j -> "SUCCESS".equals(j.getStatus())).count(), + "两个作业都显式落到 SUCCESS"); + assertEquals("SUCCESS", lastTaskStatus.get(1L)); + assertEquals("SUCCESS", lastTaskStatus.get(2L)); + } + + @Test + void test_task_037_daily_file_job_normal_repeated_operation_is_idempotent() { + // 幂等：同一结果重复提交接收，不产生重复组装作业（已存在 PENDING/SUCCESS 作业不重复入队）； + // 同一作业重复执行不产生重复成员行、版本不重复递增。 + FileResultEntity row = addResultRow(8104L, 1L, 1, SHOP_NAME, null); + receiveTask(1L, List.of(row), snapshot(8104L)); + receiveTask(1L, List.of(row), snapshot(8104L)); + + assertEquals(1, dbFileJobs.size(), "重复接收不重复入队作业"); + runAssembleJob(1L); + runAssembleJob(1L); + assertEquals(1, dbMembers.size(), "重复执行作业不产生重复成员行"); + assertEquals(1L, dbDailyFiles.get(0).getVersion(), "重复执行作业版本号不重复递增"); + } + + @Test + void test_task_037_daily_file_job_boundary_empty_input() { + // 空输入：没有成功结果时不入队作业、不创建任何资源，任务不进入组装等待。 + FileResultEntity row = addResultRow(8105L, 1L, 0, SHOP_NAME, null); + receiveTask(1L, List.of(row), snapshot(8105L)); + + assertTrue(dbFileJobs.isEmpty(), "无成功结果不入队组装作业"); + assertTrue(dbMembers.isEmpty(), "无成功结果不产生成员行"); + assertTrue(dbDailyFiles.isEmpty(), "无成功结果不创建累计文件"); + verify(ossStorageService, never()).uploadResultFile(any(), eq(MODULE_TYPE)); + } + + @Test + void test_task_037_daily_file_job_boundary_single_item() { + // 单元素：单店铺单结果走同样拆分路径，作业一入队即达终态，不依赖批量逻辑。 + FileResultEntity row = addResultRow(8106L, 1L, 1, SHOP_NAME, null); + receiveTask(1L, List.of(row), snapshot(8106L)); + + assertEquals(1, dbFileJobs.size()); + assertEquals("PENDING", dbFileJobs.get(0).getStatus()); + runAssembleJob(1L); + assertEquals(1, dbMembers.size()); + assertEquals(1L, dbDailyFiles.get(0).getVersion()); + assertEquals("SUCCESS", lastTaskStatus.get(1L)); + } + + @Test + void test_task_037_daily_file_job_boundary_limit_and_overflow() { + // 上限/超限：大量结果连续接收+执行，作业数、成员数、版本号一致递增，无丢失无重复。 + int limit = 210; + for (int i = 0; i < limit; i++) { + long resultId = nextResultId++; + long taskId = i + 1L; + receiveTask(taskId, List.of(addResultRow(resultId, taskId, 1, SHOP_NAME, null)), snapshot(resultId)); + runAssembleJob(taskId); + } + + assertEquals(limit, dbFileJobs.size(), "每个结果一个作业"); + assertEquals(limit, dbMembers.size(), "每个结果一个成员行"); + assertEquals(Long.valueOf(limit), dbDailyFiles.get(0).getVersion(), "版本号与归档次数一致"); + assertEquals(limit, dbFileJobs.stream().filter(j -> "SUCCESS".equals(j.getStatus())).count(), + "全部作业显式 SUCCESS"); + } + + @Test + void test_task_037_daily_file_job_invalid_input_rejected() { + // 非法参数：作业缺少 taskId 时拒绝执行并抛可识别异常；任务不存在时同样拒绝。 + assertThrows(BusinessException.class, () -> service.processResultFileJob( + jobEntity(0L, null, MODULE_TYPE))); + assertThrows(BusinessException.class, () -> service.processResultFileJob( + jobEntity(0L, 9999L, MODULE_TYPE))); + } + + @Test + void test_task_037_daily_file_job_dependency_failure_releases_resources() { + // 依赖失败：组装作业执行中 OSS 上传失败时抛出异常、不残留成员/累计文件； + // 恢复后同一结果可重新归档成功（错误可恢复）。 + FileResultEntity row = addResultRow(8108L, 1L, 1, SHOP_NAME, null); + receiveTask(1L, List.of(row), snapshot(8108L)); + assertEquals(1, dbFileJobs.size(), "接收入队成功"); + dbFileJobs.clear(); + + doThrow(new IllegalStateException("upload failed")) + .when(ossStorageService).uploadResultFile(any(), eq(MODULE_TYPE)); + assertThrows(IllegalStateException.class, () -> service.processResultFileJob( + jobEntity(1L, 1L, MODULE_TYPE))); + + assertTrue(dbMembers.isEmpty(), "上传失败不残留成员行"); + assertTrue(dbDailyFiles.isEmpty(), "上传失败不残留累计文件"); + + // 错误可恢复：恢复 OSS 后同一作业重新执行成功。 + lenient().when(ossStorageService.uploadResultFile(any(), eq(MODULE_TYPE))).thenAnswer(invocation -> { + String key = "oss/split/recovered-" + System.nanoTime() + ".xlsx"; + lastUploadedObjectKey = key; + return key; + }); + service.processResultFileJob(jobEntity(1L, 1L, MODULE_TYPE)); + assertEquals(1, dbMembers.size(), "恢复后同一结果重新归档成功"); + assertEquals(1, dbDailyFiles.size(), "恢复后累计文件创建成功"); + assertEquals(1L, dbDailyFiles.get(0).getVersion(), "恢复后版本号正确"); + } + + // ---- helpers ---- + + /** 接收阶段：submitResult → tryFinalizeTask → finalizeTaskWorkbook（结果落库 + 入队组装作业）。 */ + private void receiveTask(Long taskId, List rows, ShopDataCrawlResultItemVo snapshot) { + lastJobTaskId = taskId; + FileTaskEntity task = taskEntity(taskId); + taskStore.put(taskId, task); + for (FileResultEntity row : rows) { + updateRowInDb(row); + } + service.tryFinalizeTask(taskId, false); + FileTaskEntity persisted = taskStore.get(taskId); + if (persisted != null) { + lastTaskStatus.put(taskId, persisted.getStatus()); + } + } + + /** 组装阶段：worker 执行 processResultFileJob（组装），随后 markSuccess + cleanupResultFileJob + * （作业成功钩子：把文件作业状态显式反映到任务状态，无未完成作业 → 任务进入终态）。 */ + private void runAssembleJob(Long taskId) { + lastJobTaskId = taskId; + TaskFileJobEntity job = dbFileJobs.stream() + .filter(j -> Objects.equals(j.getTaskId(), taskId)) + .findFirst() + .orElseThrow(() -> new AssertionError("作业不存在 taskId=" + taskId)); + service.processResultFileJob(job); + job.setStatus("SUCCESS"); + service.cleanupResultFileJob(job); + lastTaskStatus.put(taskId, taskStore.get(taskId).getStatus()); + } + + private void markResultFinishedInDb(FileResultEntity row) { + row.setSuccess(1); + row.setErrorMessage(null); + updateRowInDb(row); + } + + private void updateRowInDb(FileResultEntity row) { + for (int i = 0; i < dbResultRows.size(); i++) { + if (Objects.equals(dbResultRows.get(i).getId(), row.getId())) { + dbResultRows.set(i, row); + return; + } + } + dbResultRows.add(row); + } + + private TaskFileJobEntity jobEntity(long jobId, Long taskId, String moduleType) { + TaskFileJobEntity job = new TaskFileJobEntity(); + job.setId(jobId); + job.setTaskId(taskId); + job.setModuleType(moduleType); + return job; + } + + private FileTaskEntity taskEntity(long taskId) { + FileTaskEntity task = new FileTaskEntity(); + task.setId(taskId); + task.setModuleType(MODULE_TYPE); + task.setStatus("RUNNING"); + task.setUserId(7L); + task.setOwnerInstanceId("instance-a"); + task.setRequestJson("{\"ownerInstanceId\":\"instance-a\"}"); + task.setResultJson("[]"); + task.setCreatedAt(LocalDateTime.now()); + task.setUpdatedAt(LocalDateTime.now()); + return task; + } + + private FileResultEntity addResultRow(long id, long taskId, int success, String shopName, String resultFileUrl) { + FileResultEntity row = new FileResultEntity(); + row.setId(id); + row.setTaskId(taskId); + row.setModuleType(MODULE_TYPE); + row.setSuccess(success); + row.setSourceFilename(shopName); + row.setSourceFileUrl("shop-id-" + id); + row.setUserId(7L); + row.setCreatedAt(LocalDateTime.now()); + row.setResultFileUrl(resultFileUrl); + dbResultRows.add(row); + return row; + } + + private ShopDataCrawlResultItemVo snapshot(long resultId) { + ShopDataCrawlResultItemVo item = new ShopDataCrawlResultItemVo(); + item.setResultId(resultId); + item.setTaskId(1L); + item.setShopName(SHOP_NAME); + item.setShopId("shop-id-" + resultId); + item.setSuccess(true); + item.setMatched(true); + item.setTaskStatus("SUCCESS"); + item.setCountryCodes(List.of("DE")); + return item; + } + + private void captureFileJobs() { + // 仿真 enqueueAssembleResult 的幂等语义：同一 (taskId, resultId) 已存在作业时不重复入队。 + lenient().doAnswer(invocation -> { + Long taskId = invocation.getArgument(0); + Long resultId = invocation.getArgument(2); + String scopeKey = invocation.getArgument(3); + TaskFileJobEntity existing = dbFileJobs.stream() + .filter(j -> Objects.equals(j.getTaskId(), taskId) + && Objects.equals(j.getResultId(), resultId)) + .findFirst() + .orElse(null); + if (existing != null) { + if ("FAILED".equals(existing.getStatus())) { + existing.setStatus("PENDING"); + } + return existing; + } + TaskFileJobEntity job = new TaskFileJobEntity(); + job.setId(jobIdSeq.incrementAndGet()); + job.setTaskId(taskId); + job.setModuleType(MODULE_TYPE); + job.setResultId(resultId); + job.setScopeKey(scopeKey); + job.setStatus("PENDING"); + job.setJobType("ASSEMBLE_RESULT"); + job.setRetryCount(0); + dbFileJobs.add(job); + enqueuedScopes.add(scopeKey); + return job; + }).when(taskFileJobService).enqueueAssembleResult(anyLong(), eq(MODULE_TYPE), anyLong(), anyString()); + lenient().when(taskFileJobService.findAssembleJob(anyLong(), eq(MODULE_TYPE), anyLong())) + .thenAnswer(invocation -> { + long taskId = invocation.getArgument(0); + long resultId = invocation.getArgument(2); + return dbFileJobs.stream() + .filter(j -> Objects.equals(j.getTaskId(), taskId) + && Objects.equals(j.getResultId(), resultId)) + .findFirst() + .orElse(null); + }); + lenient().when(taskFileJobService.countUnfinishedAssembleJobs(anyLong(), eq(MODULE_TYPE))) + .thenAnswer(invocation -> { + long taskId = invocation.getArgument(0); + return dbFileJobs.stream() + .filter(j -> Objects.equals(j.getTaskId(), taskId) + && !"SUCCESS".equals(j.getStatus())) + .count(); + }); + } + + private void captureResultUpdates() { + lenient().when(fileResultMapper.updateById(any(FileResultEntity.class))).thenAnswer(invocation -> { + FileResultEntity updated = invocation.getArgument(0); + for (int i = 0; i < dbResultRows.size(); i++) { + if (Objects.equals(dbResultRows.get(i).getId(), updated.getId())) { + dbResultRows.set(i, updated); + return 1; + } + } + dbResultRows.add(updated); + return 1; + }); + } + + private void captureMemberInserts() { + lenient().when(dailyFileService.addMemberWithPayload(anyLong(), anyLong(), anyLong(), anyString())) + .thenAnswer(invocation -> { + long dailyFileId = invocation.getArgument(0); + long taskId = invocation.getArgument(1); + long resultId = invocation.getArgument(2); + String rowPayload = invocation.getArgument(3); + boolean duplicate = dbMembers.stream().anyMatch(m -> + Objects.equals(m.getDailyFileId(), dailyFileId) + && Objects.equals(m.getResultId(), resultId)); + if (duplicate) { + return false; + } + ShopDataCrawlDailyMemberEntity member = new ShopDataCrawlDailyMemberEntity(); + member.setId(memberIdSeq.incrementAndGet()); + member.setDailyFileId(dailyFileId); + member.setTaskId(taskId); + member.setResultId(resultId); + member.setRowPayload(rowPayload); + member.setCreatedAt(BUSINESS_TIME.plusMinutes(dbMembers.size())); + dbMembers.add(member); + return true; + }); + } + + private void captureDailyFilePersistence() { + lenient().doAnswer(invocation -> { + ShopDataCrawlDailyFileEntity entity = invocation.getArgument(0); + entity.setId(fileIdSeq.incrementAndGet()); + dbDailyFiles.add(entity); + return null; + }).when(dailyFileService).insert(any(ShopDataCrawlDailyFileEntity.class)); + lenient().doAnswer(invocation -> { + ShopDataCrawlDailyFileEntity entity = invocation.getArgument(0); + for (int i = 0; i < dbDailyFiles.size(); i++) { + ShopDataCrawlDailyFileEntity existing = dbDailyFiles.get(i); + if (Objects.equals(existing.getUserId(), entity.getUserId()) + && Objects.equals(existing.getShopKeyHash(), entity.getShopKeyHash()) + && Objects.equals(existing.getBusinessDate(), entity.getBusinessDate())) { + entity.setId(existing.getId()); + dbDailyFiles.set(i, copyDailyFile(entity)); + return null; + } + } + entity.setId(fileIdSeq.incrementAndGet()); + dbDailyFiles.add(copyDailyFile(entity)); + return null; + }).when(dailyFileService).update(any(ShopDataCrawlDailyFileEntity.class)); + lenient().when(dailyFileService.listMembers(anyLong())).thenAnswer(invocation -> { + long dailyFileId = invocation.getArgument(0); + return dbMembers.stream() + .filter(m -> Objects.equals(m.getDailyFileId(), dailyFileId)) + .sorted(Comparator + .comparing(ShopDataCrawlDailyMemberEntity::getCreatedAt, + Comparator.nullsLast(Comparator.naturalOrder())) + .thenComparing(ShopDataCrawlDailyMemberEntity::getId, + Comparator.nullsLast(Comparator.naturalOrder()))) + .toList(); + }); + lenient().when(dailyFileService.containsResult(anyLong(), anyLong())).thenAnswer(invocation -> { + long dailyFileId = invocation.getArgument(0); + long resultId = invocation.getArgument(1); + return dbMembers.stream().anyMatch(m -> + Objects.equals(m.getDailyFileId(), dailyFileId) && Objects.equals(m.getResultId(), resultId)); + }); + lenient().when(dailyFileService.findMembersByResultId(anyLong())).thenAnswer(invocation -> { + long resultId = invocation.getArgument(0); + return dbMembers.stream().filter(m -> Objects.equals(m.getResultId(), resultId)).toList(); + }); + } + + private ShopDataCrawlDailyFileEntity findDailyFile(Long userId, String shopKeyHash) { + for (ShopDataCrawlDailyFileEntity f : dbDailyFiles) { + if (Objects.equals(f.getUserId(), userId) && Objects.equals(f.getShopKeyHash(), shopKeyHash)) { + return f; + } + } + return null; + } + + private ShopDataCrawlDailyFileEntity copyDailyFile(ShopDataCrawlDailyFileEntity source) { + if (source == null) { + return null; + } + ShopDataCrawlDailyFileEntity copy = new ShopDataCrawlDailyFileEntity(); + copy.setId(source.getId()); + copy.setUserId(source.getUserId()); + copy.setShopKeyHash(source.getShopKeyHash()); + copy.setShopKey(source.getShopKey()); + copy.setBusinessDate(source.getBusinessDate()); + copy.setLatestTaskId(source.getLatestTaskId()); + copy.setLatestResultId(source.getLatestResultId()); + copy.setResultFilename(source.getResultFilename()); + copy.setResultFileUrl(source.getResultFileUrl()); + copy.setResultFileSize(source.getResultFileSize()); + copy.setResultContentType(source.getResultContentType()); + copy.setRowCount(source.getRowCount()); + copy.setVersion(source.getVersion()); + copy.setLastSuccessAt(source.getLastSuccessAt()); + copy.setCreatedAt(source.getCreatedAt()); + copy.setUpdatedAt(source.getUpdatedAt()); + return copy; + } + + private FileTaskEntity copyTask(FileTaskEntity source) { + FileTaskEntity copy = new FileTaskEntity(); + copy.setId(source.getId()); + copy.setTaskNo(source.getTaskNo()); + copy.setModuleType(source.getModuleType()); + copy.setTaskMode(source.getTaskMode()); + copy.setStatus(source.getStatus()); + copy.setSourceFileCount(source.getSourceFileCount()); + copy.setSuccessFileCount(source.getSuccessFileCount()); + copy.setFailedFileCount(source.getFailedFileCount()); + copy.setRequestJson(source.getRequestJson()); + copy.setResultJson(source.getResultJson()); + copy.setErrorMessage(source.getErrorMessage()); + copy.setCreatedBy(source.getCreatedBy()); + copy.setUserId(source.getUserId()); + copy.setOwnerInstanceId(source.getOwnerInstanceId()); + copy.setCreatedAt(source.getCreatedAt()); + copy.setUpdatedAt(source.getUpdatedAt()); + copy.setFinishedAt(source.getFinishedAt()); + copy.setScheduledAt(source.getScheduledAt()); + return copy; + } +} diff --git a/backend-java/src/test/java/com/nanri/aiimage/modules/shopdatacrawl/service/ShopDataCrawlScopeMergeTest.java b/backend-java/src/test/java/com/nanri/aiimage/modules/shopdatacrawl/service/ShopDataCrawlScopeMergeTest.java index ca1b4423..4aa78820 100644 --- a/backend-java/src/test/java/com/nanri/aiimage/modules/shopdatacrawl/service/ShopDataCrawlScopeMergeTest.java +++ b/backend-java/src/test/java/com/nanri/aiimage/modules/shopdatacrawl/service/ShopDataCrawlScopeMergeTest.java @@ -52,6 +52,7 @@ import static org.mockito.ArgumentMatchers.anyLong; import static org.mockito.ArgumentMatchers.anyString; import static org.mockito.ArgumentMatchers.eq; import static org.mockito.ArgumentMatchers.nullable; +import static org.mockito.Mockito.doAnswer; import static org.mockito.Mockito.lenient; import static org.mockito.Mockito.mock; import static org.mockito.Mockito.never; @@ -257,6 +258,35 @@ class ShopDataCrawlScopeMergeTest { verify(taskScopeStateMapper, never()).selectOne(any()); } + @Test + void test_task_029_submit_preserves_previous_countries_across_submissions() { + // 回归：客户端先提交一个国家的部分结果（未完结），再提交另一个国家并标记完结。 + // 首次提交国家被清空后，第二次提交的 payload 只含 UK，修复前 DE 会在快照中永久丢失； + // 修复后 mergePayloadIntoSnapshot 按国家合并，DE 必须保留、且第二次完成后正常收尾。 + Map mergedByShop = new LinkedHashMap<>(); + when(taskCacheService.getShopMergedPayload(anyLong(), anyString())).thenAnswer( + invocation -> mergedByShop.get(invocation.getArgument(1))); + doAnswer(invocation -> { + mergedByShop.put(invocation.getArgument(1), invocation.getArgument(2)); + return null; + }).when(taskCacheService).saveShopMergedPayload(anyLong(), anyString(), any(ShopDataCrawlShopPayloadDto.class)); + + givenRunningTask(1205L, 2205L); + + // 第一次提交：只带 DE 国家，未完结（生产上第一次提交的 shopDone=false） + ShopDataCrawlShopPayloadDto first = legacyChunk(false, "DE", row("2026-07-25", "B001")); + service.submitResult(task.getId(), request(first)); + // 第二次提交：只带 UK 国家，标记完结（生产上第二次提交的 shopDone=true 收尾） + ShopDataCrawlShopPayloadDto second = legacyChunk(true, "UK", row("2026-07-26", "B002")); + service.submitResult(task.getId(), request(second)); + + assertEquals(1, result.getSuccess(), "任务应收尾成功"); + String json = task.getResultJson(); + assertTrue(json.contains("\"DE\""), "前次提交的 DE 国家必须保留: " + json); + assertTrue(json.contains("\"UK\""), "后次提交的 UK 国家必须写入: " + json); + assertTrue(json.indexOf("B001") < json.indexOf("B002")); + } + @Test void test_task_029_chunk_merge_boundary_single_item() { // 单元素：单分片 1/1，一次 scope 查询 + 一次 insert 即齐集完成。