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

fix(lastleg): unblock inbound plan handoff

Browse Source
This commit is contained in:
ouyangxq
2026-08-27 15:56:14 +08:00
parent adabec13d3
commit 0f2e201c38
28 changed files with 378 additions and 103 deletions
Download Patch File Download Diff File Expand all files Collapse all files
firstleg/src/main/java/com/baiyue/firstleg/dto/LoadingPickDtos.java
+7 -1
View File
@@ -125,12 +125,18 @@ public final class LoadingPickDtos {
Long operatorId, String operatorName, LocalDateTime createdAt) {
}
public record CompletionCommand(Long id, String operationId, String status, Integer expectedTaskVersion, String remark,
Integer attemptCount, LocalDateTime nextRetryAt, String failureStage,
String failureCode, String requestId, String traceId,
LocalDateTime completedAt) {
}
public record Task(Long id, Long loadingPlanId, Long containerBookingId, Long inboundId,
Integer inboundVersion, Integer loadingVersion, String cabinetNo,
String currentContainerNo, String sourceType, String recordStatus,
Long mergedIntoContainerBookingId, String taskStatus,
String workflowStatus, Integer version, Assignment assignment,
Review review, List<PickItem> items) {
Review review, CompletionCommand completionCommand, List<PickItem> items) {
}
/** PDA 工作台只展示当前作业员可领取或已领取的进行中任务。 */
firstleg/src/main/java/com/baiyue/firstleg/mapper/LoadingCompletedEventMapper.java
+3 -1
View File
@@ -15,6 +15,7 @@ public interface LoadingCompletedEventMapper {
@Select("""
SELECT p.id AS loading_plan_id, p.plan_no, p.container_booking_id, p.version AS aggregate_version,
b.container_no, b.cabinet_no, b.warehouse_id, b.warehouse_name,
b.customer_id AS booking_customer_id, b.customer_name,
i.account_id, i.inbound_id AS source_inbound_id,
COALESCE(o.root_inbound_id, o.id) AS root_inbound_id,
o.order_id AS source_order_id, o.order_no AS source_order_no,
@@ -79,7 +80,8 @@ public interface LoadingCompletedEventMapper {
class Row {
private Long loadingPlanId; private String planNo; private Long containerBookingId;
private Integer aggregateVersion; private String containerNo; private String cabinetNo;
private Long warehouseId; private String warehouseName; private Long accountId;
private Long warehouseId; private String warehouseName; private Long bookingCustomerId;
private String customerName; private Long accountId;
private Long sourceInboundId; private Long rootInboundId; private Long sourceOrderId;
private Long loadingPlanItemId;
private String sourceOrderNo; private Long rootOrderId; private String inboundCode;
firstleg/src/main/java/com/baiyue/firstleg/mapper/LoadingCompletionCommandMapper.java
+3
View File
@@ -33,6 +33,9 @@ public interface LoadingCompletionCommandMapper extends BaseMapper<LoadingComple
""")
LoadingCompletionCommandEntity findActiveByTask(@Param("taskId") long taskId);
@Select("SELECT * FROM fl_loading_completion_command WHERE pick_task_id=#{taskId} ORDER BY created_at DESC LIMIT 1")
LoadingCompletionCommandEntity findLatestByTask(@Param("taskId") long taskId);
@Select("""
SELECT * FROM fl_loading_completion_command
WHERE command_status IN ('PROCESSING','RETRYABLE') AND next_retry_at<=CURRENT_TIMESTAMP(3)
firstleg/src/main/java/com/baiyue/firstleg/repository/LoadingCompletionCommandRepository.java
+1
View File
@@ -15,6 +15,7 @@ public class LoadingCompletionCommandRepository {
public int insert(LoadingCompletionCommandEntity command) { return mapper.insertCommand(command); }
public LoadingCompletionCommandEntity operation(String operationId) { return mapper.findByOperation(operationId); }
public LoadingCompletionCommandEntity active(long taskId) { return mapper.findActiveByTask(taskId); }
public LoadingCompletionCommandEntity latest(long taskId) { return mapper.findLatestByTask(taskId); }
public List<LoadingCompletionCommandEntity> due(int limit) { return mapper.findDue(limit); }
public boolean claim(long id, String worker, LocalDateTime leaseUntil) { return mapper.claim(id, worker, leaseUntil) == 1; }
public boolean renew(long id, String worker, LocalDateTime leaseUntil) { return mapper.renew(id, worker, leaseUntil) == 1; }
firstleg/src/main/java/com/baiyue/firstleg/service/LoadingCompletedOutboxService.java
+21 -2
View File
@@ -37,7 +37,7 @@ import java.util.Map;
import java.util.Objects;
import java.util.Set;
/** 在装柜完成事务内按客户写跨区业务 Outbox。 */
/** 在装柜完成事务内按客户及可信客户名称快照写跨区业务 Outbox。 */
@Service
public class LoadingCompletedOutboxService {
private static final long HANDOVER_VERSION = 2L;
@@ -49,6 +49,7 @@ public class LoadingCompletedOutboxService {
public void publish(long planId) {
List<Row> rows = snapshotMapper.rows(planId);
if (rows.isEmpty()) throw new BizException(ResultCode.CONFLICT, "装柜计划没有最终实装入仓单，不能生成尾程交接事件");
validateCustomerSnapshots(rows);
Map<Long, List<String>> cartonMarks = cartonLabelPools(rows);
List<PalletRow> palletRows = snapshotMapper.palletRows(planId);
rows.stream().collect(java.util.stream.Collectors.groupingBy(Row::getAccountId,
@@ -104,10 +105,28 @@ public class LoadingCompletedOutboxService {
first.getDestinationWarehouseCode(), first.getDestinationWarehouseName(), sumInt(sorted, Row::getPackageCount),
sumNullableInt(sorted, Row::getProductQuantity), "ONE_CLICK_DROPSHIP".equals(orderType)
? pallets.size() : 0, sum(sorted, Row::getGrossWeightKg), sum(sorted, Row::getNetWeightKg),
sum(sorted, Row::getVolumeCbm), items, string(first.getAccountId()), orderType, pallets,
sum(sorted, Row::getVolumeCbm), items, string(first.getAccountId()), first.getCustomerName().trim(), orderType, pallets,
destinationType(sorted));
}
private void validateCustomerSnapshots(List<Row> rows) {
for (Row row : rows) {
if (row.getAccountId() == null || !Objects.equals(row.getAccountId(), row.getBookingCustomerId())) {
throw new BizException(ResultCode.CONFLICT, "装柜计划客户与柜次客户快照不一致，不能生成尾程交接事件");
}
if (row.getCustomerName() == null || row.getCustomerName().isBlank()) {
throw new BizException(ResultCode.CONFLICT, "柜次缺少客户名称快照，不能生成尾程交接事件");
}
}
rows.stream().collect(java.util.stream.Collectors.groupingBy(Row::getAccountId)).forEach((accountId, values) -> {
Set<String> names = values.stream().map(Row::getCustomerName).map(String::trim)
.collect(java.util.stream.Collectors.toSet());
if (names.size() != 1) {
throw new BizException(ResultCode.CONFLICT, "同一客户事件存在多个客户名称快照");
}
});
}
/** 国内按实装箱数从可用标签池稳定随机抽取箱唛，重复发布同一计划时结果不变。 */
private Map<Long, List<String>> cartonLabelPools(List<Row> rows) {
Map<Long, List<String>> result = new java.util.LinkedHashMap<>();
firstleg/src/main/java/com/baiyue/firstleg/service/LoadingPickService.java
+7 -1
View File
@@ -16,6 +16,7 @@ import com.baiyue.firstleg.dto.LoadingPickDtos.AssignRequest;
import com.baiyue.firstleg.dto.LoadingPickDtos.ReviewRequest;
import com.baiyue.firstleg.dto.LoadingPickDtos.CompleteLoadingRequest;
import com.baiyue.firstleg.dto.LoadingPickDtos.Operation;
import com.baiyue.firstleg.dto.LoadingPickDtos.CompletionCommand;
import com.baiyue.firstleg.dto.LoadingPickDtos.BatchRequest;
import com.baiyue.firstleg.dto.LoadingPickDtos.BatchItemRequest;
import com.baiyue.firstleg.dto.LoadingPickDtos.InventoryResource;
@@ -837,12 +838,17 @@ public class LoadingPickService {
.filter(value -> detailInboundId.equals(value.inboundId())).findFirst().orElse(null);
var inbound = owner == null || owner.tenantId() == null || owner.inboundId() == null
? null : repository.inbound(owner.tenantId(), owner.inboundId());
LoadingCompletionCommandEntity command = completionCommands.latest(task.getId());
CompletionCommand completionCommand = command == null ? null : new CompletionCommand(command.getId(),
command.getOperationId(), command.getCommandStatus(), command.getExpectedTaskVersion(), command.getRemark(),
command.getAttemptCount(), command.getNextRetryAt(), command.getFailureStage(), command.getFailureCode(),
command.getFailureRequestId(), command.getFailureTraceId(), command.getCompletedAt());
return new Task(task.getId(), task.getLoadingPlanId(), task.getContainerBookingId(),
owner == null ? null : owner.inboundId(), inbound == null ? null : inbound.version(),
inbound == null ? null : inbound.version(), task.getCabinetNo(),
booking.getContainerNo(), booking.getSourceType(), booking.getRecordStatus(),
booking.getMergedIntoContainerBookingId(), task.getTaskStatus(),
workflowStatus, task.getVersion(), assignment, review, items);
workflowStatus, task.getVersion(), assignment, review, completionCommand, items);
}
private List<BatchLine> validateBatchLines(LoadingPickTaskEntity task, BatchRequest request) {
firstleg/src/main/resources/openapi/phase3-pda-convergence-v1.yaml
+17
View File
@@ -141,6 +141,7 @@ components:
mergedIntoContainerBookingId: {type: integer, format: int64, nullable: true, description: Always null for the surviving ACTIVE operational container; populated only on hidden MERGED source history}
taskStatus: {type: string, enum: [LOADING, COMPLETING, COMPLETION_FAILED, COMPLETED]}
workflowStatus: {type: string, enum: [UNASSIGNED, IN_PROGRESS, AWAITING_REVIEW, PROCESSING, COMPLETION_FAILED, COMPLETED]}
completionCommand: {$ref: '#/components/schemas/CompletionCommand', nullable: true}
version: {type: integer, minimum: 0}
assignment: {type: object, nullable: true}
review: {type: object, nullable: true}
@@ -236,6 +237,22 @@ components:
expectedTaskVersion: {type: integer, minimum: 0}
operationId: {type: string, maxLength: 128}
remark: {type: string, maxLength: 500}
CompletionCommand:
type: object
required: [id, operationId, status, expectedTaskVersion, attemptCount]
properties:
id: {type: integer, format: int64}
operationId: {type: string, maxLength: 128, description: Retry must reuse this exact identity}
status: {type: string, enum: [PROCESSING, RETRYABLE, FAILED, COMPLETED]}
expectedTaskVersion: {type: integer, minimum: 0, description: Retry must preserve this request-hashed version}
remark: {type: string, maxLength: 500, nullable: true, description: Retry must preserve this request-hashed remark}
attemptCount: {type: integer, minimum: 0}
nextRetryAt: {type: string, format: date-time, nullable: true}
failureStage: {type: string, nullable: true}
failureCode: {type: string, nullable: true}
requestId: {type: string, nullable: true}
traceId: {type: string, nullable: true}
completedAt: {type: string, format: date-time, nullable: true}
CompleteLoadingRequest:
type: object
required: [expectedTaskVersion, operationId]
firstleg/src/test/java/com/baiyue/firstleg/repository/SmartLoadingSqlContractTest.java
+1 -1
View File
@@ -308,7 +308,7 @@ class SmartLoadingSqlContractTest {
.contains("container_booking_id BIGINT UNSIGNED NOT NULL")
.contains("cabinet_no VARCHAR(64) NOT NULL")
.contains("container_no VARCHAR(64) NULL")
.contains("task_status VARCHAR(16) NOT NULL DEFAULT 'LOADING'")
.contains("task_status VARCHAR(24) NOT NULL DEFAULT 'LOADING'")
.contains("UNIQUE KEY uk_loading_pick_task_plan (loading_plan_id)")
.contains("UNIQUE KEY uk_loading_pick_task_booking (container_booking_id)")
.contains("KEY idx_loading_pick_container (container_no, task_status, started_at)")
firstleg/src/test/java/com/baiyue/firstleg/service/LoadingCompletedOutboxServiceTest.java
+35 -2
View File
@@ -59,6 +59,7 @@ class LoadingCompletedOutboxServiceTest {
assertThat(first.payload().at("/orders/0/items/0/skuCode").asText()).isEqualTo("SKU101");
assertThat(first.payload().at("/orders/0/items/0/cartonMarks/0").asText()).isEqualTo("BOX-101");
assertThat(first.payload().at("/orders/0/destinationType").asText()).isEqualTo("OVERSEAS");
assertThat(first.payload().at("/orders/0/customerName").asText()).isEqualTo("客户-11");
assertThat(rows.get(0).getEventKey()).isEqualTo("FIRSTLEG_LOADING_COMPLETED:90:11:2");
assertThat(first.eventVersion()).isEqualTo(2);
}
@@ -72,6 +73,22 @@ class LoadingCompletedOutboxServiceTest {
.hasMessageContaining("缺少 SKU");
}
@Test void rejectsMissingTrustedBookingCustomerName() {
Row value = row(11, 100, 101);
value.setCustomerName(" ");
when(snapshots.rows(90L)).thenReturn(List.of(value));
assertThatThrownBy(() -> service.publish(90L)).hasMessageContaining("缺少客户名称快照");
}
@Test void rejectsBookingCustomerIdThatDoesNotMatchAccountEvent() {
Row value = row(11, 100, 101);
value.setBookingCustomerId(22L);
when(snapshots.rows(90L)).thenReturn(List.of(value));
assertThatThrownBy(() -> service.publish(90L)).hasMessageContaining("客户与柜次客户快照不一致");
}
@Test void fbaFreezesPackageCountWithoutRequiringCartonLabels() throws Exception {
Row value = row(11, 100, 101);
value.setDestinationType("FBA");
@@ -118,6 +135,19 @@ class LoadingCompletedOutboxServiceTest {
.isEqualTo(second.payload().at("/orders/0/items/0/cartonMarks"));
}
@Test void immutableCustomerNameParticipatesInPayloadHash() {
Row value = row(11, 100, 101);
when(snapshots.rows(90L)).thenReturn(List.of(value));
service.publish(90L);
value.setCustomerName("客户甲-更名快照");
service.publish(90L);
ArgumentCaptor<InboundOutboxEntity> captor = ArgumentCaptor.forClass(InboundOutboxEntity.class);
verify(outbox, org.mockito.Mockito.times(2)).insert(captor.capture());
assertThat(captor.getAllValues().get(0).getPayloadHash())
.isNotEqualTo(captor.getAllValues().get(1).getPayloadHash());
}
@Test void fillsCollapsedMeasurementDetailFromUnusedInboundLabelsWithoutDuplicates() throws Exception {
Row first = row(11, 100, 101); first.setPackageCount(1);
Row second = row(11, 100, 101); second.setLoadingPlanItemId(102L); second.setPackageCount(2);
@@ -144,7 +174,9 @@ class LoadingCompletedOutboxServiceTest {
.getAnnotation(Select.class).value()).replaceAll("\\s+", " ");
assertThat(rowsSql).contains("p.plan_status = 'COMPLETED'")
.contains("o.process_status = 'LOADED'");
.contains("o.process_status = 'LOADED'")
.contains("b.customer_id AS booking_customer_id")
.contains("b.customer_name");
for (String sql : List.of(rowsSql, palletsSql)) {
assertThat(sql)
.contains("actual_pick.loading_plan_item_id = i.id")
@@ -219,7 +251,8 @@ class LoadingCompletedOutboxServiceTest {
private Row row(long account,long inbound,long detail) {
Row r=new Row(); r.setLoadingPlanId(90L); r.setPlanNo("P90"); r.setContainerBookingId(91L);
r.setAggregateVersion(8); r.setContainerNo("CONT"); r.setCabinetNo("CAB"); r.setWarehouseId(92L);
r.setWarehouseName("WH"); r.setAccountId(account); r.setSourceInboundId(inbound); r.setRootInboundId(99L);
r.setWarehouseName("WH"); r.setBookingCustomerId(account); r.setCustomerName("客户-" + account);
r.setAccountId(account); r.setSourceInboundId(inbound); r.setRootInboundId(99L);
r.setSourceOrderId(inbound+1000); r.setSourceOrderNo("FL"+inbound); r.setRootOrderId(1099L);
r.setInboundCode("RK"+inbound); r.setLoadingPlanItemId(detail); r.setDestinationWarehouseId(93L); r.setDestinationWarehouseCode("ONT8");
r.setDestinationWarehouseName("ONT8"); r.setSourceInboundDetailId(detail); r.setSkuCode("SKU"+detail);
firstleg/src/test/java/com/baiyue/firstleg/service/TaillegRealProducerAcceptanceTest.java
+6 -2
View File
@@ -50,12 +50,16 @@ class TaillegRealProducerAcceptanceTest {
assertEquals(2, jdbc.queryForObject("SELECT JSON_LENGTH(JSON_EXTRACT(payload_json,'$.payload.orders')) " +
"FROM fl_inbound_outbox WHERE event_key=?", Integer.class,
eventKey));
assertEquals("真实验收客户", jdbc.queryForObject(
"SELECT JSON_UNQUOTE(JSON_EXTRACT(payload_json,'$.payload.orders[0].customerName')) " +
"FROM fl_inbound_outbox WHERE event_key=?", String.class, eventKey));
}
private void seedCompletedLoadingPlan() {
jdbc.update("INSERT IGNORE INTO fl_container_booking(id,booking_no,container_no,cabinet_no,warehouse_id," +
"warehouse_name,container_type,created_by,updated_by) VALUES(?,?,?,?,?,?,?,1,1)",
BOOKING, MARKER + "-BOOK", MARKER + "-CNTR", MARKER + "-CAB", 92L, "真实验收海外仓", "40HQ");
"warehouse_name,customer_id,customer_name,container_type,created_by,updated_by) VALUES(?,?,?,?,?,?,?,?,?,1,1)",
BOOKING, MARKER + "-BOOK", MARKER + "-CNTR", MARKER + "-CAB", 92L, "真实验收海外仓",
CUSTOMER, "真实验收客户", "40HQ");
jdbc.update("INSERT IGNORE INTO fl_loading_plan(id,plan_no,container_booking_id,plan_status,created_by," +
"updated_by,version) VALUES(?,?,?,'LOADING',1,1,0)", PLAN, MARKER + "-PLAN", BOOKING);
seedInbound(FIRST_INBOUND, 990022221L, MARKER + "-FIRST-IN", "FBA_SEA", "STANDARD", "LOADED", "ONT8");