--- a/api/admin-worker/src/main/java/cn/jojo/cc/worker/task/EnglishBatchPublishTask.java +++ b/api/admin-worker/src/main/java/cn/jojo/cc/worker/task/EnglishBatchPublishTask.java @@ -144,33 +144,36 @@ String raw = StringUtils.trimToNull(jobParam); if (raw == null) { return new Param(true, true, 0); } if (StringUtils.isNumeric(raw)) { - return new Param(true, true, Integer.parseInt(raw)); + return new Param(true, true, positiveLimit(raw)); } String type = TYPE_ALL; int limit = 0; - for (String segment : raw.split(",")) { + boolean parsedAny = false; + for (String segment : raw.split(",", -1)) { String pair = StringUtils.trimToEmpty(segment); if (pair.isEmpty()) { continue; } + parsedAny = true; String[] kv = pair.split("=", 2); String key = StringUtils.trimToEmpty(kv[0]); String value = kv.length > 1 ? StringUtils.trimToEmpty(kv[1]) : ""; if ("limit".equals(key)) { if (!StringUtils.isNumeric(value)) { throw new IllegalArgumentException("limit 必须是正整数：" + pair); } - limit = Integer.parseInt(value); + limit = positiveLimit(value); } else if ("type".equals(key)) { type = value; } else { throw new IllegalArgumentException("无法识别的参数：" + pair); } } + if (!parsedAny) { throw new IllegalArgumentException("参数不包含有效键值"); } if (TYPE_WORDS.equals(type)) { return new Param(true, false, limit); } if (TYPE_PHRASES.equals(type)) { return new Param(false, true, limit); @@ -178,7 +181,14 @@ if (!TYPE_ALL.equals(type)) { throw new IllegalArgumentException("type 只支持 words / phrases / all：" + type); } return new Param(true, true, limit); } + private static int positiveLimit(String value) { + if (!value.matches("[1-9][0-9]*")) { + throw new IllegalArgumentException("limit 必须是严格正整数"); + } + return Integer.parseInt(value); + } + } } --- /dev/null +++ b/api/admin-worker/src/main/java/cn/jojo/cc/worker/task/EnglishDictOutboxJobHandler.java @@ -0,0 +1,21 @@ +package cn.jojo.cc.worker.task; + +import cn.jojo.cc.biz.service.english.dict.EnglishDictOutboxBizService; +import com.xxl.job.core.biz.model.ReturnT; +import com.xxl.job.core.handler.annotation.XxlJob; +import jakarta.annotation.Resource; +import org.springframework.stereotype.Component; + +/** 部署后必须在 XXL-Job 配置 englishDictOutboxReplay，每分钟执行。 */ +@Component +public class EnglishDictOutboxJobHandler { + @Resource + private EnglishDictOutboxBizService service; + + @XxlJob("englishDictOutboxReplay") + public ReturnT replay(String ignored) { + int failed = service.replay(100); + return new ReturnT<>(failed == 0 ? 200 : ReturnT.FAIL_CODE, + "本轮投递失败=" + failed + "；待办按 retry_after 自动恢复，broker 确认不等于消费者已失效"); + } +} --- a/application/src/main/java/cn/jojo/cc/biz/cache/english/EnglishPhraseIndexCacheService.java +++ b/application/src/main/java/cn/jojo/cc/biz/cache/english/EnglishPhraseIndexCacheService.java @@ -184,10 +184,21 @@ * 一个动作（一轮分词 / 一次下发解析）只能用一张表，中途换表会产出"前半旧版、后半新版"的混合结果。

