super/crawler-plugin
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

完善后端轮询记忆紫鸟店铺逻辑

Browse Source
This commit is contained in:
super
2026-04-02 12:26:48 +08:00
parent 0a8202788b
commit d3671206e1
7 changed files with 383 additions and 62 deletions
Download Patch File Download Diff File Expand all files Collapse all files

18
backend-java/src/main/java/com/nanri/aiimage/modules/brand/service/BrandTaskService.java
View File

@@ -1053,6 +1053,7 @@ public class BrandTaskService {
for (BrandCrawlTaskEntity task : runningTasks) {
Map<Object, Object> progress = brandTaskProgressCacheService.getProgress(task.getId());
if (progress.isEmpty()) {
failStaleRunningTask(task.getId(), "任务进度缓存已过期，任务已自动失败");
continue;
}
long lastHeartbeatAt = 0L;
@@ -1061,21 +1062,26 @@ public class BrandTaskService {
} catch (Exception ignored) {
}
if (lastHeartbeatAt <= 0) {
failStaleRunningTask(task.getId(), "任务心跳信息缺失，任务已自动失败");
continue;
}
LocalDateTime lastHeartbeat = LocalDateTime.ofInstant(Instant.ofEpochMilli(lastHeartbeatAt), ZoneId.systemDefault());
if (lastHeartbeat.isAfter(threshold)) {
continue;
}
brandCrawlTaskMapper.update(null, new LambdaUpdateWrapper<BrandCrawlTaskEntity>()
.eq(BrandCrawlTaskEntity::getId, task.getId())
.eq(BrandCrawlTaskEntity::getStatus, STATUS_RUNNING)
.set(BrandCrawlTaskEntity::getStatus, STATUS_FAILED)
.set(BrandCrawlTaskEntity::getErrorMessage, "前端长时间无响应，任务已自动失败"));
brandTaskProgressCacheService.markFailed(task.getId(), "前端长时间无响应，任务已自动失败");
failStaleRunningTask(task.getId(), "前端长时间无响应，任务已自动失败");
}
}
private void failStaleRunningTask(Long taskId, String message) {
brandCrawlTaskMapper.update(null, new LambdaUpdateWrapper<BrandCrawlTaskEntity>()
.eq(BrandCrawlTaskEntity::getId, taskId)
.eq(BrandCrawlTaskEntity::getStatus, STATUS_RUNNING)
.set(BrandCrawlTaskEntity::getStatus, STATUS_FAILED)
.set(BrandCrawlTaskEntity::getErrorMessage, message));
brandTaskProgressCacheService.markFailed(taskId, message);
}
private record ParsedBrandFile(String sheetName, List<String> columns, List<Map<String, Object>> rows,
List<String> uniqueBrands) {
}

3
backend-java/src/main/java/com/nanri/aiimage/modules/deletebrand/model/cache/DeleteBrandParsedFileCacheDto.java vendored
View File

