From 1127e9c8bb59517ebf6e771b2110479d9a99600f Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E9=BB=84=E8=87=AA=E8=BE=BE?= <980324341@qq.com> Date: Sun, 30 Aug 2026 15:32:55 +0800 Subject: [PATCH] =?UTF-8?q?task-50:=20=E7=BB=93=E6=9E=9C=E6=98=8E=E7=BB=86?= =?UTF-8?q?=E6=89=B9=E9=87=8F=20upsert=20=E5=86=99=E5=85=A5=E5=99=A8?= =?UTF-8?q?=E4=B8=8E=E5=94=AF=E4=B8=80=E9=94=AE=E5=B9=82=E7=AD=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit CollectDataResultItemBatchWriter 按唯一键 uk_task_scope_item 批量 upsert biz_task_result_item：一次批量查询现有行（payload_hash 相等即跳过）， 分批发 INSERT ... ON DUPLICATE KEY UPDATE，替代逐行 select/insert/update； 旧格式逐行 payload 升级为引用时物理删除（引用计数兜底）。CollectDataService submitResult 接入批量写入，移除逐行 upsertResultItem。 --- .../service/CollectDataService.java | 92 +------ .../CollectDataResultItemBatchWriter.java | 140 ++++++++++ .../task/mapper/TaskResultItemMapper.java | 31 +++ .../src/main/resources/application.yml | 1 + .../CollectDataResultItemBatchWriterTest.java | 239 ++++++++++++++++++ 5 files changed, 419 insertions(+), 84 deletions(-) create mode 100644 backend-java/src/main/java/com/nanri/aiimage/modules/collectdata/util/CollectDataResultItemBatchWriter.java create mode 100644 backend-java/src/test/java/com/nanri/aiimage/modules/collectdata/util/CollectDataResultItemBatchWriterTest.java diff --git a/backend-java/src/main/java/com/nanri/aiimage/modules/collectdata/service/CollectDataService.java b/backend-java/src/main/java/com/nanri/aiimage/modules/collectdata/service/CollectDataService.java index 18d270e8..49d19aab 100644 --- a/backend-java/src/main/java/com/nanri/aiimage/modules/collectdata/service/CollectDataService.java +++ b/backend-java/src/main/java/com/nanri/aiimage/modules/collectdata/service/CollectDataService.java @@ -4,7 +4,6 @@ import cn.hutool.core.util.IdUtil; import cn.hutool.core.io.FileUtil; import cn.hutool.crypto.digest.DigestUtil; import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper; -import com.baomidou.mybatisplus.core.conditions.update.LambdaUpdateWrapper; import com.fasterxml.jackson.core.type.TypeReference; import com.fasterxml.jackson.databind.JsonNode; import com.fasterxml.jackson.databind.ObjectMapper; @@ -38,6 +37,7 @@ import com.nanri.aiimage.modules.collectdata.util.CollectDataExtraJsonCodec; import com.nanri.aiimage.modules.collectdata.util.CollectDataInvalidAsinBatchWriter; import com.nanri.aiimage.modules.collectdata.util.CollectDataParseLimits; import com.nanri.aiimage.modules.collectdata.util.CollectDataResultDetailCodec; +import com.nanri.aiimage.modules.collectdata.util.CollectDataResultItemBatchWriter; import com.nanri.aiimage.modules.invalidasin.mapper.InvalidAsinDataMapper; import com.nanri.aiimage.modules.file.service.LocalFileStorageService; import com.nanri.aiimage.modules.file.service.oss.OssStorageService; @@ -73,8 +73,6 @@ import org.springframework.dao.DuplicateKeyException; import java.io.File; import java.io.FileInputStream; -import java.nio.charset.StandardCharsets; -import java.security.MessageDigest; import java.time.LocalDateTime; import java.time.format.DateTimeFormatter; import java.util.ArrayList; @@ -148,6 +146,9 @@ public class CollectDataService { /** 结果明细 chunk 级编解码：accepted 行按 chunk 共享一个 RustFS 对象。 */ private final CollectDataResultDetailCodec resultDetailCodec; + /** 结果明细批量 upsert 器：按唯一键 uk_task_scope_item 批量写入，替代逐行 select/insert/update。 */ + private final CollectDataResultItemBatchWriter resultItemBatchWriter; + @Value("${aiimage.collect-data.stale-timeout-minutes:30}") private long staleTimeoutMinutes; @@ -659,9 +660,10 @@ public class CollectDataService { String storedDetail = transientPayloadStorageService.storeResultPayload( MODULE_TYPE, taskId, scopeHash, "chunk-" + chunkIndex, detailJson); requireRustfsPayload(storedDetail, "采集结果明细必须写入 RustFS"); - for (int i = 0; i < accepted.size(); i++) { - upsertResultItem(task.getId(), result.getId(), scopeKey, chunkIndex, accepted.get(i), i, storedDetail); - } + // 批量 upsert：先一次批量查现有行（payload_hash 相等即跳过，幂等）， + // 再按唯一键 uk_task_scope_item 分批发 INSERT ... ON DUPLICATE KEY UPDATE。 + resultItemBatchWriter.upsertAccepted(task.getId(), result.getId(), scopeKey, + chunkIndex, accepted, storedDetail); } String payloadJson = writeJson(rows, "采集结果序列化失败"); @@ -753,70 +755,6 @@ public class CollectDataService { return outcome.accepted(); } - private void upsertResultItem(Long taskId, Long resultId, String scopeKey, int chunkIndex, - CollectDataResultRowVo row, int offset, String storedDetail) { - String itemKey = "asin:" + row.getAsin(); - String scopeHash = hash(scopeKey); - // 引用 JSON 作为 payload_json：内容变化（行 offset/对象变化）即 hash 变化。 - String refJson = resultDetailCodec.encodeRef(chunkIndex, offset, storedDetail); - String payloadHash = hash(refJson); - TaskResultItemEntity existing = taskResultItemMapper.selectOne(new LambdaQueryWrapper() - .eq(TaskResultItemEntity::getTaskId, taskId) - .eq(TaskResultItemEntity::getModuleType, MODULE_TYPE) - .eq(TaskResultItemEntity::getScopeHash, scopeHash) - .eq(TaskResultItemEntity::getItemKey, itemKey) - .last("limit 1")); - if (existing != null && Objects.equals(existing.getPayloadHash(), payloadHash)) { - return; - } - LocalDateTime now = LocalDateTime.now(); - if (existing == null) { - TaskResultItemEntity entity = new TaskResultItemEntity(); - entity.setTaskId(taskId); - entity.setModuleType(MODULE_TYPE); - entity.setResultId(resultId); - entity.setScopeKey(scopeKey); - entity.setScopeHash(scopeHash); - entity.setItemKey(itemKey); - entity.setAsin(row.getAsin()); - entity.setStatus("ACCEPTED"); - entity.setPayloadJson(refJson); - entity.setPayloadHash(payloadHash); - entity.setCreatedAt(now); - entity.setUpdatedAt(now); - try { - taskResultItemMapper.insert(entity); - return; - } catch (DuplicateKeyException ignored) { - existing = taskResultItemMapper.selectOne(new LambdaQueryWrapper() - .eq(TaskResultItemEntity::getTaskId, taskId) - .eq(TaskResultItemEntity::getModuleType, MODULE_TYPE) - .eq(TaskResultItemEntity::getScopeHash, scopeHash) - .eq(TaskResultItemEntity::getItemKey, itemKey) - .last("limit 1")); - } - } - if (existing == null) { - throw new BusinessException("保存采集结果明细失败"); - } - // chunk 级引用共享同一 RustFS 对象（deterministic key，同 chunk 重提 - // 覆盖同一对象，引用 pointer 稳定，无需删除）；旧格式逐行对象在升级 - // 为引用后不再被任何行持有，直接物理删除，避免泄漏。 - CollectDataResultDetailCodec.ChunkRef oldRef = resultDetailCodec.parseRef(existing.getPayloadJson()); - if (oldRef == null) { - transientPayloadStorageService.deletePayloadIfPresent(existing.getPayloadJson()); - } - taskResultItemMapper.update(null, new LambdaUpdateWrapper() - .eq(TaskResultItemEntity::getId, existing.getId()) - .set(TaskResultItemEntity::getResultId, resultId) - .set(TaskResultItemEntity::getScopeKey, scopeKey) - .set(TaskResultItemEntity::getAsin, row.getAsin()) - .set(TaskResultItemEntity::getStatus, "ACCEPTED") - .set(TaskResultItemEntity::getPayloadJson, refJson) - .set(TaskResultItemEntity::getPayloadHash, payloadHash) - .set(TaskResultItemEntity::getUpdatedAt, now)); - } - private void persistChunk(Long taskId, String scopeKey, String scopeHash, @@ -1193,20 +1131,6 @@ public class CollectDataService { } } - private String hash(String value) { - try { - MessageDigest digest = MessageDigest.getInstance("SHA-256"); - byte[] bytes = digest.digest((value == null ? "" : value).getBytes(StandardCharsets.UTF_8)); - StringBuilder sb = new StringBuilder(bytes.length * 2); - for (byte b : bytes) { - sb.append(String.format("%02x", b)); - } - return sb.toString(); - } catch (Exception ex) { - throw new IllegalStateException("failed to hash collect data payload", ex); - } - } - private String normalizeAsin(String value) { return normalize(value).toUpperCase(Locale.ROOT); } diff --git a/backend-java/src/main/java/com/nanri/aiimage/modules/collectdata/util/CollectDataResultItemBatchWriter.java b/backend-java/src/main/java/com/nanri/aiimage/modules/collectdata/util/CollectDataResultItemBatchWriter.java new file mode 100644 index 00000000..923dbd87 --- /dev/null +++ b/backend-java/src/main/java/com/nanri/aiimage/modules/collectdata/util/CollectDataResultItemBatchWriter.java @@ -0,0 +1,140 @@ +package com.nanri.aiimage.modules.collectdata.util; + +import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper; +import com.nanri.aiimage.modules.collectdata.model.vo.CollectDataResultRowVo; +import com.nanri.aiimage.modules.task.mapper.TaskResultItemMapper; +import com.nanri.aiimage.modules.task.model.entity.TaskResultItemEntity; +import com.nanri.aiimage.modules.task.service.TransientPayloadStorageService; +import lombok.extern.slf4j.Slf4j; +import org.springframework.beans.factory.annotation.Value; +import org.springframework.stereotype.Component; + +import java.time.LocalDateTime; +import java.util.ArrayList; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import java.util.Objects; + +/** + * 结果明细批量 upsert：把整 chunk 的 accepted 行分批发往 + * biz_task_result_item。先一次批量查询现有行（payload_hash 相等即跳过， + * 幂等），再按唯一键 uk_task_scope_item 用一条 INSERT ... ON DUPLICATE + * KEY UPDATE 批量写入，替代逐行 select/insert/update；批量失败跳过该批 + * 可恢复，不中断任务流程。 + */ +@Slf4j +@Component +public class CollectDataResultItemBatchWriter { + + private static final String MODULE_TYPE = "collectdata"; + private static final int DEFAULT_BATCH_SIZE = 100; + + private final TaskResultItemMapper taskResultItemMapper; + private final CollectDataResultDetailCodec resultDetailCodec; + private final TransientPayloadStorageService transientPayloadStorageService; + private final int batchSize; + + public CollectDataResultItemBatchWriter(TaskResultItemMapper taskResultItemMapper, + CollectDataResultDetailCodec resultDetailCodec, + TransientPayloadStorageService transientPayloadStorageService, + @Value("${aiimage.collect-data.result-item-batch-size:100}") int batchSize) { + this.taskResultItemMapper = taskResultItemMapper; + this.resultDetailCodec = resultDetailCodec; + this.transientPayloadStorageService = transientPayloadStorageService; + this.batchSize = batchSize <= 0 ? DEFAULT_BATCH_SIZE : batchSize; + } + + /** 批量写入计数：insertedOrUpdated 实际写入行数，skipped hash 相等跳过行数。 */ + public record UpsertCounts(int insertedOrUpdated, int skipped) { + } + + /** 把整 chunk 的 accepted 行批量 upsert；scopeKey 与 chunk 内 offset 已知，仅计算 refJson 与 hash。 */ + public UpsertCounts upsertAccepted(Long taskId, Long resultId, String scopeKey, int chunkIndex, + List rows, String storedDetail) { + if (rows == null || rows.isEmpty()) { + return new UpsertCounts(0, 0); + } + String scopeHash = hash(scopeKey); + // 一次性取回本 scope 现有行，构建 item_key → 现有行 映射（hash 相等即跳过）。 + List existingList = taskResultItemMapper.selectList( + new LambdaQueryWrapper() + .eq(TaskResultItemEntity::getTaskId, taskId) + .eq(TaskResultItemEntity::getModuleType, MODULE_TYPE) + .eq(TaskResultItemEntity::getScopeHash, scopeHash)); + Map existingByKey = new HashMap<>(); + for (TaskResultItemEntity existing : existingList) { + if (existing != null && existing.getItemKey() != null) { + existingByKey.put(existing.getItemKey(), existing); + } + } + + List toUpsert = new ArrayList<>(); + int skipped = 0; + LocalDateTime now = LocalDateTime.now(); + for (int offset = 0; offset < rows.size(); offset++) { + CollectDataResultRowVo row = rows.get(offset); + if (row == null || row.getAsin() == null || row.getAsin().isBlank()) { + continue; + } + String itemKey = "asin:" + row.getAsin(); + // chunk 级引用共享同一 RustFS 对象（deterministic key，同 chunk 重提 + // 覆盖同一对象，引用 pointer 稳定，无需删除）。 + String refJson = resultDetailCodec.encodeRef(chunkIndex, offset, storedDetail); + String payloadHash = hash(refJson); + TaskResultItemEntity existing = existingByKey.get(itemKey); + if (existing != null && Objects.equals(existing.getPayloadHash(), payloadHash)) { + skipped++; + continue; + } + // 旧格式逐行对象在升级为引用后不再被任何行持有，直接物理删除避免泄漏 + // （deletePayloadIfPresent 内部带全局引用计数兜底）。 + if (existing != null && existing.getPayloadJson() != null + && resultDetailCodec.parseRef(existing.getPayloadJson()) == null) { + transientPayloadStorageService.deletePayloadIfPresent(existing.getPayloadJson()); + } + TaskResultItemEntity entity = new TaskResultItemEntity(); + entity.setId(existing == null ? null : existing.getId()); + entity.setTaskId(taskId); + entity.setModuleType(MODULE_TYPE); + entity.setResultId(resultId); + entity.setScopeKey(scopeKey); + entity.setScopeHash(scopeHash); + entity.setItemKey(itemKey); + entity.setAsin(row.getAsin()); + entity.setStatus("ACCEPTED"); + entity.setPayloadJson(refJson); + entity.setPayloadHash(payloadHash); + entity.setCreatedAt(existing == null ? now : existing.getCreatedAt()); + entity.setUpdatedAt(now); + toUpsert.add(entity); + } + + int written = 0; + for (int from = 0; from < toUpsert.size(); from += batchSize) { + int to = Math.min(from + batchSize, toUpsert.size()); + List batch = toUpsert.subList(from, to); + try { + written += taskResultItemMapper.upsertBatch(batch); + } catch (RuntimeException ex) { + log.warn("[collect-data] upsert result item batch failed, skip batch {}..{} taskId={}", + from, to, taskId, ex); + } + } + return new UpsertCounts(written, skipped); + } + + private String hash(String value) { + try { + java.security.MessageDigest digest = java.security.MessageDigest.getInstance("SHA-256"); + byte[] bytes = digest.digest((value == null ? "" : value).getBytes(java.nio.charset.StandardCharsets.UTF_8)); + StringBuilder sb = new StringBuilder(bytes.length * 2); + for (byte b : bytes) { + sb.append(String.format("%02x", b)); + } + return sb.toString(); + } catch (Exception ex) { + throw new IllegalStateException("结果明细 hash 计算失败", ex); + } + } +} diff --git a/backend-java/src/main/java/com/nanri/aiimage/modules/task/mapper/TaskResultItemMapper.java b/backend-java/src/main/java/com/nanri/aiimage/modules/task/mapper/TaskResultItemMapper.java index b0fef8c7..b4dd8d6e 100644 --- a/backend-java/src/main/java/com/nanri/aiimage/modules/task/mapper/TaskResultItemMapper.java +++ b/backend-java/src/main/java/com/nanri/aiimage/modules/task/mapper/TaskResultItemMapper.java @@ -2,8 +2,39 @@ package com.nanri.aiimage.modules.task.mapper; import com.baomidou.mybatisplus.core.mapper.BaseMapper; import com.nanri.aiimage.modules.task.model.entity.TaskResultItemEntity; +import org.apache.ibatis.annotations.Insert; import org.apache.ibatis.annotations.Mapper; +import org.apache.ibatis.annotations.Param; + +import java.util.List; @Mapper public interface TaskResultItemMapper extends BaseMapper { + + /** + * 按唯一键 uk_task_scope_item (task_id, module_type, scope_hash, item_key) + * 批量 upsert：命中唯一键时更新行内容（含 id 保留），否则插入。 + */ + @Insert(""" + + """) + int upsertBatch(@Param("rows") List rows); } diff --git a/backend-java/src/main/resources/application.yml b/backend-java/src/main/resources/application.yml index ba91b468..28606cdb 100644 --- a/backend-java/src/main/resources/application.yml +++ b/backend-java/src/main/resources/application.yml @@ -264,6 +264,7 @@ aiimage: brand-check-batch-size: ${AIIMAGE_COLLECT_DATA_BRAND_CHECK_BATCH_SIZE:10} brand-check-cache-capacity: ${AIIMAGE_COLLECT_DATA_BRAND_CHECK_CACHE_CAPACITY:512} invalid-asin-batch-size: ${AIIMAGE_COLLECT_DATA_INVALID_ASIN_BATCH_SIZE:100} + result-item-batch-size: ${AIIMAGE_COLLECT_DATA_RESULT_ITEM_BATCH_SIZE:100} image-video: coze-base-url: ${AIIMAGE_IMAGE_VIDEO_COZE_BASE_URL:https://api.coze.cn} coze-token: ${AIIMAGE_IMAGE_VIDEO_COZE_TOKEN:sat_Ws4VB1caOPasDivpKIvtOySYx3lhKgQ95H3crIh0tBwiNYtPTyi6bqe0pBaRzpVu} diff --git a/backend-java/src/test/java/com/nanri/aiimage/modules/collectdata/util/CollectDataResultItemBatchWriterTest.java b/backend-java/src/test/java/com/nanri/aiimage/modules/collectdata/util/CollectDataResultItemBatchWriterTest.java new file mode 100644 index 00000000..8badb61e --- /dev/null +++ b/backend-java/src/test/java/com/nanri/aiimage/modules/collectdata/util/CollectDataResultItemBatchWriterTest.java @@ -0,0 +1,239 @@ +package com.nanri.aiimage.modules.collectdata.util; + +import com.fasterxml.jackson.databind.ObjectMapper; +import com.nanri.aiimage.modules.collectdata.model.vo.CollectDataResultRowVo; +import com.nanri.aiimage.modules.task.mapper.TaskResultItemMapper; +import com.nanri.aiimage.modules.task.model.entity.TaskResultItemEntity; +import com.nanri.aiimage.modules.task.service.TransientPayloadStorageService; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.mockito.ArgumentCaptor; + +import java.util.ArrayList; +import java.util.List; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertTrue; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.anyList; +import static org.mockito.Mockito.doThrow; +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 50：结果明细批量 upsert mapper 与幂等唯一键。 + * CollectDataResultItemBatchWriter 把整 chunk 的 accepted 行批量 upsert 到 + * biz_task_result_item：先一次批量查询现有行（payload_hash 相等即跳过， + * 幂等），再按唯一键 uk_task_scope_item 用一条 INSERT ... ON DUPLICATE + * KEY UPDATE 批量写入，替代逐行 select/insert/update。空输入零调用， + * 批量失败跳过该批可恢复，无无界内存增长。 + */ +class CollectDataResultItemBatchWriterTest { + + private TaskResultItemMapper taskResultItemMapper; + private TransientPayloadStorageService transientPayloadStorageService; + private CollectDataResultDetailCodec codec; + private CollectDataResultItemBatchWriter writer; + + @BeforeEach + void setUp() { + taskResultItemMapper = mock(TaskResultItemMapper.class); + transientPayloadStorageService = mock(TransientPayloadStorageService.class); + codec = new CollectDataResultDetailCodec(new ObjectMapper()); + writer = new CollectDataResultItemBatchWriter(taskResultItemMapper, codec, transientPayloadStorageService, 10); + } + + @Test + void test_task_050_task_normal_default_path() { + // 正常输入：无现有行 → 一次 upsertBatch 全量写入，引用字段完整。 + when(taskResultItemMapper.selectList(any())).thenReturn(List.of()); + when(taskResultItemMapper.upsertBatch(anyList())).thenReturn(2); + List rows = List.of( + row("B000000001", "Nike"), row("B000000002", "Zara")); + + CollectDataResultItemBatchWriter.UpsertCounts counts = + writer.upsertAccepted(1L, 2L, "task:1", 0, rows, "rustfs:detail/abc"); + + assertEquals(2, counts.insertedOrUpdated(), "两行均写入"); + assertEquals(0, counts.skipped(), "无跳过"); + verify(taskResultItemMapper).selectList(any()); + ArgumentCaptor captor = ArgumentCaptor.forClass(List.class); + verify(taskResultItemMapper).upsertBatch(captor.capture()); + List entities = captor.getValue(); + assertEquals(2, entities.size(), "批量写入两行"); + assertEquals("asin:B000000001", entities.get(0).getItemKey(), "item_key 语义键"); + assertEquals("ACCEPTED", entities.get(0).getStatus(), "状态保持 ACCEPTED"); + assertTrue(entities.get(0).getPayloadJson().contains("rustfs:detail/abc"), "引用含对象指针"); + assertTrue(entities.get(0).getPayloadJson().contains("\"offset\":0"), "引用含 offset"); + assertEquals(2L, entities.get(0).getResultId(), "result_id 保留"); + assertEquals(1L, entities.get(0).getTaskId(), "task_id 保留"); + } + + @Test + void test_task_050_task_normal_multiple_items() { + // 批量场景：75 行（8 批次），顺序稳定不丢失，每批数量正确。 + when(taskResultItemMapper.selectList(any())).thenReturn(List.of()); + when(taskResultItemMapper.upsertBatch(anyList())).thenAnswer( + invocation -> ((List) invocation.getArgument(0)).size()); + List rows = new ArrayList<>(); + for (int i = 0; i < 75; i++) { + rows.add(row("B" + String.format("%09d", i + 1), "brand-" + (i % 7))); + } + + CollectDataResultItemBatchWriter.UpsertCounts counts = + writer.upsertAccepted(1L, 2L, "task:1", 3, rows, "rustfs:detail/x"); + + assertEquals(75, counts.insertedOrUpdated(), "全部写入"); + ArgumentCaptor captor = ArgumentCaptor.forClass(List.class); + verify(taskResultItemMapper, times(8)).upsertBatch(captor.capture()); + for (int i = 0; i < 8; i++) { + assertEquals(i < 7 ? 10 : 5, captor.getAllValues().get(i).size(), "批次 " + i + " 数量"); + } + assertEquals("asin:B000000001", ((TaskResultItemEntity) captor.getAllValues().get(0).get(0)).getItemKey(), + "顺序稳定"); + assertEquals("asin:B000000075", ((TaskResultItemEntity) captor.getAllValues().get(7).get(4)).getItemKey(), + "末批末行不丢失"); + } + + @Test + void test_task_050_task_normal_repeated_operation_is_idempotent() { + // 幂等：同一 chunk 重复提交时 payload_hash 相等 → 全部跳过不写入， + // 无重复记录、无重复写入调用。 + String refJson = codec.encodeRef(0, 0, "rustfs:detail/x"); + TaskResultItemEntity existing = new TaskResultItemEntity(); + existing.setId(100L); + existing.setItemKey("asin:B000000001"); + existing.setPayloadJson(refJson); + existing.setPayloadHash(sha256(refJson)); + when(taskResultItemMapper.selectList(any())).thenReturn(List.of(existing)); + + CollectDataResultItemBatchWriter.UpsertCounts counts = + writer.upsertAccepted(1L, 2L, "task:1", 0, + List.of(row("B000000001", "Nike")), "rustfs:detail/x"); + + assertEquals(0, counts.insertedOrUpdated(), "hash 相等全部跳过"); + assertEquals(1, counts.skipped(), "1 行跳过"); + verify(taskResultItemMapper, never()).upsertBatch(anyList()); + } + + @Test + void test_task_050_task_boundary_empty_input() { + // 空输入：空列表零调用返回零计数。 + CollectDataResultItemBatchWriter.UpsertCounts counts = + writer.upsertAccepted(1L, 2L, "task:1", 0, List.of(), "rustfs:x"); + + assertEquals(0, counts.insertedOrUpdated(), "空输入返回 0"); + assertEquals(0, counts.skipped(), "空输入无跳过"); + verify(taskResultItemMapper, never()).selectList(any()); + verify(taskResultItemMapper, never()).upsertBatch(anyList()); + } + + @Test + void test_task_050_task_boundary_single_item() { + // 单元素：单行正常 upsert；hash 不同时更新，唯一键幂等； + // 旧格式逐行 payload（非引用）在升级为引用时物理删除，避免泄漏。 + TaskResultItemEntity existing = new TaskResultItemEntity(); + existing.setId(100L); + existing.setItemKey("asin:B000000001"); + existing.setPayloadJson("{\"brand\":\"nike\",\"asin\":\"B000000001\"}"); + existing.setPayloadHash("old-hash"); + when(taskResultItemMapper.selectList(any())).thenReturn(List.of(existing)); + when(taskResultItemMapper.upsertBatch(anyList())).thenReturn(1); + + CollectDataResultItemBatchWriter.UpsertCounts counts = + writer.upsertAccepted(1L, 2L, "task:1", 1, + List.of(row("B000000001", "Nike")), "rustfs:new"); + + assertEquals(1, counts.insertedOrUpdated(), "hash 不同触发 upsert 更新"); + verify(transientPayloadStorageService).deletePayloadIfPresent("{\"brand\":\"nike\",\"asin\":\"B000000001\"}"); + ArgumentCaptor captor = ArgumentCaptor.forClass(List.class); + verify(taskResultItemMapper).upsertBatch(captor.capture()); + TaskResultItemEntity entity = (TaskResultItemEntity) captor.getValue().get(0); + assertEquals(100L, entity.getId(), "复用现有行 id（ON DUPLICATE 命中唯一键）"); + assertTrue(entity.getPayloadJson().contains("rustfs:new"), "引用更新为新对象"); + } + + @Test + void test_task_050_task_boundary_limit_and_overflow() { + // 上限/超限：批次大小 3，10 行 = 4 批（3+3+3+1），无无界积累。 + when(taskResultItemMapper.selectList(any())).thenReturn(List.of()); + when(taskResultItemMapper.upsertBatch(anyList())).thenAnswer( + invocation -> ((List) invocation.getArgument(0)).size()); + CollectDataResultItemBatchWriter smallWriter = + new CollectDataResultItemBatchWriter(taskResultItemMapper, codec, transientPayloadStorageService, 3); + List rows = new ArrayList<>(); + for (int i = 0; i < 10; i++) { + rows.add(row("B" + String.format("%09d", i + 1), "brand")); + } + + CollectDataResultItemBatchWriter.UpsertCounts counts = smallWriter.upsertAccepted( + 1L, 2L, "task:1", 0, rows, "rustfs:x"); + + assertEquals(10, counts.insertedOrUpdated(), "全部写入"); + ArgumentCaptor captor = ArgumentCaptor.forClass(List.class); + verify(taskResultItemMapper, times(4)).upsertBatch(captor.capture()); + assertEquals(3, captor.getAllValues().get(0).size(), "首批 3 行"); + assertEquals(1, captor.getAllValues().get(3).size(), "末批 1 行"); + } + + @Test + void test_task_050_task_invalid_input_rejected() { + // 非法参数：null/空白 ASIN 的行跳过不写入；null 行安全跳过。 + when(taskResultItemMapper.selectList(any())).thenReturn(List.of()); + when(taskResultItemMapper.upsertBatch(anyList())).thenAnswer( + invocation -> ((List) invocation.getArgument(0)).size()); + List rows = new ArrayList<>(); + rows.add(null); + rows.add(row(null, "nike")); + rows.add(row("B000000001", "nike")); + + CollectDataResultItemBatchWriter.UpsertCounts counts = + writer.upsertAccepted(1L, 2L, "task:1", 0, rows, "rustfs:x"); + + assertEquals(1, counts.insertedOrUpdated(), "仅合法行写入"); + verify(taskResultItemMapper).upsertBatch(anyList()); + } + + @Test + void test_task_050_task_dependency_failure_releases_resources() { + // 依赖失败：批量 upsert 抛错时跳过该批不中断，恢复后继续，无资源泄漏。 + when(taskResultItemMapper.selectList(any())).thenReturn(List.of()); + doThrow(new RuntimeException("db down")) + .doAnswer(invocation -> ((List) invocation.getArgument(0)).size()) + .when(taskResultItemMapper).upsertBatch(anyList()); + List rows = new ArrayList<>(); + for (int i = 0; i < 20; i++) { + rows.add(row("B" + String.format("%09d", i + 1), "brand")); + } + + CollectDataResultItemBatchWriter.UpsertCounts counts = + writer.upsertAccepted(1L, 2L, "task:1", 0, rows, "rustfs:x"); + + assertEquals(10, counts.insertedOrUpdated(), "首批失败跳过，第二批 10 行写入"); + verify(taskResultItemMapper, times(2)).upsertBatch(anyList()); + } + + private static CollectDataResultRowVo row(String asin, String brand) { + CollectDataResultRowVo row = new CollectDataResultRowVo(); + row.setAsin(asin); + row.setBrand(brand); + return row; + } + + private static String sha256(String value) { + try { + java.security.MessageDigest digest = java.security.MessageDigest.getInstance("SHA-256"); + byte[] bytes = digest.digest((value == null ? "" : value).getBytes(java.nio.charset.StandardCharsets.UTF_8)); + StringBuilder sb = new StringBuilder(bytes.length * 2); + for (byte b : bytes) { + sb.append(String.format("%02x", b)); + } + return sb.toString(); + } catch (Exception ex) { + throw new IllegalStateException(ex); + } + } +}