diff --git a/firstleg/src/main/java/com/baiyue/firstleg/client/OrderFulfillmentCompletionClient.java b/firstleg/src/main/java/com/baiyue/firstleg/client/OrderFulfillmentCompletionClient.java index 8833d0115..1e8a0e4b1 100644 --- a/firstleg/src/main/java/com/baiyue/firstleg/client/OrderFulfillmentCompletionClient.java +++ b/firstleg/src/main/java/com/baiyue/firstleg/client/OrderFulfillmentCompletionClient.java @@ -15,7 +15,7 @@ public interface OrderFulfillmentCompletionClient { ApiResponse complete(@RequestHeader("X-Internal-Service-Token") String token, @RequestBody Request request); - record Request(String tenantId, String orderId, String orderNo, String inboundCode, + record Request(String tenantId, String orderId, String orderNo, String inboundCode, String rootOrderId, String deliveryNo, Instant completedAt, String eventId, String payloadHash) { } - record Result(Long orderId, String status, boolean replayed) { } + record Result(Long orderId, String status, boolean replayed, Long rootOrderId, String rootStatus) { } } diff --git a/firstleg/src/main/java/com/baiyue/firstleg/controller/InternalInboundOrderController.java b/firstleg/src/main/java/com/baiyue/firstleg/controller/InternalInboundOrderController.java index 24b464e85..b8f0ced52 100644 --- a/firstleg/src/main/java/com/baiyue/firstleg/controller/InternalInboundOrderController.java +++ b/firstleg/src/main/java/com/baiyue/firstleg/controller/InternalInboundOrderController.java @@ -47,6 +47,7 @@ public class InternalInboundOrderController { private OrderDeclarationSyncService declarationSync; @Resource private OrderRemarksSyncService remarksSync; + @Resource private com.baiyue.firstleg.service.OrderFulfillmentSnapshotService fulfillmentSnapshot; @Value("${baiyue.security.internal-service-token}") private String internalToken; private byte[] expectedToken; @@ -56,6 +57,14 @@ public class InternalInboundOrderController { expectedToken = internalToken.getBytes(StandardCharsets.UTF_8); } + @PostMapping("/order-fulfillment-completed") + public ApiResponse completeFulfillmentSnapshot( + @RequestHeader(value = "X-Internal-Service-Token", required = false) String token, + @Valid @RequestBody com.baiyue.firstleg.dto.OrderFulfillmentCompleted command) { + requireToken(token); + return ApiResponse.success(fulfillmentSnapshot.complete(command)); + } + @PostMapping("/order-approved") public ApiResponse createFromApprovedOrder( @RequestHeader(value = "X-Internal-Service-Token", required = false) String token, diff --git a/firstleg/src/main/java/com/baiyue/firstleg/dto/OrderFulfillmentCompleted.java b/firstleg/src/main/java/com/baiyue/firstleg/dto/OrderFulfillmentCompleted.java new file mode 100644 index 000000000..d9cb5cfef --- /dev/null +++ b/firstleg/src/main/java/com/baiyue/firstleg/dto/OrderFulfillmentCompleted.java @@ -0,0 +1,11 @@ +package com.baiyue.firstleg.dto; + +import jakarta.validation.constraints.NotBlank; +import jakarta.validation.constraints.NotNull; +import jakarta.validation.constraints.Positive; +import java.time.LocalDateTime; + +/** Order authoritative final state, delivered through its transactional Outbox. */ +public record OrderFulfillmentCompleted(@Positive long tenantId, @Positive long orderId, + @NotBlank String orderNo, @NotBlank String inboundCode, + @NotNull LocalDateTime completedAt) { } diff --git a/firstleg/src/main/java/com/baiyue/firstleg/service/OrderFulfillmentSnapshotService.java b/firstleg/src/main/java/com/baiyue/firstleg/service/OrderFulfillmentSnapshotService.java new file mode 100644 index 000000000..e6590b88b --- /dev/null +++ b/firstleg/src/main/java/com/baiyue/firstleg/service/OrderFulfillmentSnapshotService.java @@ -0,0 +1,32 @@ +package com.baiyue.firstleg.service; + +import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper; +import com.baiyue.common.core.api.ResultCode; +import com.baiyue.common.core.exception.BizException; +import com.baiyue.firstleg.dto.OrderFulfillmentCompleted; +import com.baiyue.firstleg.entity.InboundOrderEntity; +import com.baiyue.firstleg.mapper.InboundOrderMapper; +import lombok.RequiredArgsConstructor; +import org.springframework.stereotype.Service; +import org.springframework.transaction.annotation.Transactional; + +@Service +@RequiredArgsConstructor +public class OrderFulfillmentSnapshotService { + private final InboundOrderMapper inbounds; + + @Transactional + public int complete(OrderFulfillmentCompleted command) { + var rows = inbounds.selectList(new LambdaQueryWrapper() + .eq(InboundOrderEntity::getTenantId, command.tenantId()) + .eq(InboundOrderEntity::getOrderId, command.orderId()) + .eq(InboundOrderEntity::getDeleted, false).orderByAsc(InboundOrderEntity::getId).last("FOR UPDATE")); + if (rows.isEmpty()) throw new BizException(ResultCode.NOT_FOUND, "来源订单入仓单不存在"); + if (rows.stream().anyMatch(row -> !command.orderNo().equals(row.getOrderNo()) + || !command.inboundCode().equals(row.getInboundCode()))) + throw new BizException(ResultCode.CONFLICT, "履约完成快照与来源订单标识不一致"); + if (rows.stream().anyMatch(row -> !"DELIVERED".equals(row.getProcessStatus()))) + throw new BizException(ResultCode.CONFLICT, "来源订单仍有未签收入仓单"); + return inbounds.markOrderSnapshotCompleted(command.tenantId(), command.orderId(), command.completedAt()); + } +} diff --git a/firstleg/src/main/java/com/baiyue/firstleg/service/OverseasDeliveryCompletedConsumer.java b/firstleg/src/main/java/com/baiyue/firstleg/service/OverseasDeliveryCompletedConsumer.java index cafd14e10..9362d35b4 100644 --- a/firstleg/src/main/java/com/baiyue/firstleg/service/OverseasDeliveryCompletedConsumer.java +++ b/firstleg/src/main/java/com/baiyue/firstleg/service/OverseasDeliveryCompletedConsumer.java @@ -84,20 +84,33 @@ public class OverseasDeliveryCompletedConsumer { long sourceOrderId = positive(payload.sourceOrderId(), "来源订单 ID 无效"); if (inboundOrders.countNotDeliveredByOrder(tenantId, sourceOrderId) == 0) { var response = orderClient.complete(internalToken, new OrderFulfillmentCompletionClient.Request( - event.tenantId(), payload.sourceOrderId(), payload.sourceOrderNo(), payload.inboundCode(), + event.tenantId(), payload.sourceOrderId(), payload.sourceOrderNo(), payload.inboundCode(), payload.rootOrderId(), payload.deliveryNo(), payload.completedAt(), event.eventId(), event.payloadHash())); if (response == null || !response.successful() || response.data() == null - || !"COMPLETED".equals(response.data().status())) { - throw new IllegalStateException("Order 服务未确认来源订单完成"); + || !Objects.equals(response.data().orderId(), sourceOrderId) + || !Objects.equals(response.data().rootOrderId(), positive(payload.rootOrderId(), "履约根订单 ID 无效")) + || !completionStatus(response.data().rootStatus()) + || !("COMPLETED".equals(response.data().status()) + || (response.data().rootOrderId() == sourceOrderId + && Objects.equals(response.data().status(), response.data().rootStatus()) + && completionStatus(response.data().status())))) { + throw new IllegalStateException("Order 服务未确认来源订单履约事实"); } - inboundOrders.markOrderSnapshotCompleted(tenantId, sourceOrderId, - LocalDateTime.ofInstant(payload.completedAt(), ZoneOffset.UTC)); + LocalDateTime completedAt = LocalDateTime.ofInstant(payload.completedAt(), ZoneOffset.UTC); + if ("COMPLETED".equals(response.data().status())) + inboundOrders.markOrderSnapshotCompleted(tenantId, sourceOrderId, completedAt); + if (response.data().rootOrderId() != sourceOrderId && "COMPLETED".equals(response.data().rootStatus())) + inboundOrders.markOrderSnapshotCompleted(tenantId, response.data().rootOrderId(), completedAt); } inboxRow.setStatus("PROCESSED"); inboxRow.setProcessedAt(LocalDateTime.now(ZoneOffset.UTC)); inbox.updateById(inboxRow); return DeliveryAck.of(event, DeliveryAck.Outcome.APPLIED); } + private static boolean completionStatus(String status) { + return status != null && java.util.Set.of("APPROVED", "ACCEPTED", "PROCESSING", "COMPLETED").contains(status); + } + private void validate(EventEnvelope event) { event.requireValid(); if (!EventTypes.OVERSEAS_DELIVERY_COMPLETED.equals(event.eventType()) || event.eventVersion() != 1 @@ -114,7 +127,7 @@ public class OverseasDeliveryCompletedConsumer { OverseasDeliveryCompletedV1 value = objectMapper.treeToValue(event.payload(), OverseasDeliveryCompletedV1.class); if (value.completedAt() == null || !StringUtils.hasText(value.deliveryNo()) || !StringUtils.hasText(value.sourceOrderId()) || !StringUtils.hasText(value.sourceOrderNo()) - || !StringUtils.hasText(value.inboundCode())) throw new EventProtocolException("字段不完整"); + || !StringUtils.hasText(value.inboundCode()) || !StringUtils.hasText(value.rootOrderId())) throw new EventProtocolException("字段不完整"); return value; } catch (Exception exception) { throw new EventProtocolException("海外派送完成事件负载无效", exception); diff --git a/firstleg/src/main/resources/openapi/firstleg-v1.yaml b/firstleg/src/main/resources/openapi/firstleg-v1.yaml index 1d9490311..026d8a674 100644 --- a/firstleg/src/main/resources/openapi/firstleg-v1.yaml +++ b/firstleg/src/main/resources/openapi/firstleg-v1.yaml @@ -303,6 +303,29 @@ paths: schema: {$ref: '#/components/schemas/OrderDeclarationUpdatedEvent'} responses: '200': {description: 成功、幂等重放或旧版本已忽略；冲突时响应体 code=A0409} + /internal/firstleg/v1/inbounds/order-fulfillment-completed: + post: + summary: 接收 Order 履约最终完成并幂等同步入仓订单状态摘要 + description: >- + 由 Order 事务 Outbox 可靠投递，签收聚合和最后有效余量取消都使用同一正式完成事实。 + 按客户、来源订单号及入仓码核验全部有效入仓已为 DELIVERED 后更新摘要；重复调用不重复递增版本。 + security: + - internalServiceToken: [] + requestBody: + required: true + content: + application/json: + schema: + type: object + required: [tenantId, orderId, orderNo, inboundCode, completedAt] + properties: + tenantId: {type: integer, format: int64, minimum: 1} + orderId: {type: integer, format: int64, minimum: 1} + orderNo: {type: string, minLength: 1} + inboundCode: {type: string, minLength: 1} + completedAt: {type: string, format: date-time} + responses: + '200': {description: data 返回实际更新入仓摘要数量，幂等重放为 0；未签收或来源不符时拒绝写入} /internal/firstleg/v1/inbounds/order-remarks-updated: post: summary: 同步 Order 权威协作备注到未冻结预装明细 diff --git a/firstleg/src/test/java/com/baiyue/firstleg/controller/InternalInboundOrderReferenceControllerTest.java b/firstleg/src/test/java/com/baiyue/firstleg/controller/InternalInboundOrderReferenceControllerTest.java index 5cd1aec90..63985272b 100644 --- a/firstleg/src/test/java/com/baiyue/firstleg/controller/InternalInboundOrderReferenceControllerTest.java +++ b/firstleg/src/test/java/com/baiyue/firstleg/controller/InternalInboundOrderReferenceControllerTest.java @@ -39,9 +39,21 @@ class InternalInboundOrderReferenceControllerTest { private OrderDeclarationSyncService declarationSync; @MockBean private OrderRemarksSyncService remarksSync; + @MockBean private com.baiyue.firstleg.service.OrderFulfillmentSnapshotService fulfillmentSnapshot; @Resource private InternalInboundOrderController controller; + @Test + void completionSnapshotRequiresInternalTokenAndDelegatesExactCommand() { + var command = new com.baiyue.firstleg.dto.OrderFulfillmentCompleted(11, 21, + "FL-21", "IN-21", LocalDateTime.of(2026, 9, 30, 12, 0)); + assertThatThrownBy(() -> controller.completeFulfillmentSnapshot("wrong", command)).isInstanceOf(BizException.class); + org.mockito.Mockito.verifyNoInteractions(fulfillmentSnapshot); + when(fulfillmentSnapshot.complete(command)).thenReturn(0); + assertThat(controller.completeFulfillmentSnapshot("internal-token", command).data()).isZero(); + verify(fulfillmentSnapshot).complete(command); + } + @Test void shouldHandleCase01() { var request = new InboundOrderReference.BatchRequest(10001L, List.of(77L, 77L, 78L)); diff --git a/firstleg/src/test/java/com/baiyue/firstleg/service/OrderFulfillmentSnapshotServiceTest.java b/firstleg/src/test/java/com/baiyue/firstleg/service/OrderFulfillmentSnapshotServiceTest.java new file mode 100644 index 000000000..8d125f39b --- /dev/null +++ b/firstleg/src/test/java/com/baiyue/firstleg/service/OrderFulfillmentSnapshotServiceTest.java @@ -0,0 +1,52 @@ +package com.baiyue.firstleg.service; + +import com.baomidou.mybatisplus.core.MybatisConfiguration; +import com.baomidou.mybatisplus.core.metadata.TableInfoHelper; +import com.baiyue.common.core.exception.BizException; +import com.baiyue.firstleg.dto.OrderFulfillmentCompleted; +import com.baiyue.firstleg.entity.InboundOrderEntity; +import com.baiyue.firstleg.mapper.InboundOrderMapper; +import org.apache.ibatis.builder.MapperBuilderAssistant; +import org.junit.jupiter.api.BeforeAll; +import org.junit.jupiter.api.Test; +import java.time.LocalDateTime; +import java.util.List; +import static org.assertj.core.api.Assertions.*; +import static org.mockito.Mockito.*; +import static org.mockito.ArgumentMatchers.*; + +class OrderFulfillmentSnapshotServiceTest { + @BeforeAll static void metadata() { + TableInfoHelper.initTableInfo(new MapperBuilderAssistant(new MybatisConfiguration(), "snapshot-test"), InboundOrderEntity.class); + } + @Test void completedAndReplaySyncOnlyAuthenticatedSource() { + var mapper = mock(InboundOrderMapper.class); + when(mapper.selectList(any())).thenReturn(List.of(row("FL21", "IN21", "DELIVERED"))); + when(mapper.markOrderSnapshotCompleted(eq(11L),eq(21L),any())).thenReturn(1,0); + var service = new OrderFulfillmentSnapshotService(mapper); + assertThat(service.complete(command())).isEqualTo(1); + assertThat(service.complete(command())).isZero(); + var query = org.mockito.ArgumentCaptor.forClass(com.baomidou.mybatisplus.core.conditions.Wrapper.class); + verify(mapper, times(2)).selectList(query.capture()); + assertThat(query.getValue().getSqlSegment()).contains("tenant_id", "order_id", "deleted", "FOR UPDATE"); + } + @Test void unDeliveredFragmentBlocksAllSnapshotWrites() { + var mapper = mock(InboundOrderMapper.class); + when(mapper.selectList(any())).thenReturn(List.of(row("FL21", "IN21", "DELIVERED"), row("FL21", "IN21", "LOADED"))); + assertThatThrownBy(() -> new OrderFulfillmentSnapshotService(mapper).complete(command())).isInstanceOf(BizException.class); + verify(mapper,never()).markOrderSnapshotCompleted(anyLong(),anyLong(),any()); + } + @Test void missingOrForeignSnapshotBlocksAllWrites() { + var mapper = mock(InboundOrderMapper.class); + when(mapper.selectList(any())).thenReturn(List.of(), List.of(row("OTHER", "IN21", "DELIVERED"))); + var service = new OrderFulfillmentSnapshotService(mapper); + assertThatThrownBy(() -> service.complete(command())).isInstanceOf(BizException.class); + assertThatThrownBy(() -> service.complete(command())).isInstanceOf(BizException.class); + verify(mapper,never()).markOrderSnapshotCompleted(anyLong(),anyLong(),any()); + } + private OrderFulfillmentCompleted command() { return new OrderFulfillmentCompleted(11,21,"FL21","IN21", LocalDateTime.of(2026,9,30,12,0)); } + private InboundOrderEntity row(String orderNo, String inboundCode, String process) { + var row = new InboundOrderEntity(); row.setId(1L);row.setTenantId(11L);row.setOrderId(21L); + row.setOrderNo(orderNo);row.setInboundCode(inboundCode);row.setProcessStatus(process); return row; + } +} diff --git a/firstleg/src/test/java/com/baiyue/firstleg/service/OverseasDeliveryCompletedConsumerTest.java b/firstleg/src/test/java/com/baiyue/firstleg/service/OverseasDeliveryCompletedConsumerTest.java index 215339e11..e8f3091b2 100644 --- a/firstleg/src/test/java/com/baiyue/firstleg/service/OverseasDeliveryCompletedConsumerTest.java +++ b/firstleg/src/test/java/com/baiyue/firstleg/service/OverseasDeliveryCompletedConsumerTest.java @@ -36,7 +36,7 @@ class OverseasDeliveryCompletedConsumerTest { when(inbounds.completeOverseasDelivery(eq(88L), eq(101L), any())).thenReturn(1); when(inbounds.countNotDeliveredByOrder(88L, 201L)).thenReturn(0); when(orders.complete(eq("token"), any())).thenReturn(ApiResponse.success( - new OrderFulfillmentCompletionClient.Result(201L, "COMPLETED", false))); + new OrderFulfillmentCompletionClient.Result(201L, "COMPLETED", false, 201L, "COMPLETED"))); consumer.consume(event(json)); @@ -72,10 +72,48 @@ class OverseasDeliveryCompletedConsumerTest { verify(inbounds, never()).markOrderSnapshotCompleted(anyLong(), anyLong(), any()); } - private EventEnvelope event(ObjectMapper json) { + @Test + void acceptsRootShipmentFactWithoutPrematureCompletedSnapshot() { + verifyCompletionResponse(new OrderFulfillmentCompletionClient.Result(201L, "PROCESSING", false, + 201L, "PROCESSING"), false, false); + } + + @Test + void marksOnlyChildWhenRootStillHasOutstandingBranch() { + verifyCompletionResponse(new OrderFulfillmentCompletionClient.Result(201L, "COMPLETED", false, + 200L, "PROCESSING"), true, false); + } + + @Test + void marksChildAndRootOnLastSignedBranch() { + verifyCompletionResponse(new OrderFulfillmentCompletionClient.Result(201L, "COMPLETED", false, + 200L, "COMPLETED"), true, true); + } + + private void verifyCompletionResponse(OrderFulfillmentCompletionClient.Result result, + boolean sourceCompleted, boolean rootCompleted) { + InboundOrderMapper inbounds = mock(InboundOrderMapper.class); + OrderFulfillmentCompletionClient orders = mock(OrderFulfillmentCompletionClient.class); + ObjectMapper json = new ObjectMapper().registerModule(new JavaTimeModule()); + OverseasDeliveryCompletedConsumer consumer = new OverseasDeliveryCompletedConsumer(inbounds, + mock(InboundStatusHistoryMapper.class), mock(ContainerEventInboxMapper.class), orders, json); + ReflectionTestUtils.setField(consumer, "internalToken", "token"); + InboundOrderEntity inbound = new InboundOrderEntity(); + inbound.setId(101L); inbound.setTenantId(88L); inbound.setOrderId(201L); + inbound.setOrderNo("FL-201"); inbound.setInboundCode("IN-1"); inbound.setDestinationCode("ONT8"); + inbound.setProcessStatus("DELIVERED"); + when(inbounds.selectActiveByIdForUpdate(101L)).thenReturn(inbound); + when(orders.complete(eq("token"), any())).thenReturn(ApiResponse.success(result)); + consumer.consume(event(json, result.rootOrderId().toString())); + verify(inbounds, times(sourceCompleted ? 1 : 0)).markOrderSnapshotCompleted(eq(88L), eq(201L), any()); + verify(inbounds, times(rootCompleted ? 1 : 0)).markOrderSnapshotCompleted(eq(88L), eq(200L), any()); + } + + private EventEnvelope event(ObjectMapper json) { return event(json, "201"); } + private EventEnvelope event(ObjectMapper json, String rootOrderId) { Instant time = Instant.parse("2026-09-15T08:00:00Z"); var payload = new OverseasDeliveryCompletedV1("1", "DL-ONT8", "11", "LM-11", - "101", "IN-1", "201", "FL-201", "201", "ONT8", time); + "101", "IN-1", "201", "FL-201", rootOrderId, "ONT8", time); var node = json.valueToTree(payload); return new EventEnvelope("00000000-0000-4000-8000-000000000001", EventTypes.OVERSEAS_DELIVERY_COMPLETED, 1, time, diff --git a/order/pom.xml b/order/pom.xml index c9a6c1cf0..82ae0dd70 100644 --- a/order/pom.xml +++ b/order/pom.xml @@ -5,6 +5,7 @@ com.baiyue.firstlegbaiyue-firstleg-platform1.0.0-SNAPSHOT../pom.xml order-service + com.h2databaseh2test com.baiyue.sharedbaiyue-common-audit${project.version} org.apache.rocketmqrocketmq-spring-boot-starter com.baiyue.sharedbaiyue-common-openapi${project.version} diff --git a/order/src/main/java/com/baiyue/order/client/FirstlegInboundClient.java b/order/src/main/java/com/baiyue/order/client/FirstlegInboundClient.java index 739505aa2..fec2ffc9a 100644 --- a/order/src/main/java/com/baiyue/order/client/FirstlegInboundClient.java +++ b/order/src/main/java/com/baiyue/order/client/FirstlegInboundClient.java @@ -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 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 create(@RequestHeader("X-Internal-Service-Token") String token, @RequestBody OrderApprovedInboundCommand command); diff --git a/order/src/main/java/com/baiyue/order/mapper/OrderMapper.java b/order/src/main/java/com/baiyue/order/mapper/OrderMapper.java index 2fccbe52a..acb108c1f 100644 --- a/order/src/main/java/com/baiyue/order/mapper/OrderMapper.java +++ b/order/src/main/java/com/baiyue/order/mapper/OrderMapper.java @@ -12,9 +12,17 @@ import java.util.Map; */ @Mapper public interface OrderMapper extends BaseMapper{ + @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 lockFulfillmentChildren(@Param("tenantId") long tenantId, + @Param("rootOrderId") long rootOrderId); + @Select(""" SELECT COUNT(*) AS total_submitted_orders, COALESCE(SUM(CASE WHEN submitted_at >= #{monthStart} diff --git a/order/src/main/java/com/baiyue/order/mapper/OrderStatusHistoryMapper.java b/order/src/main/java/com/baiyue/order/mapper/OrderStatusHistoryMapper.java index 47de0ff24..053862a10 100644 --- a/order/src/main/java/com/baiyue/order/mapper/OrderStatusHistoryMapper.java +++ b/order/src/main/java/com/baiyue/order/mapper/OrderStatusHistoryMapper.java @@ -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{} +public interface OrderStatusHistoryMapper extends BaseMapper{ + @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); +} diff --git a/order/src/main/java/com/baiyue/order/service/OrderApplicationService.java b/order/src/main/java/com/baiyue/order/service/OrderApplicationService.java index 2e72adc96..afc691774 100644 --- a/order/src/main/java/com/baiyue/order/service/OrderApplicationService.java +++ b/order/src/main/java/com/baiyue/order/service/OrderApplicationService.java @@ -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); } /** 提交订单的幂等入口。 */ diff --git a/order/src/main/java/com/baiyue/order/service/OrderFulfillmentCompletionService.java b/order/src/main/java/com/baiyue/order/service/OrderFulfillmentCompletionService.java index 41ef25794..f68c73ead 100644 --- a/order/src/main/java/com/baiyue/order/service/OrderFulfillmentCompletionService.java +++ b/order/src/main/java/com/baiyue/order/service/OrderFulfillmentCompletionService.java @@ -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 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() - .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) { } } diff --git a/order/src/main/java/com/baiyue/order/service/OrderInboundOutboxPublisher.java b/order/src/main/java/com/baiyue/order/service/OrderInboundOutboxPublisher.java index 37cd57cbf..69a4beec5 100644 --- a/order/src/main/java/com/baiyue/order/service/OrderInboundOutboxPublisher.java +++ b/order/src/main/java/com/baiyue/order/service/OrderInboundOutboxPublisher.java @@ -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( diff --git a/order/src/main/resources/openapi/order-v1.yaml b/order/src/main/resources/openapi/order-v1.yaml index 9a84ea738..b27d06374 100644 --- a/order/src/main/resources/openapi/order-v1.yaml +++ b/order/src/main/resources/openapi/order-v1.yaml @@ -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: diff --git a/order/src/test/java/com/baiyue/order/service/OrderApplicationServiceStateTest.java b/order/src/test/java/com/baiyue/order/service/OrderApplicationServiceStateTest.java index 8a47198db..490879d19 100644 --- a/order/src/test/java/com/baiyue/order/service/OrderApplicationServiceStateTest.java +++ b/order/src/test/java/com/baiyue/order/service/OrderApplicationServiceStateTest.java @@ -37,6 +37,7 @@ import static org.mockito.Mockito.*; @DirtiesContext(classMode = DirtiesContext.ClassMode.AFTER_EACH_TEST_METHOD) class OrderApplicationServiceStateTest { @MockBean private OrderRepository contextRepository; + @MockBean private OrderFulfillmentCompletionService contextCompletion; @MockBean private OrderEligibilityGuard contextEligibility; @MockBean private OrderMasterDataGuard contextMasterData; @MockBean private OrderPricingChannelGuard contextPricingChannelGuard; @@ -48,6 +49,24 @@ class OrderApplicationServiceStateTest { static class TestConfig { @Bean ObjectMapper objectMapper() { return new ObjectMapper().findAndRegisterModules(); } } + @Test + void cancellationLocksRootBeforeSourceWriteAndThenAggregates() { + OrderRepository repository = mock(OrderRepository.class); + OrderEntity order = persistedOrder(OrderStatus.APPROVED); + order.setOrderNo("FL-1"); + when(repository.findOwned(10001L, 1L)).thenReturn(order); + when(repository.saveOrder(10001L, order)).thenReturn(1); + OrderFulfillmentCompletionService completion = mock(OrderFulfillmentCompletionService.class); + var app = service(repository, mock(OrderEligibilityGuard.class), mock(OrderMasterDataGuard.class)); + ReflectionTestUtils.setField(app, "fulfillmentCompletion", completion); + app.transit(10001, 2, "u", 1, OrderStatus.CANCELLED, "取消余量"); + var sequence = org.mockito.Mockito.inOrder(repository, completion); + sequence.verify(completion).lockCancellationRoot(10001, 1); + sequence.verify(repository).findOwned(10001, 1); + sequence.verify(repository).saveOrder(10001, order); + sequence.verify(completion).afterCancellation(10001, 1); + } + @Test void unapprovedTenantCanCreateDraft() { OrderRepository repository = mock(OrderRepository.class); diff --git a/order/src/test/java/com/baiyue/order/service/OrderFulfillmentCompletionServiceTest.java b/order/src/test/java/com/baiyue/order/service/OrderFulfillmentCompletionServiceTest.java index 8651dbc72..c6c9304f9 100644 --- a/order/src/test/java/com/baiyue/order/service/OrderFulfillmentCompletionServiceTest.java +++ b/order/src/test/java/com/baiyue/order/service/OrderFulfillmentCompletionServiceTest.java @@ -29,6 +29,7 @@ import static org.mockito.Mockito.when; class OrderFulfillmentCompletionServiceTest { @Mock private OrderMapper orders; @Mock private OrderStatusHistoryMapper histories; + @Mock private com.baiyue.order.mapper.OutboxEventMapper outbox; @InjectMocks private OrderFulfillmentCompletionService service; @BeforeAll @@ -45,19 +46,26 @@ class OrderFulfillmentCompletionServiceTest { order.setTenantId(11L); order.setOrderNo("FL201"); order.setInboundCode("IN-201"); + order.setOrderRole("MASTER"); order.setStatus(OrderStatus.PROCESSING); order.setVersion(7); + when(orders.findOwned(11L, 201L)).thenReturn(order); when(orders.lockOwned(11L, 201L)).thenReturn(order); when(orders.update(isNull(), any(LambdaUpdateWrapper.class))).thenReturn(1); when(histories.insert(any(OrderStatusHistoryEntity.class))).thenReturn(1); + when(outbox.insert(any(com.baiyue.order.entity.OutboxEventEntity.class))).thenReturn(1); + org.springframework.test.util.ReflectionTestUtils.setField(service, "json", + new com.fasterxml.jackson.databind.ObjectMapper().findAndRegisterModules()); var result = service.complete(new OrderFulfillmentCompletionService.Request( - "11", "201", "FL201", "IN-201", "DL-201", Instant.parse("2026-09-15T12:00:00Z"), + "11", "201", "FL201", "IN-201", "201", "DL-201", Instant.parse("2026-09-15T12:00:00Z"), "event-201", "sha256:test")); assertThat(result.status()).isEqualTo("COMPLETED"); ArgumentCaptor history = ArgumentCaptor.forClass(OrderStatusHistoryEntity.class); - verify(histories).insert(history.capture()); - assertThat(history.getValue().getAction()).isEqualTo("OVERSEAS_DELIVERED").hasSizeLessThanOrEqualTo(24); + verify(histories, org.mockito.Mockito.times(2)).insert(history.capture()); + assertThat(history.getAllValues()).extracting(OrderStatusHistoryEntity::getAction) + .containsExactly("OVERSEAS_DELIVERED", "FULFILLMENT_COMPLETED"); + assertThat(history.getAllValues()).allSatisfy(row -> assertThat(row.getAction()).hasSizeLessThanOrEqualTo(24)); } } diff --git a/order/src/test/java/com/baiyue/order/service/OrderFulfillmentCompletionTransactionTest.java b/order/src/test/java/com/baiyue/order/service/OrderFulfillmentCompletionTransactionTest.java new file mode 100644 index 000000000..fb4c2e312 --- /dev/null +++ b/order/src/test/java/com/baiyue/order/service/OrderFulfillmentCompletionTransactionTest.java @@ -0,0 +1,222 @@ +package com.baiyue.order.service; + +import com.baomidou.mybatisplus.core.MybatisConfiguration; +import com.baomidou.mybatisplus.extension.spring.MybatisSqlSessionFactoryBean; +import com.baiyue.common.core.exception.BizException; +import com.baiyue.order.mapper.OrderMapper; +import com.baiyue.order.mapper.OrderStatusHistoryMapper; +import org.h2.jdbcx.JdbcDataSource; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.mybatis.spring.SqlSessionTemplate; +import org.springframework.aop.framework.ProxyFactory; +import org.springframework.jdbc.core.JdbcTemplate; +import org.springframework.jdbc.datasource.DataSourceTransactionManager; +import org.springframework.test.util.ReflectionTestUtils; +import org.springframework.transaction.annotation.AnnotationTransactionAttributeSource; +import org.springframework.transaction.interceptor.TransactionInterceptor; + +import java.time.Instant; +import java.util.UUID; +import java.util.concurrent.*; + +import static org.assertj.core.api.Assertions.*; + +/** Runs production Mapper SQL and service writes with a real local transaction manager. */ +class OrderFulfillmentCompletionTransactionTest { + private JdbcTemplate db; + private OrderFulfillmentCompletionService service; + private org.springframework.transaction.support.TransactionTemplate transactions; + private static final Instant ROOT_AT = Instant.parse("2026-09-30T08:00:00Z"); + + @BeforeEach + void prepare() throws Exception { + JdbcDataSource source = new JdbcDataSource(); + source.setURL("jdbc:h2:mem:" + UUID.randomUUID() + ";MODE=MySQL;DB_CLOSE_DELAY=-1;LOCK_TIMEOUT=10000"); + db = new JdbcTemplate(source); + db.execute("CREATE TABLE ord_order (id BIGINT PRIMARY KEY, tenant_id BIGINT, order_no VARCHAR(80), " + + "inbound_code VARCHAR(80), order_role VARCHAR(16), parent_order_id BIGINT, root_order_id BIGINT, " + + "status VARCHAR(20), completed_at TIMESTAMP, version INT, deleted BOOLEAN DEFAULT FALSE, " + + "updated_by BIGINT, updated_at TIMESTAMP)"); + db.execute("CREATE TABLE ord_order_status_history (id BIGINT PRIMARY KEY, tenant_id BIGINT, order_id BIGINT, " + + "from_status VARCHAR(20), to_status VARCHAR(20), action VARCHAR(24), reason VARCHAR(200), " + + "operator_id BIGINT, operator_name VARCHAR(80), occurred_at TIMESTAMP)"); + db.execute("CREATE TABLE ord_outbox_event (id BIGINT PRIMARY KEY, tenant_id BIGINT, aggregate_type VARCHAR(80), " + + "aggregate_id BIGINT, event_type VARCHAR(80), payload_json CLOB, publish_status VARCHAR(20), " + + "retry_count INT, created_at TIMESTAMP)"); + MybatisConfiguration configuration = new MybatisConfiguration(); + configuration.setMapUnderscoreToCamelCase(true); + configuration.addMapper(OrderMapper.class); + configuration.addMapper(OrderStatusHistoryMapper.class); + configuration.addMapper(com.baiyue.order.mapper.OutboxEventMapper.class); + MybatisSqlSessionFactoryBean factory = new MybatisSqlSessionFactoryBean(); + factory.setDataSource(source); factory.setConfiguration(configuration); + SqlSessionTemplate session = new SqlSessionTemplate(factory.getObject()); + OrderFulfillmentCompletionService target = new OrderFulfillmentCompletionService(); + ReflectionTestUtils.setField(target, "orders", session.getMapper(OrderMapper.class)); + ReflectionTestUtils.setField(target, "histories", session.getMapper(OrderStatusHistoryMapper.class)); + ReflectionTestUtils.setField(target, "outbox", session.getMapper(com.baiyue.order.mapper.OutboxEventMapper.class)); + ReflectionTestUtils.setField(target, "json", new com.fasterxml.jackson.databind.ObjectMapper().findAndRegisterModules()); + transactions = new org.springframework.transaction.support.TransactionTemplate(new DataSourceTransactionManager(source)); + ProxyFactory proxy = new ProxyFactory(target); + proxy.setProxyTargetClass(true); + proxy.addAdvice(new TransactionInterceptor(new DataSourceTransactionManager(source), + new AnnotationTransactionAttributeSource())); + service = (OrderFulfillmentCompletionService) proxy.getProxy(); + master(201, 11); + } + + @Test + void rootOwnShipmentKeepsRootOpenUntilAllChildrenDeliverAndReplaysDoNotRewrite() { + child(202, "APPROVED"); child(203, "ACCEPTED"); + var root = complete(201, ROOT_AT); + assertThat(root.status()).isEqualTo("PROCESSING"); + assertThat(root.rootStatus()).isEqualTo("PROCESSING"); + assertThat(history(201, "OVERSEAS_DELIVERED")).isEqualTo(1); + assertThat(history(201, "FULFILLMENT_COMPLETED")).isZero(); + assertThat(complete(201, ROOT_AT).replayed()).isTrue(); + complete(203, ROOT_AT.plusSeconds(20)); + assertThat(status(201)).isEqualTo("PROCESSING"); + var last = complete(202, ROOT_AT.plusSeconds(10)); + assertThat(last.rootOrderId()).isEqualTo(201L); + assertThat(last.rootStatus()).isEqualTo("COMPLETED"); + assertThat(db.queryForObject("SELECT completed_at FROM ord_order WHERE id=201", java.sql.Timestamp.class) + .toLocalDateTime().toInstant(java.time.ZoneOffset.UTC)).isEqualTo(ROOT_AT.plusSeconds(20)); + complete(202, ROOT_AT.plusSeconds(10)); complete(203, ROOT_AT.plusSeconds(20)); complete(201, ROOT_AT); + assertThat(history(201, "OVERSEAS_DELIVERED")).isEqualTo(1); + assertThat(history(201, "FULFILLMENT_COMPLETED")).isEqualTo(1); + assertThat(history(202, "OVERSEAS_DELIVERED")).isEqualTo(1); + assertThat(history(203, "OVERSEAS_DELIVERED")).isEqualTo(1); + } + + @Test + void allChildrenFirstCannotCompleteRootWithoutItsOwnSignedFact() { + child(202, "APPROVED"); child(203, "APPROVED"); + complete(202, ROOT_AT.plusSeconds(10)); complete(203, ROOT_AT.plusSeconds(20)); + assertThat(status(201)).isEqualTo("PROCESSING"); + assertThat(history(201, "OVERSEAS_DELIVERED")).isZero(); + assertThat(complete(201, ROOT_AT).status()).isEqualTo("COMPLETED"); + assertThat(history(201, "FULFILLMENT_COMPLETED")).isEqualTo(1); + } + + @Test + void onlyExplicitCancelledAndDeletedChildrenAreNonEffective() { + child(202, "CANCELLED"); child(203, "DRAFT"); + db.update("UPDATE ord_order SET deleted=TRUE WHERE id=203"); + master(900, 12); + db.update("INSERT INTO ord_order (id,tenant_id,order_no,inbound_code,order_role,parent_order_id,root_order_id,status,version) " + + "VALUES (901,12,'OTHER','OTHER-IN','CHILD',201,201,'DRAFT',0)"); + assertThat(complete(201, ROOT_AT).status()).isEqualTo("COMPLETED"); + assertThatThrownBy(() -> complete(202, ROOT_AT)).isInstanceOf(BizException.class); + assertThat(history(202, "OVERSEAS_DELIVERED")).isZero(); + } + + @Test + void draftOrRejectedEffectiveBranchBlocksAggregate() { + child(202, "REJECTED"); + assertThat(complete(201, ROOT_AT).rootStatus()).isEqualTo("PROCESSING"); + assertThat(history(201, "FULFILLMENT_COMPLETED")).isZero(); + } + + @Test + void malformedChildIdentityRollsBackNewFactAndChildCompletion() { + child(202, "APPROVED"); + db.update("UPDATE ord_order SET root_order_id=999 WHERE id=202"); + assertThatThrownBy(() -> complete(201, ROOT_AT)).isInstanceOf(BizException.class); + assertThat(history(201, "OVERSEAS_DELIVERED")).isZero(); + assertThat(status(201)).isEqualTo("PROCESSING"); + assertThatThrownBy(() -> complete(202, ROOT_AT)).isInstanceOf(BizException.class); + assertThat(status(202)).isEqualTo("APPROVED"); + } + + @Test + void failedAggregateUpdateRollsBackSourceCompletionAndHistory() { + child(202, "APPROVED"); + complete(201, ROOT_AT); + db.execute("ALTER TABLE ord_order ADD CONSTRAINT refuse_root_completion " + + "CHECK (id<>201 OR status<>'COMPLETED')"); + assertThatThrownBy(() -> complete(202, ROOT_AT.plusSeconds(10))).isInstanceOf(RuntimeException.class); + assertThat(status(202)).isEqualTo("APPROVED"); + assertThat(history(202, "OVERSEAS_DELIVERED")).isZero(); + assertThat(history(201, "FULFILLMENT_COMPLETED")).isZero(); + } + + @Test + void concurrentLastChildrenSerializeOnRootAndProduceSingleFinalHistory() throws Exception { + child(202, "APPROVED"); child(203, "APPROVED"); + complete(201, ROOT_AT); + ExecutorService workers = Executors.newFixedThreadPool(2); + CountDownLatch start = new CountDownLatch(1); + try { + Future first = workers.submit(() -> { await(start); complete(202, ROOT_AT.plusSeconds(10)); }); + Future second = workers.submit(() -> { await(start); complete(203, ROOT_AT.plusSeconds(20)); }); + start.countDown(); first.get(15, TimeUnit.SECONDS); second.get(15, TimeUnit.SECONDS); + } finally { workers.shutdownNow(); } + assertThat(status(201)).isEqualTo("COMPLETED"); + assertThat(history(201, "FULFILLMENT_COMPLETED")).isEqualTo(1); + assertThat(history(202, "OVERSEAS_DELIVERED")).isEqualTo(1); + assertThat(history(203, "OVERSEAS_DELIVERED")).isEqualTo(1); + } + + @Test + void lastRemainderCancellationFinishesSignedRootAndPublishesSnapshotExactlyOnce() { + child(202, "APPROVED"); + complete(201, ROOT_AT); + transactions.executeWithoutResult(tx -> { + service.lockCancellationRoot(11, 202); + db.update("UPDATE ord_order SET status='CANCELLED' WHERE id=202"); + service.afterCancellation(11, 202); + }); + assertThat(status(201)).isEqualTo("COMPLETED"); + transactions.executeWithoutResult(tx -> service.afterCancellation(11, 202)); + assertThat(history(201, "FULFILLMENT_COMPLETED")).isEqualTo(1); + assertThat(db.queryForObject("SELECT COUNT(*) FROM ord_outbox_event WHERE aggregate_id=201", Integer.class)).isEqualTo(1); + assertThat(db.queryForObject("SELECT payload_json FROM ord_outbox_event WHERE aggregate_id=201", String.class)) + .contains("FL201", "IN-201"); + } + + @Test + void cancellingAllChildrenWithoutRootSignedFactCannotCompleteRoot() { + child(202, "APPROVED"); + transactions.executeWithoutResult(tx -> { + service.lockCancellationRoot(11, 202); + db.update("UPDATE ord_order SET status='CANCELLED' WHERE id=202"); + service.afterCancellation(11, 202); + }); + assertThat(status(201)).isEqualTo("PROCESSING"); + assertThat(history(201, "FULFILLMENT_COMPLETED")).isZero(); + assertThat(db.queryForObject("SELECT COUNT(*) FROM ord_outbox_event", Integer.class)).isZero(); + } + + @Test + void foreignCustomerOrWrongSnapshotCannotWrite() { + assertThatThrownBy(() -> service.complete(new OrderFulfillmentCompletionService.Request( + "12", "201", "FL201", "IN-201", "201", "DL201", ROOT_AT, "event", "hash"))) + .isInstanceOf(BizException.class); + assertThatThrownBy(() -> service.complete(new OrderFulfillmentCompletionService.Request( + "11", "201", "WRONG", "IN-201", "201", "DL201", ROOT_AT, "event", "hash"))) + .isInstanceOf(BizException.class); + assertThat(history(201, "OVERSEAS_DELIVERED")).isZero(); + } + + private void master(long id, long tenant) { + db.update("INSERT INTO ord_order (id,tenant_id,order_no,inbound_code,order_role,status,version) " + + "VALUES (?,?,?,?,'MASTER','PROCESSING',0)", id, tenant, "FL"+id, "IN-"+id); + } + private void child(long id, String status) { + db.update("INSERT INTO ord_order (id,tenant_id,order_no,inbound_code,order_role,parent_order_id,root_order_id,status,version) " + + "VALUES (?,11,?,?,'CHILD',201,201,?,0)", id, "FL"+id, "IN-"+id, status); + } + private OrderFulfillmentCompletionService.Result complete(long id, Instant at) { + return service.complete(new OrderFulfillmentCompletionService.Request("11", Long.toString(id), + "FL"+id, "IN-"+id, "201", "DL"+id, at, "event-"+id, "hash-"+id)); + } + private String status(long id) { return db.queryForObject("SELECT status FROM ord_order WHERE id=?", String.class, id); } + private int history(long id, String action) { + return db.queryForObject("SELECT COUNT(*) FROM ord_order_status_history WHERE order_id=? AND action=?", + Integer.class, id, action); + } + private static void await(CountDownLatch latch) { + try { latch.await(); } catch (InterruptedException interrupted) { Thread.currentThread().interrupt(); throw new RuntimeException(interrupted); } + } +} diff --git a/order/src/test/java/com/baiyue/order/service/OrderIdempotencyServiceTest.java b/order/src/test/java/com/baiyue/order/service/OrderIdempotencyServiceTest.java index 6efebce49..e8f6947de 100644 --- a/order/src/test/java/com/baiyue/order/service/OrderIdempotencyServiceTest.java +++ b/order/src/test/java/com/baiyue/order/service/OrderIdempotencyServiceTest.java @@ -37,6 +37,8 @@ import static org.mockito.Mockito.when; @SpringJUnitConfig(classes = {OrderApplicationService.class, OrderIdempotencyServiceTest.TestConfig.class}) @DirtiesContext(classMode = DirtiesContext.ClassMode.AFTER_EACH_TEST_METHOD) class OrderIdempotencyServiceTest { + @MockBean private OrderFulfillmentCompletionService fulfillmentCompletion; + @MockBean private OrderRepository contextRepository; @MockBean private OrderEligibilityGuard contextEligibility; @MockBean private OrderMasterDataGuard contextMasterData; diff --git a/order/src/test/java/com/baiyue/order/service/OrderInboundOutboxPublisherTest.java b/order/src/test/java/com/baiyue/order/service/OrderInboundOutboxPublisherTest.java index a339603dd..9d9501dd5 100644 --- a/order/src/test/java/com/baiyue/order/service/OrderInboundOutboxPublisherTest.java +++ b/order/src/test/java/com/baiyue/order/service/OrderInboundOutboxPublisherTest.java @@ -206,6 +206,41 @@ class OrderInboundOutboxPublisherTest { verify(fixture.repository, never()).markOutboxPublished(anyLong(), anyLong(), any()); } + @Test + void completionOutboxPublishesAcknowledgedSnapshotIncludingIdempotentZeroUpdate() { + Fixture fixture = fixture(event("FIRST_LEG_ORDER_FULFILLMENT_COMPLETED", 11L, 21L, + completionPayload(11, 21))); + when(fixture.client.completeFulfillmentSnapshot(eq("internal-token"), any())).thenReturn(ApiResponse.success(0)); + fixture.publisher.publishApprovedOrders(); + verify(fixture.client).completeFulfillmentSnapshot(eq("internal-token"), + org.mockito.ArgumentMatchers.argThat(command -> command.orderId()==21 && command.inboundCode().equals("IN-21"))); + verify(fixture.repository).markOutboxPublished(eq(11L), eq(41L), any()); + } + + @Test + void completionOutboxFailureRetriesOriginalEventAndDoesNotPublish() { + Fixture fixture = fixture(event("FIRST_LEG_ORDER_FULFILLMENT_COMPLETED", 11L, 21L, + completionPayload(11, 21))); + when(fixture.client.completeFulfillmentSnapshot(eq("internal-token"), any())).thenThrow(new IllegalStateException("timeout")); + fixture.publisher.publishApprovedOrders(); + verify(fixture.repository).markOutboxRetry(eq(11L), eq(41L), any()); + verify(fixture.repository, never()).markOutboxPublished(anyLong(), anyLong(), any()); + } + + @Test + void completionOutboxRejectsForeignPayloadBeforeCallingFirstleg() { + Fixture fixture = fixture(event("FIRST_LEG_ORDER_FULFILLMENT_COMPLETED", 11L, 21L, + completionPayload(12, 21))); + fixture.publisher.publishApprovedOrders(); + verifyNoInteractions(fixture.client); + verify(fixture.repository).markOutboxFailed(11L, 41L); + } + + private String completionPayload(long tenant, long order) { + return "{\"tenantId\":"+tenant+",\"orderId\":"+order + +",\"orderNo\":\"FL-21\",\"inboundCode\":\"IN-21\",\"completedAt\":\"2026-09-30T08:00:00\"}"; + } + private Fixture fixture(OutboxEventEntity event) { return fixture(event, "BUY_DECLARATION"); } @@ -242,7 +277,7 @@ class OrderInboundOutboxPublisherTest { ReflectionTestUtils.setField(publisher, "repository", repository); ReflectionTestUtils.setField(publisher, "client", client); ReflectionTestUtils.setField(publisher, "pricingClient", pricingClient); - ReflectionTestUtils.setField(publisher, "json", new ObjectMapper()); + ReflectionTestUtils.setField(publisher, "json", new ObjectMapper().findAndRegisterModules()); ReflectionTestUtils.setField(publisher, "internalToken", "internal-token"); return new Fixture(publisher, repository, client, pricingClient); } diff --git a/order/src/test/java/com/baiyue/order/service/PlatformOrderPageServiceTest.java b/order/src/test/java/com/baiyue/order/service/PlatformOrderPageServiceTest.java index 49d6131eb..64f0e0cf4 100644 --- a/order/src/test/java/com/baiyue/order/service/PlatformOrderPageServiceTest.java +++ b/order/src/test/java/com/baiyue/order/service/PlatformOrderPageServiceTest.java @@ -16,6 +16,8 @@ import static org.mockito.Mockito.verify; @SpringJUnitConfig(classes = {OrderApplicationService.class, PlatformOrderPageServiceTest.TestConfig.class}) class PlatformOrderPageServiceTest { + @MockBean private OrderFulfillmentCompletionService fulfillmentCompletion; + @MockBean private OrderRepository repository; @MockBean private OrderEligibilityGuard eligibilityGuard; @MockBean private OrderMasterDataGuard masterDataGuard; diff --git a/order/src/test/java/com/baiyue/order/service/TenantOrderCountServiceTest.java b/order/src/test/java/com/baiyue/order/service/TenantOrderCountServiceTest.java index 6ce84eb99..7e64a6269 100644 --- a/order/src/test/java/com/baiyue/order/service/TenantOrderCountServiceTest.java +++ b/order/src/test/java/com/baiyue/order/service/TenantOrderCountServiceTest.java @@ -19,6 +19,8 @@ import static org.mockito.Mockito.when; @SpringJUnitConfig(classes = {OrderApplicationService.class, TenantOrderCountServiceTest.TestConfig.class}) class TenantOrderCountServiceTest { + @MockBean private OrderFulfillmentCompletionService fulfillmentCompletion; + @MockBean private OrderRepository repository; @MockBean private OrderEligibilityGuard eligibilityGuard; @MockBean private OrderMasterDataGuard masterDataGuard; diff --git a/order/src/test/java/com/baiyue/order/service/TenantOrderStatisticsServiceTest.java b/order/src/test/java/com/baiyue/order/service/TenantOrderStatisticsServiceTest.java index a9186caca..9bee3e435 100644 --- a/order/src/test/java/com/baiyue/order/service/TenantOrderStatisticsServiceTest.java +++ b/order/src/test/java/com/baiyue/order/service/TenantOrderStatisticsServiceTest.java @@ -23,6 +23,8 @@ import static org.mockito.Mockito.when; @SpringJUnitConfig(classes = {OrderApplicationService.class, TenantOrderStatisticsServiceTest.TestConfig.class}) class TenantOrderStatisticsServiceTest { + @MockBean private OrderFulfillmentCompletionService fulfillmentCompletion; + @MockBean private OrderRepository repository; @MockBean private OrderEligibilityGuard eligibilityGuard; @MockBean private OrderMasterDataGuard masterDataGuard;