This commit is contained in:
2026-08-20 15:20:05 +08:00
parent 6c7a670f1e
commit b55b049e77
30 changed files with 727 additions and 12 deletions
@@ -0,0 +1,40 @@
package com.baiyue.firstleg.controller;
import com.baiyue.common.core.api.ApiResponse;
import com.baiyue.common.core.api.ResultCode;
import com.baiyue.common.core.exception.BizException;
import com.baiyue.common.event.EventEnvelope;
import com.baiyue.firstleg.entity.ContainerBookingEntity;
import com.baiyue.firstleg.service.OverseasContainerEventConsumer;
import lombok.RequiredArgsConstructor;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.web.bind.annotation.PostMapping;
import org.springframework.web.bind.annotation.RequestBody;
import org.springframework.web.bind.annotation.RequestHeader;
import org.springframework.web.bind.annotation.RestController;
import java.nio.charset.StandardCharsets;
import java.security.MessageDigest;
@RestController
@RequiredArgsConstructor
public class InternalOverseasContainerEventController {
private final OverseasContainerEventConsumer consumer;
@Value("${baiyue.security.internal-service-token}") private String expectedToken;
@PostMapping("/internal/firstleg/v1/container-events")
public ApiResponse<ContainerBookingEntity> receive(
@RequestHeader(value = "X-Internal-Service-Token", required = false) String token,
@RequestBody EventEnvelope event) {
authenticate(token);
return ApiResponse.success(consumer.consume(event));
}
private void authenticate(String token) {
if (token == null || expectedToken == null || expectedToken.isBlank()
|| !MessageDigest.isEqual(token.getBytes(StandardCharsets.UTF_8),
expectedToken.getBytes(StandardCharsets.UTF_8))) {
throw new BizException(ResultCode.FORBIDDEN, "内部服务令牌无效");
}
}
}
@@ -54,6 +54,11 @@ public class ContainerBookingEntity {
private LocalDateTime actualPortArrivalAt;
private LocalDateTime estimatedWarehouseArrivalAt;
private LocalDateTime actualWarehouseArrivalAt;
private LocalDateTime unloadingCompletedAt;
private LocalDateTime unloadingPhotoUploadedAt;
private LocalDateTime estimatedEmptyReportedAt;
private LocalDateTime actualEmptyReportedAt;
private Long emptyReportSourceVersion;
private Long createdBy;
private String createdByName;
@TableField(fill = FieldFill.INSERT) private LocalDateTime createdAt;
@@ -0,0 +1,24 @@
package com.baiyue.firstleg.entity;
import com.baomidou.mybatisplus.annotation.TableId;
import com.baomidou.mybatisplus.annotation.TableName;
import lombok.Getter;
import lombok.Setter;
import java.time.LocalDateTime;
@Getter
@Setter
@TableName("fl_container_event_inbox")
public class ContainerEventInboxEntity {
@TableId private String eventId;
private String eventType;
private Integer eventVersion;
private String aggregateId;
private Long aggregateVersion;
private String payloadHash;
private String payloadJson;
private String status;
private LocalDateTime receivedAt;
private LocalDateTime processedAt;
}
@@ -107,6 +107,24 @@ public interface ContainerBookingMapper extends BaseMapper<ContainerBookingEntit
@Select("SELECT * FROM fl_container_booking WHERE id=#{id} AND record_status='ACTIVE' FOR UPDATE")
ContainerBookingEntity selectActiveBookingForUpdate(@Param("id") long id);
@Select("SELECT * FROM fl_container_booking WHERE container_no=#{containerNo} AND record_status='ACTIVE'")
ContainerBookingEntity selectActiveByContainerNo(@Param("containerNo") String containerNo);
@Update("""
UPDATE fl_container_booking SET unloading_completed_at=#{unloadingCompletedAt},
unloading_photo_uploaded_at=#{photoUploadedAt}, estimated_empty_reported_at=#{estimatedEmptyReportedAt},
actual_empty_reported_at=#{actualEmptyReportedAt}, empty_report_source_version=#{sourceVersion},
updated_by=0, updated_by_name='OVERSEAS_EVENT', updated_at=CURRENT_TIMESTAMP(3), version=version+1
WHERE id=#{id} AND record_status='ACTIVE'
AND (empty_report_source_version IS NULL OR empty_report_source_version &lt; #{sourceVersion})
""")
int applyEmptyReport(@Param("id") long id,
@Param("unloadingCompletedAt") java.time.LocalDateTime unloadingCompletedAt,
@Param("photoUploadedAt") java.time.LocalDateTime photoUploadedAt,
@Param("estimatedEmptyReportedAt") java.time.LocalDateTime estimatedEmptyReportedAt,
@Param("actualEmptyReportedAt") java.time.LocalDateTime actualEmptyReportedAt,
@Param("sourceVersion") long sourceVersion);
@Update("""
UPDATE fl_container_booking SET container_no=NULL, ocean_booking_id=NULL,
ocean_booking_requirement_id=NULL, customer_id=NULL, customer_name=NULL,
@@ -0,0 +1,9 @@
package com.baiyue.firstleg.mapper;
import com.baomidou.mybatisplus.core.mapper.BaseMapper;
import com.baiyue.firstleg.entity.ContainerEventInboxEntity;
import org.apache.ibatis.annotations.Mapper;
@Mapper
public interface ContainerEventInboxMapper extends BaseMapper<ContainerEventInboxEntity> {
}
@@ -0,0 +1,109 @@
package com.baiyue.firstleg.service;
import com.baiyue.common.event.EventEnvelope;
import com.baiyue.common.event.EventPayloadCanonicalizer;
import com.baiyue.common.event.EventPayloadHash;
import com.baiyue.common.event.EventTypes;
import com.baiyue.common.event.OverseasContainerUnloadedV1;
import com.baiyue.common.event.RegionCode;
import com.baiyue.firstleg.entity.ContainerBookingEntity;
import com.baiyue.firstleg.entity.ContainerEventInboxEntity;
import com.baiyue.firstleg.mapper.ContainerBookingMapper;
import com.baiyue.firstleg.mapper.ContainerEventInboxMapper;
import com.fasterxml.jackson.databind.ObjectMapper;
import lombok.RequiredArgsConstructor;
import org.springframework.dao.DuplicateKeyException;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;
import java.time.LocalDateTime;
import java.time.ZoneOffset;
@Service
@RequiredArgsConstructor
public class OverseasContainerEventConsumer {
private final ContainerBookingMapper containers;
private final ContainerEventInboxMapper inbox;
private final ObjectMapper objectMapper;
@Transactional
public ContainerBookingEntity consume(EventEnvelope event) {
validateEnvelope(event);
OverseasContainerUnloadedV1 payload = payload(event);
if (payload.containerNo() == null || payload.containerNo().isBlank()
|| !payload.containerNo().equals(event.aggregateId())) {
throw new IllegalArgumentException("事件 aggregateId 与 containerNo 不一致");
}
if (payload.unloadingCompletedAt() == null || payload.unloadingPhotoUploadedAt() == null) {
throw new IllegalArgumentException("卸柜完成时间和卸柜照片上传时间不能为空");
}
if (payload.actualEmptyReportedAt() != null
&& payload.actualEmptyReportedAt().isBefore(payload.unloadingPhotoUploadedAt())) {
throw new IllegalArgumentException("实际报空时间不能早于卸柜照片上传时间");
}
ContainerEventInboxEntity existing = inbox.selectById(event.eventId());
if (existing != null) {
if (!existing.getPayloadHash().equalsIgnoreCase(event.payloadHash())) {
throw new IllegalArgumentException("相同 eventId 对应的事件摘要冲突");
}
return requiredContainer(payload.containerNo());
}
ContainerEventInboxEntity row = inbox(event);
try {
inbox.insert(row);
} catch (DuplicateKeyException duplicate) {
ContainerEventInboxEntity raced = inbox.selectById(event.eventId());
if (raced == null || !raced.getPayloadHash().equalsIgnoreCase(event.payloadHash())) throw duplicate;
return requiredContainer(payload.containerNo());
}
ContainerBookingEntity container = requiredContainer(payload.containerNo());
LocalDateTime uploadedAt = utc(payload.unloadingPhotoUploadedAt());
containers.applyEmptyReport(container.getId(), utc(payload.unloadingCompletedAt()), uploadedAt,
uploadedAt.plusDays(2), utc(payload.actualEmptyReportedAt()), event.aggregateVersion());
row.setStatus("PROCESSED");
row.setProcessedAt(LocalDateTime.now(ZoneOffset.UTC));
inbox.updateById(row);
return requiredContainer(payload.containerNo());
}
private void validateEnvelope(EventEnvelope event) {
event.requireValid();
if (!EventTypes.OVERSEAS_CONTAINER_UNLOADED.equals(event.eventType()) || event.eventVersion() != 1) {
throw new IllegalArgumentException("不支持的海外货柜事件");
}
if (event.sourceRegion() != RegionCode.OVERSEAS || event.targetRegion() != RegionCode.CN
|| !"CONTAINER_BOOKING".equals(event.aggregateType())) {
throw new IllegalArgumentException("海外货柜事件区域或聚合类型无效");
}
if (!EventPayloadHash.matches(EventPayloadCanonicalizer.canonicalize(event.payload()), event.payloadHash())) {
throw new IllegalArgumentException("payloadHash 与事件负载不一致");
}
}
private OverseasContainerUnloadedV1 payload(EventEnvelope event) {
try { return objectMapper.treeToValue(event.payload(), OverseasContainerUnloadedV1.class); }
catch (Exception exception) { throw new IllegalArgumentException("海外货柜事件负载无效", exception); }
}
private ContainerEventInboxEntity inbox(EventEnvelope event) {
ContainerEventInboxEntity row = new ContainerEventInboxEntity();
row.setEventId(event.eventId()); row.setEventType(event.eventType()); row.setEventVersion(event.eventVersion());
row.setAggregateId(event.aggregateId()); row.setAggregateVersion(event.aggregateVersion());
row.setPayloadHash(event.payloadHash().toLowerCase(java.util.Locale.ROOT));
row.setPayloadJson(event.payload().toString()); row.setStatus("RECEIVED");
row.setReceivedAt(LocalDateTime.now(ZoneOffset.UTC));
return row;
}
private ContainerBookingEntity requiredContainer(String containerNo) {
ContainerBookingEntity value = containers.selectActiveByContainerNo(containerNo);
if (value == null) throw new IllegalArgumentException("头程货柜不存在或不可用");
return value;
}
private LocalDateTime utc(java.time.Instant value) {
return value == null ? null : LocalDateTime.ofInstant(value, ZoneOffset.UTC);
}
}
@@ -72,6 +72,11 @@ CREATE TABLE IF NOT EXISTS fl_container_booking (
actual_port_arrival_at DATETIME(3) NULL COMMENT '实际到港时间',
estimated_warehouse_arrival_at DATETIME(3) NULL COMMENT '预计到仓时间',
actual_warehouse_arrival_at DATETIME(3) NULL COMMENT '实际到仓时间',
unloading_completed_at DATETIME(3) NULL COMMENT '海外仓卸柜完成时间(UTC)',
unloading_photo_uploaded_at DATETIME(3) NULL COMMENT '海外仓卸柜照片上传成功时间(UTC)',
estimated_empty_reported_at DATETIME(3) NULL COMMENT '预计报空时间(卸柜照片上传成功时间加两天,UTC)',
actual_empty_reported_at DATETIME(3) NULL COMMENT '实际报空时间(UTC)',
empty_report_source_version BIGINT UNSIGNED NULL COMMENT '海外货柜事件聚合版本,用于阻止乱序覆盖',
created_by BIGINT UNSIGNED NOT NULL COMMENT '创建人ID',
created_by_name VARCHAR(64) NULL COMMENT '创建人名称快照',
created_at DATETIME(3) NOT NULL DEFAULT CURRENT_TIMESTAMP(3) COMMENT '创建时间',
@@ -93,6 +98,21 @@ CREATE TABLE IF NOT EXISTS fl_container_booking (
CONSTRAINT chk_container_booking_weight CHECK (max_weight_kg IS NULL OR max_weight_kg > 0)
) ENGINE=InnoDB COMMENT='平台货柜预订资料;预装状态唯一来源为装柜计划';
CREATE TABLE IF NOT EXISTS fl_container_event_inbox (
event_id VARCHAR(64) NOT NULL COMMENT '跨区事件全局唯一ID',
event_type VARCHAR(64) NOT NULL COMMENT '事件类型',
event_version INT UNSIGNED NOT NULL COMMENT '事件契约版本',
aggregate_id VARCHAR(64) NOT NULL COMMENT '货柜聚合ID',
aggregate_version BIGINT UNSIGNED NOT NULL COMMENT '货柜聚合版本',
payload_hash VARCHAR(80) NOT NULL COMMENT '规范化负载SHA-256摘要',
payload_json JSON NOT NULL COMMENT '原始事件负载',
status VARCHAR(16) NOT NULL COMMENT 'RECEIVED或PROCESSED',
received_at DATETIME(3) NOT NULL COMMENT '接收时间(UTC)',
processed_at DATETIME(3) NULL COMMENT '处理完成时间(UTC)',
PRIMARY KEY (event_id),
KEY idx_container_event_aggregate (aggregate_id, aggregate_version)
) ENGINE=InnoDB COMMENT='海外货柜事件消费Inbox';
DROP PROCEDURE IF EXISTS phase3_container_real_unit_columns;
DELIMITER $$
CREATE PROCEDURE phase3_container_real_unit_columns()
@@ -112,6 +132,31 @@ BEGIN
ALTER TABLE fl_container_booking ADD COLUMN merged_into_container_booking_id BIGINT UNSIGNED NULL
COMMENT '真实柜被模拟柜吸收后的目标稳定柜次ID' AFTER record_status;
END IF;
IF NOT EXISTS (SELECT 1 FROM information_schema.columns WHERE table_schema=DATABASE()
AND table_name='fl_container_booking' AND column_name='unloading_completed_at') THEN
ALTER TABLE fl_container_booking ADD COLUMN unloading_completed_at DATETIME(3) NULL
COMMENT '海外仓卸柜完成时间(UTC)' AFTER actual_warehouse_arrival_at;
END IF;
IF NOT EXISTS (SELECT 1 FROM information_schema.columns WHERE table_schema=DATABASE()
AND table_name='fl_container_booking' AND column_name='unloading_photo_uploaded_at') THEN
ALTER TABLE fl_container_booking ADD COLUMN unloading_photo_uploaded_at DATETIME(3) NULL
COMMENT '海外仓卸柜照片上传成功时间(UTC)' AFTER unloading_completed_at;
END IF;
IF NOT EXISTS (SELECT 1 FROM information_schema.columns WHERE table_schema=DATABASE()
AND table_name='fl_container_booking' AND column_name='estimated_empty_reported_at') THEN
ALTER TABLE fl_container_booking ADD COLUMN estimated_empty_reported_at DATETIME(3) NULL
COMMENT '预计报空时间(卸柜照片上传成功时间加两天,UTC)' AFTER unloading_photo_uploaded_at;
END IF;
IF NOT EXISTS (SELECT 1 FROM information_schema.columns WHERE table_schema=DATABASE()
AND table_name='fl_container_booking' AND column_name='actual_empty_reported_at') THEN
ALTER TABLE fl_container_booking ADD COLUMN actual_empty_reported_at DATETIME(3) NULL
COMMENT '实际报空时间(UTC)' AFTER estimated_empty_reported_at;
END IF;
IF NOT EXISTS (SELECT 1 FROM information_schema.columns WHERE table_schema=DATABASE()
AND table_name='fl_container_booking' AND column_name='empty_report_source_version') THEN
ALTER TABLE fl_container_booking ADD COLUMN empty_report_source_version BIGINT UNSIGNED NULL
COMMENT '海外货柜事件聚合版本,用于阻止乱序覆盖' AFTER actual_empty_reported_at;
END IF;
IF NOT EXISTS (SELECT 1 FROM information_schema.statistics WHERE table_schema=DATABASE()
AND table_name='fl_container_booking' AND index_name='uk_container_booking_container_no') THEN
ALTER TABLE fl_container_booking ADD UNIQUE KEY uk_container_booking_container_no (container_no);
@@ -1525,6 +1525,11 @@ components:
actualPortArrivalAt: {type: string, format: date-time, nullable: true}
estimatedWarehouseArrivalAt: {type: string, format: date-time, nullable: true}
actualWarehouseArrivalAt: {type: string, format: date-time, nullable: true}
unloadingCompletedAt: {type: string, format: date-time, nullable: true, description: 海外仓卸柜完成时间}
unloadingPhotoUploadedAt: {type: string, format: date-time, nullable: true, description: 海外仓卸柜照片上传成功时间}
estimatedEmptyReportedAt: {type: string, format: date-time, nullable: true, description: 预计报空时间,由卸柜照片上传成功时间加两天计算}
actualEmptyReportedAt: {type: string, format: date-time, nullable: true, description: 实际报空时间}
emptyReportSourceVersion: {type: string, pattern: '^[1-9][0-9]*$', nullable: true, description: 海外货柜事件聚合版本}
planStatus: {type: string, enum: [WAIT_LOADING, EDITING, LOADING, COMPLETED]}
exceptionActive: {type: boolean, description: 最新异常历史动作是否为 MARK}
destinationClearanceDocumentStatus: {$ref: '#/components/schemas/ContainerDocumentStatus'}
@@ -0,0 +1,108 @@
package com.baiyue.firstleg.service;
import com.baiyue.common.event.EventEnvelope;
import com.baiyue.common.event.EventPayloadCanonicalizer;
import com.baiyue.common.event.EventPayloadHash;
import com.baiyue.common.event.EventTypes;
import com.baiyue.common.event.OverseasContainerUnloadedV1;
import com.baiyue.common.event.RegionCode;
import com.baiyue.firstleg.entity.ContainerBookingEntity;
import com.baiyue.firstleg.entity.ContainerEventInboxEntity;
import com.baiyue.firstleg.mapper.ContainerBookingMapper;
import com.baiyue.firstleg.mapper.ContainerEventInboxMapper;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.fasterxml.jackson.datatype.jsr310.JavaTimeModule;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.extension.ExtendWith;
import org.mockito.ArgumentCaptor;
import org.mockito.Mock;
import org.mockito.junit.jupiter.MockitoExtension;
import java.time.Instant;
import java.time.LocalDateTime;
import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatThrownBy;
import static org.mockito.ArgumentMatchers.*;
import static org.mockito.Mockito.*;
@ExtendWith(MockitoExtension.class)
class OverseasContainerEventConsumerTest {
@Mock private ContainerBookingMapper containers;
@Mock private ContainerEventInboxMapper inbox;
private ObjectMapper objectMapper;
private OverseasContainerEventConsumer consumer;
@BeforeEach
void setUp() {
objectMapper = new ObjectMapper().registerModule(new JavaTimeModule());
consumer = new OverseasContainerEventConsumer(containers, inbox, objectMapper);
}
@Test
void derivesEstimatedEmptyReportedAtFromPhotoUploadAndPersistsInbox() {
ContainerBookingEntity container = container();
when(containers.selectActiveByContainerNo("MSCU1234567")).thenReturn(container);
Instant completedAt = Instant.parse("2026-08-20T03:00:00Z");
Instant uploadedAt = Instant.parse("2026-08-20T02:30:00Z");
consumer.consume(event("00000000-0000-4000-8000-000000000001", 7,
new OverseasContainerUnloadedV1("MSCU1234567", completedAt, uploadedAt, null)));
verify(containers).applyEmptyReport(eq(1001L), eq(LocalDateTime.parse("2026-08-20T03:00:00")),
eq(LocalDateTime.parse("2026-08-20T02:30:00")),
eq(LocalDateTime.parse("2026-08-22T02:30:00")), isNull(), eq(7L));
ArgumentCaptor<ContainerEventInboxEntity> captor = ArgumentCaptor.forClass(ContainerEventInboxEntity.class);
verify(inbox).insert(captor.capture());
assertThat(captor.getValue().getAggregateId()).isEqualTo("1001");
assertThat(captor.getValue().getStatus()).isEqualTo("PROCESSED");
verify(inbox).updateById(captor.getValue());
}
@Test
void repeatedEventWithSameHashDoesNotApplyAgain() {
OverseasContainerUnloadedV1 payload = new OverseasContainerUnloadedV1("MSCU1234567",
Instant.parse("2026-08-20T03:00:00Z"), Instant.parse("2026-08-20T02:30:00Z"), null);
EventEnvelope event = event("00000000-0000-4000-8000-000000000002", 7, payload);
ContainerEventInboxEntity existing = new ContainerEventInboxEntity();
existing.setPayloadHash(event.payloadHash());
when(inbox.selectById(event.eventId())).thenReturn(existing);
when(containers.selectActiveByContainerNo("MSCU1234567")).thenReturn(container());
consumer.consume(event);
verify(containers, never()).applyEmptyReport(anyLong(), any(), any(), any(), any(), anyLong());
verify(inbox, never()).insert(any(ContainerEventInboxEntity.class));
}
@Test
void rejectsTamperedPayloadHash() {
EventEnvelope valid = event("00000000-0000-4000-8000-000000000003", 7,
new OverseasContainerUnloadedV1("MSCU1234567", Instant.parse("2026-08-20T03:00:00Z"),
Instant.parse("2026-08-20T02:30:00Z"), null));
EventEnvelope tampered = new EventEnvelope(valid.eventId(), valid.eventType(), valid.eventVersion(),
valid.occurredAt(), valid.sourceRegion(), valid.targetRegion(), valid.tenantId(), valid.aggregateType(),
valid.aggregateId(), valid.aggregateVersion(), valid.handoverVersion(), valid.traceId(),
valid.correlationId(), valid.causationId(), valid.payload(), "sha256:" + "0".repeat(64));
assertThatThrownBy(() -> consumer.consume(tampered))
.isInstanceOf(IllegalArgumentException.class).hasMessageContaining("payloadHash");
verifyNoInteractions(containers, inbox);
}
private EventEnvelope event(String eventId, long aggregateVersion, OverseasContainerUnloadedV1 payload) {
var node = objectMapper.valueToTree(payload);
String hash = EventPayloadHash.sha256(EventPayloadCanonicalizer.canonicalize(node));
return new EventEnvelope(eventId, EventTypes.OVERSEAS_CONTAINER_UNLOADED, 1,
Instant.parse("2026-08-20T02:31:00Z"), RegionCode.OVERSEAS, RegionCode.CN, "1",
"CONTAINER_BOOKING", payload.containerNo(), aggregateVersion, null,
"trace-12345678", "correlation-12345678", null, node, hash);
}
private ContainerBookingEntity container() {
ContainerBookingEntity value = new ContainerBookingEntity();
value.setId(1001L); value.setContainerNo("MSCU1234567");
return value;
}
}