@@ -12,6 +12,9 @@ public class DeleteBrandParsedFileCacheDto {
private String fileKey;
private String sourceFilename;
private String shopName;
private String shopId;
private String platform;
private String openStoreUrl;
private Integer totalRows;
private List<DeleteBrandCountryGroupVo> countries = new ArrayList<>();
private List<DeleteBrandPreviewRowVo> previewRows = new ArrayList<>();

18
backend-java/src/main/java/com/nanri/aiimage/modules/deletebrand/service/DeleteBrandRunService.java
View File

@@ -151,6 +151,9 @@ public class DeleteBrandRunService {
cacheDto.setFileKey(sourceFile.getFileKey());
cacheDto.setSourceFilename(sourceFile.getOriginalFilename());
cacheDto.setShopName(item.getShopName());
cacheDto.setShopId(item.getShopId());
cacheDto.setPlatform(item.getPlatform());
cacheDto.setOpenStoreUrl(item.getOpenStoreUrl());
cacheDto.setTotalRows(parsed.totalRows());
cacheDto.setCountries(parsed.countries());
cacheDto.setPreviewRows(parsed.previewRows());
@@ -284,9 +287,9 @@ public class DeleteBrandRunService {
item.setMatchStatus(candidate.getMatchStatus());
item.setMatchMessage(candidate.getMatchMessage());
// 如果物理结果已经是成功了，说明已经处理过了，强制置为匹配成功
if (item.isSuccess()) {
if (item.isSuccess() && candidate.getShopId() != null && !candidate.getShopId().isBlank()) {
item.setMatched(true);
item.setMatchStatus("MATCHED");
item.setMatchStatus(ZiniaoShopIndexService.MATCH_STATUS_MATCHED);
item.setMatchMessage(null);
}
@@ -588,7 +591,7 @@ public class DeleteBrandRunService {
}
item.setTaskId(task.getId());
item.setTaskStatus(task.getStatus());
if ("SUCCESS".equals(task.getStatus()) && item.isSuccess()) {
if ("SUCCESS".equals(task.getStatus()) && item.isSuccess() && item.getShopId() != null && !item.getShopId().isBlank()) {
item.setMatched(true);
item.setMatchStatus(ZiniaoShopIndexService.MATCH_STATUS_MATCHED);
item.setMatchMessage(null);
@@ -1054,9 +1057,12 @@ public class DeleteBrandRunService {
item.setCountries(parsedFile.getCountries());
item.setPreviewRows(parsedFile.getPreviewRows());
item.setTruncated(false);
item.setMatched(true);
item.setMatchStatus("MATCHED");
item.setMatchMessage(null);
item.setShopId(parsedFile.getShopId());
item.setPlatform(parsedFile.getPlatform());
item.setOpenStoreUrl(parsedFile.getOpenStoreUrl());
item.setMatched(parsedFile.getShopId() != null && !parsedFile.getShopId().isBlank());
item.setMatchStatus(item.isMatched() ? ZiniaoShopIndexService.MATCH_STATUS_MATCHED : ZiniaoShopIndexService.MATCH_STATUS_PENDING);
item.setMatchMessage(item.isMatched() ? null : "店铺索引信息缺失，请重新匹配索引");
finalItems.add(item);
successCount++;
}

35
backend-java/src/main/java/com/nanri/aiimage/modules/ziniao/memory/service/ZiniaoMemoryStoreService.java
View File

@@ -7,6 +7,7 @@ import com.nanri.aiimage.modules.ziniao.memory.model.entity.ZiniaoMemoryStoreEnt
import com.fasterxml.jackson.databind.JavaType;
import com.fasterxml.jackson.databind.ObjectMapper;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;
@@ -18,8 +19,12 @@ import java.util.Optional;
@Service
@RequiredArgsConstructor
@Slf4j
public class ZiniaoMemoryStoreService {
/** 本表仅用于持久化店铺索引条目（见 ZiniaoShopIndexService）。 */
public static final String CACHE_TYPE_SHOP_INDEX_ENTRY = "SHOP_INDEX_ENTRY";
private final ZiniaoMemoryStoreMapper ziniaoMemoryStoreMapper;
private final ObjectMapper objectMapper;
@@ -80,6 +85,21 @@ public class ZiniaoMemoryStoreService {
.toList();
}
/**
* 未过期行（含 cache_key），用于索引巡检、修复与去重后的 stale 标记。
*/
public List<ZiniaoMemoryStoreEntity> listAliveEntitiesByType(String cacheType, int limit) {
String normalizedType = normalizeRequired(cacheType, "cacheType 不能为空");
int safeLimit = Math.max(limit, 1);
LocalDateTime now = LocalDateTime.now();
List<ZiniaoMemoryStoreEntity> entities = ziniaoMemoryStoreMapper.selectList(new LambdaQueryWrapper<ZiniaoMemoryStoreEntity>()
.eq(ZiniaoMemoryStoreEntity::getCacheType, normalizedType)
.gt(ZiniaoMemoryStoreEntity::getExpiresAt, now)
.orderByAsc(ZiniaoMemoryStoreEntity::getCacheKey)
.last("LIMIT " + safeLimit));
return entities == null ? List.of() : entities;
}
@Transactional
public void put(String cacheType, String cacheKey, Object payload, Duration ttl) {
String normalizedType = normalizeRequired(cacheType, "cacheType 不能为空");
@@ -105,19 +125,32 @@ public class ZiniaoMemoryStoreService {
entity.setCreatedAt(now);
entity.setUpdatedAt(now);
ziniaoMemoryStoreMapper.insert(entity);
if (CACHE_TYPE_SHOP_INDEX_ENTRY.equals(normalizedType)) {
log.info("[ziniao-shop-index-db] insert cacheKey={} expiresAt={} bytes={}",
normalizedKey, expiresAt, payloadJson.length());
}
return;
}
entity.setPayloadJson(payloadJson);
entity.setExpiresAt(expiresAt);
entity.setUpdatedAt(now);
ziniaoMemoryStoreMapper.updateById(entity);
if (CACHE_TYPE_SHOP_INDEX_ENTRY.equals(normalizedType)) {
log.info("[ziniao-shop-index-db] update cacheKey={} id={} expiresAt={} bytes={}",
normalizedKey, entity.getId(), expiresAt, payloadJson.length());
}
}
@Transactional
public void delete(String cacheType, String cacheKey) {
ZiniaoMemoryStoreEntity entity = findOne(cacheType, cacheKey);
String normalizedType = normalizeRequired(cacheType, "cacheType 不能为空");
String normalizedKey = normalizeRequired(cacheKey, "cacheKey 不能为空");
ZiniaoMemoryStoreEntity entity = findOne(normalizedType, normalizedKey);
if (entity != null) {
ziniaoMemoryStoreMapper.deleteById(entity.getId());
if (CACHE_TYPE_SHOP_INDEX_ENTRY.equals(normalizedType)) {
log.info("[ziniao-shop-index-db] delete cacheKey={} id={}", entity.getCacheKey(), entity.getId());
}
}
}

114
backend-java/src/main/java/com/nanri/aiimage/modules/ziniao/memory/service/ZiniaoTransientCacheService.java Normal file
View File

@@ -0,0 +1,114 @@
package com.nanri.aiimage.modules.ziniao.memory.service;
import com.fasterxml.jackson.databind.JavaType;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.nanri.aiimage.common.exception.BusinessException;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Service;
import java.time.Duration;
import java.time.LocalDateTime;
import java.util.List;
import java.util.Objects;
import java.util.Optional;
import java.util.concurrent.ConcurrentHashMap;
/**
* 进程内缓存：紫鸟 token/员工/店铺列表等短期数据。
* 不写入 biz_ziniao_memory_store，避免与「店铺索引落库」混用同一张表。
*/
@Service
@RequiredArgsConstructor
@Slf4j
public class ZiniaoTransientCacheService {
private static final String SEP = "::";
private final ObjectMapper objectMapper;
private final ConcurrentHashMap<String, Holder> map = new ConcurrentHashMap<>();
public <T> Optional<T> get(String cacheType, String cacheKey, Class<T> valueType) {
Holder holder = getHolder(cacheType, cacheKey);
if (holder == null) {
return Optional.empty();
}
try {
return Optional.ofNullable(objectMapper.readValue(holder.json, valueType));
} catch (Exception ex) {
throw new BusinessException("读取紫鸟进程缓存失败");
}
}
public <T> Optional<T> get(String cacheType, String cacheKey, JavaType javaType) {
Holder holder = getHolder(cacheType, cacheKey);
if (holder == null) {
return Optional.empty();
}
try {
@SuppressWarnings("unchecked")
T value = (T) objectMapper.readValue(holder.json, javaType);
return Optional.ofNullable(value);
} catch (Exception ex) {
throw new BusinessException("读取紫鸟进程缓存失败");
}
}
public <E> Optional<List<E>> getList(String cacheType, String cacheKey, Class<E> elementType) {
JavaType type = objectMapper.getTypeFactory().constructCollectionType(List.class, elementType);
return get(cacheType, cacheKey, type);
}
public void put(String cacheType, String cacheKey, Object payload, Duration ttl) {
String normalizedType = normalizeRequired(cacheType, "cacheType 不能为空");
String normalizedKey = normalizeRequired(cacheKey, "cacheKey 不能为空");
if (ttl == null || ttl.isZero() || ttl.isNegative()) {
throw new BusinessException("ttl 不合法");
}
try {
String json = objectMapper.writeValueAsString(payload);
LocalDateTime expiresAt = LocalDateTime.now().plusSeconds(ttl.getSeconds());
map.put(compoundKey(normalizedType, normalizedKey), new Holder(json, expiresAt));
log.trace("[ziniao-transient] put type={} keyLen={} ttlSec={}", normalizedType, normalizedKey.length(), ttl.getSeconds());
} catch (Exception ex) {
throw new BusinessException("写入紫鸟进程缓存失败");
}
}
public void delete(String cacheType, String cacheKey) {
String normalizedType = normalizeRequired(cacheType, "cacheType 不能为空");
String normalizedKey = normalizeRequired(cacheKey, "cacheKey 不能为空");
map.remove(compoundKey(normalizedType, normalizedKey));
log.trace("[ziniao-transient] delete type={}", normalizedType);
}
private Holder getHolder(String cacheType, String cacheKey) {
String normalizedType = normalizeRequired(cacheType, "cacheType 不能为空");
String normalizedKey = normalizeRequired(cacheKey, "cacheKey 不能为空");
Holder holder = map.get(compoundKey(normalizedType, normalizedKey));
if (holder == null) {
return null;
}
if (holder.expiresAt == null || !holder.expiresAt.isAfter(LocalDateTime.now())) {
map.remove(compoundKey(normalizedType, normalizedKey));
return null;
}
return holder;
}
private static String compoundKey(String cacheType, String cacheKey) {
return cacheType + SEP + cacheKey;
}
private static String normalizeRequired(String value, String message) {
String normalized = Objects.toString(value, "").trim();
if (normalized.isEmpty()) {
throw new BusinessException(message);
}
return normalized;
}
private record Holder(String json, LocalDateTime expiresAt) {
}
}

52
backend-java/src/main/java/com/nanri/aiimage/modules/ziniao/service/ZiniaoAuthService.java
View File

@@ -3,7 +3,7 @@ package com.nanri.aiimage.modules.ziniao.service;
import com.nanri.aiimage.common.exception.BusinessException;
import com.nanri.aiimage.config.ZiniaoProperties;
import com.nanri.aiimage.modules.ziniao.client.ZiniaoClient;
import com.nanri.aiimage.modules.ziniao.memory.service.ZiniaoMemoryStoreService;
import com.nanri.aiimage.modules.ziniao.memory.service.ZiniaoTransientCacheService;
import com.nanri.aiimage.modules.ziniao.model.cache.ZiniaoSessionCacheDto;
import com.nanri.aiimage.modules.ziniao.model.cache.ZiniaoShopCacheDto;
import com.nanri.aiimage.modules.ziniao.service.ZiniaoApiKeyProvider;
@@ -151,23 +151,30 @@ public class ZiniaoAuthService {
|| message.contains("userId存在无效的参数值");
}
public void evictUserStoresCacheForIndex(String apiKey, Long companyId, Long userId) {
if (apiKey == null || apiKey.isBlank() || companyId == null || companyId <= 0 || userId == null || userId <= 0) {
return;
}
ziniaoTransientCacheService.delete(CACHE_TYPE_USER_STORES, buildApiKeyHash(apiKey) + ":" + companyId + ":" + userId);
}
public void evictInvalidUserForIndex(String apiKey, Long companyId, Long userId) {
if (apiKey == null || apiKey.isBlank() || companyId == null || companyId <= 0 || userId == null || userId <= 0) {
return;
}
String staffCacheKey = buildApiKeyHash(apiKey) + ":" + companyId;
ziniaoMemoryStoreService.getList(CACHE_TYPE_STAFF_LIST, staffCacheKey, ZiniaoStaffItemVo.class)
ziniaoTransientCacheService.getList(CACHE_TYPE_STAFF_LIST, staffCacheKey, ZiniaoStaffItemVo.class)
.ifPresent(staff -> {
List<ZiniaoStaffItemVo> filtered = staff.stream()
.filter(item -> item != null && !userId.equals(item.getUserId()))
.toList();
if (filtered.isEmpty()) {
ziniaoMemoryStoreService.delete(CACHE_TYPE_STAFF_LIST, staffCacheKey);
ziniaoTransientCacheService.delete(CACHE_TYPE_STAFF_LIST, staffCacheKey);
} else {
ziniaoMemoryStoreService.put(CACHE_TYPE_STAFF_LIST, staffCacheKey, filtered, STAFF_LIST_CACHE_TTL);
ziniaoTransientCacheService.put(CACHE_TYPE_STAFF_LIST, staffCacheKey, filtered, STAFF_LIST_CACHE_TTL);
}
});
ziniaoMemoryStoreService.delete(CACHE_TYPE_USER_STORES, buildApiKeyHash(apiKey) + ":" + companyId + ":" + userId);
ziniaoTransientCacheService.delete(CACHE_TYPE_USER_STORES, buildApiKeyHash(apiKey) + ":" + companyId + ":" + userId);
}
private boolean isSkippableUserStoresError(BusinessException ex) {
@@ -196,7 +203,7 @@ public class ZiniaoAuthService {
private final ZiniaoProperties ziniaoProperties;
private final ZiniaoClient ziniaoClient;
private final ZiniaoSessionCacheService ziniaoSessionCacheService;
private final ZiniaoMemoryStoreService ziniaoMemoryStoreService;
private final ZiniaoTransientCacheService ziniaoTransientCacheService;
private final ZiniaoApiKeyProvider ziniaoApiKeyProvider;
public ZiniaoSessionVo getSession(String sessionId, Long userId) {
@@ -393,14 +400,14 @@ public class ZiniaoAuthService {
String normalizedApiKey = requireText(apiKey, "紫鸟 apiKey 未配置");
String cacheKey = buildApiKeyHash(normalizedApiKey);
Long cached = ziniaoMemoryStoreService.get(CACHE_TYPE_COMPANY_ID, cacheKey, Long.class).orElse(null);
Long cached = ziniaoTransientCacheService.get(CACHE_TYPE_COMPANY_ID, cacheKey, Long.class).orElse(null);
if (cached != null && cached > 0) {
log.info("[ziniao-company] hit cache, keyHash={}, companyId={}", shortKeyHash(apiKey), cached);
return cached;
}
log.info("[ziniao-company] cache miss, requesting upstream, keyHash={}", shortKeyHash(apiKey));
Long companyId = ziniaoClient.getCompanyIdByApiKey(normalizedApiKey);
ziniaoMemoryStoreService.put(CACHE_TYPE_COMPANY_ID, cacheKey, companyId, COMPANY_ID_CACHE_TTL);
ziniaoTransientCacheService.put(CACHE_TYPE_COMPANY_ID, cacheKey, companyId, COMPANY_ID_CACHE_TTL);
log.info("[ziniao-company] cached upstream result, keyHash={}, companyId={}", shortKeyHash(apiKey), companyId);
return companyId;
}
@@ -408,7 +415,7 @@ public class ZiniaoAuthService {
private List<ZiniaoStaffItemVo> getOrLoadStaff(String apiKey, Long companyId) {
String cacheKey = buildApiKeyHash(apiKey) + ":" + companyId;
List<ZiniaoStaffItemVo> cached = ziniaoMemoryStoreService.getList(CACHE_TYPE_STAFF_LIST, cacheKey, ZiniaoStaffItemVo.class)
List<ZiniaoStaffItemVo> cached = ziniaoTransientCacheService.getList(CACHE_TYPE_STAFF_LIST, cacheKey, ZiniaoStaffItemVo.class)
.orElse(null);
if (cached != null) {
log.info("[ziniao-staff] hit cache, keyHash={}, companyId={}, size={}", shortKeyHash(apiKey), companyId, cached.size());
@@ -416,28 +423,43 @@ public class ZiniaoAuthService {
}
log.info("[ziniao-staff] cache miss, requesting upstream, keyHash={}, companyId={}", shortKeyHash(apiKey), companyId);
List<ZiniaoStaffItemVo> staff = ziniaoClient.listStaff(apiKey, companyId);
ziniaoMemoryStoreService.put(CACHE_TYPE_STAFF_LIST, cacheKey, staff, STAFF_LIST_CACHE_TTL);
ziniaoTransientCacheService.put(CACHE_TYPE_STAFF_LIST, cacheKey, staff, STAFF_LIST_CACHE_TTL);
log.info("[ziniao-staff] cached upstream result, keyHash={}, companyId={}, size={}", shortKeyHash(apiKey), companyId, staff.size());
return staff;
}
private List<ZiniaoShopCacheDto> getOrLoadUserStores(String apiKey, Long companyId, Long userId) {
String cacheKey = buildApiKeyHash(apiKey) + ":" + companyId + ":" + userId;
List<ZiniaoShopCacheDto> cached = ziniaoMemoryStoreService.getList(CACHE_TYPE_USER_STORES, cacheKey, ZiniaoShopCacheDto.class)
List<ZiniaoShopCacheDto> cached = ziniaoTransientCacheService.getList(CACHE_TYPE_USER_STORES, cacheKey, ZiniaoShopCacheDto.class)
.orElse(null);
if (cached != null) {
log.info("[ziniao-stores] hit cache, keyHash={}, companyId={}, userId={}, size={}", shortKeyHash(apiKey), companyId, userId, cached.size());
return cached;
}
log.info("[ziniao-stores] cache miss, requesting upstream, keyHash={}, companyId={}, userId={}", shortKeyHash(apiKey), companyId, userId);
List<ZiniaoShopCacheDto> stores = ziniaoClient.listUserStores(apiKey, companyId, userId);
ziniaoMemoryStoreService.put(CACHE_TYPE_USER_STORES, cacheKey, stores, USER_STORES_CACHE_TTL);
List<ZiniaoShopCacheDto> stores = ziniaoClient.listUserStores(apiKey, companyId, userId).stream()
.filter(item -> item != null && item.getShopId() != null && !item.getShopId().isBlank())
.collect(java.util.stream.Collectors.collectingAndThen(
java.util.stream.Collectors.toMap(
ZiniaoShopCacheDto::getShopId,
item -> item,
(left, right) -> left,
java.util.LinkedHashMap::new
),
map -> new java.util.ArrayList<>(map.values())
));
if (stores.isEmpty()) {
ziniaoTransientCacheService.delete(CACHE_TYPE_USER_STORES, cacheKey);
log.info("[ziniao-stores] upstream returned empty stores, skip cache, keyHash={}, companyId={}, userId={}", shortKeyHash(apiKey), companyId, userId);
return stores;
}
ziniaoTransientCacheService.put(CACHE_TYPE_USER_STORES, cacheKey, stores, USER_STORES_CACHE_TTL);
log.info("[ziniao-stores] cached upstream result, keyHash={}, companyId={}, userId={}, size={}", shortKeyHash(apiKey), companyId, userId, stores.size());
return stores;
}
private StoreMatchResult getCachedShopMatch(String apiKey, Long companyId, String normalizedShopName) {
StoreMatchResult cached = ziniaoMemoryStoreService.get(CACHE_TYPE_SHOP_MATCH, buildShopMatchCacheKey(apiKey, companyId, normalizedShopName), StoreMatchResult.class)
StoreMatchResult cached = ziniaoTransientCacheService.get(CACHE_TYPE_SHOP_MATCH, buildShopMatchCacheKey(apiKey, companyId, normalizedShopName), StoreMatchResult.class)
.orElse(null);
if (cached == null || !cached.matched() || cached.shopId() == null || cached.matchedUserId() == null) {
return null;
@@ -446,7 +468,7 @@ public class ZiniaoAuthService {
}
private void cacheShopMatch(String apiKey, Long companyId, String normalizedShopName, StoreMatchResult result, Duration ttl) {
ziniaoMemoryStoreService.put(CACHE_TYPE_SHOP_MATCH, buildShopMatchCacheKey(apiKey, companyId, normalizedShopName), result, ttl);
ziniaoTransientCacheService.put(CACHE_TYPE_SHOP_MATCH, buildShopMatchCacheKey(apiKey, companyId, normalizedShopName), result, ttl);
}
private String buildShopMatchCacheKey(String apiKey, Long companyId, String normalizedShopName) {

205
backend-java/src/main/java/com/nanri/aiimage/modules/ziniao/service/ZiniaoShopIndexService.java
View File

@@ -1,8 +1,11 @@
package com.nanri.aiimage.modules.ziniao.service;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.nanri.aiimage.common.exception.BusinessException;
import com.nanri.aiimage.config.ZiniaoProperties;
import com.nanri.aiimage.modules.ziniao.memory.model.entity.ZiniaoMemoryStoreEntity;
import com.nanri.aiimage.modules.ziniao.memory.service.ZiniaoMemoryStoreService;
import com.nanri.aiimage.modules.ziniao.memory.service.ZiniaoTransientCacheService;
import com.nanri.aiimage.modules.ziniao.model.cache.ZiniaoShopIndexEntryDto;
import com.nanri.aiimage.modules.ziniao.model.cache.ZiniaoShopIndexRefreshCursorDto;
import com.nanri.aiimage.modules.ziniao.model.cache.ZiniaoShopCacheDto;
@@ -17,6 +20,7 @@ import java.security.MessageDigest;
import java.time.Duration;
import java.time.Instant;
import java.util.ArrayList;
import java.util.HashSet;
import java.util.LinkedHashMap;
import java.util.LinkedHashSet;
import java.util.List;
@@ -38,21 +42,26 @@ public class ZiniaoShopIndexService {
public static final String MATCH_STATUS_CONFLICT = "CONFLICT";
public static final String MATCH_STATUS_STALE = "INDEX_STALE";
private static final String CACHE_TYPE_SHOP_INDEX_ENTRY = "SHOP_INDEX_ENTRY";
private static final String CACHE_TYPE_SHOP_INDEX_SCOPE_SNAPSHOT = "SHOP_INDEX_SCOPE_SNAPSHOT";
private static final String CACHE_TYPE_SHOP_INDEX_REFRESH_CURSOR = "SHOP_INDEX_REFRESH_CURSOR";
/** 有 shopId 时唯一键，避免同一店铺因不同员工/哈希产生多行。 */
private static final String SHOP_ENTRY_KEY_SHOP_PREFIX = "s:";
/** 无 shopId（冲突占位等）时仍按规范化店名存一行。 */
private static final String SHOP_ENTRY_KEY_NAME_PREFIX = "n:";
private static final Duration DEFAULT_ENTRY_TTL = Duration.ofHours(12);
private static final Duration DEFAULT_CURSOR_TTL = Duration.ofHours(24);
private static final int DEFAULT_REFRESH_BATCH_SIZE = 100;
private static final int DEFAULT_LIST_BY_TYPE_LIMIT = 5000;
private static final int SHOP_INDEX_LIST_LIMIT = 10000;
private final ZiniaoMemoryStoreService ziniaoMemoryStoreService;
private final ZiniaoTransientCacheService ziniaoTransientCacheService;
private final ZiniaoApiKeyProvider ziniaoApiKeyProvider;
private final ZiniaoAuthService ziniaoAuthService;
private final ZiniaoProperties ziniaoProperties;
private final ObjectMapper objectMapper;
public ZiniaoShopIndexRefreshCursorDto getRefreshCursor() {
ZiniaoShopIndexRefreshCursorDto cursor = ziniaoMemoryStoreService
ZiniaoShopIndexRefreshCursorDto cursor = ziniaoTransientCacheService
.get(CACHE_TYPE_SHOP_INDEX_REFRESH_CURSOR, "global", ZiniaoShopIndexRefreshCursorDto.class)
.orElse(null);
if (cursor == null) {
@@ -70,9 +79,7 @@ public class ZiniaoShopIndexService {
return pendingResult("店铺名为空，无法匹配索引");
}
ZiniaoShopIndexEntryDto entry = ziniaoMemoryStoreService
.get(CACHE_TYPE_SHOP_INDEX_ENTRY, normalizedShopName, ZiniaoShopIndexEntryDto.class)
.orElse(null);
ZiniaoShopIndexEntryDto entry = findShopIndexEntryForLookup(normalizedShopName);
if (entry == null) {
return pendingResult("店铺索引未命中，请等待后台刷新");
}
@@ -119,9 +126,11 @@ public class ZiniaoShopIndexService {
cursor.setScopeKey("global");
cursor.setStatus("RUNNING");
cursor.setLastStartedAt(now);
ziniaoMemoryStoreService.put(CACHE_TYPE_SHOP_INDEX_REFRESH_CURSOR, "global", cursor, DEFAULT_CURSOR_TTL);
ziniaoTransientCacheService.put(CACHE_TYPE_SHOP_INDEX_REFRESH_CURSOR, "global", cursor, DEFAULT_CURSOR_TTL);
log.info("[ziniao-index] refresh started");
Map<String, List<ZiniaoShopIndexEntryDto>> grouped = new LinkedHashMap<>();
Map<String, Long> storesFingerprintToUserId = new LinkedHashMap<>();
Set<Long> allInvalidUserIds = new LinkedHashSet<>();
int refreshBatchSize = resolveRefreshBatchSize();
try {
@@ -156,7 +165,25 @@ public class ZiniaoShopIndexService {
log.warn("[ziniao-index] skip user stores, companyId={}, userId={}, msg={}", companyId, userId, ex.getMessage());
continue;
}
ziniaoMemoryStoreService.put(
if (stores == null || stores.isEmpty()) {
ziniaoTransientCacheService.delete(
CACHE_TYPE_SHOP_INDEX_SCOPE_SNAPSHOT,
buildScopeSnapshotKey(apiKey, companyId, userId)
);
continue;
}
String storesFingerprint = buildStoresFingerprint(stores);
Long duplicatedUserId = storesFingerprintToUserId.putIfAbsent(storesFingerprint, userId);
if (duplicatedUserId != null) {
ziniaoTransientCacheService.delete(
CACHE_TYPE_SHOP_INDEX_SCOPE_SNAPSHOT,
buildScopeSnapshotKey(apiKey, companyId, userId)
);
ziniaoAuthService.evictUserStoresCacheForIndex(apiKey, companyId, userId);
log.info("[ziniao-index] skip duplicated user stores snapshot, companyId={}, userId={}, duplicatedUserId={}", companyId, userId, duplicatedUserId);
continue;
}
ziniaoTransientCacheService.put(
CACHE_TYPE_SHOP_INDEX_SCOPE_SNAPSHOT,
buildScopeSnapshotKey(apiKey, companyId, userId),
stores,
@@ -186,7 +213,7 @@ public class ZiniaoShopIndexService {
}
if (!invalidUserIds.isEmpty()) {
for (Long invalidUserId : invalidUserIds) {
ziniaoMemoryStoreService.delete(
ziniaoTransientCacheService.delete(
CACHE_TYPE_SHOP_INDEX_SCOPE_SNAPSHOT,
buildScopeSnapshotKey(apiKey, companyId, invalidUserId)
);
@@ -194,6 +221,7 @@ public class ZiniaoShopIndexService {
}
}
List<ZiniaoShopIndexEntryDto> roundEntries = new ArrayList<>();
for (Map.Entry<String, List<ZiniaoShopIndexEntryDto>> groupedEntry : grouped.entrySet()) {
String normalizedShopName = groupedEntry.getKey();
List<ZiniaoShopIndexEntryDto> candidates = groupedEntry.getValue();
@@ -201,9 +229,32 @@ public class ZiniaoShopIndexService {
if (candidates.size() > 1) {
entryToStore = buildConflictEntry(normalizedShopName, candidates, now);
}
ziniaoMemoryStoreService.put(CACHE_TYPE_SHOP_INDEX_ENTRY, normalizedShopName, entryToStore, resolveEntryTtl());
roundEntries.add(entryToStore);
}
markMissingEntriesAsStale(grouped.keySet(), now);
Map<String, ZiniaoShopIndexEntryDto> byShopId = new LinkedHashMap<>();
Map<String, ZiniaoShopIndexEntryDto> byNameOnly = new LinkedHashMap<>();
for (ZiniaoShopIndexEntryDto e : roundEntries) {
String sid = e.getShopId();
if (sid != null && !sid.isBlank()) {
byShopId.merge(sid, e, this::mergeDuplicateShopIndexEntries);
} else if (e.getNormalizedShopName() != null && !e.getNormalizedShopName().isBlank()) {
byNameOnly.merge(e.getNormalizedShopName(), e, this::mergeDuplicateShopIndexEntries);
}
}
Set<String> activeCacheKeys = new HashSet<>();
for (ZiniaoShopIndexEntryDto e : byShopId.values()) {
String key = SHOP_ENTRY_KEY_SHOP_PREFIX + e.getShopId();
activeCacheKeys.add(key);
ziniaoMemoryStoreService.put(ZiniaoMemoryStoreService.CACHE_TYPE_SHOP_INDEX_ENTRY, key, e, resolveEntryTtl());
}
for (ZiniaoShopIndexEntryDto e : byNameOnly.values()) {
String key = SHOP_ENTRY_KEY_NAME_PREFIX + e.getNormalizedShopName();
activeCacheKeys.add(key);
ziniaoMemoryStoreService.put(ZiniaoMemoryStoreService.CACHE_TYPE_SHOP_INDEX_ENTRY, key, e, resolveEntryTtl());
}
log.info("[ziniao-index] refresh persist shopIndex uniqueShopId={} nameOnly={} groupedNames={} activeRows={}",
byShopId.size(), byNameOnly.size(), grouped.size(), activeCacheKeys.size());
markMissingEntriesAsStale(activeCacheKeys, now);
cursor.setStatus("SUCCESS");
cursor.setMessage(allInvalidUserIds.isEmpty()
@@ -213,12 +264,14 @@ public class ZiniaoShopIndexService {
cursor.setSampleInvalidUserIds(allInvalidUserIds.stream().limit(20).toList());
cursor.setLastFinishedAt(now);
cursor.setLastSuccessAt(now);
ziniaoMemoryStoreService.put(CACHE_TYPE_SHOP_INDEX_REFRESH_CURSOR, "global", cursor, DEFAULT_CURSOR_TTL);
ziniaoTransientCacheService.put(CACHE_TYPE_SHOP_INDEX_REFRESH_CURSOR, "global", cursor, DEFAULT_CURSOR_TTL);
log.info("[ziniao-index] refresh success invalidUsersSkipped={}", allInvalidUserIds.size());
} catch (Exception ex) {
cursor.setStatus("FAILED");
cursor.setMessage(ex.getMessage());
cursor.setLastFinishedAt(now);
ziniaoMemoryStoreService.put(CACHE_TYPE_SHOP_INDEX_REFRESH_CURSOR, "global", cursor, DEFAULT_CURSOR_TTL);
ziniaoTransientCacheService.put(CACHE_TYPE_SHOP_INDEX_REFRESH_CURSOR, "global", cursor, DEFAULT_CURSOR_TTL);
log.warn("[ziniao-index] refresh failed status=FAILED msg={}", ex.getMessage());
if (ex instanceof BusinessException businessException) {
throw businessException;
}
@@ -227,7 +280,8 @@ public class ZiniaoShopIndexService {
}
public void invalidateIndex() {
ziniaoMemoryStoreService.delete(CACHE_TYPE_SHOP_INDEX_REFRESH_CURSOR, "global");
ziniaoTransientCacheService.delete(CACHE_TYPE_SHOP_INDEX_REFRESH_CURSOR, "global");
log.info("[ziniao-index] cursor invalidated (transient only; shop rows unchanged)");
}
public String normalizeShopName(String value) {
@@ -292,20 +346,85 @@ public class ZiniaoShopIndexService {
return pendingResult(defaultMessage);
}
private void markMissingEntriesAsStale(Set<String> activeNames, long now) {
List<ZiniaoShopIndexEntryDto> existingEntries = ziniaoMemoryStoreService.listByType(
CACHE_TYPE_SHOP_INDEX_ENTRY,
ZiniaoShopIndexEntryDto.class,
DEFAULT_LIST_BY_TYPE_LIMIT
);
if (existingEntries.isEmpty()) {
return;
private ZiniaoShopIndexEntryDto findShopIndexEntryForLookup(String normalizedShopName) {
String prefixedKey = SHOP_ENTRY_KEY_NAME_PREFIX + normalizedShopName;
ZiniaoShopIndexEntryDto entry = ziniaoMemoryStoreService
.get(ZiniaoMemoryStoreService.CACHE_TYPE_SHOP_INDEX_ENTRY, prefixedKey, ZiniaoShopIndexEntryDto.class)
.orElse(null);
if (entry != null) {
log.debug("[ziniao-shop-index] lookup hit=prefixedName cacheKey={}", prefixedKey);
return entry;
}
for (ZiniaoShopIndexEntryDto existingEntry : existingEntries) {
if (existingEntry == null || existingEntry.getNormalizedShopName() == null || existingEntry.getNormalizedShopName().isBlank()) {
entry = ziniaoMemoryStoreService
.get(ZiniaoMemoryStoreService.CACHE_TYPE_SHOP_INDEX_ENTRY, normalizedShopName, ZiniaoShopIndexEntryDto.class)
.orElse(null);
if (entry != null) {
log.info("[ziniao-shop-index] lookup hit=legacyKey please migrate cacheKey={}", normalizedShopName);
return entry;
}
List<ZiniaoShopIndexEntryDto> rows = ziniaoMemoryStoreService.listByType(
ZiniaoMemoryStoreService.CACHE_TYPE_SHOP_INDEX_ENTRY,
ZiniaoShopIndexEntryDto.class,
SHOP_INDEX_LIST_LIMIT
);
for (ZiniaoShopIndexEntryDto row : rows) {
if (row == null) {
continue;
}
if (activeNames.contains(existingEntry.getNormalizedShopName())) {
if (normalizedShopName.equals(row.getNormalizedShopName())
|| normalizedShopName.equals(normalizeShopName(row.getShopName()))) {
log.info("[ziniao-shop-index] lookup hit=scanByShopName shopId={} normalizedInRow={}",
row.getShopId(), row.getNormalizedShopName());
return row;
}
}
log.info("[ziniao-shop-index] lookup miss normalizedName={}", normalizedShopName);
return null;
}
private ZiniaoShopIndexEntryDto mergeDuplicateShopIndexEntries(ZiniaoShopIndexEntryDto a, ZiniaoShopIndexEntryDto b) {
if (a == null) {
return b;
}
if (b == null) {
return a;
}
if (STATUS_CONFLICT.equals(a.getStatus()) != STATUS_CONFLICT.equals(b.getStatus())) {
return STATUS_CONFLICT.equals(a.getStatus()) ? a : b;
}
if (STATUS_CONFLICT.equals(a.getStatus())) {
int ca = a.getCandidateCount() == null ? 0 : a.getCandidateCount();
int cb = b.getCandidateCount() == null ? 0 : b.getCandidateCount();
return cb > ca ? b : a;
}
long ta = a.getLastRefreshedAt() != null ? a.getLastRefreshedAt() : (a.getLastSeenAt() != null ? a.getLastSeenAt() : 0L);
long tb = b.getLastRefreshedAt() != null ? b.getLastRefreshedAt() : (b.getLastSeenAt() != null ? b.getLastSeenAt() : 0L);
return tb >= ta ? b : a;
}
private void markMissingEntriesAsStale(Set<String> activeCacheKeys, long now) {
List<ZiniaoMemoryStoreEntity> entities = ziniaoMemoryStoreService.listAliveEntitiesByType(
ZiniaoMemoryStoreService.CACHE_TYPE_SHOP_INDEX_ENTRY,
SHOP_INDEX_LIST_LIMIT
);
if (entities.isEmpty()) {
return;
}
int staleCount = 0;
for (ZiniaoMemoryStoreEntity entity : entities) {
String rowKey = entity.getCacheKey();
if (activeCacheKeys.contains(rowKey)) {
continue;
}
ZiniaoShopIndexEntryDto existingEntry;
try {
existingEntry = objectMapper.readValue(entity.getPayloadJson(), ZiniaoShopIndexEntryDto.class);
} catch (Exception ex) {
log.warn("[ziniao-index] skip corrupt shop_index row cacheKey={}", rowKey);
continue;
}
if (existingEntry == null || existingEntry.getNormalizedShopName() == null
|| existingEntry.getNormalizedShopName().isBlank()) {
continue;
}
if (STATUS_CONFLICT.equals(existingEntry.getStatus())
@@ -313,18 +432,18 @@ public class ZiniaoShopIndexService {
&& (existingEntry.getShopName() == null || existingEntry.getShopName().isBlank())
&& (existingEntry.getApiKeyHash() == null || existingEntry.getApiKeyHash().isBlank())
&& (existingEntry.getCompanyId() == null || existingEntry.getCompanyId() <= 0)) {
ziniaoMemoryStoreService.delete(CACHE_TYPE_SHOP_INDEX_ENTRY, existingEntry.getNormalizedShopName());
ziniaoMemoryStoreService.delete(ZiniaoMemoryStoreService.CACHE_TYPE_SHOP_INDEX_ENTRY, rowKey);
log.info("[ziniao-index] removed ghost conflict shop_index row cacheKey={}", rowKey);
continue;
}
existingEntry.setStatus(STATUS_STALE);
existingEntry.setMessage("店铺索引已过期，请等待后台刷新");
existingEntry.setLastRefreshedAt(now);
ziniaoMemoryStoreService.put(
CACHE_TYPE_SHOP_INDEX_ENTRY,
existingEntry.getNormalizedShopName(),
existingEntry,
resolveEntryTtl()
);
ziniaoMemoryStoreService.put(ZiniaoMemoryStoreService.CACHE_TYPE_SHOP_INDEX_ENTRY, rowKey, existingEntry, resolveEntryTtl());
staleCount++;
}
if (staleCount > 0) {
log.info("[ziniao-index] marked stale shop_index rows count={}", staleCount);
}
}
@@ -370,9 +489,10 @@ public class ZiniaoShopIndexService {
if (left == null || right == null) {
return false;
}
if (left.getShopId() != null && right.getShopId() != null) {
if (left.getShopId() != null && !left.getShopId().isBlank()
&& right.getShopId() != null && !right.getShopId().isBlank()) {
// 同一 apiKey + 公司 + shopId：多员工可见同店，紫鸟 shopId 相同，保留任一员工即可
return Objects.equals(left.getShopId(), right.getShopId())
&& Objects.equals(left.getMatchedUserId(), right.getMatchedUserId())
&& Objects.equals(left.getCompanyId(), right.getCompanyId())
&& Objects.equals(left.getApiKeyHash(), right.getApiKeyHash());
}
@@ -382,6 +502,23 @@ public class ZiniaoShopIndexService {
&& Objects.equals(left.getApiKeyHash(), right.getApiKeyHash());
}
private String buildStoresFingerprint(List<ZiniaoShopCacheDto> stores) {
if (stores == null || stores.isEmpty()) {
return "";
}
StringBuilder sb = new StringBuilder();
stores.stream()
.filter(Objects::nonNull)
.sorted((left, right) -> Objects.toString(left.getShopId(), "").compareTo(Objects.toString(right.getShopId(), "")))
.forEach(item -> sb.append(Objects.toString(item.getShopId(), ""))
.append('|')
.append(Objects.toString(item.getShopName(), ""))
.append('|')
.append(Objects.toString(item.getPlatform(), ""))
.append(';'));
return sb.toString();
}
private String buildScopeSnapshotKey(String apiKey, Long companyId, Long userId) {
return buildApiKeyHash(apiKey) + ":" + companyId + ":" + userId;
}