[ysch-flow] S3.3~S3.5 notice 迁移试点:手写 Handler 换声明式 Adapter
build / build-push (push) Successful in 1m45s

事件→状态机推进改由 NoticeFlowBizAdapter 一次声明、starter 的 FlowBizCallbackBridge 执行;
纪律逐条等价搬运(行锁读单、未绑流程/单据不存在/脏 bizId 吞掉回 200、last_instance_version
乱序二道门、audit_status 终态不回退与重复终态不重跑副作用、抄送逐人留痕)。落库换成新增
窄列方法 updateAuditFromFlow/updateAuditResultFromFlow/markPublished,不再整实体 save
(apply 回填与并发回调互打的 E2E 教训);通过即发布 + 短信后置留在 onPassed 钩子,
未到定时点不发由 SmsCatchupJob 补发的口径不变。抄送按 F7 显式开镜像(本应用有自家列表)。

发起编排上交 FlowSubmitTemplate(AFTER_COMMIT/独立线程/串行/duplicated 按成功/失败静默
保持待提交/bizUrl 按 detail-route 自动拼/变量契约预检),出站仍走本应用 FlowClient 以保住
enabled=false 降级与 legacy base-url 回落;守卫在异步路径由 spec.preflight 于出站线程判定。
attempt 归属核对刻意未接(notice bizId 无 attempt 段,接上会丢首轮 submitted,保守处理)。

NoticeFlowCallbackHandler 只摘 @Service 保留代码作回退绳:恢复=加回 @Service + 删 yaml
的 yh.flow-biz 块(两者互斥,上下文已有手写 Handler 时桥不装配)。
mvn -o test 全绿:142 例(含新增 NoticeFlowBizAdapterTest 16 例等价性断言与原 21 例手写
Handler 用例);受影响 3 个测试的构造改用与生产同源的 FlowSubmitTemplates.forTests。
This commit is contained in:
chenxing
2026-10-17 23:44:35 +08:00
parent d0af9d481c
commit 986a875279
9 changed files with 592 additions and 82 deletions
@@ -40,6 +40,36 @@ public interface NoticeRepository extends JpaRepository<Notice, Integer>,
int markApproving(@org.springframework.data.repository.query.Param("id") Integer id, int markApproving(@org.springframework.data.repository.query.Param("id") Integer id,
@org.springframework.data.repository.query.Param("instanceId") Long instanceId); @org.springframework.data.repository.query.Param("instanceId") Long instanceId);
/**
* 回调推进在途态的窄列更新(2026-10-17 ysch-flow Adapter.persist 落点):只写 audit_status +
* last_instance_version,绝不整实体 save——同事务里 apply 返回的 markApproving 与并发推送
* (E2E 教训:整实体盲写会把已落定的 20 打回 10 并覆盖结果快照)。
*/
@org.springframework.transaction.annotation.Transactional
@org.springframework.data.jpa.repository.Modifying
@org.springframework.data.jpa.repository.Query(
"update Notice n set n.auditStatus = :status, n.lastInstanceVersion = :version where n.id = :id")
int updateAuditFromFlow(@org.springframework.data.repository.query.Param("id") Integer id,
@org.springframework.data.repository.query.Param("status") Integer status,
@org.springframework.data.repository.query.Param("version") Integer version);
/** 回调落终态:审核码 + 版本 + 结果明细快照(flow_result_json,详情页进度卡片数据源) */
@org.springframework.transaction.annotation.Transactional
@org.springframework.data.jpa.repository.Modifying
@org.springframework.data.jpa.repository.Query(
"update Notice n set n.auditStatus = :status, n.lastInstanceVersion = :version, "
+ "n.flowResultJson = :result where n.id = :id")
int updateAuditResultFromFlow(@org.springframework.data.repository.query.Param("id") Integer id,
@org.springframework.data.repository.query.Param("status") Integer status,
@org.springframework.data.repository.query.Param("version") Integer version,
@org.springframework.data.repository.query.Param("result") String result);
/** 审核通过=正式发布(status 0→1,用户端自此可见);窄列更新,与审核码同事务但分行 */
@org.springframework.transaction.annotation.Transactional
@org.springframework.data.jpa.repository.Modifying
@org.springframework.data.jpa.repository.Query("update Notice n set n.status = 1 where n.id = :id")
int markPublished(@org.springframework.data.repository.query.Param("id") Integer id);
/** /**
* 短信发送原子认领(§3.2 防重复核心):置 sms_sent_at,仅当当前为 NULL(未发)时生效。 * 短信发送原子认领(§3.2 防重复核心):置 sms_sent_at,仅当当前为 NULL(未发)时生效。
* 返回 0=已被发送过(并发/重复调度双发的闸门)。发送失败由调用方 clearSmsSent 归还。 * 返回 0=已被发送过(并发/重复调度双发的闸门)。发送失败由调用方 clearSmsSent 归还。
@@ -0,0 +1,183 @@
package com.yunhai.notice.service.flow;
import com.yunhai.notice.entity.Notice;
import com.yunhai.notice.entity.NoticeCc;
import com.yunhai.notice.enums.NoticeStatusEnum;
import com.yunhai.notice.model.flow.FlowModels;
import com.yunhai.notice.repository.NoticeCcRepository;
import com.yunhai.notice.repository.NoticeRepository;
import com.yunhai.notice.service.sms.SmsSendService;
import com.yunhai.platform.flowcallback.FlowCallbackRegistrar;
import com.yunhai.platform.openapi.flow.model.FlowApplyBody;
import com.yunhai.platform.openapi.flow.model.FlowCallbackEvent;
import com.yunhai.platform.yschflow.FlowApplyGateway;
import com.yunhai.platform.yschflow.FlowBizAdapter;
import com.yunhai.platform.yschflow.FlowBizIds;
import com.yunhai.platform.yschflow.FlowBizProperties;
import com.yunhai.platform.yschflow.FlowStateUpdate;
import com.yunhai.platform.yschflow.FlowSubmitTemplate;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.ObjectProvider;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.util.StringUtils;
import java.time.LocalDateTime;
import java.util.Map;
/**
* 通知侧语义接入声明(2026-10-17 ysch-flow T3 迁移试点):把 {@code NoticeFlowCallbackHandler}
* 手写版的审核状态机推进改成一次声明,纪律(乱序/终态不回退/脏消息/抄送)由 starter 内置。
*
* <p><b>语义逐条对应(迁移即等价,不改行为)</b>:
* <ul>
* <li>行锁串行化:loader 用 {@code findByIdForUpdate}(并发推送排队,防丢更新);</li>
* <li>未绑流程/单据不存在:loader 返回 null → starter 按确定性脏消息吞掉(回 200 不重推);</li>
* <li>单据级乱序二道门:{@code recordInstanceVersion=last_instance_version};</li>
* <li>终态不回退:{@code status=audit_status},starter 判定后终态行的非终态/重复/冲突事件一律忽略;</li>
* <li>窄列原子更新:persist 走 {@code updateAuditFromFlow/updateAuditResultFromFlow},
* 绝不整实体 save(E2E 教训:apply 返回的 markApproving 与回调并发,盲写会把已落定的 20 打回 10);</li>
* <li>审核通过=正式发布 + 短信后置:onPassed 钩子(status 非 1 才置 1 并补发短信,
* 未到定时点不发——由 SmsCatchupJob 到点补发,BP-R5-2 口径);</li>
* <li>抄送留痕:notice 有自家「抄送我的」列表,故显式 {@code onCc} 镜像到 notice_cc
* (F7 默认不镜像,本应用属"确需自家展示才开 mirror"的那一类)。</li>
* </ul>
*
* <p><b>刻意未接的一项</b>:{@code recordInstanceId}/{@code recordBizId} 不接线——notice 的
* bizId 恒为 {@code notice.id}(无 attempt 段),迁移前也不做实例归属核对;接上会让"apply 已成功、
* markApproving 未提交"窗口内的首轮 submitted 事件被丢弃。按《任务书》§0.5 保守方向(少改、
* 语义等价)保留原判定面,attempt 归属留给 notice 引入重提换 attempt 的那一版再开。</p>
*
* <p>回退绳:删掉 yaml 的 {@code yh.flow-biz} 块并给 {@code NoticeFlowCallbackHandler} 加回
* {@code @Service},即回到 2026-09-20 的手写 Handler 形态(两者互斥:上下文里已存在手写
* FlowCallbackHandler 时 starter 的桥不装配)。</p>
*/
@Slf4j
@Configuration
public class NoticeFlowBizAdapter {
@Bean
public FlowBizAdapter<Notice> noticeFlowBiz(NoticeRepository noticeRepository,
NoticeCcRepository noticeCcRepository,
SmsSendService smsSendService,
FlowBizProperties properties) {
return FlowBizAdapter.<Notice>builder()
.bizType(FlowCenterClient.BIZ_TYPE)
// 状态码以 yaml yh.flow-biz.states 为单一真源(避免两处码表)
.states(properties.resolveStates())
.loader(bizId -> loadForUpdate(noticeRepository, bizId))
.status(Notice::getAuditStatus)
.recordInstanceVersion(Notice::getLastInstanceVersion)
.persist((notice, update) -> persistAudit(noticeRepository, notice, update))
.onPassed((notice, event) -> publishAndSms(noticeRepository, smsSendService, notice))
.onCc((notice, event) -> recordCc(noticeCcRepository, notice, event))
.build();
}
/**
* 发起编排模板:出站仍走本应用既有 {@link FlowClient}(保留 enabled=false 降级与
* legacy {@code notice.flow.base-url} 回落两条现状语义),AFTER_COMMIT/独立线程/串行/
* duplicated 按成功等编排纪律交给 starter。
*/
@Bean
public FlowSubmitTemplate flowSubmitTemplate(FlowBizProperties properties, FlowClient flowClient,
ObjectProvider<FlowCallbackRegistrar> registrarProvider) {
return new FlowSubmitTemplate(properties, gateway(flowClient), null, registrarProvider);
}
/**
* 出站网关:starter 组装的请求体转投本应用既有 {@link FlowClient}(保留 enabled=false 降级与
* legacy base-url 回落),公开供单测复用同一根线(测试替身与生产装配零分叉)。
*/
public static FlowApplyGateway gateway(FlowClient flowClient) {
return new FlowApplyGateway() {
@Override
public com.yunhai.platform.openapi.flow.model.FlowApplyResult apply(long schId, FlowApplyBody body) {
return flowClient.apply(Integer.valueOf((int) schId), toNoticeBody(body));
}
};
}
/** starter 组装的请求体转成本应用契约类型(字段逐项照搬,Notice 未用发起时指派,assigns 不透传) */
private static FlowModels.ApplyBody toNoticeBody(FlowApplyBody source) {
FlowModels.ApplyBody target = new FlowModels.ApplyBody();
target.setFlowGuid(source.getFlowGuid());
target.setBizType(source.getBizType());
target.setBizId(source.getBizId());
target.setBizTitle(source.getBizTitle());
target.setBizUrl(source.getBizUrl());
target.setBizSnapshot(source.getBizSnapshot());
target.setSubmitterId(source.getSubmitterId());
target.setSubmitterName(source.getSubmitterName());
target.setFormData(source.getFormData());
return target;
}
/** 行锁读单据;bizId 非法 / 通知不存在 / 未绑流程 → null(确定性丢弃,回 200) */
private static Notice loadForUpdate(NoticeRepository repository, String bizId) {
Integer noticeId = FlowBizIds.asInteger(bizId);
if (noticeId == null) {
throw new IllegalArgumentException("bizId 非通知主键:" + bizId);
}
Notice notice = repository.findByIdForUpdate(noticeId).orElse(null);
if (notice == null || !StringUtils.hasText(notice.getFlowGuid())) {
return null;
}
return notice;
}
private static void persistAudit(NoticeRepository repository, Notice notice, FlowStateUpdate update) {
Integer version = update.getInstanceVersion() == null
? notice.getLastInstanceVersion() : update.getInstanceVersion();
if (update.isTerminal()) {
repository.updateAuditResultFromFlow(notice.getId(), update.getCode(), version, update.getSnapshotJson());
} else {
repository.updateAuditFromFlow(notice.getId(), update.getCode(), version);
}
}
/** 审核通过=正式发布:置 status=1(用户端自此可见)+ 短信后置(只在本次真的从未发布→已发布时补发) */
private static void publishAndSms(NoticeRepository repository, SmsSendService smsSendService, Notice notice) {
boolean justPublished = notice.getStatus() == null
|| notice.getStatus() != NoticeStatusEnum.PUBLISHED.getCode();
if (!justPublished) {
return;
}
repository.markPublished(notice.getId());
if (notice.getSmsFlag() == null || notice.getSmsFlag() != 1) {
return;
}
// 与直发链路 trySendSms 的 isImmediatePublish 语义对齐:定时未到点的通知此刻不发
// (可见性已生效),到点由 SmsCatchupJob 补发(§3.2 修 BP-R5-2 短信永久丢失缺口)
if (notice.getTimingSendTime() != null && notice.getTimingSendTime().isAfter(LocalDateTime.now())) {
return;
}
smsSendService.sendOnce(notice);
}
/**
* 抄送留痕(逐人一行,不动审核状态机):data 缺 ccUserId 属上游脏数据,记日志跳过(回 200),
* 落库异常照旧外抛回 5xx 重试。重放由 starter 的 eventId 幂等拦截。
*/
private static void recordCc(NoticeCcRepository ccRepository, Notice notice, FlowCallbackEvent event) {
Map<String, Object> data = event.getData();
Object ccUserId = data == null ? null : data.get("ccUserId");
if (ccUserId == null || String.valueOf(ccUserId).isEmpty()) {
log.warn("[notice-flow] cc.sent 缺 ccUserId(丢弃):notice={} eventId={}",
notice.getId(), event.getEventId());
return;
}
NoticeCc cc = new NoticeCc();
cc.setSchId(notice.getSchId());
cc.setNoticeId(notice.getId());
cc.setFlowInstanceId(event.getInstanceId());
cc.setNodeId(data.get("nodeId") == null ? null : String.valueOf(data.get("nodeId")));
cc.setNodeName(data.get("nodeName") == null ? null : String.valueOf(data.get("nodeName")));
cc.setCcUserId(Integer.valueOf(String.valueOf(ccUserId)));
cc.setCcUserName(data.get("ccUserName") == null ? null : String.valueOf(data.get("ccUserName")));
cc.setSentTime(LocalDateTime.now());
ccRepository.save(cc);
log.info("[notice-flow] 抄送留痕:notice={} instance={} ccUser={} node={}",
notice.getId(), event.getInstanceId(), cc.getCcUserId(), cc.getNodeName());
}
}
@@ -9,7 +9,6 @@ import com.yunhai.platform.flowcallback.FlowCallbackHandler;
import com.yunhai.platform.openapi.flow.model.FlowCallbackEvent; import com.yunhai.platform.openapi.flow.model.FlowCallbackEvent;
import lombok.RequiredArgsConstructor; import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j; import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Service;
import org.springframework.util.StringUtils; import org.springframework.util.StringUtils;
import java.time.LocalDateTime; import java.time.LocalDateTime;
@@ -24,9 +23,14 @@ import java.util.Map;
* 2026-09-24)。行锁串行化并发推送,终态不回退守卫,终态 data 快照落 flow_result_json(详情页进度卡片)。 * 2026-09-24)。行锁串行化并发推送,终态不回退守卫,终态 data 快照落 flow_result_json(详情页进度卡片)。
* <br>应答语义不变:本方法抛任何异常 → starter 回 5xx(flow-center 持续重试追平,免对账); * <br>应答语义不变:本方法抛任何异常 → starter 回 5xx(flow-center 持续重试追平,免对账);
* 正常返回(含通知不存在/未绑流程/未知事件等不可重试场景)→ 200。 * 正常返回(含通知不存在/未绑流程/未知事件等不可重试场景)→ 200。
*
* <p><b>2026-10-17 T3 迁移:bean 已停用,代码原样保留作回退绳。</b>状态机推进改由
* {@link NoticeFlowBizAdapter} 声明、starter 的 {@code FlowBizCallbackBridge} 执行。
* 恢复本形态只需两步(语义与迁移前逐字节一致):① 给本类加回 {@code @Service}(连同其 import);
* ② 删除 yaml 的 {@code yh.flow-biz} 块——两者互斥(上下文已存在手写 FlowCallbackHandler 时
* starter 的桥不装配),故不会双写。</p>
*/ */
@Slf4j @Slf4j
@Service
@RequiredArgsConstructor @RequiredArgsConstructor
public class NoticeFlowCallbackHandler implements FlowCallbackHandler { public class NoticeFlowCallbackHandler implements FlowCallbackHandler {
@@ -14,15 +14,14 @@ import com.yunhai.notice.utils.ReceiveParser;
import com.yunhai.platform.apistd.BizException; import com.yunhai.platform.apistd.BizException;
import com.yunhai.platform.auth.spring.CurrentBoaUser; import com.yunhai.platform.auth.spring.CurrentBoaUser;
import com.yunhai.platform.openapi.base.YschBaseOpenApi; import com.yunhai.platform.openapi.base.YschBaseOpenApi;
import com.yunhai.platform.openapi.core.JumpTarget; import com.yunhai.platform.yschflow.FlowApplySpec;
import com.yunhai.platform.yschflow.FlowSubmitTemplate;
import com.yunhai.platform.utils.exception.ErrorCode; import com.yunhai.platform.utils.exception.ErrorCode;
import com.yunhai.platform.utils.model.BoaUser; import com.yunhai.platform.utils.model.BoaUser;
import lombok.RequiredArgsConstructor; import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j; import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Service; import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional; import org.springframework.transaction.annotation.Transactional;
import org.springframework.transaction.support.TransactionSynchronizationAdapter;
import org.springframework.transaction.support.TransactionSynchronizationManager;
import org.springframework.util.StringUtils; import org.springframework.util.StringUtils;
import java.util.ArrayList; import java.util.ArrayList;
@@ -46,54 +45,25 @@ public class NoticeFlowService {
private final FlowClient flowClient; private final FlowClient flowClient;
private final NoticeFlowProperties properties; private final NoticeFlowProperties properties;
private final ObjectMapper objectMapper; private final ObjectMapper objectMapper;
private final com.yunhai.platform.flowcallback.FlowCallbackRegistrar callbackRegistrar;
private final CurrentBoaUser currentBoaUser; private final CurrentBoaUser currentBoaUser;
private final NoticeFlowGuard noticeFlowGuard; private final NoticeFlowGuard noticeFlowGuard;
private final FlowSubmitTemplate flowSubmitTemplate;
/** /**
* 注册后事务提交了再提交审核(D9:apply 出站不进 DB 事务,防 HTTP 长调用拖住连接); * 提交审核(D9 AFTER_COMMIT 出站,编排纪律交 {@link FlowSubmitTemplate},2026-10-17 T3 迁移):
* 无活动事务时立即执行(测试/兜底)。 * 出站动作在事务提交后的独立线程执行,提交人身份与单据快照在请求线程先取好。
* <p>E2E 实测教训:afterCommit 回调若留在原请求线程执行,OSIV 绑定的 EntityManager * <p>E2E 实测教训(迁移前本类自带线程池,语义不变):afterCommit 回调若留在原请求线程执行,
* 已随提交关闭,后续 repository 写会话静默丢失(有成功日志但库无行)——故提交动作 * OSIV 绑定的 EntityManager 已随提交关闭,后续 repository 写会话静默丢失(有成功日志但库无行);
* 一律切到独立线程(干净事务 + 无 ThreadLocal 残留),提交人身份在请求线程先捕获透传。 * 「能不能提交」的守卫也必须在提交后的状态上判定,故随 spec 的 preflight 一起搬到出站线程。</p>
*/ */
public void submitAfterCommit(Integer noticeId) { public void submitAfterCommit(Integer noticeId) {
final BoaUser user = currentUser(); final BoaUser user = currentUser();
Runnable task = () -> trySubmitQuietly(noticeId, user); final Notice notice = requireNotice(noticeId);
if (TransactionSynchronizationManager.isSynchronizationActive()) { flowSubmitTemplate.submitAfterCommit(noticeId,
TransactionSynchronizationManager.registerSynchronization(new TransactionSynchronizationAdapter() { specOf(notice, user).preflight(() -> checkSubmittable(notice, user)));
@Override
public void afterCommit() {
SUBMIT_EXECUTOR.submit(task);
}
});
} else {
task.run();
}
} }
/** 提交动作专用单线程池(串行化避免并发双 apply;uk 幂等兜底) */ /** 提交/重提审核(幂等入口:编辑页重提按钮);失败当场抛业务异常给调用方 */
private static final java.util.concurrent.ExecutorService SUBMIT_EXECUTOR =
java.util.concurrent.Executors.newSingleThreadExecutor(r -> {
Thread t = new Thread(r, "notice-flow-submit");
t.setDaemon(true);
return t;
});
/** 提交失败静默保持 0 待提交(行已落库不回滚;列表"待提交"态即提示,编辑页可重推) */
private void trySubmitQuietly(Integer noticeId, BoaUser user) {
try {
trySubmitFlow(noticeId, user);
} catch (Exception e) {
log.warn("[notice-flow] 提交审核失败(保持待提交,可重推):notice={} {}", noticeId, e.toString());
}
}
/**
* 提交/重提审核(幂等入口:编辑页重提按钮 / create 后置共用)。
* audit=0/30/40 可提交;10 审批中拒绝;apply 成功回填 flow_instance_id + audit=10。
* apply 失败抛业务异常,notice 行保持 0 待提交(不回滚已落库的编辑内容)。
*/
public void trySubmitFlow(Integer noticeId) { public void trySubmitFlow(Integer noticeId) {
trySubmitFlow(noticeId, currentUser()); trySubmitFlow(noticeId, currentUser());
} }
@@ -101,6 +71,15 @@ public class NoticeFlowService {
/** 提交/重提(带访问者——控制器直传登录人,BP-R5-3 守卫入口) */ /** 提交/重提(带访问者——控制器直传登录人,BP-R5-3 守卫入口) */
public void trySubmitFlow(Integer noticeId, BoaUser user) { public void trySubmitFlow(Integer noticeId, BoaUser user) {
Notice notice = requireNotice(noticeId); Notice notice = requireNotice(noticeId);
checkSubmittable(notice, user);
flowSubmitTemplate.submit(noticeId, specOf(notice, user));
}
/**
* 提交前置校验(同步入口当场抛、异步入口由 spec.preflight 在出站线程抛后静默保持待提交):
* BP-R5-3 三分支守卫 + 未绑流程 + 审批中禁止重复提交。
*/
private void checkSubmittable(Notice notice, BoaUser user) {
// BP-R5-3:Trace/TrySubmit 原先仅验存在性,任意登录用户可对他人通知触发重提——补 // BP-R5-3:Trace/TrySubmit 原先仅验存在性,任意登录用户可对他人通知触发重提——补
// NoticeFlowGuard 三分支(提交人/管理员/办理人),与 Detail 的守卫口径对齐 // NoticeFlowGuard 三分支(提交人/管理员/办理人),与 Detail 的守卫口径对齐
noticeFlowGuard.requireUnpublishedVisible(notice, user); noticeFlowGuard.requireUnpublishedVisible(notice, user);
@@ -111,47 +90,34 @@ public class NoticeFlowService {
if (audit != null && audit == FlowAuditStatusEnum.APPROVING.getCode()) { if (audit != null && audit == FlowAuditStatusEnum.APPROVING.getCode()) {
throw new BizException(ErrorCode.FORBIDDEN, "审批中,不可重复提交,请先撤回"); throw new BizException(ErrorCode.FORBIDDEN, "审批中,不可重复提交,请先撤回");
} }
ensureRegistered(notice);
FlowModels.ApplyBody body = buildApplyBody(notice, user);
FlowModels.ApplyResultVo result = flowClient.apply(notice.getSchId(), body);
// E2E 实测教训:此处曾整实体 save——apply 出站期间回调事件可能已并发推进
// (submitted 先行置 10 / completed 置 20),旧快照盲写会把状态打回;改为窄列原子更新,
// 且 audit=20(已通过)不回退(apply 返回晚于 completed 送达的合法场景)
noticeRepository.markApproving(noticeId, result.getInstanceId());
log.info("[notice-flow] 提交审核成功:notice={} instance={} duplicated={}",
noticeId, result.getInstanceId(), result.isDuplicated());
} }
/** /**
* 回调注册(2026-09-20 迁 starter 的 {@link com.yunhai.platform.flowcallback.FlowCallbackRegistrar}): * 发起编排入参(契约冻结:bizId=notice.id、bizType=notice;bizUrl 不再手工传,
* 仅本地镜像行(flow_callback_reg)缺失/失效时注册并落 secret。<b>绝不重复调用</b>—— * 由 starter 按 {@code yh.flow-biz.detail-route} 拼装,产物与迁移前的
* flow-center 重复注册会轮换 secret;失配恢复手段=删除镜像行后重新保存绑定。 * {@code JumpTarget.page("list/detail").param("serid", id)} 逐字节相同)。
* 标题/表单/快照都取自请求线程读到的已提交实体值,出站前不落库。
*/ */
void ensureRegistered(Notice notice) { private FlowApplySpec specOf(Notice notice, BoaUser user) {
callbackRegistrar.ensureRegistered(notice.getSchId(), notice.getFlowGuid()); return FlowApplySpec.of(String.valueOf(notice.getId()), notice.getFlowGuid(),
() -> notice.getTitle(), () -> buildFormData(notice), () -> buildSnapshot(notice))
.bizType(FlowCenterClient.BIZ_TYPE)
.schId(notice.getSchId() == null ? null : Long.valueOf(notice.getSchId()))
.submitter(user == null ? null : user.getId(),
user == null ? null : (StringUtils.hasText(user.getName()) ? user.getName() : user.getAccount()))
// E2E 实测教训:apply 出站期间回调可能已并发推进,回填必须是窄列原子 UPDATE,
// 且 audit=20(已通过)不回退——守卫在 markApproving 的 case-when 里
.onApplied((instanceId, duplicated) -> noticeRepository.markApproving(notice.getId(), instanceId))
.onSubmitError((bizKey, error) -> log.warn("[notice-flow] 提交审核失败(保持待提交,可重推):"
+ "notice={} {}", bizKey, String.valueOf(error)));
} }
/** apply 出参组装(契约冻结:bizId=notice.id、bizUrl=裸相对路径、快照只收摘要不收富文本正文; /** formData 键与变量契约同源(yh.flow-biz.variables=[deptId, deptName],发送前由 starter 本地预检) */
* formData 带发文部门(2026-09-24)——条件路由判断端点优先读它免查库,缺字段回落查表同口径。 private Map<String, Object> buildFormData(Notice notice) {
* bizUrl 自 2026-09-26 起为裸相对路径(不含域名/ctx/#,接收方按页面归属拼绝对地址),
* Notice 不再需要配置自己的前端基址(自指地址优化 Phase 2) */
FlowModels.ApplyBody buildApplyBody(Notice notice, BoaUser user) {
FlowModels.ApplyBody body = new FlowModels.ApplyBody();
body.setFlowGuid(notice.getFlowGuid());
body.setBizType(FlowCenterClient.BIZ_TYPE);
body.setBizId(String.valueOf(notice.getId()));
body.setBizTitle(notice.getTitle());
body.setBizJump(JumpTarget.page("list/detail").param("serid", notice.getId()));
body.setBizSnapshot(buildSnapshot(notice));
Map<String, Object> formData = new LinkedHashMap<>(); Map<String, Object> formData = new LinkedHashMap<>();
formData.put("deptId", notice.getDeptId()); formData.put("deptId", notice.getDeptId());
formData.put("deptName", notice.getDeptName()); formData.put("deptName", notice.getDeptName());
body.setFormData(formData); return formData;
if (user != null) {
body.setSubmitterId(user.getId());
body.setSubmitterName(StringUtils.hasText(user.getName()) ? user.getName() : user.getAccount());
}
return body;
} }
/** 业务摘要:原键照旧(流程中心透传、嵌入页不展示)+ 约定键 summaryFields(嵌入页「业务信息」唯一渲染源) */ /** 业务摘要:原键照旧(流程中心透传、嵌入页不展示)+ 约定键 summaryFields(嵌入页「业务信息」唯一渲染源) */
@@ -0,0 +1,289 @@
package com.yunhai.notice.service.flow;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.yunhai.notice.entity.Notice;
import com.yunhai.notice.entity.NoticeCc;
import com.yunhai.notice.repository.NoticeCcRepository;
import com.yunhai.notice.repository.NoticeRepository;
import com.yunhai.notice.service.sms.SmsSendService;
import com.yunhai.platform.openapi.flow.model.FlowCallbackEvent;
import com.yunhai.platform.yschflow.FlowBizAdapter;
import com.yunhai.platform.yschflow.FlowBizCallbackBridge;
import com.yunhai.platform.yschflow.FlowBizProperties;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.DisplayName;
import org.junit.jupiter.api.Test;
import org.mockito.ArgumentCaptor;
import java.time.LocalDateTime;
import java.util.LinkedHashMap;
import java.util.Map;
import java.util.Optional;
import static org.junit.jupiter.api.Assertions.assertDoesNotThrow;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.ArgumentMatchers.anyInt;
import static org.mockito.ArgumentMatchers.anyString;
import static org.mockito.ArgumentMatchers.eq;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.never;
import static org.mockito.Mockito.times;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
/**
* 迁移等价性锁定(S3.3):{@link NoticeFlowBizAdapter} 声明 + starter 桥执行,必须与停用的
* {@link NoticeFlowCallbackHandler} 手写版语义逐条一致——窄列更新、乱序二道门、终态不回退、
* 通过即发布+短信后置、抄送逐人留痕、脏消息回 200 不重推。
*/
@DisplayName("Notice 语义接入声明 - 与迁移前手写 Handler 等价")
class NoticeFlowBizAdapterTest {
private static final Integer NOTICE_ID = 7;
private static final Integer BOUND_VERSION = 3;
private NoticeRepository noticeRepository;
private NoticeCcRepository ccRepository;
private SmsSendService smsSendService;
private FlowBizCallbackBridge<Notice> bridge;
private Notice notice;
@BeforeEach
void setUp() {
noticeRepository = mock(NoticeRepository.class);
ccRepository = mock(NoticeCcRepository.class);
smsSendService = mock(SmsSendService.class);
notice = boundNotice();
when(noticeRepository.findByIdForUpdate(NOTICE_ID)).thenReturn(Optional.of(notice));
FlowBizAdapter<Notice> adapter = new NoticeFlowBizAdapter()
.noticeFlowBiz(noticeRepository, ccRepository, smsSendService, properties());
bridge = new FlowBizCallbackBridge<>(adapter, new ObjectMapper(), properties());
}
/** 与 application.yml 的 yh.flow-biz 同值 */
private static FlowBizProperties properties() {
FlowBizProperties properties = new FlowBizProperties();
properties.setBizType("notice");
properties.getStates().setApproving(10);
properties.getStates().setPassed(20);
properties.getStates().setRejected(30);
properties.getStates().setWithdrawn(40);
return properties;
}
private static Notice boundNotice() {
Notice entity = new Notice();
entity.setId(NOTICE_ID);
entity.setSchId(1);
entity.setFlowGuid("fg-1");
entity.setAuditStatus(10);
entity.setLastInstanceVersion(BOUND_VERSION);
entity.setStatus(0);
entity.setSmsFlag(1);
return entity;
}
private static FlowCallbackEvent event(String code, Integer version, Map<String, Object> data) {
FlowCallbackEvent event = new FlowCallbackEvent();
event.setEventId("evt-" + code + "-" + version);
event.setEvent(code);
event.setBizType("notice");
event.setBizId(String.valueOf(NOTICE_ID));
event.setInstanceId(900L);
event.setInstanceVersion(version);
event.setData(data);
return event;
}
private static Map<String, Object> terminalData() {
Map<String, Object> data = new LinkedHashMap<>();
data.put("opinion", "同意");
data.put("result", "通过");
return data;
}
// ==================== 窄列推进 ====================
@Test
void approvingEvents_useNarrowColumnUpdate_neverWholeEntitySave() {
bridge.onEvent(event("flow.submitted", 4, null));
bridge.onEvent(event("flow.task.created", 5, null));
bridge.onEvent(event("flow.task.handled", 6, null));
verify(noticeRepository).updateAuditFromFlow(NOTICE_ID, 10, 4);
verify(noticeRepository).updateAuditFromFlow(NOTICE_ID, 10, 5);
verify(noticeRepository).updateAuditFromFlow(NOTICE_ID, 10, 6);
verify(noticeRepository, never()).save(any(Notice.class));
verify(noticeRepository, never()).saveAndFlush(any(Notice.class));
}
@Test
void eventWithoutVersion_keepsRecordedVersionInsteadOfNullingIt() {
bridge.onEvent(event("flow.task.handled", null, null));
verify(noticeRepository).updateAuditFromFlow(NOTICE_ID, 10, BOUND_VERSION);
}
@Test
void completed_writesTerminalSnapshotAndPublishesAndSendsSms() {
bridge.onEvent(event("flow.completed", 7, terminalData()));
ArgumentCaptor<String> json = ArgumentCaptor.forClass(String.class);
verify(noticeRepository).updateAuditResultFromFlow(eq(NOTICE_ID), eq(20), eq(7), json.capture());
assertEquals("{\"opinion\":\"同意\",\"result\":\"通过\"}", json.getValue());
verify(noticeRepository).markPublished(NOTICE_ID);
verify(smsSendService).sendOnce(notice);
}
@Test
void rejected_andWithdrawn_writeTerminalWithoutPublish() {
bridge.onEvent(event("flow.rejected", 7, terminalData()));
verify(noticeRepository).updateAuditResultFromFlow(eq(NOTICE_ID), eq(30), eq(7), anyString());
notice.setAuditStatus(10);
bridge.onEvent(event("flow.withdrawn", 8, terminalData()));
verify(noticeRepository).updateAuditResultFromFlow(eq(NOTICE_ID), eq(40), eq(8), anyString());
verify(noticeRepository, never()).markPublished(anyInt());
verify(smsSendService, never()).sendOnce(any(Notice.class));
}
// ==================== 发布与短信后置的三种现状分支 ====================
@Test
void alreadyPublished_noSecondPublishNoSms() {
notice.setStatus(1);
bridge.onEvent(event("flow.completed", 7, terminalData()));
verify(noticeRepository, never()).markPublished(anyInt());
verify(smsSendService, never()).sendOnce(any(Notice.class));
}
@Test
void smsFlagOff_publishesButSendsNoSms() {
notice.setSmsFlag(0);
bridge.onEvent(event("flow.completed", 7, terminalData()));
verify(noticeRepository).markPublished(NOTICE_ID);
verify(smsSendService, never()).sendOnce(any(Notice.class));
}
@Test
void timingNotReached_publishesButSendsNoSmsYet() {
notice.setTimingSendTime(LocalDateTime.now().plusHours(2));
bridge.onEvent(event("flow.completed", 7, terminalData()));
verify(noticeRepository).markPublished(NOTICE_ID);
verify(smsSendService, never()).sendOnce(any(Notice.class));
}
// ==================== 纪律:不回退 / 不重复副作用 / 乱序 / 脏消息 ====================
@Test
void terminalRow_ignoresNonTerminalEvent() {
notice.setAuditStatus(20);
bridge.onEvent(event("flow.task.created", 8, null));
verify(noticeRepository, never()).updateAuditFromFlow(anyInt(), anyInt(), any());
}
@Test
void terminalRow_ignoresRepeatedCompleted_soSmsFiresOnce() {
bridge.onEvent(event("flow.completed", 7, terminalData()));
notice.setAuditStatus(20);
bridge.onEvent(event("flow.completed", 7, terminalData()));
verify(noticeRepository, times(1)).updateAuditResultFromFlow(anyInt(), anyInt(), anyInt(), anyString());
verify(smsSendService, times(1)).sendOnce(any(Notice.class));
}
@Test
void staleVersion_droppedByRecordGate() {
bridge.onEvent(event("flow.task.handled", 2, null));
verify(noticeRepository, never()).updateAuditFromFlow(anyInt(), anyInt(), any());
}
@Test
void unboundNotice_swallowedAndWritesNothing() {
notice.setFlowGuid(null);
assertDoesNotThrow(() -> bridge.onEvent(event("flow.completed", 7, terminalData())));
assertWritesNothing();
}
@Test
void missingNotice_swallowedAndWritesNothing() {
when(noticeRepository.findByIdForUpdate(NOTICE_ID)).thenReturn(Optional.empty());
assertDoesNotThrow(() -> bridge.onEvent(event("flow.completed", 7, terminalData())));
assertWritesNothing();
}
@Test
void dirtyBizId_swallowedAndNotEvenLookedUp() {
FlowCallbackEvent dirty = event("flow.completed", 7, terminalData());
dirty.setBizId("abc");
assertDoesNotThrow(() -> bridge.onEvent(dirty));
verify(noticeRepository, never()).findByIdForUpdate(anyInt());
assertWritesNothing();
}
private void assertWritesNothing() {
verify(noticeRepository, never()).updateAuditFromFlow(anyInt(), anyInt(), any());
verify(noticeRepository, never()).updateAuditResultFromFlow(anyInt(), anyInt(), any(), anyString());
verify(noticeRepository, never()).markPublished(anyInt());
}
// ==================== 抄送镜像(notice 有自家"抄送我的"列表) ====================
@Test
void ccSent_recordsRowPerPerson_withoutTouchingAudit() {
Map<String, Object> data = new LinkedHashMap<>();
data.put("ccUserId", 301);
data.put("ccUserName", "李四");
data.put("nodeId", "n3");
data.put("nodeName", "抄送分管领导");
bridge.onEvent(event("flow.cc.sent", 8, data));
ArgumentCaptor<NoticeCc> captor = ArgumentCaptor.forClass(NoticeCc.class);
verify(ccRepository).save(captor.capture());
NoticeCc row = captor.getValue();
assertEquals(Integer.valueOf(1), row.getSchId());
assertEquals(NOTICE_ID, row.getNoticeId());
assertEquals(Long.valueOf(900L), row.getFlowInstanceId());
assertEquals(Integer.valueOf(301), row.getCcUserId());
assertEquals("李四", row.getCcUserName());
assertEquals("n3", row.getNodeId());
assertEquals("抄送分管领导", row.getNodeName());
verify(noticeRepository, never()).updateAuditFromFlow(anyInt(), anyInt(), any());
verify(noticeRepository, never()).updateAuditResultFromFlow(anyInt(), anyInt(), any(), anyString());
}
@Test
void ccSentWithoutCcUserId_skippedAsDirtyData() {
Map<String, Object> data = new LinkedHashMap<>();
data.put("nodeName", "抄送分管领导");
assertDoesNotThrow(() -> bridge.onEvent(event("flow.cc.sent", 8, data)));
verify(ccRepository, never()).save(any(NoticeCc.class));
}
@Test
void unknownEvent_swallowed() {
assertDoesNotThrow(() -> bridge.onEvent(event("flow.route.decided", 8, terminalData())));
verify(noticeRepository, never()).updateAuditFromFlow(anyInt(), anyInt(), any());
}
}
@@ -138,8 +138,9 @@ class NoticeFlowPublishTest {
noticeFlowService = new NoticeFlowService(noticeRepository, bindingRepository, noticeFlowService = new NoticeFlowService(noticeRepository, bindingRepository,
mock(UpFileRepository.class), flowClient, props, new ObjectMapper(), mock(UpFileRepository.class), flowClient, props, new ObjectMapper(),
mock(com.yunhai.platform.flowcallback.FlowCallbackRegistrar.class), currentBoaUser, currentBoaUser,
new NoticeFlowGuard(flowClient, props, mock(com.yunhai.notice.repository.NoticeCcRepository.class))); new NoticeFlowGuard(flowClient, props, mock(com.yunhai.notice.repository.NoticeCcRepository.class)),
com.yunhai.notice.support.FlowSubmitTemplates.forTests(flowClient));
noticeService = new NoticeService(); noticeService = new NoticeService();
ReflectionTestUtils.setField(noticeService, "noticeUserMsgNotifier", ReflectionTestUtils.setField(noticeService, "noticeUserMsgNotifier",
@@ -47,8 +47,8 @@ class NoticeFlowTraceTest {
service = new NoticeFlowService(noticeRepository, service = new NoticeFlowService(noticeRepository,
mock(com.yunhai.notice.repository.NoticeFlowBindingRepository.class), mock(com.yunhai.notice.repository.NoticeFlowBindingRepository.class),
mock(UpFileRepository.class), flowClient, props, new ObjectMapper(), mock(UpFileRepository.class), flowClient, props, new ObjectMapper(),
mock(com.yunhai.platform.flowcallback.FlowCallbackRegistrar.class), mock(CurrentBoaUser.class), new NoticeFlowGuard(flowClient, props, mock(com.yunhai.notice.repository.NoticeCcRepository.class)),
mock(CurrentBoaUser.class), new NoticeFlowGuard(flowClient, props, mock(com.yunhai.notice.repository.NoticeCcRepository.class))); com.yunhai.notice.support.FlowSubmitTemplates.forTests(flowClient));
bound = new Notice(); bound = new Notice();
bound.setId(42); bound.setId(42);
@@ -42,8 +42,9 @@ class NoticeFlowWithdrawTest {
props.setEnabled(true); props.setEnabled(true);
service = new NoticeFlowService(noticeRepository, service = new NoticeFlowService(noticeRepository,
mock(NoticeFlowBindingRepository.class), mock(UpFileRepository.class), flowClient, props, mock(NoticeFlowBindingRepository.class), mock(UpFileRepository.class), flowClient, props,
new ObjectMapper(), mock(com.yunhai.platform.flowcallback.FlowCallbackRegistrar.class), new ObjectMapper(), mock(CurrentBoaUser.class),
mock(CurrentBoaUser.class), new NoticeFlowGuard(flowClient, props, mock(com.yunhai.notice.repository.NoticeCcRepository.class))); new NoticeFlowGuard(flowClient, props, mock(com.yunhai.notice.repository.NoticeCcRepository.class)),
com.yunhai.notice.support.FlowSubmitTemplates.forTests(flowClient));
approving = new Notice(); approving = new Notice();
approving.setId(42); approving.setId(42);
@@ -0,0 +1,36 @@
package com.yunhai.notice.support;
import com.yunhai.notice.service.flow.FlowClient;
import com.yunhai.notice.service.flow.NoticeFlowBizAdapter;
import com.yunhai.platform.yschflow.FlowBizProperties;
import com.yunhai.platform.yschflow.FlowSubmitTemplate;
import java.util.Arrays;
/**
* 单测用发起模板:与生产装配同源(同一个 {@link NoticeFlowBizAdapter#gateway},出站仍落到
* 测试替身 FlowClient),只是不走真线程池(执行器传 null;单元测试无活动事务,
* {@code submitAfterCommit} 走内联分支,断言可确定性观察)。
*/
public final class FlowSubmitTemplates {
private FlowSubmitTemplates() {
}
public static FlowSubmitTemplate forTests(FlowClient flowClient) {
return new FlowSubmitTemplate(properties(), NoticeFlowBizAdapter.gateway(flowClient), null, null);
}
/** 与 application.yml 的 yh.flow-biz 块同值(测试里不引 yml,改这里必须同步 yml) */
private static FlowBizProperties properties() {
FlowBizProperties properties = new FlowBizProperties();
properties.setBizType("notice");
properties.getStates().setApproving(10);
properties.getStates().setPassed(20);
properties.getStates().setRejected(30);
properties.getStates().setWithdrawn(40);
properties.setVariables(Arrays.asList("deptId", "deptName"));
properties.setDetailRoute("/list/detail?serid={bizId}");
return properties;
}
}