diff --git a/backend-java/src/main/java/com/nanri/aiimage/modules/shopdatacrawl/model/entity/ShopDataCrawlDailyMemberEntity.java b/backend-java/src/main/java/com/nanri/aiimage/modules/shopdatacrawl/model/entity/ShopDataCrawlDailyMemberEntity.java index 783975c3..ca155932 100644 --- a/backend-java/src/main/java/com/nanri/aiimage/modules/shopdatacrawl/model/entity/ShopDataCrawlDailyMemberEntity.java +++ b/backend-java/src/main/java/com/nanri/aiimage/modules/shopdatacrawl/model/entity/ShopDataCrawlDailyMemberEntity.java @@ -16,5 +16,7 @@ public class ShopDataCrawlDailyMemberEntity { private Long dailyFileId; private Long taskId; private Long resultId; + /** 结果快照 JSON（数据层增量模型：整表重建时按成员行累积，不再读回旧累计对象）。 */ + private String rowPayload; private LocalDateTime createdAt; } diff --git a/backend-java/src/main/java/com/nanri/aiimage/modules/shopdatacrawl/service/ShopDataCrawlDailyFileService.java b/backend-java/src/main/java/com/nanri/aiimage/modules/shopdatacrawl/service/ShopDataCrawlDailyFileService.java index 75518836..57f51013 100644 --- a/backend-java/src/main/java/com/nanri/aiimage/modules/shopdatacrawl/service/ShopDataCrawlDailyFileService.java +++ b/backend-java/src/main/java/com/nanri/aiimage/modules/shopdatacrawl/service/ShopDataCrawlDailyFileService.java @@ -138,7 +138,7 @@ public class ShopDataCrawlDailyFileService { .eq(ShopDataCrawlDailyMemberEntity::getResultId, resultId)) > 0; } - public boolean addMember(Long dailyFileId, Long taskId, Long resultId) { + public boolean addMemberWithPayload(Long dailyFileId, Long taskId, Long resultId, String rowPayload) { if (dailyFileId == null || dailyFileId <= 0 || taskId == null || taskId <= 0 || resultId == null || resultId <= 0) { return false; @@ -147,6 +147,7 @@ public class ShopDataCrawlDailyFileService { member.setDailyFileId(dailyFileId); member.setTaskId(taskId); member.setResultId(resultId); + member.setRowPayload(rowPayload); member.setCreatedAt(currentBusinessDateTime()); try { dailyMemberMapper.insert(member); @@ -156,6 +157,10 @@ public class ShopDataCrawlDailyFileService { } } + public boolean addMember(Long dailyFileId, Long taskId, Long resultId) { + return addMemberWithPayload(dailyFileId, taskId, resultId, null); + } + public List listMembers(Long dailyFileId) { if (dailyFileId == null || dailyFileId <= 0) { return List.of(); @@ -183,6 +188,17 @@ public class ShopDataCrawlDailyFileService { .in(ShopDataCrawlDailyMemberEntity::getResultId, resultIds)); } + /** 累计文件跨天滚动时把旧文件成员行（含 row_payload）迁移到新文件，保留跨天携带的行。 */ + public void reassignMembers(Long fromDailyFileId, Long toDailyFileId) { + if (fromDailyFileId == null || fromDailyFileId <= 0 || toDailyFileId == null || toDailyFileId <= 0) { + return; + } + ShopDataCrawlDailyMemberEntity update = new ShopDataCrawlDailyMemberEntity(); + update.setDailyFileId(toDailyFileId); + dailyMemberMapper.update(update, new LambdaQueryWrapper() + .eq(ShopDataCrawlDailyMemberEntity::getDailyFileId, fromDailyFileId)); + } + public long countObjectReferences(String objectKey) { if (objectKey == null || objectKey.isBlank()) { return 0L; 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 82f8b4c5..7497d3e7 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 @@ -1762,7 +1762,7 @@ public class ShopDataCrawlTaskService { try { persistedResult = executeShortTransaction(() -> persistDailyAggregation( task, row, userId, shopKey, shopKeyHash, businessDate, - preparation, artifact)); + preparation, artifact, snapshot)); if (persistedResult.discardUploadedObject() && artifact.uploaded()) { deleteObjectQuietly(artifact.objectKey()); } @@ -1832,35 +1832,103 @@ public class ShopDataCrawlTaskService { "shop-data-crawl-result", String.valueOf(task.getId()), "daily-" + UUID.randomUUID())); - File baseXlsx = FileUtil.file(workRoot, "base.xlsx"); File outputXlsx = FileUtil.file(workRoot, filename); try { - if (baseDailyFile != null && !blank(existingObjectKey)) { - try { - Files.write(baseXlsx.toPath(), ossStorageService.readObjectBytes(existingObjectKey)); - } catch (Exception ex) { - throw new BusinessException("读取累计文件失败: " + safeMessage(ex)); - } - int rowCount = excelAssemblyService.replaceCountriesWorkbook(baseXlsx, outputXlsx, List.of(snapshot)); - String objectKey = ossStorageService.uploadResultFile(outputXlsx, MODULE_TYPE); - if (blank(objectKey)) { - throw new BusinessException("累计文件上传后未返回文件地址"); - } - return new DailyWorkbookArtifact(objectKey, outputXlsx.length(), true, filename, rowCount); - } - excelAssemblyService.writeWorkbook(outputXlsx, List.of(snapshot)); + // Task 35：数据层增量模型。整表从成员行（row_payload）累积重建， + // 不再读回旧累计对象（readObjectBytes）并整表重写（replaceCountriesWorkbook）； + // 历史成员行无 payload 时按结果快照兜底，兼容旧归档数据。 + List accumulatedItems = buildDailyFileFromData(baseDailyFile, List.of(snapshot)); + int rowCount = excelAssemblyService.writeWorkbook(outputXlsx, accumulatedItems); String objectKey = ossStorageService.uploadResultFile(outputXlsx, MODULE_TYPE); if (blank(objectKey)) { throw new BusinessException("累计文件上传后未返回文件地址"); } - return new DailyWorkbookArtifact(objectKey, outputXlsx.length(), true, filename, addedRowCount); + return new DailyWorkbookArtifact(objectKey, outputXlsx.length(), true, filename, rowCount); } finally { - FileUtil.del(baseXlsx); FileUtil.del(outputXlsx); FileUtil.del(workRoot); } } + /** + * 从数据层累积重建每日累计文件的快照列表：既有成员行按 (createdAt, id) 升序读取， + * 优先用行级 payload；payload 缺失（历史数据）时按结果快照兜底；新结果追加在末尾。 + */ + private List buildDailyFileFromData(ShopDataCrawlDailyFileEntity baseDailyFile, + List appended) { + List accumulated = new ArrayList<>(); + if (baseDailyFile != null && baseDailyFile.getId() != null) { + for (ShopDataCrawlDailyMemberEntity member : sortedDailyMembers(dailyFileService.listMembers(baseDailyFile.getId()))) { + ShopDataCrawlResultItemVo item = snapshotFromPayload(member); + if (item == null) { + FileResultEntity result = fileResultMapper.selectById(member.getResultId()); + item = result == null ? null : loadSnapshotForDailyMember(result); + } + if (item == null) { + throw new BusinessException("无法读取累计文件中的结果数据，请重试文件任务"); + } + accumulated.add(item); + } + } + for (ShopDataCrawlResultItemVo item : appended == null ? List.of() : appended) { + if (item != null) { + accumulated.add(item); + } + } + return accumulated; + } + + private ShopDataCrawlResultItemVo snapshotFromPayload(ShopDataCrawlDailyMemberEntity member) { + if (member == null || blank(member.getRowPayload())) { + return null; + } + try { + return objectMapper.readValue(member.getRowPayload(), ShopDataCrawlResultItemVo.class); + } catch (Exception ex) { + log.warn("[shop-data-crawl] daily member payload 解析失败 member={} msg={}", + member.getId(), safeMessage(ex)); + return null; + } + } + + private String snapshotPayload(ShopDataCrawlResultItemVo snapshot) { + if (snapshot == null) { + return null; + } + try { + return objectMapper.writeValueAsString(snapshot); + } catch (Exception ex) { + log.warn("[shop-data-crawl] 累计文件成员 payload 序列化失败 result={} msg={}", + snapshot.getResultId(), safeMessage(ex)); + return null; + } + } + + private List sortedDailyMembers(List memberRows) { + List members = new ArrayList<>(memberRows == null ? List.of() : memberRows); + members.removeIf(Objects::isNull); + members.sort(Comparator + .comparing(ShopDataCrawlDailyMemberEntity::getCreatedAt, + Comparator.nullsLast(Comparator.naturalOrder())) + .thenComparing(ShopDataCrawlDailyMemberEntity::getId, + Comparator.nullsLast(Comparator.naturalOrder()))); + return members; + } + + /** 累计文件跨天滚动时，把旧文件成员行（含 row_payload）原样迁到新文件，保留跨天携带的行。 */ + private void reassignOlderMembers(ShopDataCrawlDailyFileEntity newDailyFile, + List olderFiles) { + if (newDailyFile == null || newDailyFile.getId() == null) { + return; + } + for (ShopDataCrawlDailyFileEntity older : olderFiles == null ? List.of() : olderFiles) { + if (older == null || older.getId() == null) { + continue; + } + dailyFileService.reassignMembers(older.getId(), newDailyFile.getId()); + } + } + private DailyAggregationResult persistDailyAggregation(FileTaskEntity task, FileResultEntity row, Long userId, @@ -1868,7 +1936,8 @@ public class ShopDataCrawlTaskService { String shopKeyHash, LocalDate businessDate, DailyAggregationPreparation preparation, - DailyWorkbookArtifact artifact) { + DailyWorkbookArtifact artifact, + ShopDataCrawlResultItemVo snapshot) { ShopDataCrawlDailyFileEntity dailyFile = dailyFileService.findForUpdate( userId, shopKeyHash, businessDate); if (handleExistingDailyMembership(row, dailyFile)) { @@ -1923,10 +1992,12 @@ public class ShopDataCrawlTaskService { } else { dailyFileService.update(dailyFile); } - if (!dailyFileService.addMember(dailyFile.getId(), row.getTaskId(), row.getId())) { + if (!dailyFileService.addMemberWithPayload(dailyFile.getId(), row.getTaskId(), row.getId(), + snapshotPayload(snapshot))) { throw new BusinessException("结果已归档，请重试文件任务"); } + reassignOlderMembers(dailyFile, olderFiles); for (ShopDataCrawlDailyFileEntity older : olderFiles) { dailyFileService.deleteDailyFile(older.getId()); } diff --git a/backend-java/src/main/resources/db/V93__shop_data_crawl_daily_member_row_payload.sql b/backend-java/src/main/resources/db/V93__shop_data_crawl_daily_member_row_payload.sql new file mode 100644 index 00000000..281f13f8 --- /dev/null +++ b/backend-java/src/main/resources/db/V93__shop_data_crawl_daily_member_row_payload.sql @@ -0,0 +1,3 @@ +ALTER TABLE `biz_shop_data_crawl_daily_member` + ADD COLUMN `row_payload` MEDIUMTEXT NULL DEFAULT NULL COMMENT '结果快照 JSON（数据层增量模型：整表重建按成员行累积，旧数据为 NULL 时按结果快照兜底）' AFTER `result_id`; + diff --git a/backend-java/src/test/java/com/nanri/aiimage/modules/shopdatacrawl/service/ShopDataCrawlDailyFileIncrementalTest.java b/backend-java/src/test/java/com/nanri/aiimage/modules/shopdatacrawl/service/ShopDataCrawlDailyFileIncrementalTest.java new file mode 100644 index 00000000..0ea85f9f --- /dev/null +++ b/backend-java/src/test/java/com/nanri/aiimage/modules/shopdatacrawl/service/ShopDataCrawlDailyFileIncrementalTest.java @@ -0,0 +1,538 @@ +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 java.util.concurrent.atomic.AtomicReference; + +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.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.verify; +import static org.mockito.Mockito.when; + +/** + * Task 35：将每日累计文件改为数据层增量模型，避免每次下载并重写完整 XLSX。 + * 新结果归档时把该结果的快照 JSON（row_payload，V93 迁移新增列）写入 daily_member 行； + * 重新生成整表时按成员行（createdAt,id 升序）从数据层累积重建快照列表， + * 只做一次 writeWorkbook + 上传，不再读回旧累计对象（readObjectBytes）并整表重写 + * （replaceCountriesWorkbook）。历史成员行无 payload 时按结果数据兜底重建，兼容旧数据。 + */ +@ExtendWith(MockitoExtension.class) +class ShopDataCrawlDailyFileIncrementalTest { + + private static final String MODULE_TYPE = "SHOP_DATA_CRAWL"; + private static final String SHOP_NAME = "shop-a"; + + @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); + } + + @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; + + /** 内存中的结果行表（listTaskRows 读取源 + updateById 回写目标）。 */ + private final List dbResultRows = new ArrayList<>(); + /** 内存中的 daily_file 表。 */ + private final List dbDailyFiles = new ArrayList<>(); + /** 内存中的 daily_member 表（row_payload 落在成员行上）。 */ + private final List dbMembers = new ArrayList<>(); + private final AtomicLong memberIdSeq = new AtomicLong(1000); + private long nextResultId = 7100; + private String lastUploadedObjectKey; + /** 最近一次整表组装时交给 writeWorkbook 的快照列表（验证数据层累积与顺序）。 */ + private List lastAssembledItems = List.of(); + /** 最近一次从 dailyFileService.acquireLock 获取的锁句柄（验证失败路径释放）。 */ + private final AtomicReference lastLock = new AtomicReference<>(); + private Long lastJobTaskId; + + @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(); + memberIdSeq.set(1000); + nextResultId = 7100; + lastUploadedObjectKey = null; + lastAssembledItems = List.of(); + lastLock.set(null); + lastJobTaskId = null; + + 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(taskFileJobService.findAssembleJob(anyLong(), eq(MODULE_TYPE), anyLong())).thenReturn(null); + 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())).thenReturn(null); + lenient().when(fileTaskMapper.updateById(any(FileTaskEntity.class))).thenReturn(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/daily/" + System.nanoTime() + ".xlsx"; + lastUploadedObjectKey = key; + return key; + }); + lenient().when(ossStorageService.readObjectBytes(anyString())).thenReturn(new byte[0]); + lenient().doNothing().when(ossStorageService).deleteObject(anyString()); + + // 每次整表组装都捕获交给 writeWorkbook 的快照列表（数据层累积内容与顺序）。 + lenient().doAnswer(invocation -> { + lastAssembledItems = new ArrayList<>(invocation.getArgument(1)); + return lastAssembledItems.size(); + }).when(excelAssemblyService).writeWorkbook(any(), any()); + lenient().when(excelAssemblyService.countRows(any())).thenAnswer(invocation -> { + List items = invocation.getArgument(0); + return items == null ? 0 : items.size(); + }); + + // 店铺级锁：每次返回独立 mock 句柄，供失败路径验证 close()。 + lenient().when(dailyFileService.acquireLock(anyLong(), anyString())).thenAnswer(invocation -> { + TaskDistributedLockService.LockHandle handle = mock(TaskDistributedLockService.LockHandle.class); + lastLock.set(handle); + return handle; + }); + lenient().when(dailyFileService.currentBusinessDate()).thenReturn(LocalDate.of(2026, 8, 29)); + lenient().when(dailyFileService.currentBusinessDateTime()).thenReturn(LocalDateTime.of(2026, 8, 29, 12, 0)); + 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.findForUpdate(anyLong(), anyString(), any())) + .thenAnswer(invocation -> 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).deleteMembersForResults(any()); + captureDailyFilePersistence(); + captureMemberInserts(); + captureResultUpdates(); + + // 结果行读取：listTaskRows 按 taskId+moduleType 过滤并升序；其他查询返回空。 + 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_035_daily_file_normal_default_path() { + // 正常路径：新结果归档时不再下载并重写旧累计对象，而是把结果快照 + // payload 写入成员行（数据层增量），新对象只由本次快照生成。 + FileResultEntity row = addResultRow(7101L, 1L, 1, SHOP_NAME, null); + processJob(1L, List.of(row), snapshot(7101L, SHOP_NAME, 3)); + + ShopDataCrawlDailyMemberEntity member = soleMember(); + assertNotNull(member, "累计文件归档后存在成员行"); + assertNotNull(member.getRowPayload(), "成员行写入行级 payload"); + assertTrue(member.getRowPayload().contains(SHOP_NAME), "payload 是结果快照的 JSON 序列化"); + assertTrue(member.getRowPayload().contains("\"resultId\":7101"), "payload 携带结果标识"); + assertNotNull(lastUploadedObjectKey, "增量路径上传了新对象"); + assertEquals(1, lastAssembledItems.size(), "新对象只由本次结果快照生成"); + verify(ossStorageService, never()).readObjectBytes(anyString()); + verify(excelAssemblyService, never()).replaceCountriesWorkbook(any(), any(), any()); + } + + @Test + void test_task_035_daily_file_normal_multiple_items() { + // 多结果批量场景：每个结果各自归档，成员行按创建顺序累积，payload 齐全且顺序稳定。 + processJob(1L, List.of(addResultRow(7102L, 1L, 1, SHOP_NAME, null)), snapshot(7102L, SHOP_NAME, 2)); + processJob(2L, List.of(addResultRow(7103L, 2L, 1, SHOP_NAME, null)), snapshot(7103L, SHOP_NAME, 5)); + + List members = dbMembers.stream() + .sorted(Comparator.comparing(ShopDataCrawlDailyMemberEntity::getResultId)) + .toList(); + assertEquals(2, members.size(), "两个结果各有一个成员行"); + assertEquals(7102L, members.get(0).getResultId()); + assertEquals(7103L, members.get(1).getResultId()); + assertTrue(members.get(0).getRowPayload().contains("\"resultId\":7102"), "首个结果 payload 齐全"); + assertTrue(members.get(1).getRowPayload().contains("\"resultId\":7103"), "后续结果 payload 齐全"); + + // 整表重建直接来自数据层：一次 writeWorkbook，绝不读回旧对象。 + triggerRebuild(); + assertEquals(3, lastAssembledItems.size(), "整表重建累积全部成员快照"); + List assembledResultIds = lastAssembledItems.stream() + .map(ShopDataCrawlResultItemVo::getResultId).toList(); + assertEquals(List.of(7102L, 7103L, 900L), assembledResultIds, "累积顺序稳定：按成员创建顺序 + 新结果"); + verify(ossStorageService, never()).readObjectBytes(anyString()); + verify(excelAssemblyService, never()).replaceCountriesWorkbook(any(), any(), any()); + } + + @Test + void test_task_035_daily_file_normal_repeated_operation_is_idempotent() { + // 幂等：同一结果重复归档不产生第二个成员行、不重复上传对象； + // 已归档结果再次提交走 alreadyArchived 快捷路径，不触碰数据层。 + FileResultEntity row = addResultRow(7104L, 1L, 1, SHOP_NAME, null); + processJob(1L, List.of(row), snapshot(7104L, SHOP_NAME, 2)); + int membersAfterFirst = dbMembers.size(); + String objectAfterFirst = lastUploadedObjectKey; + processJob(1L, List.of(row), snapshot(7104L, SHOP_NAME, 2)); + + assertEquals(1, membersAfterFirst, "首次归档只有一个成员行"); + assertEquals(1, dbMembers.size(), "重复归档不产生重复成员行"); + assertEquals(objectAfterFirst, lastUploadedObjectKey, "重复归档复用既有对象，不重复上传"); + verify(ossStorageService, never()).readObjectBytes(anyString()); + } + + @Test + void test_task_035_daily_file_boundary_empty_input() { + // 空输入：没有任何成功结果时安全跳过，不创建成员、不上传对象。 + FileResultEntity row = addResultRow(7105L, 1L, 0, SHOP_NAME, null); + processJob(1L, List.of(row), snapshot(7105L, SHOP_NAME, 0)); + + assertTrue(dbMembers.isEmpty(), "无成功结果不创建成员行"); + assertTrue(dbDailyFiles.isEmpty(), "无成功结果不创建累计文件"); + verify(ossStorageService, never()).uploadResultFile(any(), eq(MODULE_TYPE)); + } + + @Test + void test_task_035_daily_file_boundary_single_item() { + // 单元素：单结果归档不依赖批量路径，成员行与累计文件各一。 + FileResultEntity row = addResultRow(7106L, 1L, 1, SHOP_NAME, null); + processJob(1L, List.of(row), snapshot(7106L, SHOP_NAME, 1)); + + assertEquals(1, dbMembers.size(), "单结果一个成员行"); + assertEquals(1, dbDailyFiles.size(), "单结果一个累计文件"); + assertNotNull(dbDailyFiles.get(0).getResultFileUrl()); + assertEquals(SHOP_NAME, dbDailyFiles.get(0).getShopKey()); + assertEquals(1, lastAssembledItems.size(), "单结果组装一次"); + } + + @Test + void test_task_035_daily_file_boundary_limit_and_overflow() { + // 上限/超限：达到批量上限（dbSelectBatchSize=200）并超出 10 个后， + // 成员行全部保留且无重复，整表重建从数据层累积全部快照，不发生无界内存增长。 + lenient().when(taskPressureProperties.getDbSelectBatchSize()).thenReturn(200); + int limit = 210; + for (int i = 0; i < limit; i++) { + long resultId = nextResultId++; + long taskId = i + 1L; + FileResultEntity row = addResultRow(resultId, taskId, 1, SHOP_NAME, null); + processJob(taskId, List.of(row), snapshot(resultId, SHOP_NAME, 1)); + } + + assertEquals(limit, dbMembers.size(), "超过上限数量的成员行全部保留，无丢失"); + assertEquals(limit, dbMembers.stream() + .map(ShopDataCrawlDailyMemberEntity::getResultId).distinct().count(), "成员结果无重复"); + assertEquals(limit, lastAssembledItems.size(), "整表重建按数据层累积全部快照"); + assertEquals(limit, lastAssembledItems.stream() + .map(ShopDataCrawlResultItemVo::getResultId).distinct().count(), "组装快照无重复"); + verify(ossStorageService, never()).readObjectBytes(anyString()); + } + + @Test + void test_task_035_daily_file_invalid_input_rejected() { + // 非法参数：空 job 直接拒绝；结果全部失败时抛出可识别的 BusinessException。 + assertThrows(BusinessException.class, () -> service.processResultFileJob(null), + "空 job 抛出项目约定异常"); + + FileResultEntity row = addResultRow(7107L, 1L, 2, SHOP_NAME, null); + Exception ex = assertThrows(BusinessException.class, + () -> processJob(1L, List.of(row), null)); + assertTrue(ex.getMessage().contains("没有可生成的店铺数据抓取结果"), "无成功结果时错误消息可识别"); + assertTrue(dbMembers.isEmpty(), "失败路径不留下成员行"); + assertTrue(dbDailyFiles.isEmpty(), "失败路径不留下累计文件"); + } + + @Test + void test_task_035_daily_file_dependency_failure_releases_resources() { + // 依赖失败：Excel 生成失败时错误可恢复，店铺级锁已释放、 + // 临时文件已清理、不残留成员/累计文件、不上传对象。 + doThrow(new BusinessException("模板不可用")) + .when(excelAssemblyService).writeWorkbook(any(), any()); + FileResultEntity row = addResultRow(7108L, 1L, 1, SHOP_NAME, null); + assertThrows(BusinessException.class, + () -> processJob(1L, List.of(row), snapshot(7108L, SHOP_NAME, 1))); + + assertTrue(dbMembers.isEmpty(), "生成失败不残留成员行"); + assertTrue(dbDailyFiles.isEmpty(), "生成失败不残留累计文件"); + assertNotNull(lastLock.get(), "失败路径已获取店铺级锁"); + verify(lastLock.get()).close(); + assertTrue(lastUploadedObjectKey == null, "失败路径不上传对象"); + } + + // ---- helpers ---- + + 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, String shopName, int rows) { + ShopDataCrawlResultItemVo item = new ShopDataCrawlResultItemVo(); + item.setResultId(resultId); + item.setTaskId(1L); + item.setShopName(shopName); + item.setShopId("shop-id-" + resultId); + item.setSuccess(true); + item.setMatched(true); + item.setTaskStatus("SUCCESS"); + item.setCountryCodes(List.of("DE")); + return item; + } + + private void processJob(long jobTaskId, List rows, ShopDataCrawlResultItemVo snapshot) { + lastJobTaskId = jobTaskId; + TaskFileJobEntity job = new TaskFileJobEntity(); + job.setId(jobTaskId); + job.setTaskId(jobTaskId); + job.setModuleType(MODULE_TYPE); + FileTaskEntity task = taskEntity(jobTaskId); + lenient().when(fileTaskMapper.selectById(jobTaskId)).thenReturn(task); + lenient().when(taskResultItemService.listResultSnapshots(eq(jobTaskId), eq(MODULE_TYPE), any())) + .thenReturn(snapshot == null ? List.of() : List.of(snapshot)); + service.processResultFileJob(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; + } + + /** 模拟 fileResultMapper.updateById 对内存结果行的回写。 */ + 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; + }); + } + + /** 模拟 dailyMemberMapper 插入（含 row_payload）：写入内存成员表并回填 id，重复唯一键返回 false。 */ + 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(LocalDateTime.of(2026, 8, 29, 12, 0).plusMinutes(dbMembers.size())); + dbMembers.add(member); + return true; + }); + } + + /** 模拟 daily_file 与 daily_member 读取：findForUpdate 命中内存表，listMembers 升序返回。 */ + private void captureDailyFilePersistence() { + lenient().doAnswer(invocation -> { + ShopDataCrawlDailyFileEntity entity = invocation.getArgument(0); + entity.setId(5000L + dbDailyFiles.size() + 1); + 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, entity); + return null; + } + } + entity.setId(5000L + dbDailyFiles.size() + 1); + dbDailyFiles.add(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 ShopDataCrawlDailyMemberEntity soleMember() { + return dbMembers.size() == 1 ? dbMembers.get(0) : null; + } + + /** 对既有累计文件再归档一个结果（固定 resultId=900），触发一次从数据层累积的整表重建。 */ + private void triggerRebuild() { + FileResultEntity row = addResultRow(900L, 900L, 1, SHOP_NAME, null); + processJob(900L, List.of(row), snapshot(900L, SHOP_NAME, 1)); + } +} diff --git a/backend-java/src/test/java/com/nanri/aiimage/modules/shopdatacrawl/service/ShopDataCrawlTaskServiceRetentionTest.java b/backend-java/src/test/java/com/nanri/aiimage/modules/shopdatacrawl/service/ShopDataCrawlTaskServiceRetentionTest.java index 65e5dad5..bea9bd54 100644 --- a/backend-java/src/test/java/com/nanri/aiimage/modules/shopdatacrawl/service/ShopDataCrawlTaskServiceRetentionTest.java +++ b/backend-java/src/test/java/com/nanri/aiimage/modules/shopdatacrawl/service/ShopDataCrawlTaskServiceRetentionTest.java @@ -132,7 +132,7 @@ class ShopDataCrawlTaskServiceRetentionTest { when(dailyFileService.findMembersByResultId(RESULT_ID)).thenReturn(List.of()); when(dailyFileService.findOlder(USER_ID, "hash-1", BUSINESS_DATE)).thenReturn(List.of()); when(dailyFileService.countObjectReferences(anyString())).thenReturn(0L); - when(dailyFileService.addMember(anyLong(), anyLong(), anyLong())).thenReturn(true); + when(dailyFileService.addMemberWithPayload(anyLong(), anyLong(), anyLong(), anyString())).thenReturn(true); when(transactionManager.getTransaction(any())).thenReturn(transactionStatus); doAnswer(invocation -> { ShopDataCrawlDailyFileEntity entity = invocation.getArgument(0); @@ -146,12 +146,13 @@ class ShopDataCrawlTaskServiceRetentionTest { when(fileResultMapper.selectList(any())).thenReturn(List.of(currentRow), List.of(currentRow)); when(dailyFileService.findForUpdate(USER_ID, "hash-1", BUSINESS_DATE)).thenReturn(null); when(ossStorageService.uploadResultFile(any(), eq(MODULE_TYPE))).thenReturn("result/new.xlsx"); + when(excelAssemblyService.writeWorkbook(any(), any())).thenReturn(1); service.processResultFileJob(job); verify(excelAssemblyService).writeWorkbook(any(), eq(List.of(snapshot))); verify(excelAssemblyService, never()).replaceCountriesWorkbook(any(), any(), any()); - verify(dailyFileService).addMember(301L, TASK_ID, RESULT_ID); + verify(dailyFileService).addMemberWithPayload(eq(301L), eq(TASK_ID), eq(RESULT_ID), anyString()); assertEquals("result/new.xlsx", currentRow.getResultFileUrl()); assertEquals(1, currentRow.getRowCount()); @@ -169,22 +170,31 @@ class ShopDataCrawlTaskServiceRetentionTest { FileTaskEntity previousTask = task(); previousTask.setId(100L); ShopDataCrawlDailyFileEntity daily = daily("result/old.xlsx", 2); + ShopDataCrawlResultItemVo previousSnapshot = snapshot(200L, 100L); + ShopDataCrawlDailyMemberEntity previousMember = member(301L, 100L, 200L, BUSINESS_TIME.minusMinutes(10)); + previousMember.setRowPayload("{\"resultId\":200,\"taskId\":100,\"shopName\":\"Demo Shop\"," + + "\"shopId\":\"shop-1\",\"success\":true,\"countryResults\":[]}"); when(fileResultMapper.selectList(any())).thenReturn(List.of(currentRow), List.of(previous, currentRow)); when(fileTaskMapper.selectBatchIds(List.of(100L))).thenReturn(List.of(previousTask)); when(dailyFileService.findForUpdate(USER_ID, "hash-1", BUSINESS_DATE)).thenReturn(daily); when(dailyFileService.containsResult(301L, RESULT_ID)).thenReturn(false); - when(ossStorageService.readObjectBytes("result/old.xlsx")).thenReturn(new byte[]{1, 2, 3}); + when(dailyFileService.listMembers(301L)).thenReturn(List.of(previousMember)); + when(taskResultItemService.getResultSnapshot( + 100L, MODULE_TYPE, 200L, ShopDataCrawlResultItemVo.class)).thenReturn(previousSnapshot); when(ossStorageService.uploadResultFile(any(), eq(MODULE_TYPE))).thenReturn("result/new.xlsx"); - when(excelAssemblyService.replaceCountriesWorkbook(any(), any(), any())).thenReturn(3); + when(excelAssemblyService.writeWorkbook(any(), any())).thenReturn(2); service.processResultFileJob(job); - verify(excelAssemblyService).replaceCountriesWorkbook(any(), any(), eq(List.of(snapshot))); + // Task 35：整表从数据层成员行（row_payload + 本次快照）累积重建，不再读回旧对象。 + verify(excelAssemblyService).writeWorkbook(any(), eq(List.of(previousSnapshot, snapshot))); + verify(excelAssemblyService, never()).replaceCountriesWorkbook(any(), any(), any()); + verify(ossStorageService, never()).readObjectBytes(anyString()); verify(dailyFileService).update(daily); verify(ossStorageService).deleteObject("result/old.xlsx"); assertNull(previous.getResultFileUrl()); assertEquals("result/new.xlsx", currentRow.getResultFileUrl()); - assertEquals(3, currentRow.getRowCount()); + assertEquals(2, currentRow.getRowCount()); } @Test @@ -212,13 +222,13 @@ class ShopDataCrawlTaskServiceRetentionTest { return 1; }).when(excelAssemblyService).countRows(any()); doAnswer(invocation -> { - assertFalse(transactionActive.get(), "workbook download must run outside the database transaction"); - return new byte[]{1, 2, 3}; - }).when(ossStorageService).readObjectBytes("result/old.xlsx"); + assertFalse(transactionActive.get(), "daily member data load must run outside the database transaction"); + return List.of(); + }).when(dailyFileService).listMembers(anyLong()); doAnswer(invocation -> { assertFalse(transactionActive.get(), "workbook assembly must run outside the database transaction"); return 3; - }).when(excelAssemblyService).replaceCountriesWorkbook(any(), any(), any()); + }).when(excelAssemblyService).writeWorkbook(any(), any()); doAnswer(invocation -> { assertFalse(transactionActive.get(), "workbook upload must run outside the database transaction"); return "result/new.xlsx"; @@ -264,7 +274,7 @@ class ShopDataCrawlTaskServiceRetentionTest { verify(excelAssemblyService, never()).writeWorkbook(any(), any()); verify(excelAssemblyService, never()).replaceCountriesWorkbook(any(), any(), any()); verify(ossStorageService, never()).uploadResultFile(any(), anyString()); - verify(dailyFileService).addMember(301L, TASK_ID, RESULT_ID); + verify(dailyFileService).addMemberWithPayload(eq(301L), eq(TASK_ID), eq(RESULT_ID), anyString()); assertEquals("result/current.xlsx", currentRow.getResultFileUrl()); assertEquals(10L, currentRow.getResultFileSize()); assertEquals(3, currentRow.getRowCount()); @@ -283,7 +293,9 @@ class ShopDataCrawlTaskServiceRetentionTest { when(fileResultMapper.selectCount(any())).thenReturn(1L); when(dailyFileService.findForUpdate(USER_ID, "hash-1", BUSINESS_DATE)).thenReturn(daily); when(dailyFileService.containsResult(301L, RESULT_ID)).thenReturn(false); - when(ossStorageService.readObjectBytes("result/old.xlsx")).thenReturn(new byte[]{1, 2, 3}); + when(dailyFileService.listMembers(301L)).thenReturn(List.of()); + when(taskResultItemService.getResultSnapshot( + 100L, MODULE_TYPE, 200L, ShopDataCrawlResultItemVo.class)).thenReturn(snapshot(200L, 100L)); when(ossStorageService.uploadResultFile(any(), eq(MODULE_TYPE))).thenReturn("result/new.xlsx"); service.processResultFileJob(job); @@ -323,18 +335,21 @@ class ShopDataCrawlTaskServiceRetentionTest { when(fileResultMapper.selectList(any())).thenReturn(List.of(currentRow), List.of(previous, currentRow)); when(dailyFileService.findForUpdate(USER_ID, "hash-1", BUSINESS_DATE)).thenReturn(null); when(dailyFileService.findOlder(USER_ID, "hash-1", BUSINESS_DATE)).thenReturn(List.of(yesterday)); - when(ossStorageService.readObjectBytes("result/yesterday.xlsx")).thenReturn(new byte[]{1, 2, 3}); + when(dailyFileService.listMembers(300L)).thenReturn(List.of()); + when(excelAssemblyService.writeWorkbook(any(), any())).thenReturn(1); when(ossStorageService.uploadResultFile(any(), eq(MODULE_TYPE))).thenReturn("result/today.xlsx"); - when(excelAssemblyService.replaceCountriesWorkbook(any(), any(), any())).thenReturn(5); service.processResultFileJob(job); - verify(excelAssemblyService, never()).writeWorkbook(any(), any()); - verify(excelAssemblyService).replaceCountriesWorkbook(any(), any(), eq(List.of(snapshot))); + // Task 35：新一天跨天滚动改为数据层整表重建 + 旧成员迁移，不再读回旧对象重写。 + verify(excelAssemblyService).writeWorkbook(any(), eq(List.of(snapshot))); + verify(excelAssemblyService, never()).replaceCountriesWorkbook(any(), any(), any()); + verify(ossStorageService, never()).readObjectBytes(anyString()); verify(dailyFileService).deleteDailyFile(300L); + verify(dailyFileService).reassignMembers(300L, 301L); verify(ossStorageService).deleteObject("result/yesterday.xlsx"); assertEquals("result/today.xlsx", currentRow.getResultFileUrl()); - assertEquals(5, currentRow.getRowCount()); + assertEquals(1, currentRow.getRowCount()); assertNull(previous.getResultFileUrl()); } @@ -346,8 +361,8 @@ class ShopDataCrawlTaskServiceRetentionTest { when(fileResultMapper.selectList(any())).thenReturn(List.of(currentRow)); when(dailyFileService.findForUpdate(USER_ID, "hash-1", BUSINESS_DATE)).thenReturn(null); when(dailyFileService.findOlder(USER_ID, "hash-1", BUSINESS_DATE)).thenReturn(List.of(yesterday)); - when(ossStorageService.readObjectBytes("result/yesterday.xlsx")).thenReturn(new byte[]{1, 2, 3}); - when(excelAssemblyService.replaceCountriesWorkbook(any(), any(), any())).thenReturn(4); + when(dailyFileService.listMembers(300L)).thenReturn(List.of()); + when(excelAssemblyService.writeWorkbook(any(), any())).thenReturn(4); doThrow(new IllegalStateException("upload failed")) .when(ossStorageService).uploadResultFile(any(), eq(MODULE_TYPE)); @@ -365,8 +380,8 @@ class ShopDataCrawlTaskServiceRetentionTest { when(fileResultMapper.selectList(any())).thenReturn(List.of(currentRow)); when(dailyFileService.findForUpdate(USER_ID, "hash-1", BUSINESS_DATE)).thenReturn(null); when(dailyFileService.findOlder(USER_ID, "hash-1", BUSINESS_DATE)).thenReturn(List.of(yesterday)); - when(ossStorageService.readObjectBytes("result/yesterday.xlsx")).thenReturn(new byte[]{1, 2, 3}); - when(excelAssemblyService.replaceCountriesWorkbook(any(), any(), any())).thenReturn(4); + when(dailyFileService.listMembers(300L)).thenReturn(List.of()); + when(excelAssemblyService.writeWorkbook(any(), any())).thenReturn(4); when(ossStorageService.uploadResultFile(any(), eq(MODULE_TYPE))).thenReturn("result/new.xlsx"); AtomicInteger commitCount = new AtomicInteger(); doAnswer(invocation -> {