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 b1f439bd..477d62dd 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 @@ -86,6 +86,8 @@ public class ShopDataCrawlTaskService { private static final String INTERRUPTED_MESSAGE = "Python 在该店铺结果提交完成前中断"; private static final String PARTIAL_RESULT_MESSAGE = "Python 中断，已保留已回传的部分数据"; private static final String RESULT_CHUNK_SCOPE_PREFIX = "result-chunks:"; + /** 国家结果行去重键字段分隔符（控制字符，字段值 trim 后不可能包含）。 */ + private static final String ROW_KEY_SEPARATOR = ""; private final FileTaskMapper fileTaskMapper; private final FileResultMapper fileResultMapper; @@ -1604,9 +1606,22 @@ public class ShopDataCrawlTaskService { map.put(item.getCountry(), item); continue; } - List merged = new ArrayList<>(existing.getItems() == null ? List.of() : existing.getItems()); + List merged = existing.getItems() == null ? new ArrayList<>() : new ArrayList<>(existing.getItems()); + Set seen = new HashSet<>(); + for (ShopDataCrawlRowDto old : merged) { + String key = rowDedupKey(old); + if (key != null) { + seen.add(key); + } + } for (ShopDataCrawlRowDto row : item.getItems() == null ? List.of() : item.getItems()) { - if (row != null && merged.stream().noneMatch(old -> sameRow(old, row))) merged.add(copyRow(row)); + if (row == null) { + continue; + } + String key = rowDedupKey(row); + if (key != null && seen.add(key)) { + merged.add(copyRow(row)); + } } existing.setItems(merged); } @@ -2366,8 +2381,13 @@ public class ShopDataCrawlTaskService { item.setCountry(normalizeCountry(source.getCountry())); List rows = new ArrayList<>(); if (source.getItems() != null) { + Set seen = new HashSet<>(); for (ShopDataCrawlRowDto sourceRow : source.getItems()) { - if (sourceRow != null && !rowEmpty(sourceRow) && rows.stream().noneMatch(old -> sameRow(old, sourceRow))) { + if (sourceRow == null || rowEmpty(sourceRow)) { + continue; + } + String key = rowDedupKey(sourceRow); + if (key != null && seen.add(key)) { rows.add(copyRow(sourceRow)); } } @@ -2455,13 +2475,25 @@ public class ShopDataCrawlTaskService { && Objects.equals(trim(left.getRecommendedOffer()), trim(right.getRecommendedOffer())); } + /** 国家结果行稳定去重键：与 sameRow 的 10 字段 trim 比较语义等价，用于 O(1) 去重。 */ + static String rowDedupKey(ShopDataCrawlRowDto row) { + if (row == null) { + return null; + } + return trim(row.getDate()) + ROW_KEY_SEPARATOR + trim(row.getAsin()) + ROW_KEY_SEPARATOR + trim(row.getBrand()) + + ROW_KEY_SEPARATOR + trim(row.getCommodityImage()) + ROW_KEY_SEPARATOR + trim(row.getInventorySales()) + + ROW_KEY_SEPARATOR + trim(row.getSalesRank()) + ROW_KEY_SEPARATOR + trim(row.getPageViews()) + + ROW_KEY_SEPARATOR + trim(row.getUnitsSold()) + ROW_KEY_SEPARATOR + trim(row.getPrice()) + + ROW_KEY_SEPARATOR + trim(row.getRecommendedOffer()); + } + private boolean rowEmpty(ShopDataCrawlRowDto row) { return row == null || (blank(row.getDate()) && blank(row.getAsin()) && blank(row.getBrand()) && blank(row.getCommodityImage()) && blank(row.getInventorySales()) && blank(row.getSalesRank()) && blank(row.getPageViews()) && blank(row.getUnitsSold()) && blank(row.getPrice()) && blank(row.getRecommendedOffer())); } - private String trim(String value) { + private static String trim(String value) { return value == null ? "" : value.trim(); } diff --git a/backend-java/src/test/java/com/nanri/aiimage/modules/shopdatacrawl/service/ShopDataCrawlRowDedupKeyTest.java b/backend-java/src/test/java/com/nanri/aiimage/modules/shopdatacrawl/service/ShopDataCrawlRowDedupKeyTest.java new file mode 100644 index 00000000..1a029d59 --- /dev/null +++ b/backend-java/src/test/java/com/nanri/aiimage/modules/shopdatacrawl/service/ShopDataCrawlRowDedupKeyTest.java @@ -0,0 +1,654 @@ +package com.nanri.aiimage.modules.shopdatacrawl.service; + +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.dto.ShopDataCrawlCountryResultDto; +import com.nanri.aiimage.modules.shopdatacrawl.model.dto.ShopDataCrawlRowDto; +import com.nanri.aiimage.modules.shopdatacrawl.model.dto.ShopDataCrawlShopPayloadDto; +import com.nanri.aiimage.modules.shopdatacrawl.model.dto.ShopDataCrawlSubmitResultRequest; +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.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 org.springframework.dao.DuplicateKeyException; + +import java.time.LocalDateTime; +import java.util.ArrayList; +import java.util.Comparator; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; +import java.util.Objects; + +import static org.junit.jupiter.api.Assertions.assertEquals; +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.anyLong; +import static org.mockito.ArgumentMatchers.anyString; +import static org.mockito.ArgumentMatchers.eq; +import static org.mockito.ArgumentMatchers.nullable; +import static org.mockito.Mockito.lenient; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; + +/** + * Task 30：为国家结果行建立稳定去重键，替换线性重复扫描。 + * mergeCountryResults / copyCountryResults 原用 sameRow 的 O(n²) noneMatch 扫描去重； + * 实现改为按 10 个 trim 后字段构造稳定去重键（date|asin|brand|commodityImage|inventorySales| + * salesRank|pageViews|unitsSold|price|recommendedOffer），LinkedHashSet 一次遍历去重， + * 保留首现顺序与语义等价性（去重结果与 sameRow 逐行比较完全一致）。 + */ +@ExtendWith(MockitoExtension.class) +class ShopDataCrawlRowDedupKeyTest { + + private static final String MODULE_TYPE = "SHOP_DATA_CRAWL"; + private static final String SHOP_NAME = "Demo Shop"; + + @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); + } + + @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 storedChunks = new ArrayList<>(); + private final List storedScopes = new ArrayList<>(); + private final Map rustfsPayloads = new LinkedHashMap<>(); + private FileTaskEntity task; + private FileResultEntity result; + private int nextPayloadId; + private boolean insertFails; + private boolean scopeUpdateFails; + + @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); + + storedChunks.clear(); + storedScopes.clear(); + rustfsPayloads.clear(); + nextPayloadId = 0; + insertFails = false; + scopeUpdateFails = false; + + 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.getAllShopMergedPayload(anyLong())).thenReturn(Map.of()); + lenient().when(taskResultItemService.listResultSnapshots(anyLong(), eq(MODULE_TYPE), any())).thenReturn(List.of()); + lenient().when(taskFileJobService.findAssembleJob(anyLong(), eq(MODULE_TYPE), anyLong())).thenReturn(null); + lenient().when(excelAssemblyService.countRows(any())).thenReturn(1); + + lenient().when(fileTaskMapper.selectById(anyLong())).thenAnswer(invocation -> { + Long taskId = invocation.getArgument(0); + return task != null && Objects.equals(taskId, task.getId()) ? task : null; + }); + lenient().when(fileTaskMapper.updateById(any(FileTaskEntity.class))).thenReturn(1); + lenient().when(fileResultMapper.selectById(anyLong())).thenAnswer(invocation -> { + Long resultId = invocation.getArgument(0); + return result != null && Objects.equals(resultId, result.getId()) ? result : null; + }); + lenient().when(fileResultMapper.selectList(any())).thenAnswer(invocation -> { + LambdaQueryWrapper query = invocation.getArgument(0); + Long taskId = queryLong(query); + return result != null && Objects.equals(taskId, result.getTaskId()) ? List.of(result) : List.of(); + }); + lenient().when(fileResultMapper.updateById(any(FileResultEntity.class))).thenReturn(1); + + lenient().when(ziniaoShopSwitchService.normalizeShopName(nullable(String.class))).thenAnswer(invocation -> { + String value = invocation.getArgument(0); + return value == null ? "" : value.trim(); + }); + + configureTransientPayloadStorage(); + configureChunkMapper(); + configureScopeMapper(); + } + + @Test + void test_task_030_task_normal_default_path() { + // 正常路径：同一国家内重复行按稳定去重键合并，保留首现顺序； + // 全部 10 个字段被保留（copyRow 语义），无重复行。 + givenRunningTask(1301L, 2301L); + + service.submitResult(task.getId(), request(chunk(1, 1, "DE", + row("2026-07-25", "B001"), + row("2026-07-26", "B002"), + sameRow("2026-07-25", "B001")))); + + List rows = finalRows(); + assertEquals(2, rows.size(), "重复行只保留一份"); + assertEquals("B001", rows.get(0).getAsin(), "首现顺序保留"); + assertEquals("B002", rows.get(1).getAsin()); + assertEquals("https://m.media-amazon.com/images/I/B001.jpg", rows.get(0).getCommodityImage()); + assertEquals("10", rows.get(0).getInventorySales()); + assertEquals("20", rows.get(0).getSalesRank()); + assertEquals("30", rows.get(0).getPageViews()); + assertEquals("40", rows.get(0).getUnitsSold()); + assertEquals("50", rows.get(0).getPrice()); + assertEquals("60", rows.get(0).getRecommendedOffer()); + + // 稳定去重键：语义相同（含空白差异）的行键相等，null 行安全。 + assertEquals(ShopDataCrawlTaskService.rowDedupKey(row("2026-07-25", "B001")), + ShopDataCrawlTaskService.rowDedupKey(sameRow("2026-07-25", "B001"))); + assertEquals(ShopDataCrawlTaskService.rowDedupKey(row("2026-07-25", "B001")), + ShopDataCrawlTaskService.rowDedupKey(paddedRow("2026-07-25", "B001"))); + assertEquals(null, ShopDataCrawlTaskService.rowDedupKey(null)); + } + + @Test + void test_task_030_task_normal_multiple_items() { + // 批量场景：同一国家跨多个分片合并去重（跨分片重复行只保留首现）， + // 国家顺序与行顺序稳定（国家首次出现顺序、行首次出现顺序）。 + givenRunningTask(1302L, 2302L); + + service.submitResult(task.getId(), request(chunk(1, 3, "DE", + row("2026-07-25", "B001"), + row("2026-07-26", "B002")))); + service.submitResult(task.getId(), request(chunk(2, 3, "DE", + row("2026-07-26", "B002"), + row("2026-07-27", "B003")))); + service.submitResult(task.getId(), request(chunk(3, 3, "UK", + row("2026-07-27", "B003"), + row("2026-07-28", "B004")))); + + List countries = finalCountries(); + assertEquals(2, countries.size(), "国家按首次出现顺序"); + assertEquals("DE", countries.get(0).getCountry()); + assertEquals("UK", countries.get(1).getCountry()); + assertEquals(List.of("B001", "B002", "B003"), asins(countries.get(0).getItems()), "跨分片重复合并"); + assertEquals(List.of("B003", "B004"), asins(countries.get(1).getItems())); + } + + @Test + void test_task_030_task_normal_repeated_operation_is_idempotent() { + // 幂等：同一分片重放不会放大结果（命中唯一键，payload 清理，行不重复）。 + givenRunningTask(1303L, 2303L); + ShopDataCrawlSubmitResultRequest request = request(chunk(1, 2, "DE", + row("2026-07-25", "B001"), + row("2026-07-26", "B002"))); + + service.submitResult(task.getId(), request); + service.submitResult(task.getId(), request); + service.submitResult(task.getId(), request(chunk(2, 2, "DE", + row("2026-07-26", "B002"), + row("2026-07-27", "B003")))); + + assertEquals(List.of("B001", "B002", "B003"), asins(finalRows()), "重放不放大结果"); + assertEquals(1, storedScopes.size(), "scope 状态只有一行"); + assertEquals(2, storedChunks.size(), "分片行只有两份"); + } + + @Test + void test_task_030_task_boundary_empty_input() { + // 空输入：无可处理数据的分片在去重之前被拒，无结果行、无资源创建。 + givenRunningTask(1304L, 2304L); + + ShopDataCrawlShopPayloadDto emptyItems = new ShopDataCrawlShopPayloadDto(); + emptyItems.setShopName(SHOP_NAME); + emptyItems.setChunkIndex(1); + emptyItems.setChunkTotal(1); + emptyItems.setCountryResults(List.of(countryWithItems("DE", List.of()))); + BusinessException error = assertThrows(BusinessException.class, () -> + service.submitResult(task.getId(), request(emptyItems))); + assertTrue(error.getMessage().contains("内容为空")); + + assertEquals(0, storedChunks.size()); + assertEquals(0, rustfsPayloads.size()); + assertEquals(0, storedScopes.size()); + } + + @Test + void test_task_030_task_boundary_single_item() { + // 单元素：单国家单行，去重键单元素路径结果正确、顺序稳定。 + givenRunningTask(1305L, 2305L); + + service.submitResult(task.getId(), request(chunk(1, 1, "DE", row("2026-07-25", "B001")))); + + List rows = finalRows(); + assertEquals(1, rows.size()); + assertEquals("B001", rows.get(0).getAsin()); + assertEquals("2026-07-25", rows.get(0).getDate()); + } + + @Test + void test_task_030_task_boundary_limit_and_overflow() { + // 上限/超限：大量行合并只保留唯一行（每行唯一键不同）， + // 去重为一次遍历，行数正确、顺序稳定，无重复。 + givenRunningTask(1306L, 2306L); + List rows = new ArrayList<>(); + for (int i = 0; i < 200; i++) { + rows.add(row(String.format("2026-07-%02d", i % 28 + 1), "B" + String.format("%04d", i))); + } + rows.add(rows.get(0)); + rows.add(rows.get(42)); + rows.add(rows.get(199)); + + service.submitResult(task.getId(), request(chunk(1, 1, "UK", rows))); + List finalRows = finalRows(); + assertEquals(200, finalRows.size(), "200 唯一行 + 3 个重复行只保留 200 行"); + for (int i = 0; i < finalRows.size(); i++) { + assertEquals("B" + String.format("%04d", i), finalRows.get(i).getAsin(), "首现顺序稳定"); + } + } + + @Test + void test_task_030_task_invalid_input_rejected() { + // 非法参数：分片元数据非法在去重之前拒绝；含空国家/空行的分片不产生任何结果行。 + givenRunningTask(1307L, 2307L); + + BusinessException zeroIndex = assertThrows(BusinessException.class, () -> + service.submitResult(task.getId(), request(chunk(0, 1, "DE", row("2026-07-25", "B001"))))); + assertTrue(zeroIndex.getMessage().contains("chunk_index")); + + BusinessException zeroTotal = assertThrows(BusinessException.class, () -> + service.submitResult(task.getId(), request(chunk(1, 0, "DE", row("2026-07-25", "B001"))))); + assertTrue(zeroTotal.getMessage().contains("chunk_total")); + + BusinessException empty = assertThrows(BusinessException.class, () -> + service.submitResult(task.getId(), request(blankRowsChunk(1, 1)))); + assertTrue(empty.getMessage().contains("内容为空")); + + assertEquals(0, storedChunks.size(), "非法输入不落库"); + assertEquals(0, storedScopes.size()); + assertEquals(0, rustfsPayloads.size()); + } + + @Test + void test_task_030_task_dependency_failure_releases_resources() { + // 依赖失败：chunk 插入异常 → 无结果行、payload 清理，恢复后重试成功； + // scope 更新失败 → 回滚分片行与 payload，恢复后重试成功，去重结果不变。 + givenRunningTask(1308L, 2308L); + + insertFails = true; + assertThrows(RuntimeException.class, () -> + service.submitResult(task.getId(), request(chunk(1, 2, "DE", + row("2026-07-25", "B001"))))); + assertEquals(0, storedChunks.size()); + assertEquals(0, rustfsPayloads.size()); + assertEquals(0, storedScopes.size()); + insertFails = false; + + service.submitResult(task.getId(), request(chunk(1, 2, "DE", + row("2026-07-25", "B001"), + row("2026-07-26", "B002")))); + + scopeUpdateFails = true; + assertThrows(RuntimeException.class, () -> + service.submitResult(task.getId(), request(chunk(2, 2, "DE", + row("2026-07-26", "B002"), + row("2026-07-27", "B003"))))); + assertEquals(1, storedChunks.size(), "状态写入失败回滚分片行"); + assertEquals(1, rustfsPayloads.size(), "状态写入失败清理 payload"); + assertEquals(1, storedScopes.get(0).getReceivedChunkCount(), "状态写入失败计数器未推进"); + scopeUpdateFails = false; + + service.submitResult(task.getId(), request(chunk(2, 2, "DE", + row("2026-07-26", "B002"), + row("2026-07-27", "B003")))); + + assertEquals(List.of("B001", "B002", "B003"), asins(finalRows()), "恢复后去重结果不变"); + assertEquals(1, storedScopes.size(), "scope 行始终只有一行"); + assertEquals(2, storedScopes.get(0).getReceivedChunkCount()); + assertEquals(1, storedScopes.get(0).getCompleted()); + } + + private List finalRows() { + return finalCountries().get(0).getItems(); + } + + private List finalCountries() { + return parseResultJson().get(0).getCountryResults(); + } + + private List parseResultJson() { + try { + return objectMapper.readValue(task.getResultJson(), + objectMapper.getTypeFactory().constructCollectionType(List.class, ShopDataCrawlResultItemVo.class)); + } catch (Exception ex) { + throw new IllegalStateException("解析最终结果 JSON 失败", ex); + } + } + + private List asins(List rows) { + List result = new ArrayList<>(); + if (rows == null) { + return result; + } + for (ShopDataCrawlRowDto row : rows) { + result.add(row.getAsin()); + } + return result; + } + + private void configureTransientPayloadStorage() { + lenient().when(transientPayloadStorageService.isSharedWriteEnabled()).thenReturn(true); + lenient().when(transientPayloadStorageService.storeChunkPayloadVersioned( + anyString(), anyLong(), anyString(), any(), anyString())).thenAnswer(invocation -> { + String pointer = "rustfs:payload-" + (++nextPayloadId); + rustfsPayloads.put(pointer, invocation.getArgument(4)); + return pointer; + }); + lenient().when(transientPayloadStorageService.extractPointer(any())).thenAnswer(invocation -> { + String value = invocation.getArgument(0); + if (value == null) { + return null; + } + return value.startsWith("rustfs:") || value.startsWith("local:") || value.startsWith("oss:") + ? value : null; + }); + lenient().when(transientPayloadStorageService.resolvePayload(any(), any())).thenAnswer(invocation -> { + String pointer = invocation.getArgument(0); + String payload = rustfsPayloads.get(pointer); + if (payload == null) { + throw new IllegalStateException("missing test RustFS payload: " + pointer); + } + return payload; + }); + lenient().doAnswer(invocation -> { + rustfsPayloads.remove(invocation.getArgument(0)); + return null; + }).when(transientPayloadStorageService).deletePayloadIfPresent(any()); + } + + private void configureChunkMapper() { + lenient().when(taskChunkMapper.insert(any(TaskChunkEntity.class))).thenAnswer(invocation -> { + if (insertFails) { + throw new RuntimeException("db down"); + } + TaskChunkEntity chunk = invocation.getArgument(0); + boolean duplicate = storedChunks.stream().anyMatch(existing -> + Objects.equals(existing.getTaskId(), chunk.getTaskId()) + && Objects.equals(existing.getScopeHash(), chunk.getScopeHash()) + && Objects.equals(existing.getChunkIndex(), chunk.getChunkIndex())); + if (duplicate) { + throw new DuplicateKeyException("duplicate chunk key: " + chunk.getScopeHash() + "/" + chunk.getChunkIndex()); + } + chunk.setId((long) storedChunks.size() + 1L); + storedChunks.add(chunk); + return 1; + }); + lenient().when(taskChunkMapper.selectOne(any())).thenAnswer(invocation -> { + LambdaQueryWrapper query = invocation.getArgument(0); + Long taskId = queryLong(query); + String scopeHash = queryScopeHash(query); + Integer chunkIndex = queryInteger(query); + return storedChunks.stream() + .filter(chunk -> matchesChunk(chunk, taskId, scopeHash, chunkIndex)) + .findFirst() + .orElse(null); + }); + lenient().when(taskChunkMapper.selectList(any())).thenAnswer(invocation -> { + LambdaQueryWrapper query = invocation.getArgument(0); + Long taskId = queryLong(query); + String scopeHash = queryScopeHash(query); + return storedChunks.stream() + .filter(chunk -> matchesChunk(chunk, taskId, scopeHash, null)) + .sorted(Comparator.comparing(TaskChunkEntity::getChunkIndex)) + .toList(); + }); + lenient().when(taskChunkMapper.delete(any())).thenAnswer(invocation -> { + LambdaQueryWrapper query = invocation.getArgument(0); + Long taskId = queryLong(query); + String scopeHash = queryScopeHash(query); + Integer chunkIndex = queryInteger(query); + int before = storedChunks.size(); + storedChunks.removeIf(chunk -> matchesChunk(chunk, taskId, scopeHash, chunkIndex)); + return before - storedChunks.size(); + }); + } + + private void configureScopeMapper() { + lenient().when(taskScopeStateMapper.insert(any(TaskScopeStateEntity.class))).thenAnswer(invocation -> { + TaskScopeStateEntity scope = invocation.getArgument(0); + scope.setId((long) storedScopes.size() + 1L); + storedScopes.add(scope); + return 1; + }); + lenient().when(taskScopeStateMapper.selectOne(any())).thenAnswer(invocation -> { + LambdaQueryWrapper query = invocation.getArgument(0); + Long taskId = queryLong(query); + String scopeHash = queryScopeHash(query); + return storedScopes.stream() + .filter(scope -> Objects.equals(taskId, scope.getTaskId())) + .filter(scope -> scopeHash == null || Objects.equals(scopeHash, scope.getScopeHash())) + .findFirst() + .map(ShopDataCrawlRowDedupKeyTest::copyScope) + .orElse(null); + }); + lenient().when(taskScopeStateMapper.updateById(any(TaskScopeStateEntity.class))).thenAnswer(invocation -> { + if (scopeUpdateFails) { + throw new RuntimeException("scope update down"); + } + TaskScopeStateEntity updated = invocation.getArgument(0); + for (int i = 0; i < storedScopes.size(); i++) { + if (Objects.equals(storedScopes.get(i).getId(), updated.getId())) { + storedScopes.set(i, copyScope(updated)); + } + } + return 1; + }); + } + + private static TaskScopeStateEntity copyScope(TaskScopeStateEntity source) { + TaskScopeStateEntity copy = new TaskScopeStateEntity(); + copy.setId(source.getId()); + copy.setTaskId(source.getTaskId()); + copy.setModuleType(source.getModuleType()); + copy.setScopeKey(source.getScopeKey()); + copy.setScopeHash(source.getScopeHash()); + copy.setChunkTotal(source.getChunkTotal()); + copy.setReceivedChunkCount(source.getReceivedChunkCount()); + copy.setCompleted(source.getCompleted()); + copy.setLastChunkAt(source.getLastChunkAt()); + copy.setLastError(source.getLastError()); + copy.setCreatedAt(source.getCreatedAt()); + copy.setUpdatedAt(source.getUpdatedAt()); + return copy; + } + + private boolean matchesChunk(TaskChunkEntity chunk, Long taskId, String scopeHash, Integer chunkIndex) { + return (taskId == null || Objects.equals(taskId, chunk.getTaskId())) + && (scopeHash == null || Objects.equals(scopeHash, chunk.getScopeHash())) + && (chunkIndex == null || Objects.equals(chunkIndex, chunk.getChunkIndex())); + } + + private Long queryLong(LambdaQueryWrapper query) { + query.getSqlSegment(); + return query.getParamNameValuePairs().values().stream() + .filter(Long.class::isInstance) + .map(Long.class::cast) + .findFirst() + .orElse(null); + } + + private Integer queryInteger(LambdaQueryWrapper query) { + query.getSqlSegment(); + return query.getParamNameValuePairs().values().stream() + .filter(Integer.class::isInstance) + .map(Integer.class::cast) + .findFirst() + .orElse(null); + } + + private String queryScopeHash(LambdaQueryWrapper query) { + query.getSqlSegment(); + return query.getParamNameValuePairs().values().stream() + .filter(String.class::isInstance) + .map(String.class::cast) + .filter(value -> value.length() == 64) + .findFirst() + .orElse(null); + } + + private void givenRunningTask(long taskId, long resultId) { + task = new FileTaskEntity(); + task.setId(taskId); + task.setUserId(7L); + task.setModuleType(MODULE_TYPE); + task.setStatus("RUNNING"); + task.setRequestJson("{\"ownerInstanceId\":\"instance-a\"}"); + task.setResultJson("[]"); + task.setCreatedAt(LocalDateTime.now()); + task.setUpdatedAt(LocalDateTime.now()); + + result = new FileResultEntity(); + result.setId(resultId); + result.setTaskId(taskId); + result.setUserId(7L); + result.setModuleType(MODULE_TYPE); + result.setSourceFilename(SHOP_NAME); + result.setSourceFileUrl("shop-1"); + result.setSuccess(-1); + result.setCreatedAt(LocalDateTime.now()); + } + + private ShopDataCrawlSubmitResultRequest request(ShopDataCrawlShopPayloadDto payload) { + ShopDataCrawlSubmitResultRequest request = new ShopDataCrawlSubmitResultRequest(); + request.setShops(List.of(payload)); + return request; + } + + private ShopDataCrawlShopPayloadDto chunk(int chunkIndex, + int chunkTotal, + String country, + List rows) { + ShopDataCrawlShopPayloadDto payload = new ShopDataCrawlShopPayloadDto(); + payload.setShopName(SHOP_NAME); + payload.setChunkIndex(chunkIndex); + payload.setChunkTotal(chunkTotal); + payload.setCountryResults(List.of(countryWithItems(country, rows))); + return payload; + } + + private ShopDataCrawlShopPayloadDto chunk(int chunkIndex, + int chunkTotal, + String country, + ShopDataCrawlRowDto... rows) { + return chunk(chunkIndex, chunkTotal, country, List.of(rows)); + } + + private ShopDataCrawlShopPayloadDto blankRowsChunk(int chunkIndex, int chunkTotal) { + ShopDataCrawlShopPayloadDto payload = new ShopDataCrawlShopPayloadDto(); + payload.setShopName(SHOP_NAME); + payload.setChunkIndex(chunkIndex); + payload.setChunkTotal(chunkTotal); + ShopDataCrawlCountryResultDto country = new ShopDataCrawlCountryResultDto(); + country.setCountry("DE"); + ShopDataCrawlRowDto blank = new ShopDataCrawlRowDto(); + blank.setDate(""); + blank.setAsin(" "); + country.setItems(List.of(blank)); + payload.setCountryResults(List.of(country)); + return payload; + } + + private ShopDataCrawlCountryResultDto countryWithItems(String country, List items) { + ShopDataCrawlCountryResultDto result = new ShopDataCrawlCountryResultDto(); + result.setCountry(country); + result.setItems(items); + return result; + } + + private ShopDataCrawlRowDto row(String date, String asin) { + ShopDataCrawlRowDto row = new ShopDataCrawlRowDto(); + row.setDate(date); + row.setAsin(asin); + row.setCommodityImage("https://m.media-amazon.com/images/I/" + asin + ".jpg"); + row.setInventorySales("10"); + row.setSalesRank("20"); + row.setPageViews("30"); + row.setUnitsSold("40"); + row.setPrice("50"); + row.setRecommendedOffer("60"); + return row; + } + + private ShopDataCrawlRowDto sameRow(String date, String asin) { + return row(date, asin); + } + + private ShopDataCrawlRowDto paddedRow(String date, String asin) { + ShopDataCrawlRowDto row = new ShopDataCrawlRowDto(); + row.setDate(" " + date + " "); + row.setAsin(" " + asin + " "); + row.setCommodityImage(" https://m.media-amazon.com/images/I/" + asin + ".jpg "); + row.setInventorySales(" 10 "); + row.setSalesRank(" 20 "); + row.setPageViews(" 30 "); + row.setUnitsSold(" 40 "); + row.setPrice(" 50 "); + row.setRecommendedOffer(" 60 "); + return row; + } +}