t17:oapi v2 增量拉取 + RabbitMQ 事件（可选依赖、信封十字段闭集、软删防回归断言）
- GET /oapi/form/answers（契约 §6.1/C9/C11/C12）：域校验走 readDomainForm（他校与不存在同一条
"表单不存在"，V54；软删表单照读，§5.4/P32），行集恒带 sch_id。增量口径按冻结状态二分且不接受混用
（混用给可诊断拒绝，而不是静默返回一个"看起来正常"的空窗口）：
冻结=append-only + after=answer_uuid（服务端按 uk 换算库内 id，id 绝不出响应，V53，墓碑照常下发）；
非冻结=updated_since + 复合键 (update_time, answer_uuid)，续拉游标 {ISO8601}|{answer_uuid}（P126）。
limit 1~2000 越界拒不夹；nextCursor 只在这一页填满时给（宁可末尾多要一次空页，不可少要一次）。
- 行字段是 §6.1 的 15 键闭集（逐字段 @JsonProperty，外层 rows/nextCursor 按 C9 原文 camelCase——
命名差异是契约原文，不加"顺手统一"的适配层）；meta.ip/ua 只匿名行有值（C12），登录行两项 null。
- 时间线格式降到整秒并对齐列精度（DATETIME 秒粒度；带纳秒的游标与列值比不出等值，
同事务重读还会经一级缓存拿到内存对象）：取舍方向是"宁可重复投递也不漏行"，
重复投递正是 C9 明写的 at-least-once（P127）。IT 侧 flush+clear 把读取摆回生产形态。
- 新 FcEventPublisher（§6.2/V62/V63/V64）：RABBITMQ_HOST 未注入=整体 no-op + 启动一条说明日志；
RabbitTemplate 走 ObjectProvider（可选依赖不该在缺 bean 时拖垮上下文）；exchange fc.events.v1
惰性 durable topic 声明（声明失败放回标志位允许重试）；routing key fc.answer.{app_code}.{action}；
afterCommit 发布且信封在注册前按实体快照组好；发布失败只 warn 不抛（没有 outbox、没有死信）。
信封十字段闭集、不含值文档（裁决 1：MQ 是通知通道不是数据通道）。
- 发布点三处：submit 写库段后=created（代录/导入同管线自动获得）；两处编辑写库段=updated
（能走到即说明未开冻结，V61 已在前面拦）；两处软删实体 save 后=tombstoned（公共出口是发布器那一个方法）。
- 依赖与配套：pom 加 spring-boot-starter-amqp（BOM 管版本）；yml 的 spring.rabbitmq.* 全从
RABBITMQ_* env 现读，并关 rabbit 健康探针与 Boot 的 RabbitAdmin（dynamic=false）——
后者启动就拨号声明拓扑，无 broker 环境实测把上下文启动拖到分钟级，拓扑已由发布器自己声明。
测试：OapiAnswersIT 8 例（闭集无 id、口径二分互不混用、limit 边界、after 翻页+重放+墓碑可见、
V70 软删前移 update_time 才让墓碑进窗口、复合游标带 uuid 尾、他校同文面、匿名行 meta）
+ FcEventPublisherTest 5 例（V63 十字段闭集、routing key、未配置即 no-op、afterCommit 之前不发、
异常被吞）+ MqDisabledContextIT 2 例（无 broker：enabled()=false 与 /actuator/health 仍 UP）。
门禁：定向 15 例全绿（s6-gate-2.log），本段另跑过一次全量 288 例全绿（mvn-full-2.log，
按协调口径后续段不再连发全量，全量留 S8 收口）。差额 P126/P127/P128。
This commit is contained in:
chenxing
@@ -17,6 +17,7 @@ import org.springframework.web.bind.annotation.RestController;
|
||||
* GET /oapi/form/schema?schId=&uuid= → T11 新形态：该表单<b>当前已发布版</b>
|
||||
* GET /oapi/form/schema?schId=&uuid=&version= → T11 新形态：精确版本行（快照核对/重放）
|
||||
* GET /oapi/form/answer?schId=&answerId= → data=钉住版本的 schema 引用 + 值文档（契约 §9）
|
||||
* GET /oapi/form/answers?schId=&form_uuid=&游标 → v2 增量拉取（契约 v2 §6.1，T17，见方法注释）
|
||||
* </pre>
|
||||
*
|
||||
* <p>uuid 两形态的用途=给 flow-center 在<b>流程定义发布期</b>拉契约快照嵌进 def 版本行
|
||||
@@ -57,4 +58,25 @@ public class OapiFormController {
|
||||
@RequestParam("answerId") String answerId) {
|
||||
return formService.answer(schId, answerId);
|
||||
}
|
||||
|
||||
/**
|
||||
* v2 增量拉取（契约 §6.1/C9，T17）——<b>事实源</b>，MQ 事件只是降延迟。
|
||||
*
|
||||
* <pre>
|
||||
* GET /oapi/form/answers?schId=&formUuid=&after=answer_uuid[&limit=] 冻结表单：append-only + 墓碑
|
||||
* GET /oapi/form/answers?schId=&formUuid=&updated_since=ISO8601[|answer_uuid] 非冻结：updated 窗口
|
||||
* </pre>
|
||||
*
|
||||
* <p>两种游标按表单的冻结状态二分、不接受混用（拒绝文案点明这张表要哪种），{@code limit} 默认 500
|
||||
* 上限 2000；响应 {@code {rows,nextCursor}}，行字段是闭集且<b>不含库内 id</b>（V53）。
|
||||
* {@code schId} 与既有两个 oapi 端点同义=调用方声明的租户域，行集按它过滤（应用签名不替代谁能看）。</p>
|
||||
*/
|
||||
@GetMapping("/answers")
|
||||
public OapiModels.AnswersView answers(@RequestParam("schId") Long schId,
|
||||
@RequestParam("form_uuid") String formUuid,
|
||||
@RequestParam(value = "after", required = false) String after,
|
||||
@RequestParam(value = "updated_since", required = false) String updatedSince,
|
||||
@RequestParam(value = "limit", required = false) Integer limit) {
|
||||
return formService.answers(schId, formUuid, after, updatedSince, limit);
|
||||
}
|
||||
}
|
||||
|
||||
@@ -1,6 +1,8 @@
|
||||
/**
|
||||
* 对外签名面（/oapi/**，appKey 三件套验签，见 application.yml 的 yh.app-auth.server.enforce-patterns）。
|
||||
* T4 落地契约 §9 的两个读端点：{@code GET /oapi/form/schema}（内联 options + valueSchema）与
|
||||
* {@code GET /oapi/form/answer}。一期<b>无 oapi 代提交</b>（方案 §2.3 边界，答卷身份一律会话推导）。
|
||||
* {@code GET /oapi/form/answer}。T17 加第三个：{@code GET /oapi/form/answers}（契约 v2 §6.1 增量拉取，
|
||||
* <b>它是消费方的事实源</b>，MQ 事件只降延迟、不含值文档）。一期<b>无 oapi 代提交</b>
|
||||
* （方案 §2.3 边界，答卷身份一律会话推导）。
|
||||
*/
|
||||
package com.yunhai.formcenter.controller.oapi;
|
||||
|
||||
@@ -75,4 +75,92 @@ public final class OapiModels {
|
||||
private String code;
|
||||
private Integer version;
|
||||
}
|
||||
|
||||
// ==================== v2 拉取端点（§6.1，T17） ====================
|
||||
|
||||
/**
|
||||
* {@code GET /oapi/form/answers} 的 data：{@code {rows, nextCursor}}（C9）。
|
||||
*
|
||||
* <p>外层两个键按契约原文是 <b>camelCase</b>，而 {@link AnswerRow} 内部是 <b>snake_case</b>——
|
||||
* 这不是笔误而是原文如此（§6.1 的行样例 + C9 的响应样例），实现照抄，不在中间加"顺手统一"的适配层：
|
||||
* 线格式一改，已接的消费方就得跟着改（同 C5/G3 那次 breaking 的教训）。</p>
|
||||
*/
|
||||
@Getter
|
||||
@Setter
|
||||
public static class AnswersView {
|
||||
private java.util.List<AnswerRow> rows;
|
||||
/** null=已到当前窗口末尾；同游标重放返回同窗口或其后子集（C9 重放安全） */
|
||||
private String nextCursor;
|
||||
}
|
||||
|
||||
/**
|
||||
* 拉取行的<b>字段闭集</b>（§6.1，V53：库内 {@code id} 绝不出现在响应里）。
|
||||
*
|
||||
* <p>逐字段标 {@code @JsonProperty} 而不是配命名策略：这张表一旦有人"顺手"加字段，
|
||||
* {@code OapiAnswersIT} 的键集断言就会红（闭集是契约级的，不是实现细节）。
|
||||
* 软删行照常出现在这里（{@code is_deleted=1} 即墓碑，C11）。</p>
|
||||
*/
|
||||
@Getter
|
||||
@Setter
|
||||
public static class AnswerRow {
|
||||
@com.fasterxml.jackson.annotation.JsonProperty("answer_uuid")
|
||||
private String answerUuid;
|
||||
|
||||
@com.fasterxml.jackson.annotation.JsonProperty("sch_id")
|
||||
private Long schId;
|
||||
|
||||
@com.fasterxml.jackson.annotation.JsonProperty("app_code")
|
||||
private String appCode;
|
||||
|
||||
@com.fasterxml.jackson.annotation.JsonProperty("form_uuid")
|
||||
private String formUuid;
|
||||
|
||||
@com.fasterxml.jackson.annotation.JsonProperty("form_code")
|
||||
private String formCode;
|
||||
|
||||
@com.fasterxml.jackson.annotation.JsonProperty("version")
|
||||
private Integer version;
|
||||
|
||||
@com.fasterxml.jackson.annotation.JsonProperty("answer_no")
|
||||
private String answerNo;
|
||||
|
||||
/** 提交人；0=匿名行（C8） */
|
||||
@com.fasterxml.jackson.annotation.JsonProperty("submitter_id")
|
||||
private Integer submitterId;
|
||||
|
||||
/** 被代填人；自填=本人、匿名=NULL */
|
||||
@com.fasterxml.jackson.annotation.JsonProperty("subject_id")
|
||||
private Integer subjectId;
|
||||
|
||||
@com.fasterxml.jackson.annotation.JsonProperty("status")
|
||||
private String status;
|
||||
|
||||
@com.fasterxml.jackson.annotation.JsonProperty("values")
|
||||
private JsonNode values;
|
||||
|
||||
/** 匿名留痕（C12：登录答卷两项恒 null，值文档里也不出现） */
|
||||
@com.fasterxml.jackson.annotation.JsonProperty("meta")
|
||||
private AnswerRowMeta meta;
|
||||
|
||||
@com.fasterxml.jackson.annotation.JsonProperty("create_time")
|
||||
private String createTime;
|
||||
|
||||
@com.fasterxml.jackson.annotation.JsonProperty("update_time")
|
||||
private String updateTime;
|
||||
|
||||
/** 1=墓碑（C11） */
|
||||
@com.fasterxml.jackson.annotation.JsonProperty("is_deleted")
|
||||
private Integer isDeleted;
|
||||
}
|
||||
|
||||
/** 行内 {@code meta} 的两件（闭集，消费方按 {@code submitter_id==0} 认匿名行） */
|
||||
@Getter
|
||||
@Setter
|
||||
public static class AnswerRowMeta {
|
||||
@com.fasterxml.jackson.annotation.JsonProperty("ip")
|
||||
private String ip;
|
||||
|
||||
@com.fasterxml.jackson.annotation.JsonProperty("ua")
|
||||
private String ua;
|
||||
}
|
||||
}
|
||||
|
||||
@@ -68,6 +68,31 @@ public interface AnswerRepository extends JpaRepository<Answer, Long> {
|
||||
/** 我的提交段（T7）/ "我提交的"视图（T8）：按 submitter 过滤本人答卷，按 id 降序（最新在前） */
|
||||
List<Answer> findBySchIdAndSubmitterIdOrderByIdDesc(Long schId, Integer submitterId);
|
||||
|
||||
/**
|
||||
* oapi v2 拉取·冻结表单的 append-only 窗口（契约 v2 §6.1/C9/C11，T17）：
|
||||
* 按 id 升序推进、<b>不过滤 is_deleted</b>（墓碑照常下发）。
|
||||
*
|
||||
* <p>带 {@code sch_id} 是 oapi 面既有那条"本层只保证不跨校取数"的不变式（应用签名不替代谁能看）；
|
||||
* 与 {@link #findLiveByFormUuidAfterId} 的区别正是软删行——那条是补建索引用的活数据，这条是对外事实流。</p>
|
||||
*/
|
||||
@Query("select a from Answer a where a.formUuid = :formUuid and a.schId = :schId and a.id > :cursor "
|
||||
+ "order by a.id asc")
|
||||
List<Answer> findPullAfterId(@Param("formUuid") String formUuid, @Param("schId") Long schId,
|
||||
@Param("cursor") Long cursor, Pageable pageable);
|
||||
|
||||
/**
|
||||
* oapi v2 拉取·非冻结表单的 updated 窗口（§6.1/C9/C11）：复合推进键
|
||||
* {@code (update_time, answer_uuid)}——同刻多行也能定序且不漏不重（重放安全，
|
||||
* 边界行至多重复投递一次，由消费方幂等 upsert 吸收）。<b>同样含墓碑</b>：
|
||||
* 软删走实体 save（{@code @PreUpdate} 必刷 update_time），所以删除动作就在这个窗口里（V70 钉住）。
|
||||
*/
|
||||
@Query("select a from Answer a where a.formUuid = :formUuid and a.schId = :schId "
|
||||
+ "and (a.updateTime > :time or (a.updateTime = :time and a.uuid > :uuid)) "
|
||||
+ "order by a.updateTime asc, a.uuid asc")
|
||||
List<Answer> findPullUpdatedSince(@Param("formUuid") String formUuid, @Param("schId") Long schId,
|
||||
@Param("time") java.time.LocalDateTime time, @Param("uuid") String uuid,
|
||||
Pageable pageable);
|
||||
|
||||
/**
|
||||
* "关于我的"视图（T8，V18）：按 subject 过滤——别人<b>代我填</b>的答卷也在这里
|
||||
* （方案 §6.3：谁提交与关于谁是两个角色，学生看到的是"关于我"的那几条，不限于自己填的）。
|
||||
|
||||
@@ -71,6 +71,7 @@ public class AnswerAdminService {
|
||||
private final AnswerReadService readService;
|
||||
private final AnswerScopeQueryBuilder queryBuilder;
|
||||
private final TransactionTemplate transactionTemplate;
|
||||
private final FcEventPublisher events;
|
||||
|
||||
// ==================== 单条 ====================
|
||||
|
||||
@@ -145,6 +146,8 @@ public class AnswerAdminService {
|
||||
answerRepository.save(answer);
|
||||
// 索引同步重建（方案 §5.4）：仍按钉住版本的 searchable 声明写，不随新版本 schema 改变
|
||||
indexService.rebuild(answer, schemaJson);
|
||||
// 事件：非冻结编辑（冻结在上面就拒了，走不到这里）= updated（契约 v2 §6.2 发布点之一）
|
||||
events.answerUpdated(answer);
|
||||
opLog(answer.getId(), AnswerOpLog.OP_EDIT, operator.getId(),
|
||||
req.getSummary() == null ? "管理员编辑（钉住版本 v" + answer.getVersion() + "）"
|
||||
: "编辑：钉住版本 v" + answer.getVersion() + "；" + req.getSummary());
|
||||
@@ -163,6 +166,8 @@ public class AnswerAdminService {
|
||||
FormRuntimeRules.quotaKey(answer.getSubjectId(), answer.getSubmitterId()));
|
||||
opLog(answer.getId(), batched ? AnswerOpLog.OP_BATCH_SOFT_DELETE : AnswerOpLog.OP_SOFT_DELETE,
|
||||
operator.getId(), batched ? "批量软删（单条同校验、单条留痕）" : "软删（无物理删除接口）");
|
||||
// 事件：tombstoned（管理面单条与批量共用本方法，所以两个入口一处发布，§6.2）
|
||||
events.answerTombstoned(answer);
|
||||
}
|
||||
|
||||
private void doChangeStatus(BoaUser operator, Long schId, AnswerModels.ChangeStatusRequest req,
|
||||
|
||||
@@ -66,6 +66,7 @@ public class AnswerReadService {
|
||||
private final FillQuotaRepository quotaRepository;
|
||||
private final FormFileRepository fileRepository;
|
||||
private final AnswerOpLogRepository opLogRepository;
|
||||
private final FcEventPublisher events;
|
||||
|
||||
// ==================== 规格构造（四路径共用） ====================
|
||||
|
||||
@@ -322,6 +323,8 @@ public class AnswerReadService {
|
||||
defService.pinnedForData(answer.getFormUuid(), answer.getVersion()).form().getId(),
|
||||
FormRuntimeRules.quotaKey(answer.getSubjectId(), answer.getSubmitterId()));
|
||||
opLog(answer.getId(), AnswerOpLog.OP_SOFT_DELETE, user.getId(), "本人删除自己的提交（软删）");
|
||||
// 事件：tombstoned（提交人侧软删与管理面各自实体 save，共用发布器这一个出口，§6.2）
|
||||
events.answerTombstoned(answer);
|
||||
log.info("[我的提交] 软删答卷 answerId={} submitterId={}", answerId, user.getId());
|
||||
}
|
||||
|
||||
@@ -367,6 +370,8 @@ public class AnswerReadService {
|
||||
opLog(answer.getId(), AnswerOpLog.OP_EDIT, user.getId(), summary == null || summary.trim().isEmpty()
|
||||
? "本人编辑自己的提交（钉住版本 v" + answer.getVersion() + "）"
|
||||
: "本人编辑：钉住版本 v" + answer.getVersion() + "；" + summary.trim());
|
||||
// 事件：updated（与管理面编辑同一点位；能走到这里说明该表未开冻结，V61 已在上面拦过）
|
||||
events.answerUpdated(answer);
|
||||
log.info("[我的提交] 编辑答卷 answerId={} submitterId={} v={}", answerId, user.getId(),
|
||||
answer.getVersion());
|
||||
return toRow(answer, form);
|
||||
|
||||
@@ -92,6 +92,7 @@ public class AnswerService {
|
||||
private final FormFileRepository fileRepository;
|
||||
private final AnswerIndexService indexService;
|
||||
private final TransactionTemplate transactionTemplate;
|
||||
private final FcEventPublisher events;
|
||||
|
||||
/**
|
||||
* 提交答卷。
|
||||
@@ -266,6 +267,10 @@ public class AnswerService {
|
||||
// 9) 索引：只写 searchable 字段（V3/V9）
|
||||
indexService.write(answer, schemaJson);
|
||||
|
||||
// 10) 事件：created 只在"真提交"这条路上发（幂等命中在第 4 关就 return 了，不发事件）。
|
||||
// 发布器内部是 afterCommit + 失败仅 warn + 未配 broker 即 no-op，主流程零风险（§6.2 裁决 2/3）
|
||||
events.answerCreated(answer);
|
||||
|
||||
log.info("[提交]{} answer={} form={}/{} v{} submitter={} subject={} quotaKey={}",
|
||||
identity.anonymous() ? "[匿名]" : "", answer.getUuid(), answer.getAppCode(), answer.getCode(),
|
||||
answer.getVersion(), submitterId, storedSubjectId, quotaKey);
|
||||
|
||||
@@ -0,0 +1,213 @@
|
||||
package com.yunhai.formcenter.service;
|
||||
|
||||
import com.fasterxml.jackson.annotation.JsonProperty;
|
||||
import com.yunhai.formcenter.entity.Answer;
|
||||
import lombok.Getter;
|
||||
import lombok.RequiredArgsConstructor;
|
||||
import lombok.Setter;
|
||||
import lombok.extern.slf4j.Slf4j;
|
||||
import org.springframework.amqp.rabbit.core.RabbitTemplate;
|
||||
import org.springframework.beans.factory.ObjectProvider;
|
||||
import org.springframework.beans.factory.annotation.Value;
|
||||
import org.springframework.stereotype.Component;
|
||||
import org.springframework.transaction.support.TransactionSynchronization;
|
||||
import org.springframework.transaction.support.TransactionSynchronizationManager;
|
||||
|
||||
import javax.annotation.PostConstruct;
|
||||
import java.time.OffsetDateTime;
|
||||
import java.time.ZoneId;
|
||||
import java.time.format.DateTimeFormatter;
|
||||
import java.util.UUID;
|
||||
import java.util.concurrent.atomic.AtomicBoolean;
|
||||
|
||||
/**
|
||||
* 答卷事件推送（契约 v2 §6.2，T17，验收 <b>V62/V63/V64</b>）。
|
||||
*
|
||||
* <p><b>三条裁决决定了这个类的形状</b>（契约把它们标成"不可协商"）：</p>
|
||||
* <ol>
|
||||
* <li><b>MQ 是通知通道，不是数据通道</b>——信封十字段闭集（V63），<b>不含值文档</b>。值文档只有
|
||||
* 签名验过的拉取端点那一份；把值塞进总线等于把读面从"鉴权端点"扩大到"能连上 broker 的任何人"；</li>
|
||||
* <li><b>拉取是事实源，MQ 只降延迟</b>——所以这里<b>不写 outbox、不配死信</b>，发布失败只 warn
|
||||
* （丢的那条由消费方的 5 分钟增量拉取补回，V62 的"同步单向"也因此不会被消息顺序反向污染）；</li>
|
||||
* <li><b>可选依赖</b>——{@code RABBITMQ_HOST} 未注入 → 整体 no-op + 启动一条日志，服务照跑（V64；
|
||||
* 照 data-redis 那条先例：引依赖但不引运行期强耦合，配套 yml 关 rabbit 健康探针）。</li>
|
||||
* </ol>
|
||||
*
|
||||
* <p><b>发布时机=事务提交后</b>（{@code afterCommit}）：回滚掉的事务不能留下"已发布但不存在"的事件——
|
||||
* 消费方按 {@code answer_uuid} 去拉会拉不到，而它没有理由重试。同步器未激活时（无事务上下文，
|
||||
* 例如单测直接调）就地发布，不为测试特殊开事务。</p>
|
||||
*
|
||||
* <p>exchange 惰性声明（{@code durable topic fc.events.v1}）：第一次真要发的时候 declare 一次，
|
||||
* 之后不再重复。未启 broker 的环境永远走不到这一行，因此也不需要 {@code RabbitAdmin} 的自动声明链路。</p>
|
||||
*/
|
||||
@Slf4j
|
||||
@Component
|
||||
@RequiredArgsConstructor
|
||||
public class FcEventPublisher {
|
||||
|
||||
/** 拓扑：topic exchange，名与 durable 都由契约定死（§6.2） */
|
||||
public static final String EXCHANGE = "fc.events.v1";
|
||||
|
||||
/** 平台时区（与 hibernate.jdbc.time_zone / jackson.time-zone 同一条），信封时间戳按它给偏移量 */
|
||||
private static final ZoneId ZONE = ZoneId.of("Asia/Shanghai");
|
||||
private static final DateTimeFormatter OCCURRED_AT = DateTimeFormatter.ISO_OFFSET_DATE_TIME;
|
||||
|
||||
/**
|
||||
* RabbitTemplate 走 {@link ObjectProvider}：本应用"可以完全不配 MQ"，
|
||||
* 直接注入会在依赖缺失时把上下文拖垮（而可选依赖的整个前提就是"哪天 broker 就绪哪天开"）。
|
||||
*/
|
||||
private final ObjectProvider<RabbitTemplate> templates;
|
||||
|
||||
/** 是否启用的唯一判据=宿主是否注入了 RABBITMQ_HOST（yml 把它映射到 spring.rabbitmq.host） */
|
||||
@Value("${spring.rabbitmq.host:}")
|
||||
private String host;
|
||||
|
||||
private final AtomicBoolean exchangeDeclared = new AtomicBoolean(false);
|
||||
|
||||
@PostConstruct
|
||||
public void logState() {
|
||||
if (enabled()) {
|
||||
log.info("[MQ] 事件发布启用（exchange={}，host={}）", EXCHANGE, host);
|
||||
} else {
|
||||
log.info("[MQ] 未配置 RABBITMQ_HOST，事件发布禁用——拉取端点即事实源，服务照常（契约 v2 §6.2 裁决 3）");
|
||||
}
|
||||
}
|
||||
|
||||
public boolean enabled() {
|
||||
return host != null && !host.trim().isEmpty();
|
||||
}
|
||||
|
||||
// ==================== 三个发布点 ====================
|
||||
|
||||
/** 提交管线写库段之后（手工/代录/导入同管线，三者自动都在这，§3.4/V60） */
|
||||
public void answerCreated(Answer answer) {
|
||||
publish(answer, "created");
|
||||
}
|
||||
|
||||
/** 编辑写库段同点位（非冻结表单才可能走到这里——冻结双通道同关，V61） */
|
||||
public void answerUpdated(Answer answer) {
|
||||
publish(answer, "updated");
|
||||
}
|
||||
|
||||
/**
|
||||
* 软删（墓碑）。
|
||||
*
|
||||
* <p>写库点有两处实体 save（管理面 {@code doSoftDelete}（单条与批量共用）与提交人侧
|
||||
* {@code softDeleteBySubmitter}），契约要求"软删公共出口收敛一处"——本方法就是那一处：
|
||||
* 类型、routing key、信封组装只在这里定义一次，两个写库点各调它一次。</p>
|
||||
*/
|
||||
public void answerTombstoned(Answer answer) {
|
||||
publish(answer, "tombstoned");
|
||||
}
|
||||
|
||||
// ==================== 内部 ====================
|
||||
|
||||
private void publish(Answer answer, String action) {
|
||||
if (!enabled() || answer == null) {
|
||||
return;
|
||||
}
|
||||
// 事件内容在提交点前就定死（快照字段），afterCommit 只负责"发"，不再回读实体——
|
||||
// 那时事务已结束，实体可能已被后续操作改写，也可能压根不在同一持久化上下文里
|
||||
Envelope envelope = envelopeOf(answer, action);
|
||||
if (TransactionSynchronizationManager.isSynchronizationActive()) {
|
||||
TransactionSynchronizationManager.registerSynchronization(new TransactionSynchronization() {
|
||||
@Override
|
||||
public void afterCommit() {
|
||||
send(envelope, action);
|
||||
}
|
||||
});
|
||||
} else {
|
||||
send(envelope, action);
|
||||
}
|
||||
}
|
||||
|
||||
private void send(Envelope envelope, String action) {
|
||||
try {
|
||||
RabbitTemplate template = templates.getIfAvailable();
|
||||
if (template == null) {
|
||||
log.warn("[MQ] 已配置 RABBITMQ_HOST 但容器里没有 RabbitTemplate，跳过发布（不影响主流程）");
|
||||
return;
|
||||
}
|
||||
declareExchangeOnce(template);
|
||||
template.convertAndSend(EXCHANGE, routingKey(envelope.getAppCode(), action), envelope);
|
||||
} catch (Exception e) {
|
||||
// 裁决 2：broker 故障绝不阻断提交。消费方以游标拉取为事实源，这条事件丢了由下一轮增量补回
|
||||
log.warn("[MQ] 事件发布失败（忽略，拉取兜底）：type={} answer={} cause={}",
|
||||
envelope.getType(), envelope.getAnswerUuid(), e.toString());
|
||||
}
|
||||
}
|
||||
|
||||
private void declareExchangeOnce(RabbitTemplate template) {
|
||||
if (!exchangeDeclared.compareAndSet(false, true)) {
|
||||
return;
|
||||
}
|
||||
try {
|
||||
template.execute(channel -> {
|
||||
channel.exchangeDeclare(EXCHANGE, "topic", true);
|
||||
return null;
|
||||
});
|
||||
} catch (Exception e) {
|
||||
// 声明失败要允许下次再试（broker 抖动一次就永久静默是不可接受的）
|
||||
exchangeDeclared.set(false);
|
||||
throw e;
|
||||
}
|
||||
}
|
||||
|
||||
/** routing key：{@code fc.answer.{app_code}.{created|updated|tombstoned}}（§6.2 队列绑定约定） */
|
||||
public static String routingKey(String appCode, String action) {
|
||||
return "fc.answer." + appCode + "." + action;
|
||||
}
|
||||
private Envelope envelopeOf(Answer answer, String action) {
|
||||
Envelope envelope = new Envelope();
|
||||
envelope.setEventId(UUID.randomUUID().toString());
|
||||
envelope.setType("fc.answer." + action);
|
||||
envelope.setOccurredAt(OffsetDateTime.now(ZONE).format(OCCURRED_AT));
|
||||
envelope.setAppCode(answer.getAppCode());
|
||||
envelope.setSchId(answer.getSchId());
|
||||
envelope.setFormUuid(answer.getFormUuid());
|
||||
envelope.setFormCode(answer.getCode());
|
||||
envelope.setAnswerUuid(answer.getUuid());
|
||||
envelope.setVersion(answer.getVersion());
|
||||
envelope.setAnswerNo(answer.getAnswerNo());
|
||||
return envelope;
|
||||
}
|
||||
|
||||
/**
|
||||
* 消息信封（§6.2 字段闭集，V63：<b>多一个字段就该被当成违约</b>，所以这里逐字段标
|
||||
* {@link JsonProperty} 而不是靠命名策略——snake_case 是契约原文，不是本类偏好）。
|
||||
*/
|
||||
@Getter
|
||||
@Setter
|
||||
public static class Envelope {
|
||||
|
||||
@JsonProperty("event_id")
|
||||
private String eventId;
|
||||
|
||||
@JsonProperty("type")
|
||||
private String type;
|
||||
|
||||
@JsonProperty("occurred_at")
|
||||
private String occurredAt;
|
||||
|
||||
@JsonProperty("app_code")
|
||||
private String appCode;
|
||||
|
||||
@JsonProperty("sch_id")
|
||||
private Long schId;
|
||||
|
||||
@JsonProperty("form_uuid")
|
||||
private String formUuid;
|
||||
|
||||
@JsonProperty("form_code")
|
||||
private String formCode;
|
||||
|
||||
@JsonProperty("answer_uuid")
|
||||
private String answerUuid;
|
||||
|
||||
@JsonProperty("version")
|
||||
private Integer version;
|
||||
|
||||
@JsonProperty("answer_no")
|
||||
private String answerNo;
|
||||
}
|
||||
}
|
||||
@@ -12,8 +12,15 @@ import com.yunhai.platform.apistd.BizException;
|
||||
import com.yunhai.platform.utils.exception.ErrorCode;
|
||||
import lombok.RequiredArgsConstructor;
|
||||
import lombok.extern.slf4j.Slf4j;
|
||||
import org.springframework.data.domain.PageRequest;
|
||||
import org.springframework.data.domain.Pageable;
|
||||
import org.springframework.stereotype.Service;
|
||||
|
||||
import java.time.LocalDateTime;
|
||||
import java.time.format.DateTimeFormatter;
|
||||
import java.util.ArrayList;
|
||||
import java.util.List;
|
||||
|
||||
/**
|
||||
* 一期对外读接口服务（契约 §9 两端点）。
|
||||
*
|
||||
@@ -160,4 +167,183 @@ public class OapiFormService {
|
||||
throw new BizException(ErrorCode.PARAM_ERROR, "schId 必填（调用方声明的租户域，≥0）");
|
||||
}
|
||||
}
|
||||
|
||||
// ==================== v2 拉取端点（契约 §6.1/C9/C11，T17） ====================
|
||||
|
||||
/** C9：默认 500、上限 2000（超限直接拒而不是静默收窄——调用方要对自己的页大小有确定预期） */
|
||||
private static final int DEFAULT_LIMIT = 500;
|
||||
private static final int MAX_LIMIT = 2000;
|
||||
|
||||
/**
|
||||
* 游标与行内时间的线格式：{@code yyyy-MM-dd'T'HH:mm:ss}（ISO8601 本地时间、**整秒**，P126）。
|
||||
*
|
||||
* <p>刻意不用 {@code ISO_LOCAL_DATE_TIME}（它会在有亚秒时把纳秒一起打出来）：
|
||||
* 列侧是 {@code DATETIME}（秒精度，t3/t10 建表实态），带纳秒的游标与库内值永远比不出等值，
|
||||
* 于是"续拉"会既可能重复投递也可能漏行。更关键的是 Hibernate 的一级缓存会让<b>同一事务里
|
||||
* 刚写完再读</b>的那一行拿到内存对象（带纳秒）而不是列里取整后的值——纳秒游标在测试与真实
|
||||
* 的同事务重读场景下都不成立。</p>
|
||||
*
|
||||
* <p>降到整秒的取舍方向是<b>宁可重复投递也不漏</b>：窗口因此可能把"游标那一秒"的行再给一次，
|
||||
* 这正是 C9 允许的 at-least-once（消费方按 {@code answer_uuid} 幂等 upsert 吸收）；
|
||||
* 反过来若按秒截断把窗口收窄，同一秒内的变更就会被永久跳过（那才是不可恢复的）。</p>
|
||||
*/
|
||||
private static final DateTimeFormatter CURSOR_TS =
|
||||
java.time.format.DateTimeFormatter.ofPattern("yyyy-MM-dd'T'HH:mm:ss");
|
||||
|
||||
/** 首拉的时间下界（必然早于任何一行），配 uuid 空串=复合键的全域最小 */
|
||||
private static final LocalDateTime PULL_FROM_BEGINNING = LocalDateTime.of(1970, 1, 1, 0, 0, 0);
|
||||
|
||||
/**
|
||||
* 增量拉取答卷（{@code GET /oapi/form/answers}）：<b>事实源</b>，MQ 只是降延迟的旁路通知。
|
||||
*
|
||||
* <p>口径三件：</p>
|
||||
* <ul>
|
||||
* <li><b>按冻结状态二分增量口径</b>（§6.1）——冻结表单 id 升序 append-only（墓碑照常下发，
|
||||
* 游标是 {@code answer_uuid}，服务端按 uk 换算成库内 id，<b>id 绝不出响应</b>，V53）；
|
||||
* 非冻结表单走 {@code updated_since} + 复合键 {@code (update_time, answer_uuid)}。
|
||||
* 两种游标<b>不接受混用</b>：按字面判给另一个口径会静默给出一个"看起来正常"的空窗口，
|
||||
* 那比报错更难发现（这里给可诊断的拒绝，指出这张表要哪种游标）；</li>
|
||||
* <li><b>域校验照做、软删表单照读</b>：{@code form_uuid} 过 {@code readDomainForm}（本校 ∪ 平台级），
|
||||
* 他校与不存在同一句"表单不存在"（V54）；生命周期与 is_deleted 不作读条件（§5.4"停用只断填报入口"、
|
||||
* P32"历史数据溯源不能断"）；行集另带 {@code sch_id} 这条 oapi 面既有不变式——
|
||||
* 应用签名不替代谁能看，本层只保证不跨校取数；</li>
|
||||
* <li><b>重放安全</b>（C9）：同游标重复请求返回同一窗口；{@code nextCursor} 只在"这一页被填满"时给出，
|
||||
* 末尾可能多要一次空页，这比"少要一次"（漏数据）可接受。</li>
|
||||
* </ul>
|
||||
*/
|
||||
public OapiModels.AnswersView answers(Long schId, String formUuid, String after, String updatedSince,
|
||||
Integer limit) {
|
||||
requireSchId(schId);
|
||||
int size = requireLimit(limit);
|
||||
Form form = defService.readDomainForm(schId, formUuid);
|
||||
Pageable page = PageRequest.of(0, size);
|
||||
boolean frozen = Integer.valueOf(1).equals(form.getFrozen());
|
||||
|
||||
if (frozen) {
|
||||
if (notBlank(updatedSince)) {
|
||||
throw new BizException(ErrorCode.PARAM_ERROR,
|
||||
"表单已开启提交后禁编辑（append-only），增量游标是 after=answer_uuid，不接受 updated_since："
|
||||
+ form.getAppCode() + "/" + form.getCode());
|
||||
}
|
||||
Long cursor = 0L;
|
||||
if (notBlank(after)) {
|
||||
cursor = cursorIdOf(schId, form, after.trim());
|
||||
}
|
||||
List<Answer> rows = answerRepository.findPullAfterId(form.getUuid(), schId, cursor, page);
|
||||
OapiModels.AnswersView view = toAnswers(rows);
|
||||
view.setNextCursor(rows.size() == size ? rows.get(rows.size() - 1).getUuid() : null);
|
||||
return view;
|
||||
}
|
||||
|
||||
if (notBlank(after)) {
|
||||
throw new BizException(ErrorCode.PARAM_ERROR,
|
||||
"表单未开启禁编辑（可编辑即有 updated 增量），增量游标是 updated_since=ISO8601，不接受 after："
|
||||
+ form.getAppCode() + "/" + form.getCode());
|
||||
}
|
||||
Cursor cursor = cursorOf(updatedSince);
|
||||
List<Answer> rows = answerRepository.findPullUpdatedSince(form.getUuid(), schId, cursor.time,
|
||||
cursor.answerUuid, page);
|
||||
OapiModels.AnswersView view = toAnswers(rows);
|
||||
Answer last = rows.isEmpty() ? null : rows.get(rows.size() - 1);
|
||||
view.setNextCursor(rows.size() == size && last != null
|
||||
? CURSOR_TS.format(last.getUpdateTime()) + "|" + last.getUuid() : null);
|
||||
return view;
|
||||
}
|
||||
|
||||
/** 冻结表单的游标换算（对外只有 uuid，id 是库内坐标）；游标本身必须属于这张表，否则窗口起点无定义 */
|
||||
private Long cursorIdOf(Long schId, Form form, String answerUuid) {
|
||||
Answer marker = answerRepository.findByUuidAndSchId(answerUuid, schId).orElse(null);
|
||||
if (marker == null || !form.getUuid().equals(marker.getFormUuid())) {
|
||||
throw new BizException(ErrorCode.PARAM_ERROR,
|
||||
"游标答卷不存在或不属于该表单：after=" + answerUuid);
|
||||
}
|
||||
return marker.getId();
|
||||
}
|
||||
|
||||
private static OapiModels.AnswersView toAnswers(List<Answer> rows) {
|
||||
List<OapiModels.AnswerRow> items = new ArrayList<>();
|
||||
for (Answer row : rows) {
|
||||
items.add(toPullRow(row));
|
||||
}
|
||||
OapiModels.AnswersView view = new OapiModels.AnswersView();
|
||||
view.setRows(items);
|
||||
return view;
|
||||
}
|
||||
|
||||
/** 行字段闭集（§6.1）：无库内 id；{@code meta} 只有匿名行有值（C12），登录行两项都是 null */
|
||||
private static OapiModels.AnswerRow toPullRow(Answer answer) {
|
||||
OapiModels.AnswerRowMeta meta = new OapiModels.AnswerRowMeta();
|
||||
meta.setIp(answer.getAnonIp());
|
||||
meta.setUa(answer.getAnonUa());
|
||||
|
||||
OapiModels.AnswerRow row = new OapiModels.AnswerRow();
|
||||
row.setAnswerUuid(answer.getUuid());
|
||||
row.setSchId(answer.getSchId());
|
||||
row.setAppCode(answer.getAppCode());
|
||||
row.setFormUuid(answer.getFormUuid());
|
||||
row.setFormCode(answer.getCode());
|
||||
row.setVersion(answer.getVersion());
|
||||
row.setAnswerNo(answer.getAnswerNo());
|
||||
row.setSubmitterId(answer.getSubmitterId());
|
||||
row.setSubjectId(answer.getSubjectId());
|
||||
row.setStatus(answer.getStatus());
|
||||
row.setValues(FormSchemaValidator.parse(answer.getValuesJson(), "答卷值文档损坏"));
|
||||
row.setMeta(meta);
|
||||
row.setCreateTime(answer.getCreateTime() == null ? null : CURSOR_TS.format(answer.getCreateTime()));
|
||||
row.setUpdateTime(answer.getUpdateTime() == null ? null : CURSOR_TS.format(answer.getUpdateTime()));
|
||||
row.setIsDeleted(answer.getIsDeleted());
|
||||
return row;
|
||||
}
|
||||
|
||||
private static int requireLimit(Integer limit) {
|
||||
if (limit == null) {
|
||||
return DEFAULT_LIMIT;
|
||||
}
|
||||
if (limit < 1 || limit > MAX_LIMIT) {
|
||||
throw new BizException(ErrorCode.PARAM_ERROR, "limit 需在 1~" + MAX_LIMIT + " 之间（默认 " + DEFAULT_LIMIT + "）");
|
||||
}
|
||||
return limit;
|
||||
}
|
||||
|
||||
/**
|
||||
* updated 形态的游标解析（C9）。
|
||||
*
|
||||
* <p>首拉=裸 {@code <ISO8601>}（或干脆不给=从最早一行起算）；续拉必须能带回复合键的第二段
|
||||
* {@code <ISO8601>|<answer_uuid>}——否则同一时刻的并列行要么被跳过、要么被无限重放
|
||||
* （落地差额 P126：C9 只写了"复合推进键"，没写游标的线格式）。
|
||||
* 输入兼容空格分隔符（本仓其它时间串的形状），输出恒为 ISO 的 {@code T} 形态。</p>
|
||||
*/
|
||||
private static Cursor cursorOf(String raw) {
|
||||
if (raw == null || raw.trim().isEmpty()) {
|
||||
return new Cursor(PULL_FROM_BEGINNING, "");
|
||||
}
|
||||
String clean = raw.trim();
|
||||
int bar = clean.indexOf('|');
|
||||
String timePart = bar < 0 ? clean : clean.substring(0, bar).trim();
|
||||
String uuidPart = bar < 0 ? "" : clean.substring(bar + 1).trim();
|
||||
LocalDateTime time;
|
||||
try {
|
||||
time = LocalDateTime.parse(timePart.replace(' ', 'T'));
|
||||
} catch (RuntimeException e) {
|
||||
throw new BizException(ErrorCode.PARAM_ERROR,
|
||||
"updated_since 需为 ISO8601 本地时间 yyyy-MM-ddTHH:mm:ss（可带 |answer_uuid 续拉）：" + raw);
|
||||
}
|
||||
// 消费方即使自己带了纳秒过来也先抹平（列精度只到秒，见 CURSOR_TS 注释那条）
|
||||
return new Cursor(time.withNano(0), uuidPart);
|
||||
}
|
||||
|
||||
private static boolean notBlank(String raw) {
|
||||
return raw != null && !raw.trim().isEmpty();
|
||||
}
|
||||
|
||||
/** updated 游标的两段（{@code answer_uuid} 空串=该时刻从头开始） */
|
||||
private static final class Cursor {
|
||||
private final LocalDateTime time;
|
||||
private final String answerUuid;
|
||||
|
||||
private Cursor(LocalDateTime time, String answerUuid) {
|
||||
this.time = time;
|
||||
this.answerUuid = answerUuid;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user