* * @return 数据源不合法，或装载失败且处于退避窗口内时返回 {@code null}； * 调用方必须据此跳过本次动作，绝不能用空表去扫——那会把短语整块漏切成单词，并被剪枝永久固化。 */ + /** 写侧仅接收已与远端版本一致的句柄；版本读取或重装失败不落库。 */ + public PhraseIndexSnapshot freshSnapshotForWrite(DataBaseTypeEnum source) { + if (!isSourceValid(source)) { return null; } + forceVersionCheck(source); + PhraseIndexSnapshot snapshot = readySnapshot(source); + Long remote = readVersionQuietly(source); + if (snapshot == null || remote == null || snapshot.version() == UNKNOWN_VERSION + || snapshot.version() != remote) { return null; } + return snapshot; + } + public PhraseIndexSnapshot readySnapshot(DataBaseTypeEnum source) { if (!isSourceValid(source)) { return null; } LoadingCache currentCache = cache; @@ -286,10 +297,25 @@ * *

必须在数据提交之后调用：版本号先于数据可见地自增，会让其它实例重装到旧数据 * 却记下新版本号，从而永久漏掉这次变更（直到下一次变更）。现有三个调用点 * （短语编辑、发布、导入完成）都在事务提交后，新增调用方须守住这一点。

*/ + /** outbox 恢复入口：版本写入失败必须保留待办，不能吞掉。 */ + public void rebuildNowReliable(DataBaseTypeEnum source, String reason) { + if (!isSourceValid(source) || redissonHelper == null) { + throw new IllegalStateException("phrase index version service unavailable"); + } + Optional version = redissonHelper.incrAndGet(buildVersionKey(source)); + if (version == null || version.isEmpty()) { + throw new IllegalStateException("phrase index version not acknowledged"); + } + LoadingCache currentCache = cache; + if (currentCache != null) { currentCache.invalidate(source); } + lastKnownSnapshots.remove(source); + nextRetryAtRef(source).set(0L); + } + public void rebuildNow(DataBaseTypeEnum source, String reason) { if (!isSourceValid(source)) { return; } // 先自增版本号（失败只记日志，绝不能影响本机失效，更不能阻断保存/发布主流程）： --- a/application/src/main/java/cn/jojo/cc/biz/model/bo/english/PublishResultBo.java +++ b/application/src/main/java/cn/jojo/cc/biz/model/bo/english/PublishResultBo.java @@ -21,10 +21,13 @@ private Long entityId; /** PROD 是否已发布 */ private boolean prodPublished; + /** PROD 事务已登记可恢复失效待办；broker/消费完成由 outbox 作业及 MQ 日志确认。 */ + private boolean cacheInvalidationPending; + /** JY 发布状态是否已确认回写 */ private boolean jyStatusConfirmed; /** 发布期间内容被再次编辑 */ private boolean contentChangedDuringPublish; --- /dev/null +++ b/application/src/main/java/cn/jojo/cc/biz/service/english/dict/EnglishAudioCompletionTxService.java @@ -0,0 +1,48 @@ +package cn.jojo.cc.biz.service.english.dict; + +import cn.jojo.cc.biz.cache.english.EnglishDictCacheBoFactory; +import cn.jojo.cc.common.enums.DataBaseTypeEnum; +import cn.jojo.cc.core.domain.english.EnglishWordDo; +import cn.jojo.cc.core.service.english.EnglishWordCoreService; +import cn.jojo.cc.core.service.english.EnglishPhraseCoreService; +import cn.jojo.cc.core.service.english.EnglishPhraseVariantCoreService; +import cn.jojo.cc.core.service.english.EnglishDictOutboxCoreService; +import com.alibaba.fastjson.JSON; +import com.alibaba.fastjson.serializer.SerializerFeature; +import jakarta.annotation.Resource; +import org.springframework.stereotype.Service; +import org.springframework.transaction.annotation.Transactional; + +/** 回填成功与教研缓存失效/正式补发布待办原子提交。 */ +@Service +public class EnglishAudioCompletionTxService { + @Resource private EnglishWordCoreService words; + @Resource private EnglishPhraseCoreService phrases; + @Resource private EnglishPhraseVariantCoreService variants; + @Resource private EnglishDictOutboxCoreService outbox; + + @Transactional(rollbackFor = Exception.class) + public boolean complete(String table, Long id, Long resourceId) { + boolean changed; + if ("word".equals(table)) { + changed = words.markAudioCompleted(id, resourceId); + if (changed) { + EnglishWordDo word = words.getById(id, 0L); + outbox.enqueue("AUDIO_WORD", id, JSON.toJSONString( + EnglishDictCacheBoFactory.wordInvalidateBos(DataBaseTypeEnum.TEACHING, id, + null, word.getWordNormalized()), SerializerFeature.WriteClassName)); + } + } else { + Long phraseId = "phrase".equals(table) ? id + : variants.mapPhraseIdByVariantIds(java.util.List.of(id), 0L).get(id); + changed = "phrase".equals(table) ? phrases.markAudioCompleted(id, resourceId) + : variants.markAudioCompleted(id, resourceId); + if (changed && phraseId != null) { + outbox.enqueue("AUDIO_PHRASE", phraseId, JSON.toJSONString( + EnglishDictCacheBoFactory.phraseInvalidateBos(DataBaseTypeEnum.TEACHING, phraseId), + SerializerFeature.WriteClassName)); + } + } + return changed; + } +} --- a/application/src/main/java/cn/jojo/cc/biz/service/english/dict/EnglishDictImportBizServiceImpl.java +++ b/application/src/main/java/cn/jojo/cc/biz/service/english/dict/EnglishDictImportBizServiceImpl.java @@ -348,10 +348,22 @@ } String fileUri = StringUtils.trimToNull(json.getString("fileUri")); if (fileUri == null) { throw BusinessException.paramException("任务参数缺少 fileUri"); } - return new Param(fileUri, Boolean.TRUE.equals(json.getBoolean("dryRun")), - json.getInteger("limit")); + Integer limit = null; + if (json.containsKey("limit")) { + Object value = json.get("limit"); + if (!(value instanceof Integer || value instanceof Long || value instanceof java.math.BigInteger) + || !value.toString().matches("[1-9][0-9]*")) { + throw BusinessException.paramException("limit 必须是严格正整数"); + } + try { + limit = Integer.valueOf(value.toString()); + } catch (NumberFormatException e) { + throw BusinessException.paramException("limit 超出整数范围"); + } + } + return new Param(fileUri, Boolean.TRUE.equals(json.getBoolean("dryRun")), limit); } } } --- /dev/null +++ b/application/src/main/java/cn/jojo/cc/biz/service/english/dict/EnglishDictOutboxBizService.java @@ -0,0 +1,57 @@ +package cn.jojo.cc.biz.service.english.dict; + +import cn.jojo.cc.biz.cache.english.EnglishPhraseIndexCacheService; +import cn.jojo.cc.biz.service.component.AliyunRocketMQService; +import cn.jojo.cc.common.enums.CcNormalTagEnum; +import cn.jojo.cc.common.enums.DataBaseTypeEnum; +import cn.jojo.cc.common.enums.DataSourceEnum; +import cn.jojo.cc.core.bo.LoginUserBo; +import cn.jojo.cc.core.domain.english.EnglishDictOutboxDo; +import cn.jojo.cc.core.multidatasource.MultiDataSourceUtil; +import cn.jojo.cc.core.service.english.EnglishDictOutboxCoreService; +import jakarta.annotation.Resource; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.stereotype.Service; + +/** 至少一次投递：broker 确认后登记 delivered_time；崩溃窗口允许重复失效，绝不丢待办。 */ +@Service +public class EnglishDictOutboxBizService { + @Resource + private EnglishDictOutboxCoreService outbox; + @Autowired + private AliyunRocketMQService rocketMqProducerService; + @Resource + private EnglishPhraseIndexCacheService index; + @Resource + private EnglishPublishBizService publisher; + + public int replay(int limit) { + int pending = 0; + for (DataSourceEnum source : new DataSourceEnum[]{DataSourceEnum.DEFAULT, DataSourceEnum.PROD}) { + pending += MultiDataSourceUtil.switchThen(source, () -> replayCurrentSource(limit)); + } + return pending; + } + + private int replayCurrentSource(int limit) { + int failed = 0; + for (EnglishDictOutboxDo row : outbox.listDue(limit)) { + try { + rocketMqProducerService.sendN(CcNormalTagEnum.CLEAR_REDIS_COMMON_CACHE, + row.getPayload(), "english-dict-outbox-" + row.getId()); + if ("PHRASE_CACHE".equals(row.getEventType())) { + index.rebuildNowReliable(DataBaseTypeEnum.PROD, "outbox-" + row.getId()); + } else if ("AUDIO_WORD".equals(row.getEventType())) { + publisher.publishAudioWord(row.getEntityId(), new LoginUserBo(-1L, "admin")); + } else if ("AUDIO_PHRASE".equals(row.getEventType())) { + publisher.publishAudioPhrase(row.getEntityId(), new LoginUserBo(-1L, "admin")); + } + outbox.acknowledge(row.getId()); + } catch (Exception e) { + failed++; + outbox.postpone(row, e.getClass().getSimpleName() + ": " + e.getMessage()); + } + } + return failed; + } +} --- a/application/src/main/java/cn/jojo/cc/biz/service/english/dict/EnglishDictPublishProServiceHelper.java +++ b/application/src/main/java/cn/jojo/cc/biz/service/english/dict/EnglishDictPublishProServiceHelper.java @@ -49,10 +49,13 @@ */ @Component @Slf4j public class EnglishDictPublishProServiceHelper { + @Resource + private cn.jojo.cc.core.service.english.EnglishDictOutboxCoreService englishDictOutboxCoreService; + private static final EnglishDictPublishSnapshotConvert SNAPSHOT = EnglishDictPublishSnapshotConvert.INSTANCE; @Resource private EnglishWordCoreService englishWordCoreService; @@ -132,10 +135,16 @@ writes.add(deleted); } if (!writes.isEmpty()) { englishWordTranslationCoreService.saveOrUpdateBatch(writes); } + englishDictOutboxCoreService.enqueue("WORD_CACHE", prodWordId, + com.alibaba.fastjson.JSON.toJSONString( + cn.jojo.cc.biz.cache.english.EnglishDictCacheBoFactory.wordInvalidateBos( + cn.jojo.cc.common.enums.DataBaseTypeEnum.PROD, prodWordId, + oldProdNormalized, jyWord.getWordNormalized()), + com.alibaba.fastjson.serializer.SerializerFeature.WriteClassName)); return oldProdNormalized; } /** * 发布短语到 PROD：主表 + 变形差量 + 音频副本（原形与每个变体各自的音频），同一 PROD 事务。 @@ -197,10 +206,15 @@ writes.add(deleted); } if (!writes.isEmpty()) { englishPhraseVariantCoreService.saveOrUpdateBatch(writes); } + englishDictOutboxCoreService.enqueue("PHRASE_CACHE", prodPhraseId, + com.alibaba.fastjson.JSON.toJSONString( + cn.jojo.cc.biz.cache.english.EnglishDictCacheBoFactory.phraseInvalidateBos( + cn.jojo.cc.common.enums.DataBaseTypeEnum.PROD, prodPhraseId), + com.alibaba.fastjson.serializer.SerializerFeature.WriteClassName)); } /** * 公共资源副本写入 PROD，返回 {@code jyResourceId -> prodResourceId}。 * --- a/application/src/main/java/cn/jojo/cc/biz/service/english/dict/EnglishDictWriteTxService.java +++ b/application/src/main/java/cn/jojo/cc/biz/service/english/dict/EnglishDictWriteTxService.java @@ -526,11 +526,11 @@ target.setId(existingChild.getId()); target.setCreatorId(existingChild.getCreatorId()); target.setCreateTime(existingChild.getCreateTime()); } target.setWordId(wordId); - target.setPos(item.getPos()); + target.setPos(PartOfSpeechEnum.normalizeKnownCodes(item.getPos())); target.setTranslation(item.getTranslation()); target.setSortOrder(item.getSortOrder() == null ? index : item.getSortOrder()); target.setDeleteTime(0L); target.fillRequired(operator, now, item.getId() == null); writes.add(target); --- a/application/src/main/java/cn/jojo/cc/biz/service/english/dict/EnglishImportFileReader.java +++ b/application/src/main/java/cn/jojo/cc/biz/service/english/dict/EnglishImportFileReader.java @@ -158,30 +158,36 @@ } List variants = new ArrayList<>(); List variantNormals = new ArrayList<>(); Set seenVariants = new LinkedHashSet<>(); + boolean invalidVariant = false; for (String raw : cell(record, parsed, COLUMN_VARIANTS).split(VARIANT_SEPARATOR)) { String variant = StringUtils.trimToEmpty(raw); String variantNormal = TextNormalizer.normalize(variant); if (StringUtils.isBlank(variantNormal)) { continue; } String variantIllegal = illegalReason(variant); if (variantIllegal != null || variantNormal.length() > MAX_TEXT_LENGTH) { result.getInvalid().add(invalid(record.rowNo, phrase, "变形非法：" + variant)); - continue; + invalidVariant = true; + break; } // 与词头同串的变形必须丢弃（come about 的过去式仍是 come about）： // 落进变形子表会撞 Spec §7.2「短语内部变形重复」的写前拒绝，整行导入失败。 if (variantNormal.equals(normalized) || !seenVariants.add(variantNormal)) { continue; } variants.add(variant); variantNormals.add(variantNormal); } + if (invalidVariant) { + firstSeen.remove(normalized); + continue; + } PhraseRow row = new PhraseRow(); row.setRowNo(record.rowNo); row.setPhrase(phrase); row.setPhraseNormalized(normalized); row.setTranslation(translation); @@ -318,51 +324,58 @@ private List readRecords(String content) { List records = new ArrayList<>(); List cells = new ArrayList<>(); StringBuilder field = new StringBuilder(); boolean inQuotes = false; + boolean closedQuote = false; boolean fieldStarted = false; int rowNo = 1; + int recordStart = 1; for (int i = 0; i < content.length(); i++) { char c = content.charAt(i); if (inQuotes) { if (c == '"') { if (i + 1 < content.length() && content.charAt(i + 1) == '"') { - // "" 转义：落一个引号并跳过配对引号 field.append('"'); i++; } else { inQuotes = false; + closedQuote = true; } } else { - if (c == '\n') { - rowNo++; - } + if (c == '\n') { rowNo++; } field.append(c); } continue; } - if (c == '"') { - inQuotes = true; - fieldStarted = true; - } else if (c == ',') { + if (c == ',') { cells.add(field.toString()); field.setLength(0); fieldStarted = false; - } else if (c == '\r') { - // CRLF：换行由 \n 统一计数 - } else if (c == '\n') { - addRecord(records, cells, field, fieldStarted, rowNo); + closedQuote = false; + } else if (c == '\r' || c == '\n') { + if (c == '\r' && i + 1 < content.length() && content.charAt(i + 1) == '\n') { i++; } + addRecord(records, cells, field, fieldStarted, recordStart); cells = new ArrayList<>(); fieldStarted = false; - rowNo++; + closedQuote = false; + recordStart = ++rowNo; + } else if (c == '"' && !fieldStarted && !closedQuote) { + inQuotes = true; + fieldStarted = true; } else { + if (c == '"' || closedQuote) { + throw BusinessException.paramException("CSV 第 " + recordStart + " 行引号位置非法"); + } field.append(c); fieldStarted = true; } } - addRecord(records, cells, field, fieldStarted, rowNo); + if (inQuotes) { + throw BusinessException.paramException("CSV 第 " + recordStart + " 行引号未闭合"); + } + addRecord(records, cells, field, fieldStarted, recordStart); return records; } /** * 收尾一条记录；引号外且无内容的空行直接跳过（清单尾部空行是常态）。 --- a/application/src/main/java/cn/jojo/cc/biz/service/english/dict/EnglishPhraseBizServiceImpl.java +++ b/application/src/main/java/cn/jojo/cc/biz/service/english/dict/EnglishPhraseBizServiceImpl.java @@ -124,11 +124,11 @@ @Override public IPage page(EnglishPhrasePageBo bo) { Page page = new Page<>(bo.getPageNum(), bo.getPageSize()); IPage doPage = englishPhraseCoreService.page(page, - StringUtils.trimToNull(bo.getPhrase()), bo.getStatus(), 0L); + TextNormalizer.normalize(bo.getPhrase()), bo.getStatus(), 0L); List records = assemble(doPage.getRecords()); Page result = new Page<>(doPage.getCurrent(), doPage.getSize(), doPage.getTotal()); result.setRecords(records); return result; } --- a/application/src/main/java/cn/jojo/cc/biz/service/english/dict/EnglishPublishBizService.java +++ b/application/src/main/java/cn/jojo/cc/biz/service/english/dict/EnglishPublishBizService.java @@ -9,10 +9,16 @@ * *

教研库到正式库的写入只允许经过本服务，Controller 不得直接调用发布 helper； * 自动补录（Task 21）、批量发布（Task 17/18）与全量刷新（Task 22）都复用这四个方法。

*/ public interface EnglishPublishBizService { + + /** 系统音频补发布：实体锁内重新确认无人工草稿。 */ + PublishResultBo publishAudioWord(Long wordId, LoginUserBo operator); + PublishResultBo publishAudioPhrase(Long phraseId, LoginUserBo operator); + /** 自动补录恢复：实体锁内确认仍是未编辑的补录草稿。 */ + PublishResultBo publishAutoFillWord(Long wordId, LoginUserBo operator); /** * 发布单个单词到正式库，并回刷引用它的待修复题目。 * * @param wordId 教研库单词 ID --- a/application/src/main/java/cn/jojo/cc/biz/service/english/dict/EnglishPublishBizServiceImpl.java +++ b/application/src/main/java/cn/jojo/cc/biz/service/english/dict/EnglishPublishBizServiceImpl.java @@ -102,10 +102,48 @@ // 锁内重读同 publishWord（问题12）；主表与变体都重新取，避免用锁外旧快照写 PROD。 () -> doPublishPhrase(requirePhrase(phraseId), englishPhraseVariantCoreService.listByPhraseId( phraseId, 0L), operator)); } + @Override + public PublishResultBo publishAudioWord(Long wordId, LoginUserBo operator) { + EnglishWordDo snapshot = requireWord(wordId); + return englishDictWriteLockService.callWithWordLock(wordId, snapshot.getWordNormalized(), () -> { + EnglishWordDo current = requireWord(wordId); + return EnglishPublishStatusEnum.PUBLISHED.getCode().equals(current.getStatus()) + && Integer.valueOf(1).equals(current.getHasPublished()) + ? doPublishWord(current, operator) : new PublishResultBo(); + }); + } + + @Override + public PublishResultBo publishAudioPhrase(Long phraseId, LoginUserBo operator) { + EnglishPhraseDo snapshot = requirePhrase(phraseId); + return englishDictWriteLockService.callWithPhraseLocks(phraseId, snapshot.getPhraseNormalized(), + variantTextsOf(englishPhraseVariantCoreService.listByPhraseId(phraseId, 0L)), () -> { + EnglishPhraseDo current = requirePhrase(phraseId); + return EnglishPublishStatusEnum.PUBLISHED.getCode().equals(current.getStatus()) + && Integer.valueOf(1).equals(current.getHasPublished()) + ? doPublishPhrase(current, englishPhraseVariantCoreService.listByPhraseId(phraseId, 0L), operator) + : new PublishResultBo(); + }); + } + + @Override + public PublishResultBo publishAutoFillWord(Long wordId, LoginUserBo operator) { + EnglishWordDo snapshot = requireWord(wordId); + return englishDictWriteLockService.callWithWordLock(wordId, snapshot.getWordNormalized(), () -> { + EnglishWordDo current = requireWord(wordId); + boolean eligible = cn.jojo.cc.common.enums.english.DictSourceEnum.SEGMENT_AUTOFILL.getCode() + .equals(current.getSource()) + && EnglishPublishStatusEnum.UNPUBLISHED.getCode().equals(current.getStatus()) + && !Integer.valueOf(1).equals(current.getHasPublished()) + && Objects.equals(current.getCreateTime(), current.getUpdateTime()); + return eligible ? doPublishWord(current, operator) : new PublishResultBo(); + }); + } + /** * 批次失败条目明细累加（问题08）：identifier 携实体类型+ID，rowNo 不适用批量发布场景留空。 */ private static void addBatchFailure(List failures, String type, Long entityId, String reason) { EnglishTaskFailureBo item = new EnglishTaskFailureBo(); @@ -150,11 +188,11 @@ result.setScannedCount(result.getScannedCount() + 1); result.setNextCursor(word.getId()); try { PublishResultBo single = englishDictWriteLockService.callWithWordLock(word.getId(), word.getWordNormalized(), () -> doPublishWord(requireWord(word.getId()), operator)); - if (single.isProdPublished()) { + if (single.isProdPublished() && !single.isCacheInvalidationPending()) { result.setSuccessCount(result.getSuccessCount() + 1); } else { result.setFailedCount(result.getFailedCount() + 1); // 未发布成功也要带原因上浮（问题08），否则任务侧只有计数没有条目。 addBatchFailure(result.getFailures(), "WORD", word.getId(), @@ -184,11 +222,11 @@ for (EnglishPhraseDo phrase : pending) { result.setScannedCount(result.getScannedCount() + 1); result.setNextCursor(phrase.getId()); try { PublishResultBo single = publishPhrase(phrase.getId(), operator); - if (single.isProdPublished()) { + if (single.isProdPublished() && !single.isCacheInvalidationPending()) { result.setSuccessCount(result.getSuccessCount() + 1); } else { result.setFailedCount(result.getFailedCount() + 1); addBatchFailure(result.getFailures(), "PHRASE", phrase.getId(), StringUtils.defaultIfBlank(single.getFailReason(), "PROD 写入未确认")); @@ -220,10 +258,12 @@ // 返回更新前 PROD 的真实旧文本（问题05）：改名重发布时旧文本映射必须一并失效， // 否则 cat 仍反查到本词条而详情已是 dog，读时解析会把错引用下发出去。 String oldProdNormalized = englishDictPublishProServiceHelper.publishWordInPro( jyWord, jyTranslations, jyResources, operator); result.setProdPublished(true); + result.setCacheInvalidationPending(true); + result.setFailReason("PROD 已提交，缓存失效由 outbox 恢复任务确认"); try { // 发布事实与当前状态分两个生命周期：条件回写只更新当前版本的状态， // 版本已被推进时单独登记「曾发布」，禁止把 v2 标成已发布，也不丢失历史发布事实（问题12）。 boolean confirmed = englishWordCoreService.updatePublishStatusIfUnmodified(jyWordId, EnglishPublishStatusEnum.PUBLISHED.getCode(), 1, snapshotUpdateTime, @@ -249,18 +289,11 @@ } log.error("单词发布后 JY 状态回写异常：PROD 已提交，JY 保持待发布、运营需重新发布一次, wordId={}", jyWordId, e); } - // PROD 失效由“PROD 已提交”这一事实驱动，不依赖状态回写正常返回（Spec §4.7）。 - try { - Long prodWordId = CommonService.decrementAndGet(jyWordId); - invalidateProdWordCache(prodWordId, oldProdNormalized, jyWord.getWordNormalized()); - } catch (Exception e) { - log.error("单词发布后 PROD 缓存失效失败，JY 状态不回写、运营重发一次即重新触发, wordId={}", - jyWordId, e); - } + // 失效事件已在 PROD 写入事务内持久化；恢复任务同步发送，失败不丢身份。 // 单词发布不再在请求线程内同步回刷 PROD 题目：词条发布后由下发时读时解析自然点亮，无待修复行； // 由 09 的定向回刷作业按 normalized_text 扫描消费（Spec §5.7），避免锁 lease 60s 内跑无上限循环。 return result; } @@ -273,10 +306,12 @@ List jyResources = loadAliveResources(collectAudioResourceIds( jyPhrase.getAudioResourceId(), jyVariants)); englishDictPublishProServiceHelper.publishPhraseInPro(jyPhrase, jyVariants, jyResources, operator); result.setProdPublished(true); + result.setCacheInvalidationPending(true); + result.setFailReason("PROD 已提交，缓存失效由 outbox 恢复任务确认"); try { boolean confirmed = englishPhraseCoreService.updatePublishStatusIfUnmodified(jyPhraseId, EnglishPublishStatusEnum.PUBLISHED.getCode(), 1, snapshotUpdateTime, operator.getLoginUserId()); result.setJyStatusConfirmed(confirmed); @@ -298,22 +333,11 @@ } log.error("短语发布后 JY 状态回写异常：PROD 已提交，JY 保持待发布、运营需重新发布一次, phraseId={}", jyPhraseId, e); } - // PROD 失效与全量重分词通知由“PROD 已提交”驱动，不依赖状态回写正常返回（Spec §4.7）： - // 短语索引不重建、脏标不置上，学生会继续点在旧短语上，且题目分词永不重算。 - try { - Long prodPhraseId = CommonService.decrementAndGet(jyPhraseId); - invalidateProdPhraseCache(prodPhraseId); - // PROD 短语索引只在正式库发布/删除后失效重载，教研库编辑不得触发（Spec §10.5）。 - englishPhraseIndexCacheService.rebuildNow(DataBaseTypeEnum.PROD, "english-phrase-publish"); - // 分词重算不再自动触发：全量重分词只在每日低峰跑一轮，需要提前生效时由运营走按短语重分词接口。 - } catch (Exception e) { - log.error("短语发布后 PROD 失效/重分词通知失败，JY 状态不回写、运营重发一次即重新触发, phraseId={}", - jyPhraseId, e); - } + // 短语失效及跨实例索引版本推进同样由持久化 outbox 恢复。 return result; } private void invalidateProdWordCache(Long prodWordId, String oldNormalizedWord, String newNormalizedWord) { rocketMqProducerService.clearRedisCommonCache(EnglishWordDo.class, prodWordId, --- a/application/src/main/java/cn/jojo/cc/biz/service/english/dict/EnglishWordBizServiceImpl.java +++ b/application/src/main/java/cn/jojo/cc/biz/service/english/dict/EnglishWordBizServiceImpl.java @@ -111,11 +111,11 @@ @Override public IPage page(EnglishWordPageBo bo) { Page page = new Page<>(bo.getPageNum(), bo.getPageSize()); IPage doPage = englishWordCoreService.page(page, - StringUtils.trimToNull(bo.getWord()), bo.getStatus(), bo.getCompleteness(), + TextNormalizer.normalize(bo.getWord()), bo.getStatus(), bo.getCompleteness(), 0L); List records = assemble(doPage.getRecords()); Page result = new Page<>(doPage.getCurrent(), doPage.getSize(), doPage.getTotal()); result.setRecords(records); return result; --- a/application/src/main/java/cn/jojo/cc/biz/service/english/dict/SegmentAutoFillBizServiceImpl.java +++ b/application/src/main/java/cn/jojo/cc/biz/service/english/dict/SegmentAutoFillBizServiceImpl.java @@ -62,10 +62,13 @@ @Resource private EnglishDictConfig englishDictConfig; @Resource + private cn.jojo.cc.core.service.english.EnglishWordCoreService englishWordCoreService; + + @Resource private BaiduDictClient baiduDictClient; @Resource private EnglishDictWriteLockService englishDictWriteLockService; @@ -136,22 +139,19 @@ return false; } } private void autoFillOne(String normalizedWord, LoginUserBo operator) { - // 1. 百度：异常/空结果统一按空壳口径（§11.4）。 - BaiduDictResult dict = queryBaiduSafely(normalizedWord); - // 2. 音频不在补录链路里生成：建行时留空，由音频回填任务按 source=分词自动补录 统一提交。 - // 一条消息里有多个词时，这里不再逐个等合成，避免把整条消息拖成小时级。 - - // 3. 落库：在单词写入锁内完成，与人工保存/删除互斥（05 冻结口径）。 - // entityId 传 null：补录以新建为主，锁内才能幂等复用旧行，取锁时拿不到稳定实体 ID； - // 与“同一实体人工编辑”交错是已接受的窄窗口（低频，事务+条件回写兜底）。 - EnglishWordDo saved = englishDictWriteLockService.callWithWordLock(null, normalizedWord, - () -> englishDictWriteTxService.saveWordForAutoFill( - normalizedWord, dict == null ? null : dict.getPhonetic(), null, - mapTranslations(dict), operator)); + // 文本锁内先认账（含人工删除行）；已有行只走幂等恢复，不再调用百度。 + EnglishWordDo saved = englishDictWriteLockService.callWithWordLock(null, normalizedWord, () -> { + EnglishWordDo existing = englishWordCoreService.getByNormalized(normalizedWord, 0L); + if (existing == null) { existing = englishWordCoreService.getByNormalizedIgnoreDeleted(normalizedWord); } + if (existing != null) { return existing; } + BaiduDictResult dict = queryBaiduSafely(normalizedWord); + return englishDictWriteTxService.saveWordForAutoFill(normalizedWord, + dict == null ? null : dict.getPhonetic(), null, mapTranslations(dict), operator); + }); if (isSoftDeleted(saved)) { // 运营删过这个词（第四轮复审问题12 的姊妹缺口）：锁内已按“含已删除行”认过账，既不重建也不发布， // 也不做完整性提示——那一行已经不在词典里了，再喊“等业务方补全”只会误导。 log.info("单词已被运营删除，本次不重建也不发布, word={}, wordId={}", normalizedWord, saved.getId()); return; @@ -181,22 +181,22 @@ && Objects.equals(saved.getCreateTime(), saved.getUpdateTime()); if (needPublish) { rocketMqProducerService.clearRedisCommonCache(EnglishWordDo.class, saved.getId(), EnglishDictCacheBoFactory.wordInvalidateBos(DataBaseTypeEnum.TEACHING, saved.getId(), null, saved.getWordNormalized())); - englishPublishBizService.publishWord(saved.getId(), operator); + englishPublishBizService.publishAutoFillWord(saved.getId(), operator); } log.info("自动补录单词完成, word={}, wordId={}, hasPublished={}, operatorId={}", normalizedWord, saved.getId(), saved.getHasPublished(), operator == null ? null : operator.getLoginUserId()); // PRD（题保存分词+下游对接）：必填项有缺失也落库并立即发布，靠日志告知待人工完善； // 用户裁决 2026-10-27：本期只打日志，不接钉钉通知。判定只看结果行完整性： // 幂等复用到的旧行若仍是残缺态，同样属于"等业务方补全"的待办，重复提示是可接受的。 if (WordCompletenessEnum.INCOMPLETE.getCode().equals(saved.getCompleteness())) { log.warn("自动补录词条完整性有缺失，需产品联系业务方手动完善, word={}, wordId={}, phonetic={}, audio={}, translation={}", normalizedWord, saved.getId(), saved.getPhonetic(), saved.getAudioResourceId(), - dict == null ? null : dict.getTranslations()); + saved.getCompleteness()); } // 第 5 段“双库定向回刷”已整块删除：库里只存划词块的文本与类型（见 SegmentSnapshot）， // ID 在下发时按当前词典现算，词补出来后存量题目自动可点，不需要按词 × 按题重跑分词。 } --- a/application/src/main/java/cn/jojo/cc/biz/service/english/dict/impl/EnglishAudioBackfillBizServiceImpl.java +++ b/application/src/main/java/cn/jojo/cc/biz/service/english/dict/impl/EnglishAudioBackfillBizServiceImpl.java @@ -61,10 +61,13 @@ * 回填后补发布用的系统操作人：音频是机器回填的，发布也必须以系统身份落库。 */ private static final LoginUserBo SYSTEM_OPERATOR = new LoginUserBo(-1L, "admin"); @Resource + private cn.jojo.cc.biz.service.english.dict.EnglishAudioCompletionTxService englishAudioCompletionTxService; + + @Resource private EnglishDictConfig englishDictConfig; @Resource private InternalTtsClient internalTtsClient; @@ -98,22 +101,43 @@ return result; } private void submitPending(int batch, EnglishAudioBackfillResultBo result) { long createdBefore = nowSeconds() - SUBMIT_COOLDOWN_SECONDS; - for (EnglishWordDo row : englishWordCoreService.listAudioPendingSubmit(batch, createdBefore)) { + for (EnglishWordDo row : pendingWords(batch, createdBefore)) { submit(ROW_WORD, row.getId(), row.getWord(), result); } - for (EnglishPhraseDo row : englishPhraseCoreService.listAudioPendingSubmit(batch, createdBefore)) { + for (EnglishPhraseDo row : pendingPhrases(batch, createdBefore)) { submit(ROW_PHRASE, row.getId(), row.getPhrase(), result); } submitPendingVariants(batch, createdBefore, result); + } + + private List pendingWords(int batch, long before) { + RAtomicLong cursor = redissonClient.getAtomicLong(failKey("cursor:words")); + List rows = englishWordCoreService.listAudioPendingSubmitAfterId(batch, before, cursor.get()); + cursor.set(rows.size() < batch ? 0L : rows.get(rows.size() - 1).getId()); + return rows; + } + + private List pendingPhrases(int batch, long before) { + RAtomicLong cursor = redissonClient.getAtomicLong(failKey("cursor:phrases")); + List rows = englishPhraseCoreService.listAudioPendingSubmitAfterId(batch, before, cursor.get()); + cursor.set(rows.size() < batch ? 0L : rows.get(rows.size() - 1).getId()); + return rows; + } + + private List pendingVariants(int batch, long before) { + RAtomicLong cursor = redissonClient.getAtomicLong(failKey("cursor:variants")); + List rows = englishPhraseVariantCoreService.listAudioPendingSubmitAfterId(batch, before, cursor.get()); + cursor.set(rows.size() < batch ? 0L : rows.get(rows.size() - 1).getId()); + return rows; } private void submitPendingVariants(int batch, long createdBefore, EnglishAudioBackfillResultBo result) { List variants = - englishPhraseVariantCoreService.listAudioPendingSubmit(batch, createdBefore); + pendingVariants(batch, createdBefore); if (variants.isEmpty()) { return; } Set phraseIds = variants.stream() .map(EnglishPhraseVariantDo::getPhraseId) @@ -148,11 +172,10 @@ try { TtsRequest request = new TtsRequest(); request.setText(text); String taskId = internalTtsClient.submitTask(request); markSubmitted(table, id, taskId); - clearFailCount(rowKey); result.setSubmitted(result.getSubmitted() + 1); } catch (Exception e) { increaseFailCount(rowKey); result.setSubmitFailed(result.getSubmitFailed() + 1); log.warn("音频任务提交失败, row={}, text={}", rowKey, text, e); @@ -193,13 +216,13 @@ * 而这些词已经发布过，运营不会再去点一次。只给已发布行补，未发布行不自动发布（发布是人工闸门）。 */ private void publishReleased(String table, Long id) { try { if (ROW_WORD.equals(table)) { - englishPublishBizService.publishWord(id, SYSTEM_OPERATOR); + englishPublishBizService.publishAudioWord(id, SYSTEM_OPERATOR); } else if (ROW_PHRASE.equals(table)) { - englishPublishBizService.publishPhrase(id, SYSTEM_OPERATOR); + englishPublishBizService.publishAudioPhrase(id, SYSTEM_OPERATOR); } } catch (Exception e) { // 补发布失败不回滚音频回填：音频已经在教研库，运营下次发布或再跑一轮仍会带上； // 这里只留痕，避免因为一次锁竞争把已成功的回填也标记成失败 log.warn("音频回填后补发布失败, row={}:{}", table, id, e); @@ -241,10 +264,21 @@ EnglishAudioBackfillResultBo result) { if (id == null || StringUtils.isBlank(taskId)) { return false; } result.setScanned(result.getScanned() + 1); + String rowKey = table + ":" + id; + long elapsed = submitTime == null ? Long.MAX_VALUE : Math.max(0L, now - submitTime); + if (elapsed >= timeoutSeconds) { + // 总生命周期预算适用于查询异常、completed 下载失败等全部分支；取消失败也必须本地收口。 + try { internalTtsClient.cancelTask(taskId); } + catch (Exception e) { log.warn("取消超时音频任务失败, taskId={}", taskId, e); } + increaseFailCount(rowKey); + markCleared(table, id); + result.setFailed(result.getFailed() + 1); + return false; + } TtsTaskStatus status; try { status = internalTtsClient.queryTask(taskId); } catch (Exception e) { // 查状态失败按「还没好」处理：下轮再查，不在这里清任务号 @@ -252,27 +286,19 @@ result.setRunning(result.getRunning() + 1); return false; } if (status == null || status.isRunning()) { - long elapsed = submitTime == null ? Long.MAX_VALUE : now - submitTime; - if (elapsed >= timeoutSeconds) { - // 平台丢任务或长时间不出结果：先取消平台任务，再清本地在途标记，否则该行永远挂着「生成中」 - internalTtsClient.cancelTask(taskId); - markCleared(table, id); - result.setFailed(result.getFailed() + 1); - log.warn("音频任务超时清理, row={}:{}, taskId={}, elapsed={}s", table, id, taskId, elapsed); - } else { - result.setRunning(result.getRunning() + 1); - } + result.setRunning(result.getRunning() + 1); return false; } if (status.isCompleted()) { // 人工优先：在途期间运营可能已上传了音频，这时只清任务号，既不下载也不覆盖 if (currentAudioResourceId != null && currentAudioResourceId > 0) { markCleared(table, id); + clearFailCount(rowKey); result.setCompleted(result.getCompleted() + 1); log.info("音频任务完成但该行已有音频，保留人工音频并清任务号, row={}:{}", table, id); return true; } Long resourceId = englishAudioBizService.saveAudio(taskId); @@ -280,21 +306,23 @@ if (!markCompleted(table, id, resourceId)) { // 条件更新未命中：写完前的一瞬间人工补了音频（并发窗口），同样只清任务号，不回写 markCleared(table, id); log.info("音频回填时发现该行已被人工补齐，保留人工音频, row={}:{}", table, id); } + clearFailCount(rowKey); result.setCompleted(result.getCompleted() + 1); return true; } // 下载或落库失败（未就绪/源文件过期/校验失败）：保留任务号，由超时清理最终收口， // 不在这里清掉——清掉就意味着这个词再也不会自动补音频 result.setFailed(result.getFailed() + 1); log.warn("音频下载或保存失败，保留任务号待下轮, row={}:{}, taskId={}", table, id, taskId); return false; } - // failed / cancelled：终态失败，清掉在途标记，音频留空（运营仍可手工上传） + // 整个尝试失败才累计，成功 submit 不重置计数。 + increaseFailCount(rowKey); markCleared(table, id); result.setFailed(result.getFailed() + 1); log.warn("音频任务终态失败, row={}:{}, taskId={}, status={}, error={}", table, id, taskId, status.getStatus(), status.getError()); return false; @@ -320,17 +348,11 @@ englishPhraseVariantCoreService.markAudioTaskCleared(id); } } private boolean markCompleted(String table, Long id, Long resourceId) { - if (ROW_WORD.equals(table)) { - return englishWordCoreService.markAudioCompleted(id, resourceId); - } - if (ROW_PHRASE.equals(table)) { - return englishPhraseCoreService.markAudioCompleted(id, resourceId); - } - return englishPhraseVariantCoreService.markAudioCompleted(id, resourceId); + return englishAudioCompletionTxService.complete(table, id, resourceId); } private boolean isAutoSource(Integer source) { return DictSourceEnum.IMPORT.getCode().equals(source) || DictSourceEnum.SEGMENT_AUTOFILL.getCode().equals(source); --- /dev/null +++ b/application/src/main/java/cn/jojo/cc/biz/service/english/segment/EnglishSegmentSkeletonBizService.java @@ -0,0 +1,35 @@ +package cn.jojo.cc.biz.service.english.segment; + +import cn.jojo.cc.biz.service.component.CommonService; +import cn.jojo.cc.common.enums.DataSourceEnum; +import cn.jojo.cc.common.enums.english.SegmentContentTypeEnum; +import cn.jojo.cc.core.domain.english.QuestionSegmentDo; +import cn.jojo.cc.core.domain.question.library.QuestionDo; +import cn.jojo.cc.core.multidatasource.MultiDataSourceUtil; +import cn.jojo.cc.core.service.question.library.QuestionCoreService; +import jakarta.annotation.Resource; +import org.springframework.stereotype.Service; + +/** 首次发布不依赖 JY 已算成功：先保留 JY 身份，再显式映射 PROD ID。 */ +@Service +public class EnglishSegmentSkeletonBizService { + @Resource private QuestionCoreService questions; + @Resource private QuestionSegmentStateTxService state; + + public void ensurePair(Long teachingId, Long prodId) { + QuestionDo prod = MultiDataSourceUtil.switchThen(DataSourceEnum.PROD, () -> questions.getById(prodId)); + if (prod == null) { return; } + Integer type = SegmentContentTypeEnum.contentTypeOfQuestionType(prod.getQuestionType()); + if (type == null) { return; } + QuestionSegmentDo identity = MultiDataSourceUtil.switchThen(DataSourceEnum.DEFAULT, + () -> state.reserveTeachingIdentity(teachingId, type)); + if (identity == null || identity.getId() == null) { + throw new IllegalStateException("缺少教研题目，禁止在 PROD 自增创建分词身份"); + } + Long mappedId = CommonService.decrementAndGet(identity.getId()); + MultiDataSourceUtil.switchThen(DataSourceEnum.PROD, () -> { + state.installProdSkeleton(prodId, mappedId, type); + return null; + }); + } +} --- a/application/src/main/java/cn/jojo/cc/biz/service/english/segment/QuestionSegmentBizServiceImpl.java +++ b/application/src/main/java/cn/jojo/cc/biz/service/english/segment/QuestionSegmentBizServiceImpl.java @@ -20,10 +20,11 @@ import com.alibaba.fastjson.JSON; import jakarta.annotation.Resource; import lombok.extern.slf4j.Slf4j; import org.apache.commons.lang3.StringUtils; import org.springframework.stereotype.Service; +import org.springframework.transaction.annotation.Transactional; import java.util.ArrayList; import java.util.Collections; import java.util.LinkedHashSet; import java.util.List; @@ -198,32 +199,41 @@ result.setQuestionRevision(question.getUpdateTime()); return result; } @Override + @Transactional(rollbackFor = Exception.class) public SegmentResult segmentAndSave(QuestionDo question, DataBaseTypeEnum segmentSource, LoginUserBo operator) { return doSegmentAndSave(question, segmentSource, operator, false); } @Override + @Transactional(rollbackFor = Exception.class) public SegmentResult refreshOne(QuestionDo question, DataBaseTypeEnum segmentSource, LoginUserBo operator) { return doSegmentAndSave(question, segmentSource, operator, true); } @Override + @Transactional(rollbackFor = Exception.class) public SegmentResult refreshOneAllowSkip(QuestionDo question, DataBaseTypeEnum segmentSource, LoginUserBo operator) { return doSegmentAndSave(question, segmentSource, operator, false); } private SegmentResult doSegmentAndSave(QuestionDo question, DataBaseTypeEnum segmentSource, LoginUserBo operator, boolean force) { if (question == null || question.getId() == null) { return new SegmentResult(); } + // 锁真实题目行，随后只处理锁内最新版本。跨内容位写入/删除和空结果清理共享这一事务。 + QuestionDo latest = questionSegmentCoreService.lockQuestion(question.getId()); + if (latest == null || !Objects.equals(latest.getUpdateTime(), question.getUpdateTime())) { + return new SegmentResult(); + } + question = latest; if (!englishDictConfig.isSegmentExecutionEnabled()) { // 急停：不抽正文、不装索引、不查词典、不写行，也不登记失败行（暂停不是失败）。 log.warn("英文分词执行已暂停，跳过单题计算, questionId={}", question.getId()); return new SegmentResult(); } @@ -238,20 +248,23 @@ textsByType = extractPlainTextsByContentType(question); sourceTextMd5 = MD5Util.creatMd5(flatten(textsByType)); // 一次读出该题全部内容位行：剪枝、修订判定与拒绝分支都要按内容位各看自己那一行。 Map existingByType = toMapByType(questionSegmentCoreService.listByQuestionId(question.getId())); + for (QuestionSegmentDo existing : existingByType.values()) { + if (isStale(existing, question.getUpdateTime())) { return new SegmentResult(); } + } if (!force && canSkipAll(existingByType, textsByType.keySet(), question, sourceTextMd5)) { // 剪枝命中：库里已有同正文、同修订、同内容位集合的成功结果。直接回显库存，绝不重算—— // 重算会用“当前词典”得出与库存不一致的 segments，而本分支又不落库（词典变更靠每日全量刷传播）。 reuseStoredResult = true; return mergeStored(existingByType); } // 句柄必须在扫描开始前取定：既保证“整个动作只用一张表”，也避免拿着空表白扫一轮再判能不能落库。 // 旧写法是先扫完、再用全局开关判，而开关会在扫描期间被其它线程翻转， // 于是“用空表切出的结果”恰好能过库（错块以 NORMAL 落库后被剪枝永久固化）。 - PhraseIndexSnapshot snapshot = phraseIndexCacheService.readySnapshot(segmentSource); + PhraseIndexSnapshot snapshot = phraseIndexCacheService.freshSnapshotForWrite(segmentSource); if (snapshot == null) { // 索引未就绪（冷启动装载失败、处于退避窗口内）时短语必然全漏：落库会把缺失结果固化成 NORMAL 行， // 并把短语的成分词写进未命中表污染自动补录，所以本次不落新结果，等下次保存重算。 // 这里有意返回空结果而不是扫描产物：不然挂接门面会拿这批“假未命中词”去投自动补录事件。 log.warn("短语索引未就绪，本次跳过分词, questionId={}, source={}", --- a/application/src/main/java/cn/jojo/cc/biz/service/english/segment/QuestionSegmentSaveHookService.java +++ b/application/src/main/java/cn/jojo/cc/biz/service/english/segment/QuestionSegmentSaveHookService.java @@ -37,10 +37,13 @@ @Service @Slf4j public class QuestionSegmentSaveHookService { @Resource + private QuestionSegmentStateTxService questionSegmentStateTxService; + + @Resource private EnglishDictConfig englishDictConfig; @Autowired private AliyunRocketMQService rocketMqProducerService; @@ -60,11 +63,12 @@ return; } if (question == null || question.getId() == null) { return; } - // 发布本事件本身不开事务、不写库；监听器内部兜住全部异常， + questionSegmentStateTxService.invalidateAfterSave(question.getId()); + // 同库事务已把旧结果降为 INIT，随后才发提交后事件；监听器内部兜住全部异常， // 不会再有带 @Transactional 的 bean 抛异常污染本次同步事务。 applicationEventPublisher.publishEvent(new QuestionSegmentSaveRequestedEvent(this, question, operator)); } /** --- /dev/null +++ b/application/src/main/java/cn/jojo/cc/biz/service/english/segment/QuestionSegmentStateTxService.java @@ -0,0 +1,53 @@ +package cn.jojo.cc.biz.service.english.segment; + +import cn.jojo.cc.common.enums.english.SegmentContentTypeEnum; +import cn.jojo.cc.core.domain.english.QuestionSegmentDo; +import cn.jojo.cc.core.domain.question.library.QuestionDo; +import cn.jojo.cc.core.service.english.QuestionSegmentCoreService; +import jakarta.annotation.Resource; +import org.springframework.stereotype.Service; +import org.springframework.transaction.annotation.Transactional; + +/** 同库正文锁与派生状态事务；调用方必须先选定数据源。 */ +@Service +public class QuestionSegmentStateTxService { + @Resource private QuestionSegmentCoreService segments; + + @Transactional(rollbackFor = Exception.class) + public void invalidateAfterSave(Long id) { + QuestionDo current = segments.lockQuestion(id); + if (current == null) { return; } + Integer type = SegmentContentTypeEnum.contentTypeOfQuestionType(current.getQuestionType()); + if (type != null) { + segments.reserveIdentity(id, current.getQuestionKey(), type, current.getUpdateTime()); + } + segments.invalidateQuestion(id, current.getUpdateTime()); + } + + @Transactional(rollbackFor = Exception.class) + public QuestionSegmentDo reserveTeachingIdentity(Long id, Integer type) { + QuestionDo current = segments.lockQuestion(id); + if (current == null) { return null; } + return segments.reserveIdentity(id, current.getQuestionKey(), type, current.getUpdateTime()); + } + + @Transactional(rollbackFor = Exception.class) + public void installProdSkeleton(Long prodId, Long mappedId, Integer type) { + QuestionDo current = segments.lockQuestion(prodId); + if (current == null) { return; } + QuestionSegmentDo row = new QuestionSegmentDo(); + row.setId(mappedId); + row.setQuestionId(prodId); + row.setQuestionKey(current.getQuestionKey()); + row.setContentType(type); + row.setQuestionRevision(current.getUpdateTime()); + segments.insertSkeletonIfAbsent(row); + // 重新发布后正文/修订可能改变；先降级，消息迟到前不下发旧 NORMAL。 + for (QuestionSegmentDo old : segments.listByQuestionId(prodId)) { + if (!java.util.Objects.equals(old.getQuestionRevision(), current.getUpdateTime())) { + segments.invalidateQuestion(prodId, current.getUpdateTime()); + break; + } + } + } +} --- a/application/src/main/java/cn/jojo/cc/biz/service/english/segment/SegmentRefreshBizServiceImpl.java +++ b/application/src/main/java/cn/jojo/cc/biz/service/english/segment/SegmentRefreshBizServiceImpl.java @@ -48,10 +48,13 @@ public class SegmentRefreshBizServiceImpl implements SegmentRefreshBizService { private static final int PAGE_SIZE = 500; @Resource + private EnglishSegmentSkeletonBizService englishSegmentSkeletonBizService; + + @Resource private QuestionSegmentBizService questionSegmentBizService; @Resource private QuestionCoreService questionCoreService; @@ -187,16 +190,15 @@ */ private boolean refreshQuestion(QuestionDo question, DataBaseTypeEnum env, LoginUserBo operator, boolean force) { if (env == DataBaseTypeEnum.PROD && CollectionUtils.isEmpty(questionSegmentCoreService.listByQuestionId(question.getId()))) { - // 发布前初始化/历史遗留的缺行：建行归题目发布链路（decrement 主键映射，包10）， - // 刷新在此自增会造出与 JY 不对应的主键（Spec §4.4）。JY 本轮已先刷成最新， - // 题目下次发布即携带；不算失败、不重标，避免每轮空转刷全库。 - log.warn("题目在 PROD 无分词行，跳过全量刷，待题目发布时随链路同步, questionId={}, questionKey={}", - question.getId(), question.getQuestionKey()); - return false; + englishSegmentSkeletonBizService.ensurePair( + cn.jojo.cc.biz.service.component.CommonService.incrementAndGet(question.getId()), question.getId()); + if (CollectionUtils.isEmpty(questionSegmentCoreService.listByQuestionId(question.getId()))) { + return false; + } } if (force) { questionSegmentBizService.refreshOne(question, env, operator); } else { questionSegmentBizService.refreshOneAllowSkip(question, env, operator); --- a/application/src/main/java/cn/jojo/cc/biz/service/impl/question/library/QuestionBizServiceImpl.java +++ b/application/src/main/java/cn/jojo/cc/biz/service/impl/question/library/QuestionBizServiceImpl.java @@ -359,10 +359,13 @@ @Slf4j public class QuestionBizServiceImpl implements QuestionBizService { @Resource private QuestionCoreService questionCoreService; + @Resource + private cn.jojo.cc.biz.service.english.segment.EnglishSegmentSkeletonBizService englishSegmentSkeletonBizService; + /** * 保存后分词与补录挂接（Task 21）：异常由门面内部兜住，不阻断保存主链路。 */ @Resource private QuestionSegmentSaveHookService questionSegmentSaveHookService; @@ -2092,41 +2095,11 @@ } try { if (!englishDictConfig.isSegmentEnabled()) { return; } - List jyRows = buildPublishableSegments(questionId); - if (CollectionUtils.isEmpty(jyRows)) { - return; - } - MultiDataSourceUtil.switchThen(DataSourceEnum.PROD, () -> { - Map existing = new HashMap<>(); - for (QuestionSegmentDo row : questionSegmentCoreService.listByQuestionId(prodQuestionId)) { - if (row != null && row.getContentType() != null) { - existing.putIfAbsent(row.getContentType(), row); - } - } - QuestionDo prodQuestion = questionCoreService.getById(prodQuestionId); - long prodRevision = prodQuestion == null || prodQuestion.getUpdateTime() == null - ? 0L : prodQuestion.getUpdateTime(); - for (QuestionSegmentDo jyRow : jyRows) { - if (jyRow == null || jyRow.getId() == null || jyRow.getContentType() == null - || existing.containsKey(jyRow.getContentType())) { - continue; - } - QuestionSegmentDo skeleton = new QuestionSegmentDo(); - // 主键按跨库映射规则换算；正式库数据源不走教研库的 ID 生成拦截器，必须显式指定。 - skeleton.setId(CommonService.decrementAndGet(jyRow.getId())); - skeleton.setQuestionId(prodQuestionId); - skeleton.setQuestionKey(jyRow.getQuestionKey()); - skeleton.setContentType(jyRow.getContentType()); - skeleton.setQuestionRevision(prodRevision); - skeleton.setSegmentStatus(SegmentStatusEnum.INIT.getCode()); - questionSegmentCoreService.insertSkeletonIfAbsent(skeleton); - } - return null; - }); + englishSegmentSkeletonBizService.ensurePair(questionId, prodQuestionId); } catch (Exception e) { log.error("发布后补建正式库分词骨架失败（不影响发布）, questionId={}, prodQuestionId={}", questionId, prodQuestionId, e); } } --- /dev/null +++ b/application/src/test/java/cn/jojo/cc/biz/service/english/dict/EnglishAudioPublishGuardRegressionTest.java @@ -0,0 +1,28 @@ +package cn.jojo.cc.biz.service.english.dict; + +import cn.jojo.cc.core.domain.english.EnglishWordDo; +import cn.jojo.cc.core.service.english.EnglishWordCoreService; +import org.junit.jupiter.api.Test; +import org.springframework.test.util.ReflectionTestUtils; +import java.util.function.Supplier; +import static org.junit.jupiter.api.Assertions.*; +import static org.mockito.Mockito.*; +import static org.mockito.ArgumentMatchers.*; + +class EnglishAudioPublishGuardRegressionTest { + @Test void draftSavedBeforeEntityLockIsAcquiredCannotBeAutoPublished() { + var service = new EnglishPublishBizServiceImpl(); + var words = mock(EnglishWordCoreService.class); + var locks = mock(EnglishDictWriteLockService.class); + var helper = mock(EnglishDictPublishProServiceHelper.class); + var old = new EnglishWordDo(); old.setId(1L); old.setWordNormalized("cat"); old.setStatus(1); old.setHasPublished(1); + var draft = new EnglishWordDo(); draft.setId(1L); draft.setWordNormalized("cat"); draft.setStatus(0); draft.setHasPublished(1); + when(words.getById(1L, 0L)).thenReturn(old, draft); + when(locks.callWithWordLock(eq(1L), eq("cat"), any())).thenAnswer(i -> ((Supplier) i.getArgument(2)).get()); + ReflectionTestUtils.setField(service, "englishWordCoreService", words); + ReflectionTestUtils.setField(service, "englishDictWriteLockService", locks); + ReflectionTestUtils.setField(service, "englishDictPublishProServiceHelper", helper); + assertFalse(service.publishAudioWord(1L, null).isProdPublished()); + verifyNoInteractions(helper); + } +} --- /dev/null +++ b/application/src/test/java/cn/jojo/cc/biz/service/english/dict/EnglishImportBoundaryRegressionTest.java @@ -0,0 +1,41 @@ +package cn.jojo.cc.biz.service.english.dict; + +import org.junit.jupiter.api.Test; +import java.nio.charset.StandardCharsets; +import java.lang.reflect.Method; +import static org.junit.jupiter.api.Assertions.*; + +class EnglishImportBoundaryRegressionTest { + private final EnglishImportFileReader reader = new EnglishImportFileReader(); + + @Test void rejectsUnclosedQuoteBeforeAnyRowCanBeWritten() { + assertThrows(RuntimeException.class, () -> reader.readPhrases( + "phrase,translation\nlook up,\"查找\nget up,起床\n".getBytes(StandardCharsets.UTF_8))); + } + + @Test void rejectsMalformedQuotePlacement() { + for (String csv : new String[]{"phrase,translation\nlo\"ok,查找", "phrase,translation\n\"look\"x,查找"}) { + assertThrows(RuntimeException.class, () -> reader.readPhrases(csv.getBytes(StandardCharsets.UTF_8))); + } + } + + @Test void invalidVariantRejectsEntireRowAndAllowsCorrectedSameKey() { + var rows = reader.readPhrases(("phrase,translation,variants\nget up,起床,got up|起床\n" + + "get up,起床,got up|gotten up\n").getBytes(StandardCharsets.UTF_8)); + assertEquals(1, rows.getRows().size()); + assertEquals(2, rows.getRows().get(0).getVariants().size()); + assertEquals(1, rows.getInvalid().size()); + } + + @Test void onlyOmittedLimitMeansUnlimited() throws Exception { + Class type = Class.forName(EnglishDictImportBizServiceImpl.class.getName() + "$Param"); + Method parse = type.getDeclaredMethod("parse", String.class); + parse.setAccessible(true); + for (String value : new String[]{"0", "-1", "1.2", "2147483648", "null", "\"2\""}) { + assertThrows(java.lang.reflect.InvocationTargetException.class, + () -> parse.invoke(null, "{\"fileUri\":\"https://example.invalid/test.csv\",\"limit\":" + value + "}")); + } + assertNotNull(parse.invoke(null, "{\"fileUri\":\"https://example.invalid/test.csv\"}")); + assertNotNull(parse.invoke(null, "{\"fileUri\":\"https://example.invalid/test.csv\",\"limit\":2}")); + } +} --- /dev/null +++ b/application/src/test/java/cn/jojo/cc/biz/service/english/dict/impl/EnglishAudioLifecycleRegressionTest.java @@ -0,0 +1,35 @@ +package cn.jojo.cc.biz.service.english.dict.impl; + +import cn.jojo.cc.adaptor.english.InternalTtsClient; +import cn.jojo.cc.biz.model.bo.english.EnglishAudioBackfillResultBo; +import cn.jojo.cc.core.service.english.EnglishWordCoreService; +import org.junit.jupiter.api.Test; +import org.redisson.api.RedissonClient; +import org.redisson.api.RAtomicLong; +import org.springframework.test.util.ReflectionTestUtils; +import static org.junit.jupiter.api.Assertions.*; +import static org.mockito.Mockito.*; +import static org.mockito.ArgumentMatchers.*; + +class EnglishAudioLifecycleRegressionTest { + @Test void totalTimeoutClearsTaskEvenWhenCancelThrows() { + var service = new EnglishAudioBackfillBizServiceImpl(); + var tts = mock(InternalTtsClient.class); + var words = mock(EnglishWordCoreService.class); + var redis = mock(RedissonClient.class); + var counter = mock(RAtomicLong.class); + when(redis.getAtomicLong(anyString())).thenReturn(counter); + doThrow(new IllegalStateException("cancel down")).when(tts).cancelTask("task"); + ReflectionTestUtils.setField(service, "internalTtsClient", tts); + ReflectionTestUtils.setField(service, "englishWordCoreService", words); + ReflectionTestUtils.setField(service, "redissonClient", redis); + var result = new EnglishAudioBackfillResultBo(); + Boolean completed = ReflectionTestUtils.invokeMethod(service, "handleInFlight", "word", 1L, + "task", 1L, null, 10000L, 1800L, result); + assertEquals(false, completed); + assertEquals(1, result.getFailed()); + verify(words).markAudioTaskCleared(1L); + verify(counter).incrementAndGet(); + verify(tts, never()).queryTask(anyString()); + } +} --- /dev/null +++ b/application/src/test/java/cn/jojo/cc/biz/service/english/segment/QuestionSegmentRevisionRegressionTest.java @@ -0,0 +1,24 @@ +package cn.jojo.cc.biz.service.english.segment; + +import cn.jojo.cc.core.domain.question.library.QuestionDo; +import cn.jojo.cc.core.service.english.QuestionSegmentCoreService; +import cn.jojo.cc.common.enums.DataBaseTypeEnum; +import org.junit.jupiter.api.Test; +import org.springframework.test.util.ReflectionTestUtils; +import static org.mockito.Mockito.*; +import static org.mockito.ArgumentMatchers.*; + +class QuestionSegmentRevisionRegressionTest { + @Test void newerRealQuestionRejectsEntireOldTaskBeforeAnyContentTypeWriteOrDelete() { + var service = new QuestionSegmentBizServiceImpl(); + var core = mock(QuestionSegmentCoreService.class); + var latest = new QuestionDo(); latest.setId(1L); latest.setUpdateTime(2L); + var stale = new QuestionDo(); stale.setId(1L); stale.setUpdateTime(1L); + when(core.lockQuestion(1L)).thenReturn(latest); + ReflectionTestUtils.setField(service, "questionSegmentCoreService", core); + service.refreshOne(stale, DataBaseTypeEnum.TEACHING, null); + verify(core, never()).upsert(any()); + verify(core, never()).softDeleteByDo(any()); + verify(core, never()).markFailedByQuestionId(any(), any(), any(), any(), any(), any()); + } +} --- a/common/src/main/java/cn/jojo/cc/common/utils/EnglishWords.java +++ b/common/src/main/java/cn/jojo/cc/common/utils/EnglishWords.java @@ -27,11 +27,11 @@ /** * 判断可组成单词的 ASCII 字符（不含连接符：hyphen 与撇号）。 */ public static boolean isWordChar(char c) { - return isAsciiLetter(c) || isAsciiDigit(c); + return isAsciiLetter(c); } /** * 判断字符是否为词内连接符：hyphen 与撇号，两者规则完全一致。 */ @@ -115,12 +115,19 @@ if (text == null || cursor < 0) { return new Token("", Math.max(cursor, 0), Math.max(cursor, 0), false); } int length = text.length(); int current = cursor; - while (current < length && !isTokenStart(text, current)) { - current++; + while (current < length) { + int phoneticEnd = phoneticSpanEnd(text, current); + if (phoneticEnd > current) { + current = phoneticEnd; + } else if (!isTokenStart(text, current)) { + current++; + } else { + break; + } } if (current >= length) { return new Token("", length, length, false); } @@ -172,19 +179,49 @@ boolean hasLetter = false; for (int i = 0; i < token.length(); i++) { char c = token.charAt(i); if (isAsciiLetter(c)) { hasLetter = true; - } else if (isAsciiDigit(c)) { - continue; + } else if (isJoinerInsideWord(token, i)) { continue; } else { return false; } } return hasLetter; + } + + /** 只排除带明确音标标记的块；and/or 不作为音标，未标记纯 ASCII 无法可靠判断。 */ + private static int phoneticSpanEnd(String text, int start) { + char open = text.charAt(start); + if (open != '/' && open != '[') { + int end = start; + boolean identifiable = false; + while (end < text.length()) { + char c = text.charAt(end); + if (!(isAsciiLetter(c) || isIpaChar(c) || c == '\'')) { break; } + identifiable |= isIpaChar(c); + end++; + } + return identifiable ? end : start; + } + if (start > 0 && isAsciiLetter(text.charAt(start - 1))) { return start; } + char close = open == '/' ? '/' : ']'; + int end = text.indexOf(close, start + 1); + if (end <= start + 1) { return start; } + String inner = text.substring(start + 1, end); + if (inner.indexOf('\n') >= 0) { return start; } + boolean ipa = inner.chars().anyMatch(EnglishWords::isIpaChar); + boolean markedAscii = open == '/' && inner.matches("[a-zA-Z' .:-]+"); + return ipa || markedAscii ? end + 1 : start; + } + + private static boolean isIpaChar(int c) { + return (c >= 0x250 && c <= 0x2ff) || (c >= 0x1d00 && c <= 0x1d7f) + || (c >= 0x300 && c <= 0x36f) || c == 0x00e6 || c == 0x00f0 + || c == 0x00f8 || c == 0x0127 || c == 0x014b || c == 0x0153 || c == 0x03b8; } private static boolean isTokenStart(String text, int index) { char c = text.charAt(index); if (isWordChar(c)) { --- /dev/null +++ b/common/src/test/java/cn/jojo/cc/common/utils/EnglishWordsBoundaryRegressionTest.java @@ -0,0 +1,16 @@ +package cn.jojo.cc.common.utils; + +import org.junit.jupiter.api.Test; +import java.util.List; +import static org.junit.jupiter.api.Assertions.*; + +class EnglishWordsBoundaryRegressionTest { + private List words(String s) { return EnglishWords.tokenize(s).stream().map(EnglishWords.Token::text).toList(); } + @Test void digitsSeparateEnglishWords() { assertEquals(List.of("cat", "dog", "look", "up"), words("cat2dog 3look up4")); } + @Test void preservesApostrophesAndHyphens() { assertEquals(List.of("it's", "low-temperature"), words("it's low-temperature")); } + @Test void excludesIdentifiableIpaWithoutDroppingAndOr() { + assertEquals(List.of("and", "or", "cat", "dog"), words("and/or cat /kæt/ dog [dɒɡ]")); + assertEquals(List.of("ordinary", "words"), words("[ordinary words]")); + } + @Test void searchNormalizesWhitespace() { assertEquals("take off", TextNormalizer.normalize(" Take OFF ")); } +} --- /dev/null +++ b/database/597084-english-dict-outbox.sql @@ -0,0 +1,13 @@ +-- Deploy in BOTH JY and PROD before application code; no business ID auto-increment. +CREATE TABLE IF NOT EXISTS cc_english_dict_outbox ( + id VARCHAR(36) NOT NULL PRIMARY KEY, + event_type VARCHAR(32) NOT NULL, + entity_id BIGINT NOT NULL, + payload VARCHAR(8000) NOT NULL, + create_time BIGINT NOT NULL, + retry_after BIGINT NOT NULL DEFAULT 0, + attempts INT NOT NULL DEFAULT 0, + delivered_time BIGINT NOT NULL DEFAULT 0, + last_error VARCHAR(990) NULL, + KEY idx_due (retry_after, create_time, id) +) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4; --- /dev/null +++ b/infrastructure/repository/src/main/java/cn/jojo/cc/core/dao/english/EnglishDictOutboxDao.java @@ -0,0 +1,7 @@ +package cn.jojo.cc.core.dao.english; + +import cn.jojo.cc.core.domain.english.EnglishDictOutboxDo; +import com.baomidou.mybatisplus.core.mapper.BaseMapper; + +public interface EnglishDictOutboxDao extends BaseMapper { +} --- a/infrastructure/repository/src/main/java/cn/jojo/cc/core/dao/english/QuestionSegmentDao.java +++ b/infrastructure/repository/src/main/java/cn/jojo/cc/core/dao/english/QuestionSegmentDao.java @@ -12,10 +12,31 @@ /** * 题目分词结果数据访问。 */ public interface QuestionSegmentDao extends SynchBaseMapper { + + /** 题级 DB 锁串行化跨内容位写入/删除，并与正文更新互斥。 */ + @Select("SELECT * FROM cc_new_question WHERE id = #{questionId} AND delete_time = 0 FOR UPDATE") + cn.jojo.cc.core.domain.question.library.QuestionDo lockQuestion(@Param("questionId") Long questionId); + + @Insert("INSERT INTO cc_question_segment (question_id, question_key, content_type, " + + "source_text_md5, question_revision, segments, segment_status, retry_count, last_error, " + + "creator_id, create_time, operator_id, update_time, delete_time) " + + "VALUES (#{questionId}, #{questionKey}, #{contentType}, '', #{revision}, '', 2, 0, '', " + + "-1, UNIX_TIMESTAMP(), -1, UNIX_TIMESTAMP(), 0) ON DUPLICATE KEY UPDATE id = id") + int reserveIdentity(@Param("questionId") Long questionId, @Param("questionKey") String questionKey, + @Param("contentType") Integer contentType, @Param("revision") Long revision); + + @Select("SELECT * FROM cc_question_segment WHERE question_id = #{questionId} " + + "AND content_type = #{contentType} LIMIT 1") + QuestionSegmentDo getIdentity(@Param("questionId") Long questionId, @Param("contentType") Integer contentType); + + @Update("UPDATE cc_question_segment SET segment_status = 2, source_text_md5 = '', " + + "question_revision = #{revision} WHERE question_id = #{questionId} AND delete_time = 0 " + + "AND question_revision <= #{revision}") + int invalidateQuestion(@Param("questionId") Long questionId, @Param("revision") Long revision); /** * 按题目取全部内容位的分词行。 * *

表上的唯一键是 (question_id, content_type)：PRD 规定一道题只属于一个内容位 --- /dev/null +++ b/infrastructure/repository/src/main/java/cn/jojo/cc/core/domain/english/EnglishDictOutboxDo.java @@ -0,0 +1,22 @@ +package cn.jojo.cc.core.domain.english; + +import com.baomidou.mybatisplus.annotation.TableId; +import com.baomidou.mybatisplus.annotation.IdType; +import com.baomidou.mybatisplus.annotation.TableName; +import lombok.Data; + +/** 单库事务内的缓存/音频发布待办；UUID 不参与教研/正式业务 ID 映射。 */ +@Data +@TableName("cc_english_dict_outbox") +public class EnglishDictOutboxDo { + @TableId(type = IdType.INPUT) + private String id; + private String eventType; + private Long entityId; + private String payload; + private Long createTime; + private Long retryAfter; + private Integer attempts; + private Long deliveredTime; + private String lastError; +} --- /dev/null +++ b/infrastructure/repository/src/main/java/cn/jojo/cc/core/service/english/EnglishDictOutboxCoreService.java @@ -0,0 +1,61 @@ +package cn.jojo.cc.core.service.english; + +import cn.jojo.cc.core.dao.english.EnglishDictOutboxDao; +import cn.jojo.cc.core.domain.english.EnglishDictOutboxDo; +import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper; +import com.baomidou.mybatisplus.core.conditions.update.LambdaUpdateWrapper; +import jakarta.annotation.Resource; +import org.springframework.stereotype.Service; +import java.util.List; +import java.util.UUID; + +/** 不切库：事务写入者和恢复者必须显式选择各自单库。 */ +@Service +public class EnglishDictOutboxCoreService { + @Resource + private EnglishDictOutboxDao dao; + + public String enqueue(String type, Long entityId, String payload) { + if (payload == null || payload.length() > 8000) { + throw new IllegalArgumentException("outbox payload 必须在 8000 字符内"); + } + EnglishDictOutboxDo row = new EnglishDictOutboxDo(); + row.setId(UUID.randomUUID().toString()); + row.setEventType(type); + row.setEntityId(entityId); + row.setPayload(payload); + row.setCreateTime(System.currentTimeMillis() / 1000); + row.setRetryAfter(0L); + row.setAttempts(0); + row.setDeliveredTime(0L); + if (dao.insert(row) != 1) { throw new IllegalStateException("outbox insert failed"); } + return row.getId(); + } + + public List listDue(int limit) { + return dao.selectList(new LambdaQueryWrapper() + .eq(EnglishDictOutboxDo::getDeliveredTime, 0L) + .le(EnglishDictOutboxDo::getRetryAfter, System.currentTimeMillis() / 1000) + .orderByAsc(EnglishDictOutboxDo::getRetryAfter, EnglishDictOutboxDo::getCreateTime, + EnglishDictOutboxDo::getId) + .last("LIMIT " + Math.min(500, Math.max(1, limit)))); + } + + public void acknowledge(String id) { + dao.update(null, new LambdaUpdateWrapper() + .eq(EnglishDictOutboxDo::getId, id) + .set(EnglishDictOutboxDo::getDeliveredTime, System.currentTimeMillis() / 1000)); + } + + public void postpone(EnglishDictOutboxDo row, String error) { + int attempts = row.getAttempts() == null ? 1 : row.getAttempts() + 1; + dao.update(null, new LambdaUpdateWrapper() + .eq(EnglishDictOutboxDo::getId, row.getId()) + .eq(EnglishDictOutboxDo::getDeliveredTime, 0L) + .set(EnglishDictOutboxDo::getAttempts, attempts) + .set(EnglishDictOutboxDo::getRetryAfter, System.currentTimeMillis() / 1000 + + Math.min(3600, 30L << Math.min(attempts, 6))) + .set(EnglishDictOutboxDo::getLastError, + error == null ? "unknown" : error.substring(0, Math.min(990, error.length())))); + } +} --- a/infrastructure/repository/src/main/java/cn/jojo/cc/core/service/english/EnglishPhraseCoreService.java +++ b/infrastructure/repository/src/main/java/cn/jojo/cc/core/service/english/EnglishPhraseCoreService.java @@ -88,6 +88,7 @@ /** * 回填音频资源并清掉在途标记。 */ boolean markAudioCompleted(Long phraseId, Long audioResourceId); + java.util.List listAudioPendingSubmitAfterId(int limit, long createdBefore, long afterId); } --- a/infrastructure/repository/src/main/java/cn/jojo/cc/core/service/english/EnglishPhraseVariantCoreService.java +++ b/infrastructure/repository/src/main/java/cn/jojo/cc/core/service/english/EnglishPhraseVariantCoreService.java @@ -53,6 +53,7 @@ /** * 回填音频资源并清掉在途标记。 */ boolean markAudioCompleted(Long variantId, Long audioResourceId); + java.util.List listAudioPendingSubmitAfterId(int limit, long createdBefore, long afterId); } --- a/infrastructure/repository/src/main/java/cn/jojo/cc/core/service/english/EnglishWordCoreService.java +++ b/infrastructure/repository/src/main/java/cn/jojo/cc/core/service/english/EnglishWordCoreService.java @@ -86,6 +86,7 @@ /** * 回填音频资源并清掉在途标记。 */ boolean markAudioCompleted(Long wordId, Long audioResourceId); + java.util.List listAudioPendingSubmitAfterId(int limit, long createdBefore, long afterId); } --- a/infrastructure/repository/src/main/java/cn/jojo/cc/core/service/english/QuestionSegmentCoreService.java +++ b/infrastructure/repository/src/main/java/cn/jojo/cc/core/service/english/QuestionSegmentCoreService.java @@ -51,6 +51,9 @@ /** * 按主键游标分页扫当前数据源未删除分词行（升序）；全量刷新末尾的孤儿行清理用。 * 调用方负责数据源选择，本方法不跨库；逐行删除复用基类 {@code softDeleteByDo}。 */ List listUndeletedByIdGt(Long lastId, int size); + cn.jojo.cc.core.domain.question.library.QuestionDo lockQuestion(Long questionId); + QuestionSegmentDo reserveIdentity(Long questionId, String questionKey, Integer contentType, Long revision); + void invalidateQuestion(Long questionId, Long revision); } --- a/infrastructure/repository/src/main/java/cn/jojo/cc/core/service/impl/english/EnglishPhraseCoreServiceImpl.java +++ b/infrastructure/repository/src/main/java/cn/jojo/cc/core/service/impl/english/EnglishPhraseCoreServiceImpl.java @@ -248,12 +248,18 @@ .last("LIMIT " + Math.max(1, limit))); } @Override public List listAudioPendingSubmit(int limit, long createdBefore) { + return listAudioPendingSubmitAfterId(limit, createdBefore, 0L); + } + + @Override + public List listAudioPendingSubmitAfterId(int limit, long createdBefore, long afterId) { return baseMapper.selectList(new LambdaQueryWrapper() .eq(EnglishPhraseDo::getDeleteTime, 0L) + .gt(EnglishPhraseDo::getId, afterId) .in(EnglishPhraseDo::getSource, DictSourceEnum.IMPORT.getCode(), DictSourceEnum.SEGMENT_AUTOFILL.getCode()) .and(w -> w.isNull(EnglishPhraseDo::getAudioResourceId) .or().le(EnglishPhraseDo::getAudioResourceId, 0L)) .and(w -> w.isNull(EnglishPhraseDo::getAudioTaskTime) --- a/infrastructure/repository/src/main/java/cn/jojo/cc/core/service/impl/english/EnglishPhraseVariantCoreServiceImpl.java +++ b/infrastructure/repository/src/main/java/cn/jojo/cc/core/service/impl/english/EnglishPhraseVariantCoreServiceImpl.java @@ -129,12 +129,18 @@ .last("LIMIT " + Math.max(1, limit))); } @Override public List listAudioPendingSubmit(int limit, long createdBefore) { + return listAudioPendingSubmitAfterId(limit, createdBefore, 0L); + } + + @Override + public List listAudioPendingSubmitAfterId(int limit, long createdBefore, long afterId) { return baseMapper.selectList(new LambdaQueryWrapper() .eq(EnglishPhraseVariantDo::getDeleteTime, 0L) + .gt(EnglishPhraseVariantDo::getId, afterId) .and(w -> w.isNull(EnglishPhraseVariantDo::getAudioResourceId) .or().le(EnglishPhraseVariantDo::getAudioResourceId, 0L)) .and(w -> w.isNull(EnglishPhraseVariantDo::getAudioTaskTime) .or().eq(EnglishPhraseVariantDo::getAudioTaskTime, 0L)) .and(w -> w.isNull(EnglishPhraseVariantDo::getAudioTaskId) --- a/infrastructure/repository/src/main/java/cn/jojo/cc/core/service/impl/english/EnglishWordCoreServiceImpl.java +++ b/infrastructure/repository/src/main/java/cn/jojo/cc/core/service/impl/english/EnglishWordCoreServiceImpl.java @@ -255,12 +255,18 @@ .last("LIMIT " + Math.max(1, limit))); } @Override public List listAudioPendingSubmit(int limit, long createdBefore) { + return listAudioPendingSubmitAfterId(limit, createdBefore, 0L); + } + + @Override + public List listAudioPendingSubmitAfterId(int limit, long createdBefore, long afterId) { return baseMapper.selectList(new LambdaQueryWrapper() .eq(EnglishWordDo::getDeleteTime, 0L) + .gt(EnglishWordDo::getId, afterId) .in(EnglishWordDo::getSource, DictSourceEnum.IMPORT.getCode(), DictSourceEnum.SEGMENT_AUTOFILL.getCode()) .and(w -> w.isNull(EnglishWordDo::getAudioResourceId) .or().le(EnglishWordDo::getAudioResourceId, 0L)) .and(w -> w.isNull(EnglishWordDo::getAudioTaskTime) @@ -306,10 +312,18 @@ // 无条件覆盖会把人工音频静默换掉（人工优先），所以这里把空音频作为写入前提 LambdaUpdateWrapper update = new LambdaUpdateWrapper() .eq(EnglishWordDo::getId, wordId) .and(w -> w.isNull(EnglishWordDo::getAudioResourceId) .or().le(EnglishWordDo::getAudioResourceId, 0L)) + .setSql("completeness = CASE WHEN TRIM(COALESCE(word, '')) <> '' " + + "AND TRIM(COALESCE(phonetic, '')) <> '' " + + "AND EXISTS (SELECT 1 FROM cc_english_word_translation t " + + "WHERE t.word_id = cc_english_word.id AND t.delete_time = 0) " + + "AND NOT EXISTS (SELECT 1 FROM cc_english_word_translation t " + + "WHERE t.word_id = cc_english_word.id AND t.delete_time = 0 " + + "AND (TRIM(COALESCE(t.pos, '')) = '' OR TRIM(COALESCE(t.translation, '')) = '')) " + + "THEN 1 ELSE 0 END") .set(EnglishWordDo::getAudioResourceId, audioResourceId) .set(EnglishWordDo::getAudioTaskId, "") .set(EnglishWordDo::getAudioTaskTime, 0L); return baseMapper.update(null, update) > 0; } --- a/infrastructure/repository/src/main/java/cn/jojo/cc/core/service/impl/english/QuestionSegmentCoreServiceImpl.java +++ b/infrastructure/repository/src/main/java/cn/jojo/cc/core/service/impl/english/QuestionSegmentCoreServiceImpl.java @@ -31,10 +31,26 @@ implements QuestionSegmentCoreService { private static final int MAX_ERROR_LENGTH = 990; private static final int MAX_QUESTION_KEY_BATCH = 100; private static final String EMPTY_SEGMENTS = "{\"matches\":[]}"; + + @Override + public cn.jojo.cc.core.domain.question.library.QuestionDo lockQuestion(Long questionId) { + return baseMapper.lockQuestion(questionId); + } + + @Override + public QuestionSegmentDo reserveIdentity(Long questionId, String key, Integer contentType, Long revision) { + baseMapper.reserveIdentity(questionId, key == null ? "" : key, contentType, revision == null ? 0L : revision); + return baseMapper.getIdentity(questionId, contentType); + } + + @Override + public void invalidateQuestion(Long questionId, Long revision) { + baseMapper.invalidateQuestion(questionId, revision == null ? 0L : revision); + } @Override public List listByQuestionId(Long questionId) { if (questionId == null) { return Collections.emptyList();