java-project/baiyue-firstleg
4
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix(order): complete root fulfillment only after effective remainders

Browse Source
This commit is contained in:
ouyangxq
2026-09-30 13:49:40 +08:00
parent bd73c7a326
commit 13028d8a5a
25 changed files with 684 additions and 39 deletions
Download Patch File Download Diff File Expand all files Collapse all files
order/src/main/java/com/baiyue/order/client/FirstlegInboundClient.java
+7
View File
@@ -14,6 +14,13 @@ import com.baiyue.order.dto.OrderDeclarationDtos;
@FeignClient(name = "firstleg-service", contextId = "orderFirstlegInboundClient",
url = "${baiyue.clients.firstleg.url:}", path = "/internal/firstleg/v1/inbounds")
public interface FirstlegInboundClient {
@PostMapping("/order-fulfillment-completed")
ApiResponse<Integer> completeFulfillmentSnapshot(@RequestHeader("X-Internal-Service-Token") String token,
@RequestBody FulfillmentCompletedCommand command);
record FulfillmentCompletedCommand(long tenantId, long orderId, String orderNo, String inboundCode,
LocalDateTime completedAt) { }
@PostMapping("/order-approved")
ApiResponse<Long> create(@RequestHeader("X-Internal-Service-Token") String token,
@RequestBody OrderApprovedInboundCommand command);
order/src/main/java/com/baiyue/order/mapper/OrderMapper.java
+8
View File
@@ -12,9 +12,17 @@ import java.util.Map;
*/
@Mapper
public interface OrderMapper extends BaseMapper<OrderEntity>{
@Select("SELECT * FROM ord_order WHERE tenant_id=#{tenantId} AND id=#{orderId} AND deleted=0")
OrderEntity findOwned(@Param("tenantId") long tenantId, @Param("orderId") long orderId);
@Select("SELECT * FROM ord_order WHERE tenant_id=#{tenantId} AND id=#{orderId} AND deleted=0 FOR UPDATE")
OrderEntity lockOwned(@Param("tenantId") long tenantId, @Param("orderId") long orderId);
@Select("SELECT * FROM ord_order WHERE tenant_id=#{tenantId} AND parent_order_id=#{rootOrderId} "
+ "AND deleted=0 ORDER BY id FOR UPDATE")
java.util.List<OrderEntity> lockFulfillmentChildren(@Param("tenantId") long tenantId,
@Param("rootOrderId") long rootOrderId);
@Select("""
SELECT COUNT(*) AS total_submitted_orders,
COALESCE(SUM(CASE WHEN submitted_at >= #{monthStart}
order/src/main/java/com/baiyue/order/mapper/OrderStatusHistoryMapper.java
+6 -1
View File
@@ -7,4 +7,9 @@ import com.baomidou.mybatisplus.core.mapper.BaseMapper;
* 订单状态历史 Mapper，提供 {@code ord_order_status_history} 的 MyBatis-Plus 基础数据访问能力。
*/
@Mapper
public interface OrderStatusHistoryMapper extends BaseMapper<OrderStatusHistoryEntity>{}
public interface OrderStatusHistoryMapper extends BaseMapper<OrderStatusHistoryEntity>{
@org.apache.ibatis.annotations.Select("SELECT MAX(occurred_at) FROM ord_order_status_history "
+ "WHERE tenant_id=#{tenantId} AND order_id=#{orderId} AND action='OVERSEAS_DELIVERED'")
java.time.LocalDateTime overseasDeliveredAt(@org.apache.ibatis.annotations.Param("tenantId") long tenantId,
@org.apache.ibatis.annotations.Param("orderId") long orderId);
}
order/src/main/java/com/baiyue/order/service/OrderApplicationService.java
+3
View File
@@ -46,6 +46,7 @@ public class OrderApplicationService {
private static final ZoneId INBOUND_CODE_ZONE = ZoneId.of("Asia/Shanghai");
private static final DateTimeFormatter INBOUND_CODE_DATE = DateTimeFormatter.ofPattern("yyMMdd");
@Resource private OrderRepository repository;
@Resource private OrderFulfillmentCompletionService fulfillmentCompletion;
@Resource private ObjectMapper json;
@Resource private OrderEligibilityGuard eligibilityGuard;
@Resource private OrderMasterDataGuard masterDataGuard;
@@ -173,6 +174,7 @@ public class OrderApplicationService {
@Transactional
public void transit(long tenantId, long userId, String username, long id, OrderStatus target, String reason,
Integer expectedVersion) {
if (target == OrderStatus.CANCELLED) fulfillmentCompletion.lockCancellationRoot(tenantId, id);
OrderEntity order = require(tenantId, id);
if (expectedVersion != null && !expectedVersion.equals(order.getVersion())) {
throw new BizException(ResultCode.CONFLICT, "订单版本已变化，请刷新后重试");
@@ -210,6 +212,7 @@ public class OrderApplicationService {
if (target == OrderStatus.SUBMITTED || target == OrderStatus.APPROVED || target == OrderStatus.CANCELLED) {
writeOutbox(order, target);
}
if (target == OrderStatus.CANCELLED) fulfillmentCompletion.afterCancellation(tenantId, id);
}
/** 提交订单的幂等入口。 */
order/src/main/java/com/baiyue/order/service/OrderFulfillmentCompletionService.java
+115 -19
View File
@@ -25,42 +25,138 @@ public class OrderFulfillmentCompletionService {
OrderStatus.APPROVED, OrderStatus.ACCEPTED, OrderStatus.PROCESSING);
@Resource private OrderMapper orders;
@Resource private OrderStatusHistoryMapper histories;
@Resource private com.baiyue.order.mapper.OutboxEventMapper outbox;
@Resource private com.fasterxml.jackson.databind.ObjectMapper json;
@Transactional
public Result complete(Request request) {
long tenantId = positive(request.tenantId(), "客户 ID 无效");
long orderId = positive(request.orderId(), "订单 ID 无效");
OrderEntity order = orders.lockOwned(tenantId, orderId);
if (order == null) throw new BizException(ResultCode.NOT_FOUND, "来源订单不存在");
OrderEntity source = orders.findOwned(tenantId, orderId);
if (source == null) throw new BizException(ResultCode.NOT_FOUND, "来源订单不存在");
long rootId = rootId(source);
if (positive(request.rootOrderId(), "履约根订单 ID 无效") != rootId)
throw new BizException(ResultCode.CONFLICT, "派送完成事件与履约根订单不一致");
// All fulfillment writers serialize on the root before locking a child, matching split creation.
OrderEntity root = orders.lockOwned(tenantId, rootId);
if (root == null || !"MASTER".equals(root.getOrderRole()))
throw new BizException(ResultCode.CONFLICT, "来源订单履约根不存在或层级非法");
OrderEntity order = orderId == rootId ? root : orders.lockOwned(tenantId, orderId);
if (order == null || rootId(order) != rootId)
throw new BizException(ResultCode.CONFLICT, "来源订单履约层级已变化");
if (!Objects.equals(order.getOrderNo(), request.orderNo())
|| !Objects.equals(order.getInboundCode(), request.inboundCode())) {
throw new BizException(ResultCode.CONFLICT, "派送完成事件与来源订单标识不一致");
}
if (order.getStatus() == OrderStatus.COMPLETED) {
return new Result(orderId, order.getStatus().name(), true);
}
if (!COMPLETABLE.contains(order.getStatus())) {
throw new BizException(ResultCode.CONFLICT, "来源订单当前状态不允许完成: " + order.getStatus());
}
OrderStatus from = order.getStatus();
requireCompletable(order);
requireCompletable(root);
LocalDateTime completedAt = LocalDateTime.ofInstant(request.completedAt(), ZoneOffset.UTC);
LocalDateTime ownDeliveredAt = histories.overseasDeliveredAt(tenantId, orderId);
boolean replayed = ownDeliveredAt != null;
if (!replayed) {
if (orderId == rootId) {
// MASTER also carries its first physical shipment: record that fact without completing
// the aggregate while an effective remainder/other shipment is outstanding.
record(order, order.getStatus(), "OVERSEAS_DELIVERED", request.deliveryNo(), completedAt);
} else {
finish(order, completedAt, "OVERSEAS_DELIVERED", request.deliveryNo());
}
}
LocalDateTime rootDeliveredAt = orderId == rootId && !replayed
? completedAt : histories.overseasDeliveredAt(tenantId, rootId);
aggregate(root, order, rootDeliveredAt, request.deliveryNo());
return new Result(orderId, order.getStatus().name(), replayed, rootId, root.getStatus().name());
}
/** Acquire the same root lock before the cancellation writer takes a child row lock. */
@Transactional(propagation = org.springframework.transaction.annotation.Propagation.MANDATORY)
public void lockCancellationRoot(long tenantId, long orderId) {
OrderEntity source = orders.findOwned(tenantId, orderId);
if (source == null) throw new BizException(ResultCode.NOT_FOUND, "来源订单不存在");
orders.lockOwned(tenantId, rootId(source));
}
/** Called in the cancellation transaction after the valid terminal branch is persisted. */
@Transactional(propagation = org.springframework.transaction.annotation.Propagation.MANDATORY)
public void afterCancellation(long tenantId, long orderId) {
OrderEntity source = orders.findOwned(tenantId, orderId);
if (source == null || !"CHILD".equals(source.getOrderRole())) return;
OrderEntity root = orders.lockOwned(tenantId, rootId(source));
if (root == null || root.getStatus() == OrderStatus.CANCELLED) return;
requireCompletable(root);
aggregate(root, source, histories.overseasDeliveredAt(tenantId, root.getId()), "余量取消后履约汇总");
}
private void aggregate(OrderEntity root, OrderEntity order, LocalDateTime rootDeliveredAt, String deliveryNo) {
long tenantId = root.getTenantId(), rootId = root.getId(), orderId = order.getId();
if (root.getStatus() != OrderStatus.COMPLETED && rootDeliveredAt != null) {
java.util.List<OrderEntity> children = orders.lockFulfillmentChildren(tenantId, rootId);
boolean allComplete = true;
LocalDateTime aggregateAt = rootDeliveredAt;
for (OrderEntity child : children) {
if (!"CHILD".equals(child.getOrderRole()) || !Objects.equals(child.getRootOrderId(), rootId))
throw new BizException(ResultCode.CONFLICT, "余量订单履约层级非法");
// CANCELLED is the only explicit non-effective terminal branch. Unknown, draft,
// rejected or pending statuses must not silently disappear from this gate.
if (child.getStatus() == OrderStatus.CANCELLED) continue;
OrderStatus status = child.getId().equals(orderId) ? order.getStatus() : child.getStatus();
if (status != OrderStatus.COMPLETED) allComplete = false;
LocalDateTime childAt = child.getId().equals(orderId) ? order.getCompletedAt() : child.getCompletedAt();
if (childAt != null && childAt.isAfter(aggregateAt)) aggregateAt = childAt;
}
if (allComplete) finish(root, aggregateAt, "FULFILLMENT_COMPLETED", deliveryNo);
}
}
private static long rootId(OrderEntity order) {
if ("MASTER".equals(order.getOrderRole())) return order.getId();
if (!"CHILD".equals(order.getOrderRole()) || order.getRootOrderId() == null
|| !Objects.equals(order.getRootOrderId(), order.getParentOrderId()))
throw new BizException(ResultCode.CONFLICT, "来源订单履约层级非法");
return order.getRootOrderId();
}
private static void requireCompletable(OrderEntity order) {
if (order.getStatus() != OrderStatus.COMPLETED && !COMPLETABLE.contains(order.getStatus()))
throw new BizException(ResultCode.CONFLICT, "来源订单当前状态不允许完成: " + order.getStatus());
}
private void finish(OrderEntity order, LocalDateTime completedAt, String action, String deliveryNo) {
if (order.getStatus() == OrderStatus.COMPLETED) return;
int version = order.getVersion() == null ? 0 : order.getVersion();
int updated = orders.update(null, new LambdaUpdateWrapper<OrderEntity>()
.eq(OrderEntity::getTenantId, tenantId).eq(OrderEntity::getId, orderId)
.eq(OrderEntity::getTenantId, order.getTenantId()).eq(OrderEntity::getId, order.getId())
.eq(OrderEntity::getVersion, version).eq(OrderEntity::getDeleted, false)
.set(OrderEntity::getStatus, OrderStatus.COMPLETED)
.set(OrderEntity::getCompletedAt, completedAt)
.set(OrderEntity::getUpdatedBy, 0L).set(OrderEntity::getUpdatedAt, completedAt)
.setSql("version=version+1"));
if (updated != 1) throw new BizException(ResultCode.CONFLICT, "来源订单版本已变化，请重试");
record(order, OrderStatus.COMPLETED, action, deliveryNo, completedAt);
order.setStatus(OrderStatus.COMPLETED); order.setCompletedAt(completedAt); order.setVersion(version + 1);
com.baiyue.order.entity.OutboxEventEntity event = new com.baiyue.order.entity.OutboxEventEntity();
event.setId(IdWorker.getId()); event.setTenantId(order.getTenantId());
event.setAggregateType("FIRST_LEG_ORDER"); event.setAggregateId(order.getId());
event.setEventType("FIRST_LEG_ORDER_FULFILLMENT_COMPLETED");
try {
event.setPayloadJson(json.writeValueAsString(new com.baiyue.order.client.FirstlegInboundClient
.FulfillmentCompletedCommand(order.getTenantId(), order.getId(), order.getOrderNo(),
order.getInboundCode(), completedAt)));
} catch (com.fasterxml.jackson.core.JsonProcessingException failure) {
throw new BizException(ResultCode.INTERNAL_ERROR, "履约完成快照序列化失败");
}
event.setPublishStatus("PENDING"); event.setRetryCount(0); event.setCreatedAt(LocalDateTime.now(ZoneOffset.UTC));
if (outbox.insert(event) != 1) throw new BizException(ResultCode.CONFLICT, "履约完成事件写入失败");
}
private void record(OrderEntity order, OrderStatus to, String action, String deliveryNo, LocalDateTime at) {
OrderStatusHistoryEntity history = new OrderStatusHistoryEntity();
history.setId(IdWorker.getId()); history.setTenantId(tenantId); history.setOrderId(orderId);
history.setFromStatus(from); history.setToStatus(OrderStatus.COMPLETED);
history.setAction("OVERSEAS_DELIVERED");
history.setReason("海外派送单 " + request.deliveryNo() + " 已签收");
history.setOperatorId(0L); history.setOperatorName("OVERSEAS_EVENT_BRIDGE");
history.setOccurredAt(completedAt); histories.insert(history);
return new Result(orderId, OrderStatus.COMPLETED.name(), false);
history.setId(IdWorker.getId()); history.setTenantId(order.getTenantId()); history.setOrderId(order.getId());
history.setFromStatus(order.getStatus()); history.setToStatus(to); history.setAction(action);
history.setReason("海外派送单 " + deliveryNo + " 已签收");
history.setOperatorId(0L); history.setOperatorName("OVERSEAS_DELIVERY_EVENT");
history.setOccurredAt(at);
if (histories.insert(history) != 1) throw new BizException(ResultCode.CONFLICT, "派送完成历史写入失败");
}
private static long positive(String value, String message) {
@@ -69,8 +165,8 @@ public class OrderFulfillmentCompletionService {
throw new BizException(ResultCode.BAD_REQUEST, message);
}
public record Request(String tenantId, String orderId, String orderNo, String inboundCode,
public record Request(String tenantId, String orderId, String orderNo, String inboundCode, String rootOrderId,
String deliveryNo, java.time.Instant completedAt, String eventId,
String payloadHash) { }
public record Result(Long orderId, String status, boolean replayed) { }
public record Result(Long orderId, String status, boolean replayed, Long rootOrderId, String rootStatus) { }
}
order/src/main/java/com/baiyue/order/service/OrderInboundOutboxPublisher.java
+22
View File
@@ -38,6 +38,10 @@ public class OrderInboundOutboxPublisher {
publishDeclarationUpdate(event);
continue;
}
if ("FIRST_LEG_ORDER_FULFILLMENT_COMPLETED".equals(event.getEventType())) {
publishFulfillmentCompletion(event);
continue;
}
boolean preparing = "FIRST_LEG_ORDER_PREPARED".equals(event.getEventType());
if (!preparing && !"FIRST_LEG_ORDER_APPROVED".equals(event.getEventType())) {
failPermanently(event, "unsupported order event");
@@ -87,6 +91,24 @@ public class OrderInboundOutboxPublisher {
}
}
private void publishFulfillmentCompletion(OutboxEventEntity event) {
try {
var command = json.readValue(event.getPayloadJson(), FirstlegInboundClient.FulfillmentCompletedCommand.class);
if (command == null || event.getTenantId() == null || event.getAggregateId() == null
|| event.getTenantId() != command.tenantId() || event.getAggregateId() != command.orderId()) {
failPermanently(event, "completion event ownership does not match payload"); return;
}
var response = client.completeFulfillmentSnapshot(internalToken, command);
if (response == null || !response.successful() || response.data() == null)
throw new IllegalStateException("头程履约完成快照同步失败");
repository.markOutboxPublished(event.getTenantId(), event.getId(), LocalDateTime.now());
} catch (com.fasterxml.jackson.core.JsonProcessingException invalid) {
failPermanently(event, "invalid completion event payload");
} catch (Exception failure) {
repository.markOutboxRetry(event.getTenantId(), event.getId(), LocalDateTime.now().plusSeconds(10));
}
}
private void publishDeclarationUpdate(OutboxEventEntity event) {
try {
OrderDeclarationDtos.UpdatedEvent command = json.readValue(
order/src/main/resources/openapi/order-v1.yaml
+25 -4
View File
@@ -9,10 +9,13 @@ paths:
/internal/order/v1/fulfillment-completions:
post:
security: []
summary: Firstleg 回传海外派送完成并结束对应订单
summary: Firstleg 回传来源实装签收并聚合履约根订单完成
description: >-
仅接受内部服务令牌。请求必须携带海外事件已验证的来源订单 ID、订单号、入仓码、
FBA 派送单号和完成时间；只推进该派送单关联的订单，COMPLETED 重放返回成功。
FBA 派送单号和完成时间。来源根订单先持久化自身实装签收事实，只有全部有效余量子单
已完成才完成根订单；来源子单独立完成并在根锁下重新评估聚合。明确 CANCELLED 子单
不参与完成门禁，其他未完成状态均阻断。重复事实不得重复写历史。响应明确返回来源状态
status 和履约根 rootOrderId/rootStatus；根尚未聚合完成时允许保留其原状态。
parameters:
- name: X-Internal-Service-Token
in: header
@@ -24,18 +27,36 @@ paths:
application/json:
schema:
type: object
required: [tenantId, orderId, orderNo, inboundCode, deliveryNo, completedAt, eventId, payloadHash]
required: [tenantId, orderId, orderNo, inboundCode, rootOrderId, deliveryNo, completedAt, eventId, payloadHash]
properties:
tenantId: {type: string, pattern: '^[1-9][0-9]*$'}
orderId: {type: string, pattern: '^[1-9][0-9]*$'}
orderNo: {type: string}
inboundCode: {type: string}
rootOrderId: {type: string, pattern: '^[1-9][0-9]*$'}
deliveryNo: {type: string}
completedAt: {type: string, format: date-time}
eventId: {type: string, format: uuid}
payloadHash: {type: string, pattern: '^[0-9a-fA-F]{64}$'}
responses:
'200': {description: 对应订单已完成或幂等重放成功}
'200':
description: 来源实装签收事实已保存，返回来源和履约根各自状态
content:
application/json:
schema:
type: object
required: [code, data]
properties:
code: {type: string}
data:
type: object
required: [orderId, status, replayed, rootOrderId, rootStatus]
properties:
orderId: {type: integer, format: int64}
status: {type: string, enum: [APPROVED, ACCEPTED, PROCESSING, COMPLETED]}
replayed: {type: boolean}
rootOrderId: {type: integer, format: int64}
rootStatus: {type: string, enum: [APPROVED, ACCEPTED, PROCESSING, COMPLETED]}
'403': {description: 内部服务令牌无效}
/internal/order/v1/cargoes/{tenantId}/{orderId}:
get: